Add non-blocking to p2p event streams

tm2/pkg/internal/p2p/p2p.go

@@ -131,10 +131,13 @@ func MakeConnectedPeers(
 		multiplexSwitch.DialPeers(addrs...)
 
 		// Set up an exit timer
-		timer := time.NewTimer(5 * time.Second)
+		timer := time.NewTimer(1 * time.Minute)
 		defer timer.Stop()
 
-		connectedPeers := make(map[p2pTypes.ID]struct{})
+		var (
+			connectedPeers = make(map[p2pTypes.ID]struct{})
+			targetPeers    = cfg.Count - 1
+		)
 
 		for {
 			select {
@@ -143,7 +146,7 @@ func MakeConnectedPeers(
 
 				connectedPeers[ev.PeerID] = struct{}{}
 
-				if len(connectedPeers) == cfg.Count-1 {
+				if len(connectedPeers) == targetPeers {
 					return nil
 				}
 			case <-timer.C:

tm2/pkg/p2p/events/events.go

@@ -68,7 +68,12 @@ type (
 func (s *subscriptions) add(filterFn EventFilter) (string, chan Event) {
 	var (
 		id = xid.New().String()
-		ch = make(chan Event, 1)
+		// Since the event stream is non-blocking,
+		// the event buffer should be sufficiently
+		// large for most use-cases. Subscribers can
+		// handle large event load caller-side to mitigate
+		// events potentially being missed
+		ch = make(chan Event, 100)
 	)
 
 	(*s)[id] = subscription{
@@ -99,6 +104,9 @@ func (s *subscriptions) notify(event Event) {
 			continue
 		}
 
-		sub.ch <- event
+		select {
+		case sub.ch <- event:
+		default: // non-blocking
+		}
 	}
 }