From db25b307e26bba7f811a2b69d546eb8bc4f67afc Mon Sep 17 00:00:00 2001 From: Prem Chaitanya Prathi Date: Tue, 12 Sep 2023 18:04:43 +0530 Subject: [PATCH] Fix/filterv2 (#722) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: updated filterv2 protocol as per rfc, make pubsub topic optional * chore: make broadcaster optional in filter client * reverting optional pubSub topic in Filter.MessagePush * fix: revert pubSubTopic as optional and add higher level validation * Update waku/v2/protocol/filter/client.go Co-authored-by: richΛrd --------- Co-authored-by: richΛrd --- waku/v2/protocol/filter/client.go | 36 ++++++--- waku/v2/protocol/filter/filter_test.go | 15 ++-- .../protocol/filter/pb/waku_filter_v2.pb.go | 74 ++++++++++--------- .../protocol/filter/pb/waku_filter_v2.proto | 4 +- waku/v2/protocol/filter/server.go | 12 +-- waku/v2/protocol/filter/subscriptions_map.go | 2 +- 6 files changed, 81 insertions(+), 62 deletions(-) diff --git a/waku/v2/protocol/filter/client.go b/waku/v2/protocol/filter/client.go index d9448965..c8ed6620 100644 --- a/waku/v2/protocol/filter/client.go +++ b/waku/v2/protocol/filter/client.go @@ -40,7 +40,7 @@ type WakuFilterLightNode struct { cancel context.CancelFunc ctx context.Context h host.Host - broadcaster relay.Broadcaster + broadcaster relay.Broadcaster //TODO: Move the broadcast functionality outside of relay client to a higher SDK layer.s timesource timesource.Timesource metrics Metrics wg *sync.WaitGroup @@ -63,6 +63,7 @@ var errNotStarted = errors.New("not started") var errAlreadyStarted = errors.New("already started") // NewWakuFilterLightnode returns a new instance of Waku Filter struct setup according to the chosen parameter and options +// Note that broadcaster is optional. // Takes an optional peermanager if WakuFilterLightnode is being created along with WakuNode. // If using libp2p host, then pass peermanager as nil func NewWakuFilterLightNode(broadcaster relay.Broadcaster, pm *peermanager.PeerManager, @@ -159,16 +160,23 @@ func (wf *WakuFilterLightNode) onRequest(ctx context.Context) func(s network.Str wf.metrics.RecordError(decodeRPCFailure) return } - - if !wf.subscriptions.Has(s.Conn().RemotePeer(), messagePush.PubsubTopic, messagePush.WakuMessage.ContentTopic) { - logger.Warn("received messagepush with invalid subscription parameters", logging.HostID("peerID", s.Conn().RemotePeer()), zap.String("topic", messagePush.PubsubTopic), zap.String("contentTopic", messagePush.WakuMessage.ContentTopic)) + //For now returning failure, this will get addressed with autosharding changes for filter. + if messagePush.PubsubTopic == nil { + logger.Error("empty pubsub topic") + wf.metrics.RecordError(decodeRPCFailure) + return + } + if !wf.subscriptions.Has(s.Conn().RemotePeer(), *messagePush.PubsubTopic, messagePush.WakuMessage.ContentTopic) { + logger.Warn("received messagepush with invalid subscription parameters", + logging.HostID("peerID", s.Conn().RemotePeer()), zap.String("topic", *messagePush.PubsubTopic), + zap.String("contentTopic", messagePush.WakuMessage.ContentTopic)) wf.metrics.RecordError(invalidSubscriptionMessage) return } wf.metrics.RecordMessage() - wf.notify(s.Conn().RemotePeer(), messagePush.PubsubTopic, messagePush.WakuMessage) + wf.notify(s.Conn().RemotePeer(), *messagePush.PubsubTopic, messagePush.WakuMessage) logger.Info("received message push") } @@ -177,14 +185,16 @@ func (wf *WakuFilterLightNode) onRequest(ctx context.Context) func(s network.Str func (wf *WakuFilterLightNode) notify(remotePeerID peer.ID, pubsubTopic string, msg *wpb.WakuMessage) { envelope := protocol.NewEnvelope(msg, wf.timesource.Now().UnixNano(), pubsubTopic) - // Broadcasting message so it's stored - wf.broadcaster.Submit(envelope) - + if wf.broadcaster != nil { + // Broadcasting message so it's stored + wf.broadcaster.Submit(envelope) + } // Notify filter subscribers wf.subscriptions.Notify(remotePeerID, envelope) } -func (wf *WakuFilterLightNode) request(ctx context.Context, params *FilterSubscribeParameters, reqType pb.FilterSubscribeRequest_FilterSubscribeType, contentFilter ContentFilter) error { +func (wf *WakuFilterLightNode) request(ctx context.Context, params *FilterSubscribeParameters, + reqType pb.FilterSubscribeRequest_FilterSubscribeType, contentFilter ContentFilter) error { conn, err := wf.h.NewStream(ctx, params.selectedPeer, FilterSubscribeID_v20beta1) if err != nil { wf.metrics.RecordError(dialFailure) @@ -198,7 +208,7 @@ func (wf *WakuFilterLightNode) request(ctx context.Context, params *FilterSubscr request := &pb.FilterSubscribeRequest{ RequestId: hex.EncodeToString(params.requestID), FilterSubscribeType: reqType, - PubsubTopic: contentFilter.Topic, + PubsubTopic: &contentFilter.Topic, ContentTopics: contentFilter.ContentTopics, } @@ -388,7 +398,8 @@ func (wf *WakuFilterLightNode) cleanupSubscriptions(peerID peer.ID, contentFilte } // Unsubscribe is used to stop receiving messages from a peer that match a content filter -func (wf *WakuFilterLightNode) Unsubscribe(ctx context.Context, contentFilter ContentFilter, opts ...FilterUnsubscribeOption) (<-chan WakuFilterPushResult, error) { +func (wf *WakuFilterLightNode) Unsubscribe(ctx context.Context, contentFilter ContentFilter, + opts ...FilterUnsubscribeOption) (<-chan WakuFilterPushResult, error) { wf.RLock() defer wf.RUnlock() @@ -474,7 +485,8 @@ func (wf *WakuFilterLightNode) Unsubscribe(ctx context.Context, contentFilter Co } // Unsubscribe is used to stop receiving messages from a peer that match a content filter -func (wf *WakuFilterLightNode) UnsubscribeWithSubscription(ctx context.Context, sub *SubscriptionDetails, opts ...FilterUnsubscribeOption) (<-chan WakuFilterPushResult, error) { +func (wf *WakuFilterLightNode) UnsubscribeWithSubscription(ctx context.Context, sub *SubscriptionDetails, + opts ...FilterUnsubscribeOption) (<-chan WakuFilterPushResult, error) { wf.RLock() defer wf.RUnlock() diff --git a/waku/v2/protocol/filter/filter_test.go b/waku/v2/protocol/filter/filter_test.go index 2d685371..9e046780 100644 --- a/waku/v2/protocol/filter/filter_test.go +++ b/waku/v2/protocol/filter/filter_test.go @@ -68,15 +68,17 @@ func (s *FilterTestSuite) makeWakuRelay(topic string) (*relay.WakuRelay, *relay. return relay, sub, host, broadcaster } -func (s *FilterTestSuite) makeWakuFilterLightNode(start bool) *WakuFilterLightNode { +func (s *FilterTestSuite) makeWakuFilterLightNode(start bool, withBroadcaster bool) *WakuFilterLightNode { port, err := tests.FindFreePort(s.T(), "", 5) s.Require().NoError(err) host, err := tests.MakeHost(context.Background(), port, rand.Reader) s.Require().NoError(err) - - b := relay.NewBroadcaster(10) - s.Require().NoError(b.Start(context.Background())) + var b relay.Broadcaster + if withBroadcaster { + b = relay.NewBroadcaster(10) + s.Require().NoError(b.Start(context.Background())) + } filterPush := NewWakuFilterLightNode(b, nil, timesource.NewDefaultClock(), prometheus.DefaultRegisterer, s.log) filterPush.SetHost(host) s.lightNodeHost = host @@ -181,7 +183,8 @@ func (s *FilterTestSuite) SetupTest() { s.testTopic = "/waku/2/go/filter/test" s.testContentTopic = "TopicA" - s.lightNode = s.makeWakuFilterLightNode(true) + s.lightNode = s.makeWakuFilterLightNode(true, false) + //TODO: Add tests to verify broadcaster. s.relayNode, s.fullNode = s.makeWakuFilterFullNode(s.testTopic) @@ -383,7 +386,7 @@ func (s *FilterTestSuite) TestFireAndForgetAndCustomWg() { func (s *FilterTestSuite) TestStartStop() { var wg sync.WaitGroup wg.Add(2) - s.lightNode = s.makeWakuFilterLightNode(false) + s.lightNode = s.makeWakuFilterLightNode(false, false) stopNode := func() { for i := 0; i < 100000; i++ { diff --git a/waku/v2/protocol/filter/pb/waku_filter_v2.pb.go b/waku/v2/protocol/filter/pb/waku_filter_v2.pb.go index 0bea8dbc..7bc04405 100644 --- a/waku/v2/protocol/filter/pb/waku_filter_v2.pb.go +++ b/waku/v2/protocol/filter/pb/waku_filter_v2.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.26.0 -// protoc v3.21.12 +// protoc-gen-go v1.31.0 +// protoc v4.23.4 // source: waku_filter_v2.proto // 12/WAKU2-FILTER rfc: https://rfc.vac.dev/spec/12/ @@ -84,7 +84,7 @@ type FilterSubscribeRequest struct { RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` FilterSubscribeType FilterSubscribeRequest_FilterSubscribeType `protobuf:"varint,2,opt,name=filter_subscribe_type,json=filterSubscribeType,proto3,enum=pb.FilterSubscribeRequest_FilterSubscribeType" json:"filter_subscribe_type,omitempty"` // Filter criteria - PubsubTopic string `protobuf:"bytes,10,opt,name=pubsub_topic,json=pubsubTopic,proto3" json:"pubsub_topic,omitempty"` + PubsubTopic *string `protobuf:"bytes,10,opt,name=pubsub_topic,json=pubsubTopic,proto3,oneof" json:"pubsub_topic,omitempty"` ContentTopics []string `protobuf:"bytes,11,rep,name=content_topics,json=contentTopics,proto3" json:"content_topics,omitempty"` } @@ -135,8 +135,8 @@ func (x *FilterSubscribeRequest) GetFilterSubscribeType() FilterSubscribeRequest } func (x *FilterSubscribeRequest) GetPubsubTopic() string { - if x != nil { - return x.PubsubTopic + if x != nil && x.PubsubTopic != nil { + return *x.PubsubTopic } return "" } @@ -218,7 +218,7 @@ type MessagePushV2 struct { unknownFields protoimpl.UnknownFields WakuMessage *pb.WakuMessage `protobuf:"bytes,1,opt,name=waku_message,json=wakuMessage,proto3" json:"waku_message,omitempty"` - PubsubTopic string `protobuf:"bytes,2,opt,name=pubsub_topic,json=pubsubTopic,proto3" json:"pubsub_topic,omitempty"` + PubsubTopic *string `protobuf:"bytes,2,opt,name=pubsub_topic,json=pubsubTopic,proto3,oneof" json:"pubsub_topic,omitempty"` } func (x *MessagePushV2) Reset() { @@ -261,8 +261,8 @@ func (x *MessagePushV2) GetWakuMessage() *pb.WakuMessage { } func (x *MessagePushV2) GetPubsubTopic() string { - if x != nil { - return x.PubsubTopic + if x != nil && x.PubsubTopic != nil { + return *x.PubsubTopic } return "" } @@ -272,7 +272,7 @@ var File_waku_filter_v2_proto protoreflect.FileDescriptor var file_waku_filter_v2_proto_rawDesc = []byte{ 0x0a, 0x14, 0x77, 0x61, 0x6b, 0x75, 0x5f, 0x66, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x5f, 0x76, 0x32, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x02, 0x70, 0x62, 0x1a, 0x12, 0x77, 0x61, 0x6b, 0x75, - 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xc6, + 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xdc, 0x02, 0x0a, 0x16, 0x46, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1d, 0x0a, 0x0a, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x72, @@ -282,33 +282,35 @@ var file_waku_filter_v2_proto_rawDesc = []byte{ 0x74, 0x65, 0x72, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x2e, 0x46, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x54, 0x79, 0x70, 0x65, 0x52, 0x13, 0x66, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x53, - 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12, 0x21, 0x0a, 0x0c, + 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12, 0x26, 0x0a, 0x0c, 0x70, 0x75, 0x62, 0x73, 0x75, 0x62, 0x5f, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x0a, 0x20, 0x01, - 0x28, 0x09, 0x52, 0x0b, 0x70, 0x75, 0x62, 0x73, 0x75, 0x62, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x12, - 0x25, 0x0a, 0x0e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x5f, 0x74, 0x6f, 0x70, 0x69, 0x63, - 0x73, 0x18, 0x0b, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0d, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, - 0x54, 0x6f, 0x70, 0x69, 0x63, 0x73, 0x22, 0x5f, 0x0a, 0x13, 0x46, 0x69, 0x6c, 0x74, 0x65, 0x72, - 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12, 0x13, 0x0a, - 0x0f, 0x53, 0x55, 0x42, 0x53, 0x43, 0x52, 0x49, 0x42, 0x45, 0x52, 0x5f, 0x50, 0x49, 0x4e, 0x47, - 0x10, 0x00, 0x12, 0x0d, 0x0a, 0x09, 0x53, 0x55, 0x42, 0x53, 0x43, 0x52, 0x49, 0x42, 0x45, 0x10, - 0x01, 0x12, 0x0f, 0x0a, 0x0b, 0x55, 0x4e, 0x53, 0x55, 0x42, 0x53, 0x43, 0x52, 0x49, 0x42, 0x45, - 0x10, 0x02, 0x12, 0x13, 0x0a, 0x0f, 0x55, 0x4e, 0x53, 0x55, 0x42, 0x53, 0x43, 0x52, 0x49, 0x42, - 0x45, 0x5f, 0x41, 0x4c, 0x4c, 0x10, 0x03, 0x22, 0x7a, 0x0a, 0x17, 0x46, 0x69, 0x6c, 0x74, 0x65, - 0x72, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, - 0x73, 0x65, 0x12, 0x1d, 0x0a, 0x0a, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x5f, 0x69, 0x64, - 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x49, - 0x64, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x5f, 0x63, 0x6f, 0x64, 0x65, - 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x43, 0x6f, - 0x64, 0x65, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x5f, 0x64, 0x65, 0x73, - 0x63, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x44, - 0x65, 0x73, 0x63, 0x22, 0x66, 0x0a, 0x0d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x50, 0x75, - 0x73, 0x68, 0x56, 0x32, 0x12, 0x32, 0x0a, 0x0c, 0x77, 0x61, 0x6b, 0x75, 0x5f, 0x6d, 0x65, 0x73, - 0x73, 0x61, 0x67, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0f, 0x2e, 0x70, 0x62, 0x2e, - 0x57, 0x61, 0x6b, 0x75, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x52, 0x0b, 0x77, 0x61, 0x6b, - 0x75, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x21, 0x0a, 0x0c, 0x70, 0x75, 0x62, 0x73, - 0x75, 0x62, 0x5f, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, - 0x70, 0x75, 0x62, 0x73, 0x75, 0x62, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x62, 0x06, 0x70, 0x72, 0x6f, - 0x74, 0x6f, 0x33, + 0x28, 0x09, 0x48, 0x00, 0x52, 0x0b, 0x70, 0x75, 0x62, 0x73, 0x75, 0x62, 0x54, 0x6f, 0x70, 0x69, + 0x63, 0x88, 0x01, 0x01, 0x12, 0x25, 0x0a, 0x0e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x5f, + 0x74, 0x6f, 0x70, 0x69, 0x63, 0x73, 0x18, 0x0b, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0d, 0x63, 0x6f, + 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x73, 0x22, 0x5f, 0x0a, 0x13, 0x46, + 0x69, 0x6c, 0x74, 0x65, 0x72, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x54, 0x79, + 0x70, 0x65, 0x12, 0x13, 0x0a, 0x0f, 0x53, 0x55, 0x42, 0x53, 0x43, 0x52, 0x49, 0x42, 0x45, 0x52, + 0x5f, 0x50, 0x49, 0x4e, 0x47, 0x10, 0x00, 0x12, 0x0d, 0x0a, 0x09, 0x53, 0x55, 0x42, 0x53, 0x43, + 0x52, 0x49, 0x42, 0x45, 0x10, 0x01, 0x12, 0x0f, 0x0a, 0x0b, 0x55, 0x4e, 0x53, 0x55, 0x42, 0x53, + 0x43, 0x52, 0x49, 0x42, 0x45, 0x10, 0x02, 0x12, 0x13, 0x0a, 0x0f, 0x55, 0x4e, 0x53, 0x55, 0x42, + 0x53, 0x43, 0x52, 0x49, 0x42, 0x45, 0x5f, 0x41, 0x4c, 0x4c, 0x10, 0x03, 0x42, 0x0f, 0x0a, 0x0d, + 0x5f, 0x70, 0x75, 0x62, 0x73, 0x75, 0x62, 0x5f, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x22, 0x7a, 0x0a, + 0x17, 0x46, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, + 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x1d, 0x0a, 0x0a, 0x72, 0x65, 0x71, 0x75, + 0x65, 0x73, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x72, 0x65, + 0x71, 0x75, 0x65, 0x73, 0x74, 0x49, 0x64, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x74, 0x61, 0x74, 0x75, + 0x73, 0x5f, 0x63, 0x6f, 0x64, 0x65, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x73, 0x74, + 0x61, 0x74, 0x75, 0x73, 0x43, 0x6f, 0x64, 0x65, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x74, 0x61, 0x74, + 0x75, 0x73, 0x5f, 0x64, 0x65, 0x73, 0x63, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x73, + 0x74, 0x61, 0x74, 0x75, 0x73, 0x44, 0x65, 0x73, 0x63, 0x22, 0x7c, 0x0a, 0x0d, 0x4d, 0x65, 0x73, + 0x73, 0x61, 0x67, 0x65, 0x50, 0x75, 0x73, 0x68, 0x56, 0x32, 0x12, 0x32, 0x0a, 0x0c, 0x77, 0x61, + 0x6b, 0x75, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, + 0x32, 0x0f, 0x2e, 0x70, 0x62, 0x2e, 0x57, 0x61, 0x6b, 0x75, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, + 0x65, 0x52, 0x0b, 0x77, 0x61, 0x6b, 0x75, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x26, + 0x0a, 0x0c, 0x70, 0x75, 0x62, 0x73, 0x75, 0x62, 0x5f, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x02, + 0x20, 0x01, 0x28, 0x09, 0x48, 0x00, 0x52, 0x0b, 0x70, 0x75, 0x62, 0x73, 0x75, 0x62, 0x54, 0x6f, + 0x70, 0x69, 0x63, 0x88, 0x01, 0x01, 0x42, 0x0f, 0x0a, 0x0d, 0x5f, 0x70, 0x75, 0x62, 0x73, 0x75, + 0x62, 0x5f, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( @@ -385,6 +387,8 @@ func file_waku_filter_v2_proto_init() { } } } + file_waku_filter_v2_proto_msgTypes[0].OneofWrappers = []interface{}{} + file_waku_filter_v2_proto_msgTypes[2].OneofWrappers = []interface{}{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ diff --git a/waku/v2/protocol/filter/pb/waku_filter_v2.proto b/waku/v2/protocol/filter/pb/waku_filter_v2.proto index f27d65af..50ad632c 100644 --- a/waku/v2/protocol/filter/pb/waku_filter_v2.proto +++ b/waku/v2/protocol/filter/pb/waku_filter_v2.proto @@ -19,7 +19,7 @@ message FilterSubscribeRequest { FilterSubscribeType filter_subscribe_type = 2; // Filter criteria - string pubsub_topic = 10; + optional string pubsub_topic = 10; repeated string content_topics = 11; } @@ -32,5 +32,5 @@ message FilterSubscribeResponse { // Protocol identifier: /vac/waku/filter-push/2.0.0-beta1 message MessagePushV2 { WakuMessage waku_message = 1; - string pubsub_topic = 2; + optional string pubsub_topic = 2; } \ No newline at end of file diff --git a/waku/v2/protocol/filter/server.go b/waku/v2/protocol/filter/server.go index b47c8db3..1baa2c96 100644 --- a/waku/v2/protocol/filter/server.go +++ b/waku/v2/protocol/filter/server.go @@ -153,7 +153,7 @@ func (wf *WakuFilterFullNode) ping(ctx context.Context, s network.Stream, logger } func (wf *WakuFilterFullNode) subscribe(ctx context.Context, s network.Stream, logger *zap.Logger, request *pb.FilterSubscribeRequest) { - if request.PubsubTopic == "" { + if request.PubsubTopic == nil { wf.reply(ctx, s, request, http.StatusBadRequest, "pubsubtopic can't be empty") return } @@ -186,14 +186,14 @@ func (wf *WakuFilterFullNode) subscribe(ctx context.Context, s network.Stream, l } } - wf.subscriptions.Set(peerID, request.PubsubTopic, request.ContentTopics) + wf.subscriptions.Set(peerID, *request.PubsubTopic, request.ContentTopics) wf.metrics.RecordSubscriptions(wf.subscriptions.Count()) wf.reply(ctx, s, request, http.StatusOK) } func (wf *WakuFilterFullNode) unsubscribe(ctx context.Context, s network.Stream, logger *zap.Logger, request *pb.FilterSubscribeRequest) { - if request.PubsubTopic == "" { + if request.PubsubTopic == nil { wf.reply(ctx, s, request, http.StatusBadRequest, "pubsubtopic can't be empty") return } @@ -207,7 +207,7 @@ func (wf *WakuFilterFullNode) unsubscribe(ctx context.Context, s network.Stream, wf.reply(ctx, s, request, http.StatusBadRequest, fmt.Sprintf("exceeds maximum content topics: %d", MaxContentTopicsPerRequest)) } - err := wf.subscriptions.Delete(s.Conn().RemotePeer(), request.PubsubTopic, request.ContentTopics) + err := wf.subscriptions.Delete(s.Conn().RemotePeer(), *request.PubsubTopic, request.ContentTopics) if err != nil { wf.reply(ctx, s, request, http.StatusNotFound, peerHasNoSubscription) } else { @@ -273,9 +273,9 @@ func (wf *WakuFilterFullNode) pushMessage(ctx context.Context, peerID peer.ID, e zap.String("pubsubTopic", env.PubsubTopic()), zap.String("contentTopic", env.Message().ContentTopic), ) - + pubSubTopic := env.PubsubTopic() messagePush := &pb.MessagePushV2{ - PubsubTopic: env.PubsubTopic(), + PubsubTopic: &pubSubTopic, WakuMessage: env.Message(), } diff --git a/waku/v2/protocol/filter/subscriptions_map.go b/waku/v2/protocol/filter/subscriptions_map.go index 5272fa9b..167a2bcc 100644 --- a/waku/v2/protocol/filter/subscriptions_map.go +++ b/waku/v2/protocol/filter/subscriptions_map.go @@ -97,7 +97,7 @@ func (sub *SubscriptionsMap) Has(peerID peer.ID, topic string, contentTopics ... if !ok { return false } - + //TODO: Handle pubsubTopic as null // Check if pubsub topic exists subscriptions, ok := peerSubscription.subscriptionsPerTopic[topic] if !ok {