diff --git a/internal/transport/waku.go b/internal/transport/waku.go index 9acb41d..04a8a7f 100644 --- a/internal/transport/waku.go +++ b/internal/transport/waku.go @@ -46,7 +46,7 @@ func NewNode(ctx context.Context, logger *zap.Logger) *Node { return &Node{ waku: nil, ctx: ctx, - logger: logger, + logger: logger.Named("waku"), pubsubTopic: FleetName(config.Fleet()).DefaultPubsubTopic(), peerConnection: nil, roomCache: NewRoomCache(logger), @@ -68,14 +68,13 @@ func (n *Node) Initialize() error { } } - peerConnection := make(chan node.PeerConnection) + n.peerConnection = make(chan node.PeerConnection) options := []node.WakuNodeOption{ - node.WithLogger(n.logger.Named("waku")), - //node.WithDNS4Domain(), + node.WithLogger(n.logger), node.WithLogLevel(zap.DebugLevel), node.WithHostAddress(hostAddr), - node.WithConnectionNotification(peerConnection), + node.WithConnectionNotification(n.peerConnection), } if config.WakuDiscV5() { @@ -97,9 +96,7 @@ func (n *Node) Initialize() error { ) } - fleet := FleetName(config.Fleet()) - - if fleet.IsSharded() { + if FleetName(config.Fleet()).IsSharded() { options = append(options, node.WithClusterID(DefaultClusterID), ) @@ -113,8 +110,6 @@ func (n *Node) Initialize() error { } n.waku = wakuNode - n.peerConnection = peerConnection - n.logger = n.logger.Named("waku") n.connectedPeers = make(map[peer.ID]struct{}) return nil @@ -365,7 +360,7 @@ func (n *Node) watchConnectionStatus() { } else { delete(n.connectedPeers, status.PeerID) } - //count := n.waku.PeerCount() + // using manual calculation instead of n.waku.PeerCount() for simpler testing count := len(n.connectedPeers) n.notifyConnectionStatus(ConnectionStatus{ IsOnline: count > 0, diff --git a/internal/transport/waku_test.go b/internal/transport/waku_test.go index 1ecef3c..108be1f 100644 --- a/internal/transport/waku_test.go +++ b/internal/transport/waku_test.go @@ -95,6 +95,13 @@ func (s *WakuSuite) TestWatchConnectionStatus() { sub := s.node.SubscribeToConnectionStatus() + finished := make(chan struct{}) + + go func() { + s.node.watchConnectionStatus() + close(finished) + }() + sent := node.PeerConnection{ PeerID: peer.ID(gofakeit.UUID()), Connected: true, @@ -113,4 +120,11 @@ func (s *WakuSuite) TestWatchConnectionStatus() { } close(s.node.peerConnection) + + select { + case <-finished: + break + case <-time.After(500 * time.Millisecond): + s.Require().Fail("timeout waiting for connection status watch finish") + } }