Add: generational monitor

monitor.go

@@ -123,7 +123,7 @@ type MonitorEvent struct {
 }
 
 type MonitorEvents struct {
-	GenerateBy *MonitorEvent
+	GeneratedBy *MonitorEvent
 	Changes     []*MonitorEvent
 }
 
@@ -198,7 +198,7 @@ func NewMonitor(opts ...MonitorOption) *Monitor {
 }
 
 func (monitor *Monitor) monitor() {
-	changesEvents := make([]*MonitorEvent, 0, 2)
+	changesEvents := make([]*MonitorEvent, 0)
 
 	for {
 		msgs, err := monitor.conn.Receive()
@@ -217,10 +217,10 @@ func (monitor *Monitor) monitor() {
 				changesEvents = append(changesEvents, event)
 
 				monitor.eventCh <- &MonitorEvents{
-					GenerateBy: event,
+					GeneratedBy: event,
 					Changes:     changesEvents,
 				}
-				changesEvents = make([]*MonitorEvent, 0, 2)
+				changesEvents = make([]*MonitorEvent, 0)
 
 				break
 			}
@@ -298,11 +298,11 @@ func (monitor *Monitor) monitor() {
 				}
 
 				monitor.eventCh <- &MonitorEvents{
-					GenerateBy: event,
+					GeneratedBy: event,
 					Changes:     changesEvents,
 				}
 
-				changesEvents = make([]*MonitorEvent, 0, 2)
+				changesEvents = make([]*MonitorEvent, 0)
 			}
 		}
 	}
@@ -331,7 +331,28 @@ func (monitor *Monitor) Close() error {
 // calling Close on Monitor or encountering a netlink conn error while Receive.
 // Caller may receive a MonitorEventTypeOOB event which contains an error we didn't
 // handle, for now.
-func (cc *Conn) AddMonitor(monitor *Monitor) (chan *MonitorEvents, error) {
+func (cc *Conn) AddMonitor(monitor *Monitor) (chan *MonitorEvent, error) {
+	generationalEventCh, err := cc.AddGenerationalMonitor(monitor)
+
+	if err != nil {
+		return nil, err
+	}
+
+	eventCh := make(chan *MonitorEvent)
+
+	go func() {
+		defer close(eventCh)
+		for monitorEvents := range generationalEventCh {
+			for _, event := range monitorEvents.Changes {
+				eventCh <- event
+			}
+		}
+	}()
+
+	return eventCh, nil
+}
+
+func (cc *Conn) AddGenerationalMonitor(monitor *Monitor) (chan *MonitorEvents, error) {
 	conn, closer, err := cc.netlinkConn()
 	if err != nil {
 		return nil, err

monitor_test.go

@@ -21,7 +21,7 @@ func ExampleNewMonitor() {
 
 	mon := nftables.NewMonitor()
 	defer mon.Close()
-	events, err := conn.AddMonitor(mon)
+	events, err := conn.AddGenerationalMonitor(mon)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -49,7 +49,7 @@ func TestMonitor(t *testing.T) {
 
 	// default to monitor all
 	monitor := nftables.NewMonitor()
-	events, err := c.AddMonitor(monitor)
+	events, err := c.AddGenerationalMonitor(monitor)
 	if err != nil {
 		t.Fatal(err)
 	}

set.go

@@ -268,12 +268,6 @@ type Set struct {
 	KeyByteOrder binaryutil.ByteOrder
 }
 
-type SetElementsInfo struct {
-	TableName string
-	SetName   string
-	Elements  []SetElement
-}
-
 // SetElement represents a data point within a set.
 type SetElement struct {
 	Key []byte
@@ -803,15 +797,10 @@ func elementsFromMsg(fam byte, msg netlink.Message) ([]SetElement, error) {
 	}
 	ad.ByteOrder = binary.BigEndian
 
-	var info = &SetElementsInfo{}
 	var elements []SetElement
 	for ad.Next() {
 		b := ad.Bytes()
 		switch ad.Type() {
-		case unix.NFTA_SET_ELEM_LIST_TABLE:
-			info.TableName = ad.String()
-		case unix.NFTA_SET_ELEM_LIST_SET:
-			info.SetName = ad.String()
 		case unix.NFTA_SET_ELEM_LIST_ELEMENTS:
 			ad, err := netlink.NewAttributeDecoder(b)
 			if err != nil {
@@ -829,6 +818,7 @@ func elementsFromMsg(fam byte, msg netlink.Message) ([]SetElement, error) {
 			}
 		}
 	}
+
 	return elements, nil
 }