status-im
/
go-waku
mirror of https://github.com/status-im/go-waku.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

fix: stream closing/reset

This commit is contained in:
Richard Ramos 2023-10-20 20:09:20 -04:00 committed by richΛrd
parent 519fa2977a
commit 9161c4f7fe
1 changed files with 25 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
waku/v2/protocol/metadata/waku_metadata.go
View File

@ -98,16 +98,11 @@ func (wakuM *WakuMetadata) Request(ctx context.Context, peerID peer.ID) (*protoc
return nil, err
}
defer stream.Close()
defer func() {
err := stream.Reset()
if err != nil {
logger.Error("resetting connection", zap.Error(err))
}
}()
clusterID, shards, err := wakuM.getClusterAndShards()
if err != nil {
if err := stream.Reset(); err != nil {
wakuM.log.Error("resetting connection", zap.Error(err))
}
return nil, err
}
@ -121,6 +116,9 @@ func (wakuM *WakuMetadata) Request(ctx context.Context, peerID peer.ID) (*protoc
err = writer.WriteMsg(request)
if err != nil {
logger.Error("writing request", zap.Error(err))
if err := stream.Reset(); err != nil {
wakuM.log.Error("resetting connection", zap.Error(err))
}
return nil, err
}
@ -128,9 +126,14 @@ func (wakuM *WakuMetadata) Request(ctx context.Context, peerID peer.ID) (*protoc
err = reader.ReadMsg(response)
if err != nil {
logger.Error("reading response", zap.Error(err))
if err := stream.Reset(); err != nil {
wakuM.log.Error("resetting connection", zap.Error(err))
}
return nil, err
}
stream.Close()
if response.ClusterId == nil {
return nil, nil // Node is not using sharding
}
@ -144,18 +147,20 @@ func (wakuM *WakuMetadata) Request(ctx context.Context, peerID peer.ID) (*protoc
return result, nil
}
func (wakuM *WakuMetadata) onRequest(ctx context.Context) func(s network.Stream) {
return func(s network.Stream) {
defer s.Close()
logger := wakuM.log.With(logging.HostID("peer", s.Conn().RemotePeer()))
func (wakuM *WakuMetadata) onRequest(ctx context.Context) func(network.Stream) {
return func(stream network.Stream) {
logger := wakuM.log.With(logging.HostID("peer", stream.Conn().RemotePeer()))
request := &pb.WakuMetadataRequest{}
writer := pbio.NewDelimitedWriter(s)
reader := pbio.NewDelimitedReader(s, math.MaxInt32)
writer := pbio.NewDelimitedWriter(stream)
reader := pbio.NewDelimitedReader(stream, math.MaxInt32)
err := reader.ReadMsg(request)
if err != nil {
logger.Error("reading request", zap.Error(err))
if err := stream.Reset(); err != nil {
wakuM.log.Error("resetting connection", zap.Error(err))
}
return
}
@ -172,8 +177,13 @@ func (wakuM *WakuMetadata) onRequest(ctx context.Context) func(s network.Stream)
err = writer.WriteMsg(response)
if err != nil {
logger.Error("writing response", zap.Error(err))
_ = s.Reset()
if err := stream.Reset(); err != nil {
wakuM.log.Error("resetting connection", zap.Error(err))
}
return
}
stream.Close()
}
}