Add unit tests for events

tm2/pkg/p2p/events/doc.go

@@ -1,2 +1,3 @@
-// TODO add documentation, coverage
+// Package events contains a simple p2p event system implementation, that simplifies asynchronous event flows in the
+// p2p module. The event subscriptions allow for event filtering, which eases the load on the event notification flow.
 package events

tm2/pkg/p2p/events/events.go

@@ -17,6 +17,7 @@ type Events struct {
 	subscriptionsMux sync.RWMutex
 }
 
+// New creates a new event subscription manager
 func New() *Events {
 	return &Events{
 		subs: make(subscriptions),
@@ -42,7 +43,7 @@ func (es *Events) Subscribe(filterFn EventFilter) (<-chan Event, func()) {
 	return ch, unsubscribeFn
 }
 
-// Notify notifies all subscribers of an incoming event
+// Notify notifies all subscribers of an incoming event [BLOCKING]
 func (es *Events) Notify(event Event) {
 	es.subscriptionsMux.RLock()
 	defer es.subscriptionsMux.RUnlock()
@@ -98,9 +99,6 @@ func (s *subscriptions) notify(event Event) {
 			continue
 		}
 
-		select {
-		case sub.ch <- event:
-		default:
-		}
+		sub.ch <- event
 	}
 }

tm2/pkg/p2p/events/events_test.go

@@ -0,0 +1,94 @@
+package events
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/gnolang/gno/tm2/pkg/p2p/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// generateEvents generates p2p events
+func generateEvents(count int) []Event {
+	events := make([]Event, 0, count)
+
+	for i := range count {
+		var event Event
+
+		if i%2 == 0 {
+			event = PeerConnectedEvent{
+				PeerID: types.ID(fmt.Sprintf("peer-%d", i)),
+			}
+		} else {
+			event = PeerDisconnectedEvent{
+				PeerID: types.ID(fmt.Sprintf("peer-%d", i)),
+			}
+		}
+
+		events = append(events, event)
+	}
+
+	return events
+}
+
+func TestEvents_Subscribe(t *testing.T) {
+	t.Parallel()
+
+	var (
+		capturedEvents []Event
+
+		events = generateEvents(10)
+		subFn  = func(e Event) bool {
+			return e.Type() == PeerDisconnected
+		}
+	)
+
+	// Create the events manager
+	e := New()
+
+	// Subscribe to events
+	ch, unsubFn := e.Subscribe(subFn)
+	defer unsubFn()
+
+	// Listen for the events
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+
+		timeout := time.After(5 * time.Second)
+
+		for {
+			select {
+			case ev := <-ch:
+				capturedEvents = append(capturedEvents, ev)
+
+				if len(capturedEvents) == len(events)/2 {
+					return
+				}
+			case <-timeout:
+				return
+			}
+		}
+	}()
+
+	// Send out the events
+	for _, ev := range events {
+		e.Notify(ev)
+	}
+
+	wg.Wait()
+
+	// Make sure the events were captured
+	// and filtered properly
+	require.Len(t, capturedEvents, len(events)/2)
+
+	for _, ev := range capturedEvents {
+		assert.Equal(t, ev.Type(), PeerDisconnected)
+	}
+}

tm2/pkg/p2p/events/types.go

@@ -11,7 +11,7 @@ const (
 	PeerDisconnected EventType = "PeerDisconnected" // emitted when a peer disconnects
 )
 
-// Event is a p2p event
+// Event is a generic p2p event
 type Event interface {
 	// Type returns the type information for the event
 	Type() EventType
@@ -22,7 +22,7 @@ type PeerConnectedEvent struct {
 	Address types.NetAddress // the dial address of the peer
 }
 
-func (p *PeerConnectedEvent) Type() EventType {
+func (p PeerConnectedEvent) Type() EventType {
 	return PeerConnected
 }
 
@@ -32,6 +32,6 @@ type PeerDisconnectedEvent struct {
 	Reason  error            // the disconnect reason, if any
 }
 
-func (p *PeerDisconnectedEvent) Type() EventType {
+func (p PeerDisconnectedEvent) Type() EventType {
 	return PeerDisconnected
 }