Add: test monitor error

monitor.go

@@ -123,8 +123,8 @@ type MonitorEvent struct {
 }
 
 type MonitorEvents struct {
-	GenerateBy *MonitorEvent
+	GeneratedBy *MonitorEvent
-	Changes    []*MonitorEvent
+	Changes     []*MonitorEvent
 }
 
 const (
@@ -198,7 +198,7 @@ func NewMonitor(opts ...MonitorOption) *Monitor {
 }
 
 func (monitor *Monitor) monitor() {
-	changesEvents := make([]*MonitorEvent, 0, 2)
+	var changesEvents []*MonitorEvent
 
 	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,
+					Changes:     changesEvents,
 				}
-				changesEvents = make([]*MonitorEvent, 0, 2)
+				changesEvents = nil
 
 				break
 			}
@@ -298,11 +298,11 @@ func (monitor *Monitor) monitor() {
 				}
 
 				monitor.eventCh <- &MonitorEvents{
-					GenerateBy: event,
+					GeneratedBy: event,
-					Changes:    changesEvents,
+					Changes:     changesEvents,
 				}
 
-				changesEvents = make([]*MonitorEvent, 0, 2)
+				changesEvents = make([]*MonitorEvent, 0)
 			}
 		}
 	}
@@ -331,7 +331,27 @@ 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

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"log"
 	"net"
+	"os"
+	"path/filepath"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -21,7 +23,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)
 	}
@@ -46,10 +48,9 @@ func TestMonitor(t *testing.T) {
 	// Clear all rules at the beginning + end of the test.
 	c.FlushRuleset()
 	defer c.FlushRuleset()
-
 	// default to monitor all
 	monitor := nftables.NewMonitor()
-	events, err := c.AddMonitor(monitor)
+	events, err := c.AddGenerationalMonitor(monitor)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -60,6 +61,7 @@ func TestMonitor(t *testing.T) {
 	var gotRule *nftables.Rule
 	wg := sync.WaitGroup{}
 	wg.Add(1)
+	var errMonitor error
 	go func() {
 		defer wg.Done()
 		count := int32(0)
@@ -69,9 +71,17 @@ func TestMonitor(t *testing.T) {
 				return
 			}
 
+			genMsg := event.GeneratedBy.Data.(*nftables.GenMsg)
+			fileName := filepath.Base(os.Args[0])
+
+			if genMsg.ProcComm != fileName {
+				errMonitor = fmt.Errorf("procComm: %s, want: %s", genMsg.ProcComm, fileName)
+				return
+			}
+
 			for _, change := range event.Changes {
 				if change.Error != nil {
-					err = fmt.Errorf("monitor err: %s", change.Error)
+					errMonitor = fmt.Errorf("monitor err: %s", change.Error)
 					return
 				}
 
@@ -132,7 +142,13 @@ func TestMonitor(t *testing.T) {
 	if err := c.Flush(); err != nil {
 		t.Fatal(err)
 	}
+
 	wg.Wait()
+
+	if errMonitor != nil {
+		t.Fatal("monitor err", errMonitor)
+	}
+
 	if gotTable.Family != nat.Family || gotTable.Name != nat.Name {
 		t.Fatal("no want table", gotTable.Family, gotTable.Name)
 	}

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,16 +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() {
+		if ad.Type() == unix.NFTA_SET_ELEM_LIST_ELEMENTS {
-		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 {
 				return nil, err