logos-messaging/logos-messaging-go
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-messaging-go.git synced 2026-01-05 23:43:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: method to disconnect all peers and not notify (#1168)

Browse Source
This commit is contained in:
Prem Chaitanya Prathi 2024-07-24 18:17:31 +05:30 committed by GitHub
parent 58d9721026
commit a9be17fd48
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 11 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
waku/v2/node/wakunode2.go
View File

@ -799,6 +799,17 @@ func (w *WakuNode) ClosePeerByAddress(address string) error {
return w.ClosePeerById(info.ID) return w.ClosePeerById(info.ID)
} }
func (w *WakuNode) DisconnectAllPeers() {
w.host.Network().StopNotify(w.connectionNotif)
for _, peerID := range w.host.Network().Peers() {
err := w.ClosePeerById(peerID)
if err != nil {
w.log.Info("failed to close peer", zap.Stringer("peer", peerID), zap.Error(err))
}
}
w.host.Network().Notify(w.connectionNotif)
}
// ClosePeerById is used to close a connection to a peer // ClosePeerById is used to close a connection to a peer
func (w *WakuNode) ClosePeerById(id peer.ID) error { func (w *WakuNode) ClosePeerById(id peer.ID) error {
err := w.host.Network().ClosePeer(id) err := w.host.Network().ClosePeer(id)

15
waku/v2/protocol/filter/client.go
View File

@ -147,17 +147,6 @@ func (wf *WakuFilterLightNode) Stop() {
}) })
} }
func (wf *WakuFilterLightNode) unsubscribeWithoutSubscription(cf protocol.ContentFilter, peerID peer.ID) {
err := wf.request(
wf.Context(),
protocol.GenerateRequestID(),
pb.FilterSubscribeRequest_UNSUBSCRIBE_ALL,
cf, peerID)
if err != nil {
wf.log.Warn("could not unsubscribe from peer", logging.HostID("peerID", peerID), zap.Error(err))
}
}
func (wf *WakuFilterLightNode) onRequest(ctx context.Context) func(network.Stream) { func (wf *WakuFilterLightNode) onRequest(ctx context.Context) func(network.Stream) {
return func(stream network.Stream) { return func(stream network.Stream) {
peerID := stream.Conn().RemotePeer() peerID := stream.Conn().RemotePeer()
@ -168,8 +157,6 @@ func (wf *WakuFilterLightNode) onRequest(ctx context.Context) func(network.Strea
logger.Warn("received message push from unknown peer", logging.HostID("peerID", peerID)) logger.Warn("received message push from unknown peer", logging.HostID("peerID", peerID))
wf.metrics.RecordError(unknownPeerMessagePush) wf.metrics.RecordError(unknownPeerMessagePush)
//Send a wildcard unsubscribe to this peer so that further requests are not forwarded to us //Send a wildcard unsubscribe to this peer so that further requests are not forwarded to us
//This could be happening due to https://github.com/waku-org/go-waku/issues/1124
go wf.unsubscribeWithoutSubscription(protocol.ContentFilter{}, peerID)
if err := stream.Reset(); err != nil { if err := stream.Reset(); err != nil {
wf.log.Error("resetting connection", zap.Error(err)) wf.log.Error("resetting connection", zap.Error(err))
} }
@ -216,8 +203,6 @@ func (wf *WakuFilterLightNode) onRequest(ctx context.Context) func(network.Strea
cf := protocol.NewContentFilter(pubSubTopic, messagePush.WakuMessage.ContentTopic) cf := protocol.NewContentFilter(pubSubTopic, messagePush.WakuMessage.ContentTopic)
if !wf.subscriptions.Has(peerID, cf) { if !wf.subscriptions.Has(peerID, cf) {
logger.Warn("received messagepush with invalid subscription parameters") logger.Warn("received messagepush with invalid subscription parameters")
//Unsubscribe from that peer for the contentTopic, possibly due to https://github.com/waku-org/go-waku/issues/1124
go wf.unsubscribeWithoutSubscription(cf, peerID)
wf.metrics.RecordError(invalidSubscriptionMessage) wf.metrics.RecordError(invalidSubscriptionMessage)
return return
} }