status-im/go-waku
2
0
Fork 0
mirror of https://github.com/status-im/go-waku.git synced 2025-01-29 15:05:29 +00:00
Code Issues Projects Releases Wiki Activity

Use PubsubTopic naming; enforce unique ContentTopics (#750)

Browse Source
This commit is contained in:
Vitaliy Vlasov 2023-09-19 15:52:11 +03:00 committed by GitHub
parent 3d8d435502
commit 81638fe111
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 145 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
examples/chat2/chat.go
View File

@ -64,8 +64,8 @@ func NewChat(ctx context.Context, node *node.WakuNode, connNotifier <-chan node.
if options.Filter.Enable { if options.Filter.Enable {
cf := filter.ContentFilter{ cf := filter.ContentFilter{
Topic: relay.DefaultWakuTopic, PubsubTopic: relay.DefaultWakuTopic,
ContentTopics: []string{options.ContentTopic}, ContentTopics: filter.NewContentTopicSet(options.ContentTopic),
} }
var filterOpt filter.FilterSubscribeOption var filterOpt filter.FilterSubscribeOption
peerID, err := options.Filter.NodePeerID() peerID, err := options.Filter.NodePeerID()

4
examples/filter2/main.go
View File

@ -98,8 +98,8 @@ func main() {
// Send FilterRequest from light node to full node // Send FilterRequest from light node to full node
cf := filter.ContentFilter{ cf := filter.ContentFilter{
Topic: pubSubTopic.String(), PubsubTopic: pubSubTopic.String(),
ContentTopics: []string{contentTopic}, ContentTopics: filter.NewContentTopicSet(contentTopic),
} }
theFilter, err := lightNode.FilterLightnode().Subscribe(ctx, cf) theFilter, err := lightNode.FilterLightnode().Subscribe(ctx, cf)

2
flake.nix
View File

@ -29,7 +29,7 @@
]; ];
doCheck = false; doCheck = false;
# FIXME: This needs to be manually changed when updating modules. # FIXME: This needs to be manually changed when updating modules.
vendorSha256 = "sha256-eS/4YnNv2yGR+tVMq6xfx0Ntq8WosV+pTrbOb3mNYaA="; vendorSha256 = "sha256-4xChSKAkwwrFp5/ZMnhtvsR4drVfw1cLE3YXwVHeW0A=";
# Fix for 'nix run' trying to execute 'go-waku'. # Fix for 'nix run' trying to execute 'go-waku'.
meta = { mainProgram = "waku"; }; meta = { mainProgram = "waku"; };
}; };

6
library/filter.go
View File

@ -11,7 +11,7 @@ import (
) )
type filterArgument struct { type filterArgument struct {
Topic string `json:"pubsubTopic,omitempty"` PubsubTopic string `json:"pubsubTopic,omitempty"`
ContentTopics []string `json:"contentTopics,omitempty"` ContentTopics []string `json:"contentTopics,omitempty"`
} }
@ -23,8 +23,8 @@ func toContentFilter(filterJSON string) (filter.ContentFilter, error) {
} }
return filter.ContentFilter{ return filter.ContentFilter{
Topic: f.Topic, PubsubTopic: f.PubsubTopic,
ContentTopics: f.ContentTopics, ContentTopics: filter.NewContentTopicSet(f.ContentTopics...),
}, nil }, nil
} }

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

@ -22,6 +22,7 @@ import (
"github.com/waku-org/go-waku/waku/v2/protocol/relay" "github.com/waku-org/go-waku/waku/v2/protocol/relay"
"github.com/waku-org/go-waku/waku/v2/timesource" "github.com/waku-org/go-waku/waku/v2/timesource"
"go.uber.org/zap" "go.uber.org/zap"
"golang.org/x/exp/maps"
) )
// FilterPushID_v20beta1 is the current Waku Filter protocol identifier used to allow // FilterPushID_v20beta1 is the current Waku Filter protocol identifier used to allow
@ -44,8 +45,12 @@ type WakuFilterLightNode struct {
} }
type ContentFilter struct { type ContentFilter struct {
Topic string PubsubTopic string
ContentTopics []string ContentTopics ContentTopicSet
}
func (cf ContentFilter) ContentTopicsList() []string {
return maps.Keys(cf.ContentTopics)
} }
type WakuFilterPushResult struct { type WakuFilterPushResult struct {
@ -176,8 +181,8 @@ func (wf *WakuFilterLightNode) request(ctx context.Context, params *FilterSubscr
request := &pb.FilterSubscribeRequest{ request := &pb.FilterSubscribeRequest{
RequestId: hex.EncodeToString(params.requestID), RequestId: hex.EncodeToString(params.requestID),
FilterSubscribeType: reqType, FilterSubscribeType: reqType,
PubsubTopic: &contentFilter.Topic, PubsubTopic: &contentFilter.PubsubTopic,
ContentTopics: contentFilter.ContentTopics, ContentTopics: contentFilter.ContentTopicsList(),
} }
wf.log.Debug("sending FilterSubscribeRequest", zap.Stringer("request", request)) wf.log.Debug("sending FilterSubscribeRequest", zap.Stringer("request", request))
@ -220,8 +225,8 @@ func (wf *WakuFilterLightNode) Subscribe(ctx context.Context, contentFilter Cont
return nil, err return nil, err
} }
if contentFilter.Topic == "" { if contentFilter.PubsubTopic == "" {
return nil, errors.New("topic is required") return nil, errors.New("pubsub topic is required")
} }
if len(contentFilter.ContentTopics) == 0 { if len(contentFilter.ContentTopics) == 0 {
@ -248,11 +253,16 @@ func (wf *WakuFilterLightNode) Subscribe(ctx context.Context, contentFilter Cont
return nil, ErrNoPeersAvailable return nil, ErrNoPeersAvailable
} }
existingSub := wf.subscriptions.Get(params.selectedPeer, contentFilter)
if existingSub != nil {
return existingSub, nil
} else {
err := wf.request(ctx, params, pb.FilterSubscribeRequest_SUBSCRIBE, contentFilter) err := wf.request(ctx, params, pb.FilterSubscribeRequest_SUBSCRIBE, contentFilter)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return wf.subscriptions.NewSubscription(params.selectedPeer, contentFilter.Topic, contentFilter.ContentTopics), nil return wf.subscriptions.NewSubscription(params.selectedPeer, contentFilter), nil
}
} }
// FilterSubscription is used to obtain an object from which you could receive messages received via filter protocol // FilterSubscription is used to obtain an object from which you could receive messages received via filter protocol
@ -263,11 +273,11 @@ func (wf *WakuFilterLightNode) FilterSubscription(peerID peer.ID, contentFilter
return nil, err return nil, err
} }
if !wf.subscriptions.Has(peerID, contentFilter.Topic, contentFilter.ContentTopics...) { if !wf.subscriptions.Has(peerID, contentFilter.PubsubTopic, contentFilter.ContentTopicsList()...) {
return nil, errors.New("subscription does not exist") return nil, errors.New("subscription does not exist")
} }
return wf.subscriptions.NewSubscription(peerID, contentFilter.Topic, contentFilter.ContentTopics), nil return wf.subscriptions.NewSubscription(peerID, contentFilter), nil
} }
func (wf *WakuFilterLightNode) getUnsubscribeParameters(opts ...FilterUnsubscribeOption) (*FilterUnsubscribeParameters, error) { func (wf *WakuFilterLightNode) getUnsubscribeParameters(opts ...FilterUnsubscribeOption) (*FilterUnsubscribeParameters, error) {
@ -318,8 +328,8 @@ func (wf *WakuFilterLightNode) Subscriptions() []*SubscriptionDetails {
var output []*SubscriptionDetails var output []*SubscriptionDetails
for _, peerSubscription := range wf.subscriptions.items { for _, peerSubscription := range wf.subscriptions.items {
for _, subscriptionPerTopic := range peerSubscription.subscriptionsPerTopic { for _, subscriptions := range peerSubscription.subsPerPubsubTopic {
for _, subscriptionDetail := range subscriptionPerTopic { for _, subscriptionDetail := range subscriptions {
output = append(output, subscriptionDetail) output = append(output, subscriptionDetail)
} }
} }
@ -337,14 +347,14 @@ func (wf *WakuFilterLightNode) cleanupSubscriptions(peerID peer.ID, contentFilte
return return
} }
subscriptionDetailList, ok := peerSubscription.subscriptionsPerTopic[contentFilter.Topic] subscriptionDetailList, ok := peerSubscription.subsPerPubsubTopic[contentFilter.PubsubTopic]
if !ok { if !ok {
return return
} }
for subscriptionDetailID, subscriptionDetail := range subscriptionDetailList { for subscriptionDetailID, subscriptionDetail := range subscriptionDetailList {
subscriptionDetail.Remove(contentFilter.ContentTopics...) subscriptionDetail.Remove(contentFilter.ContentTopicsList()...)
if len(subscriptionDetail.ContentTopics) == 0 { if len(subscriptionDetail.ContentFilter.ContentTopics) == 0 {
delete(subscriptionDetailList, subscriptionDetailID) delete(subscriptionDetailList, subscriptionDetailID)
} else { } else {
subscriptionDetailList[subscriptionDetailID] = subscriptionDetail subscriptionDetailList[subscriptionDetailID] = subscriptionDetail
@ -352,9 +362,9 @@ func (wf *WakuFilterLightNode) cleanupSubscriptions(peerID peer.ID, contentFilte
} }
if len(subscriptionDetailList) == 0 { if len(subscriptionDetailList) == 0 {
delete(wf.subscriptions.items[peerID].subscriptionsPerTopic, contentFilter.Topic) delete(wf.subscriptions.items[peerID].subsPerPubsubTopic, contentFilter.PubsubTopic)
} else { } else {
wf.subscriptions.items[peerID].subscriptionsPerTopic[contentFilter.Topic] = subscriptionDetailList wf.subscriptions.items[peerID].subsPerPubsubTopic[contentFilter.PubsubTopic] = subscriptionDetailList
} }
} }
@ -367,8 +377,8 @@ func (wf *WakuFilterLightNode) Unsubscribe(ctx context.Context, contentFilter Co
return nil, err return nil, err
} }
if contentFilter.Topic == "" { if contentFilter.PubsubTopic == "" {
return nil, errors.New("topic is required") return nil, errors.New("pubsub topic is required")
} }
if len(contentFilter.ContentTopics) == 0 { if len(contentFilter.ContentTopics) == 0 {
@ -396,7 +406,7 @@ func (wf *WakuFilterLightNode) Unsubscribe(ctx context.Context, contentFilter Co
} }
wf.cleanupSubscriptions(peerID, contentFilter) wf.cleanupSubscriptions(peerID, contentFilter)
if len(subscriptions.subscriptionsPerTopic) == 0 { if len(subscriptions.subsPerPubsubTopic) == 0 {
delete(wf.subscriptions.items, peerID) delete(wf.subscriptions.items, peerID)
} }
@ -452,14 +462,9 @@ func (wf *WakuFilterLightNode) UnsubscribeWithSubscription(ctx context.Context,
return nil, err return nil, err
} }
var contentTopics []string
for k := range sub.ContentTopics {
contentTopics = append(contentTopics, k)
}
opts = append(opts, Peer(sub.PeerID)) opts = append(opts, Peer(sub.PeerID))
return wf.Unsubscribe(ctx, ContentFilter{Topic: sub.PubsubTopic, ContentTopics: contentTopics}, opts...) return wf.Unsubscribe(ctx, sub.ContentFilter, opts...)
} }
func (wf *WakuFilterLightNode) unsubscribeAll(ctx context.Context, opts ...FilterUnsubscribeOption) (<-chan WakuFilterPushResult, error) { func (wf *WakuFilterLightNode) unsubscribeAll(ctx context.Context, opts ...FilterUnsubscribeOption) (<-chan WakuFilterPushResult, error) {

20
waku/v2/protocol/filter/filter_test.go
View File

@ -20,6 +20,7 @@ import (
"github.com/waku-org/go-waku/waku/v2/timesource" "github.com/waku-org/go-waku/waku/v2/timesource"
"github.com/waku-org/go-waku/waku/v2/utils" "github.com/waku-org/go-waku/waku/v2/utils"
"go.uber.org/zap" "go.uber.org/zap"
"golang.org/x/exp/maps"
) )
func TestFilterSuite(t *testing.T) { func TestFilterSuite(t *testing.T) {
@ -109,7 +110,7 @@ func (s *FilterTestSuite) waitForMsg(fn func(), ch chan *protocol.Envelope) {
defer s.wg.Done() defer s.wg.Done()
select { select {
case env := <-ch: case env := <-ch:
s.Require().Equal(s.contentFilter.ContentTopics[0], env.Message().GetContentTopic()) s.Require().Equal(maps.Keys(s.contentFilter.ContentTopics)[0], env.Message().GetContentTopic())
case <-time.After(5 * time.Second): case <-time.After(5 * time.Second):
s.Require().Fail("Message timeout") s.Require().Fail("Message timeout")
case <-s.ctx.Done(): case <-s.ctx.Done():
@ -141,11 +142,8 @@ func (s *FilterTestSuite) waitForTimeout(fn func(), ch chan *protocol.Envelope)
s.wg.Wait() s.wg.Wait()
} }
func (s *FilterTestSuite) subscribe(topic string, contentTopic string, peer peer.ID) *SubscriptionDetails { func (s *FilterTestSuite) subscribe(pubsubTopic string, contentTopic string, peer peer.ID) *SubscriptionDetails {
s.contentFilter = ContentFilter{ s.contentFilter = ContentFilter{pubsubTopic, NewContentTopicSet(contentTopic)}
Topic: string(topic),
ContentTopics: []string{contentTopic},
}
subDetails, err := s.lightNode.Subscribe(s.ctx, s.contentFilter, WithPeer(peer)) subDetails, err := s.lightNode.Subscribe(s.ctx, s.contentFilter, WithPeer(peer))
s.Require().NoError(err) s.Require().NoError(err)
@ -343,10 +341,7 @@ func (s *FilterTestSuite) TestMultipleMessages() {
func (s *FilterTestSuite) TestRunningGuard() { func (s *FilterTestSuite) TestRunningGuard() {
s.lightNode.Stop() s.lightNode.Stop()
contentFilter := ContentFilter{ contentFilter := ContentFilter{"test", NewContentTopicSet("test")}
Topic: "test",
ContentTopics: []string{"test"},
}
_, err := s.lightNode.Subscribe(s.ctx, contentFilter, WithPeer(s.fullNodeHost.ID())) _, err := s.lightNode.Subscribe(s.ctx, contentFilter, WithPeer(s.fullNodeHost.ID()))
@ -361,10 +356,7 @@ func (s *FilterTestSuite) TestRunningGuard() {
} }
func (s *FilterTestSuite) TestFireAndForgetAndCustomWg() { func (s *FilterTestSuite) TestFireAndForgetAndCustomWg() {
contentFilter := ContentFilter{ contentFilter := ContentFilter{"test", NewContentTopicSet("test")}
Topic: "test",
ContentTopics: []string{"test"},
}
_, err := s.lightNode.Subscribe(s.ctx, contentFilter, WithPeer(s.fullNodeHost.ID())) _, err := s.lightNode.Subscribe(s.ctx, contentFilter, WithPeer(s.fullNodeHost.ID()))
s.Require().NoError(err) s.Require().NoError(err)

8
waku/v2/protocol/filter/subscribers_map.go
View File

@ -14,6 +14,14 @@ var ErrNotFound = errors.New("not found")
type ContentTopicSet map[string]struct{} type ContentTopicSet map[string]struct{}
func NewContentTopicSet(contentTopics ...string) ContentTopicSet {
s := make(ContentTopicSet, len(contentTopics))
for _, ct := range contentTopics {
s[ct] = struct{}{}
}
return s
}
type PeerSet map[peer.ID]struct{} type PeerSet map[peer.ID]struct{}
type PubsubTopics map[string]ContentTopicSet // pubsubTopic => contentTopics type PubsubTopics map[string]ContentTopicSet // pubsubTopic => contentTopics

44
waku/v2/protocol/filter/subscribers_map_test.go
View File

@ -10,7 +10,7 @@ import (
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
const TOPIC = "/test/topic" const PUBSUB_TOPIC = "/test/topic"
func createPeerID(t *testing.T) peer.ID { func createPeerID(t *testing.T) peer.ID {
peerId, err := test.RandPeerID() peerId, err := test.RandPeerID()
@ -29,23 +29,23 @@ func TestAppend(t *testing.T) {
subs := NewSubscribersMap(5 * time.Second) subs := NewSubscribersMap(5 * time.Second)
peerId := createPeerID(t) peerId := createPeerID(t)
subs.Set(peerId, TOPIC, []string{"topic1"}) subs.Set(peerId, PUBSUB_TOPIC, []string{"topic1"})
sub := firstSubscriber(subs, TOPIC, "topic1") sub := firstSubscriber(subs, PUBSUB_TOPIC, "topic1")
assert.NotEmpty(t, sub) assert.NotEmpty(t, sub)
// Adding to existing peer // Adding to existing peer
subs.Set(peerId, TOPIC, []string{"topic2"}) subs.Set(peerId, PUBSUB_TOPIC, []string{"topic2"})
sub = firstSubscriber(subs, TOPIC, "topic2") sub = firstSubscriber(subs, PUBSUB_TOPIC, "topic2")
assert.NotEmpty(t, sub) assert.NotEmpty(t, sub)
subs.Set(peerId, TOPIC+"2", []string{"topic1"}) subs.Set(peerId, PUBSUB_TOPIC+"2", []string{"topic1"})
sub = firstSubscriber(subs, TOPIC+"2", "topic1") sub = firstSubscriber(subs, PUBSUB_TOPIC+"2", "topic1")
assert.NotEmpty(t, sub) assert.NotEmpty(t, sub)
sub = firstSubscriber(subs, TOPIC+"2", "topic2") sub = firstSubscriber(subs, PUBSUB_TOPIC+"2", "topic2")
assert.Empty(t, sub) assert.Empty(t, sub)
} }
@ -53,19 +53,19 @@ func TestRemove(t *testing.T) {
subs := NewSubscribersMap(5 * time.Second) subs := NewSubscribersMap(5 * time.Second)
peerId := createPeerID(t) peerId := createPeerID(t)
subs.Set(peerId, TOPIC+"1", []string{"topic1", "topic2"}) subs.Set(peerId, PUBSUB_TOPIC+"1", []string{"topic1", "topic2"})
subs.Set(peerId, TOPIC+"2", []string{"topic1"}) subs.Set(peerId, PUBSUB_TOPIC+"2", []string{"topic1"})
err := subs.DeleteAll(peerId) err := subs.DeleteAll(peerId)
assert.Empty(t, err) assert.Empty(t, err)
sub := firstSubscriber(subs, TOPIC+"1", "topic1") sub := firstSubscriber(subs, PUBSUB_TOPIC+"1", "topic1")
assert.Empty(t, sub) assert.Empty(t, sub)
sub = firstSubscriber(subs, TOPIC+"1", "topic2") sub = firstSubscriber(subs, PUBSUB_TOPIC+"1", "topic2")
assert.Empty(t, sub) assert.Empty(t, sub)
sub = firstSubscriber(subs, TOPIC+"2", "topic1") sub = firstSubscriber(subs, PUBSUB_TOPIC+"2", "topic1")
assert.Empty(t, sub) assert.Empty(t, sub)
assert.False(t, subs.Has(peerId)) assert.False(t, subs.Has(peerId))
@ -81,11 +81,11 @@ func TestRemovePartial(t *testing.T) {
subs := NewSubscribersMap(5 * time.Second) subs := NewSubscribersMap(5 * time.Second)
peerId := createPeerID(t) peerId := createPeerID(t)
subs.Set(peerId, TOPIC, []string{"topic1", "topic2"}) subs.Set(peerId, PUBSUB_TOPIC, []string{"topic1", "topic2"})
err := subs.Delete(peerId, TOPIC, []string{"topic1"}) err := subs.Delete(peerId, PUBSUB_TOPIC, []string{"topic1"})
require.NoError(t, err) require.NoError(t, err)
sub := firstSubscriber(subs, TOPIC, "topic2") sub := firstSubscriber(subs, PUBSUB_TOPIC, "topic2")
assert.NotEmpty(t, sub) assert.NotEmpty(t, sub)
} }
@ -93,13 +93,13 @@ func TestRemoveBogus(t *testing.T) {
subs := NewSubscribersMap(5 * time.Second) subs := NewSubscribersMap(5 * time.Second)
peerId := createPeerID(t) peerId := createPeerID(t)
subs.Set(peerId, TOPIC, []string{"topic1", "topic2"}) subs.Set(peerId, PUBSUB_TOPIC, []string{"topic1", "topic2"})
err := subs.Delete(peerId, TOPIC, []string{"does not exist", "topic1"}) err := subs.Delete(peerId, PUBSUB_TOPIC, []string{"does not exist", "topic1"})
require.NoError(t, err) require.NoError(t, err)
sub := firstSubscriber(subs, TOPIC, "topic1") sub := firstSubscriber(subs, PUBSUB_TOPIC, "topic1")
assert.Empty(t, sub) assert.Empty(t, sub)
sub = firstSubscriber(subs, TOPIC, "does not exist") sub = firstSubscriber(subs, PUBSUB_TOPIC, "does not exist")
assert.Empty(t, sub) assert.Empty(t, sub)
err = subs.Delete(peerId, "DOES_NOT_EXIST", []string{"topic1"}) err = subs.Delete(peerId, "DOES_NOT_EXIST", []string{"topic1"})
@ -110,7 +110,7 @@ func TestSuccessFailure(t *testing.T) {
subs := NewSubscribersMap(5 * time.Second) subs := NewSubscribersMap(5 * time.Second)
peerId := createPeerID(t) peerId := createPeerID(t)
subs.Set(peerId, TOPIC, []string{"topic1", "topic2"}) subs.Set(peerId, PUBSUB_TOPIC, []string{"topic1", "topic2"})
subs.FlagAsFailure(peerId) subs.FlagAsFailure(peerId)
require.True(t, subs.IsFailedPeer(peerId)) require.True(t, subs.IsFailedPeer(peerId))
@ -118,7 +118,7 @@ func TestSuccessFailure(t *testing.T) {
subs.FlagAsFailure(peerId) subs.FlagAsFailure(peerId)
require.False(t, subs.Has(peerId)) require.False(t, subs.Has(peerId))
subs.Set(peerId, TOPIC, []string{"topic1", "topic2"}) subs.Set(peerId, PUBSUB_TOPIC, []string{"topic1", "topic2"})
subs.FlagAsFailure(peerId) subs.FlagAsFailure(peerId)
require.True(t, subs.IsFailedPeer(peerId)) require.True(t, subs.IsFailedPeer(peerId))

81
waku/v2/protocol/filter/subscriptions_map.go
View File

@ -8,6 +8,7 @@ import (
"github.com/libp2p/go-libp2p/core/peer" "github.com/libp2p/go-libp2p/core/peer"
"github.com/waku-org/go-waku/waku/v2/protocol" "github.com/waku-org/go-waku/waku/v2/protocol"
"go.uber.org/zap" "go.uber.org/zap"
"golang.org/x/exp/maps"
) )
type SubscriptionDetails struct { type SubscriptionDetails struct {
@ -19,16 +20,16 @@ type SubscriptionDetails struct {
once sync.Once once sync.Once
PeerID peer.ID PeerID peer.ID
PubsubTopic string ContentFilter ContentFilter
ContentTopics map[string]struct{}
C chan *protocol.Envelope C chan *protocol.Envelope
} }
// Map of SubscriptionDetails.ID to subscriptions
type SubscriptionSet map[string]*SubscriptionDetails type SubscriptionSet map[string]*SubscriptionDetails
type PeerSubscription struct { type PeerSubscription struct {
peerID peer.ID peerID peer.ID
subscriptionsPerTopic map[string]SubscriptionSet subsPerPubsubTopic map[string]SubscriptionSet
} }
type SubscriptionsMap struct { type SubscriptionsMap struct {
@ -44,7 +45,7 @@ func NewSubscriptionMap(logger *zap.Logger) *SubscriptionsMap {
} }
} }
func (sub *SubscriptionsMap) NewSubscription(peerID peer.ID, topic string, contentTopics []string) *SubscriptionDetails { func (sub *SubscriptionsMap) NewSubscription(peerID peer.ID, cf ContentFilter) *SubscriptionDetails {
sub.Lock() sub.Lock()
defer sub.Unlock() defer sub.Unlock()
@ -52,30 +53,25 @@ func (sub *SubscriptionsMap) NewSubscription(peerID peer.ID, topic string, conte
if !ok { if !ok {
peerSubscription = &PeerSubscription{ peerSubscription = &PeerSubscription{
peerID: peerID, peerID: peerID,
subscriptionsPerTopic: make(map[string]SubscriptionSet), subsPerPubsubTopic: make(map[string]SubscriptionSet),
} }
sub.items[peerID] = peerSubscription sub.items[peerID] = peerSubscription
} }
_, ok = peerSubscription.subscriptionsPerTopic[topic] _, ok = peerSubscription.subsPerPubsubTopic[cf.PubsubTopic]
if !ok { if !ok {
peerSubscription.subscriptionsPerTopic[topic] = make(SubscriptionSet) peerSubscription.subsPerPubsubTopic[cf.PubsubTopic] = make(SubscriptionSet)
} }
details := &SubscriptionDetails{ details := &SubscriptionDetails{
ID: uuid.NewString(), ID: uuid.NewString(),
mapRef: sub, mapRef: sub,
PeerID: peerID, PeerID: peerID,
PubsubTopic: topic,
C: make(chan *protocol.Envelope, 1024), C: make(chan *protocol.Envelope, 1024),
ContentTopics: make(map[string]struct{}), ContentFilter: ContentFilter{cf.PubsubTopic, maps.Clone(cf.ContentTopics)},
} }
for _, ct := range contentTopics { sub.items[peerID].subsPerPubsubTopic[cf.PubsubTopic][details.ID] = details
details.ContentTopics[ct] = struct{}{}
}
sub.items[peerID].subscriptionsPerTopic[topic][details.ID] = details
return details return details
} }
@ -88,7 +84,34 @@ func (sub *SubscriptionsMap) IsSubscribedTo(peerID peer.ID) bool {
return ok return ok
} }
func (sub *SubscriptionsMap) Has(peerID peer.ID, topic string, contentTopics ...string) bool { func (sub *SubscriptionsMap) Get(peerID peer.ID, cf ContentFilter) *SubscriptionDetails {
sub.RLock()
defer sub.RUnlock()
// Check if peer exits
peerSubscription, ok := sub.items[peerID]
if !ok {
return nil
}
// Check if pubsub topic exists
subscriptions, ok := peerSubscription.subsPerPubsubTopic[cf.PubsubTopic]
if !ok {
return nil
}
// Check if the content topic exists within the list of subscriptions for this peer
for _, subscription := range subscriptions {
if maps.Equal(subscription.ContentFilter.ContentTopics, cf.ContentTopics) {
return subscription
}
}
return nil
}
// Check if we have subscriptions for all (pubsubTopic, contentTopics[i]) pairs provided
func (sub *SubscriptionsMap) Has(peerID peer.ID, pubsubTopic string, contentTopics ...string) bool {
sub.RLock() sub.RLock()
defer sub.RUnlock() defer sub.RUnlock()
@ -99,7 +122,7 @@ func (sub *SubscriptionsMap) Has(peerID peer.ID, topic string, contentTopics ...
} }
//TODO: Handle pubsubTopic as null //TODO: Handle pubsubTopic as null
// Check if pubsub topic exists // Check if pubsub topic exists
subscriptions, ok := peerSubscription.subscriptionsPerTopic[topic] subscriptions, ok := peerSubscription.subsPerPubsubTopic[pubsubTopic]
if !ok { if !ok {
return false return false
} }
@ -108,7 +131,7 @@ func (sub *SubscriptionsMap) Has(peerID peer.ID, topic string, contentTopics ...
for _, ct := range contentTopics { for _, ct := range contentTopics {
found := false found := false
for _, subscription := range subscriptions { for _, subscription := range subscriptions {
_, exists := subscription.ContentTopics[ct] _, exists := subscription.ContentFilter.ContentTopics[ct]
if exists { if exists {
found = true found = true
break break
@ -121,7 +144,6 @@ func (sub *SubscriptionsMap) Has(peerID peer.ID, topic string, contentTopics ...
return true return true
} }
func (sub *SubscriptionsMap) Delete(subscription *SubscriptionDetails) error { func (sub *SubscriptionsMap) Delete(subscription *SubscriptionDetails) error {
sub.Lock() sub.Lock()
defer sub.Unlock() defer sub.Unlock()
@ -131,7 +153,7 @@ func (sub *SubscriptionsMap) Delete(subscription *SubscriptionDetails) error {
return ErrNotFound return ErrNotFound
} }
delete(peerSubscription.subscriptionsPerTopic[subscription.PubsubTopic], subscription.ID) delete(peerSubscription.subsPerPubsubTopic[subscription.ContentFilter.PubsubTopic], subscription.ID)
return nil return nil
} }
@ -141,7 +163,7 @@ func (s *SubscriptionDetails) Add(contentTopics ...string) {
defer s.Unlock() defer s.Unlock()
for _, ct := range contentTopics { for _, ct := range contentTopics {
s.ContentTopics[ct] = struct{}{} s.ContentFilter.ContentTopics[ct] = struct{}{}
} }
} }
@ -150,7 +172,7 @@ func (s *SubscriptionDetails) Remove(contentTopics ...string) {
defer s.Unlock() defer s.Unlock()
for _, ct := range contentTopics { for _, ct := range contentTopics {
delete(s.ContentTopics, ct) delete(s.ContentFilter.ContentTopics, ct)
} }
} }
@ -178,21 +200,16 @@ func (s *SubscriptionDetails) Clone() *SubscriptionDetails {
mapRef: s.mapRef, mapRef: s.mapRef,
Closed: false, Closed: false,
PeerID: s.PeerID, PeerID: s.PeerID,
PubsubTopic: s.PubsubTopic, ContentFilter: ContentFilter{s.ContentFilter.PubsubTopic, maps.Clone(s.ContentFilter.ContentTopics)},
ContentTopics: make(map[string]struct{}),
C: make(chan *protocol.Envelope), C: make(chan *protocol.Envelope),
} }
for k := range s.ContentTopics {
result.ContentTopics[k] = struct{}{}
}
return result return result
} }
func (sub *SubscriptionsMap) clear() { func (sub *SubscriptionsMap) clear() {
for _, peerSubscription := range sub.items { for _, peerSubscription := range sub.items {
for _, subscriptionSet := range peerSubscription.subscriptionsPerTopic { for _, subscriptionSet := range peerSubscription.subsPerPubsubTopic {
for _, subscription := range subscriptionSet { for _, subscription := range subscriptionSet {
subscription.closeC() subscription.closeC()
} }
@ -212,7 +229,7 @@ func (sub *SubscriptionsMap) Notify(peerID peer.ID, envelope *protocol.Envelope)
sub.RLock() sub.RLock()
defer sub.RUnlock() defer sub.RUnlock()
subscriptions, ok := sub.items[peerID].subscriptionsPerTopic[envelope.PubsubTopic()] subscriptions, ok := sub.items[peerID].subsPerPubsubTopic[envelope.PubsubTopic()]
if ok { if ok {
iterateSubscriptionSet(sub.logger, subscriptions, envelope) iterateSubscriptionSet(sub.logger, subscriptions, envelope)
} }
@ -224,7 +241,7 @@ func iterateSubscriptionSet(logger *zap.Logger, subscriptions SubscriptionSet, e
subscription.RLock() subscription.RLock()
defer subscription.RUnlock() defer subscription.RUnlock()
_, ok := subscription.ContentTopics[envelope.Message().ContentTopic] _, ok := subscription.ContentFilter.ContentTopics[envelope.Message().ContentTopic]
if !ok { // only send the msg to subscriptions that have matching contentTopic if !ok { // only send the msg to subscriptions that have matching contentTopic
return return
} }
@ -249,10 +266,10 @@ func (s *SubscriptionDetails) MarshalJSON() ([]byte, error) {
result := resultType{ result := resultType{
PeerID: s.PeerID.Pretty(), PeerID: s.PeerID.Pretty(),
PubsubTopic: s.PubsubTopic, PubsubTopic: s.ContentFilter.PubsubTopic,
} }
for c := range s.ContentTopics { for c := range s.ContentFilter.ContentTopics {
result.ContentTopics = append(result.ContentTopics, c) result.ContentTopics = append(result.ContentTopics, c)
} }

34
waku/v2/protocol/filter/subscriptions_map_test.go
View File

@ -15,23 +15,23 @@ import (
func TestSubscriptionMapAppend(t *testing.T) { func TestSubscriptionMapAppend(t *testing.T) {
fmap := NewSubscriptionMap(utils.Logger()) fmap := NewSubscriptionMap(utils.Logger())
peerID := createPeerID(t) peerID := createPeerID(t)
contentTopics := []string{"ct1", "ct2"} contentTopics := NewContentTopicSet("ct1", "ct2")
sub := fmap.NewSubscription(peerID, TOPIC, contentTopics) sub := fmap.NewSubscription(peerID, ContentFilter{PUBSUB_TOPIC, contentTopics})
_, found := sub.ContentTopics["ct1"] _, found := sub.ContentFilter.ContentTopics["ct1"]
require.True(t, found) require.True(t, found)
_, found = sub.ContentTopics["ct2"] _, found = sub.ContentFilter.ContentTopics["ct2"]
require.True(t, found) require.True(t, found)
require.False(t, sub.Closed) require.False(t, sub.Closed)
require.Equal(t, sub.PeerID, peerID) require.Equal(t, sub.PeerID, peerID)
require.Equal(t, sub.PubsubTopic, TOPIC) require.Equal(t, sub.ContentFilter.PubsubTopic, PUBSUB_TOPIC)
sub.Add("ct3") sub.Add("ct3")
_, found = sub.ContentTopics["ct3"] _, found = sub.ContentFilter.ContentTopics["ct3"]
require.True(t, found) require.True(t, found)
sub.Remove("ct3") sub.Remove("ct3")
_, found = sub.ContentTopics["ct3"] _, found = sub.ContentFilter.ContentTopics["ct3"]
require.False(t, found) require.False(t, found)
err := sub.Close() err := sub.Close()
@ -44,12 +44,12 @@ func TestSubscriptionMapAppend(t *testing.T) {
func TestSubscriptionClear(t *testing.T) { func TestSubscriptionClear(t *testing.T) {
fmap := NewSubscriptionMap(utils.Logger()) fmap := NewSubscriptionMap(utils.Logger())
contentTopics := []string{"ct1", "ct2"} contentTopics := NewContentTopicSet("ct1", "ct2")
var subscriptions = []*SubscriptionDetails{ var subscriptions = []*SubscriptionDetails{
fmap.NewSubscription(createPeerID(t), TOPIC+"1", contentTopics), fmap.NewSubscription(createPeerID(t), ContentFilter{PUBSUB_TOPIC + "1", contentTopics}),
fmap.NewSubscription(createPeerID(t), TOPIC+"2", contentTopics), fmap.NewSubscription(createPeerID(t), ContentFilter{PUBSUB_TOPIC + "2", contentTopics}),
fmap.NewSubscription(createPeerID(t), TOPIC+"3", contentTopics), fmap.NewSubscription(createPeerID(t), ContentFilter{PUBSUB_TOPIC + "3", contentTopics}),
} }
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
@ -84,9 +84,9 @@ func TestSubscriptionsNotify(t *testing.T) {
p1 := createPeerID(t) p1 := createPeerID(t)
p2 := createPeerID(t) p2 := createPeerID(t)
var subscriptions = []*SubscriptionDetails{ var subscriptions = []*SubscriptionDetails{
fmap.NewSubscription(p1, TOPIC+"1", []string{"ct1", "ct2"}), fmap.NewSubscription(p1, ContentFilter{PUBSUB_TOPIC + "1", NewContentTopicSet("ct1", "ct2")}),
fmap.NewSubscription(p2, TOPIC+"1", []string{"ct1"}), fmap.NewSubscription(p2, ContentFilter{PUBSUB_TOPIC + "1", NewContentTopicSet("ct1")}),
fmap.NewSubscription(p1, TOPIC+"2", []string{"ct1", "ct2"}), fmap.NewSubscription(p1, ContentFilter{PUBSUB_TOPIC + "2", NewContentTopicSet("ct1", "ct2")}),
} }
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@ -140,7 +140,7 @@ func TestSubscriptionsNotify(t *testing.T) {
go failOnReceive(ctx, 2) go failOnReceive(ctx, 2)
time.Sleep(200 * time.Millisecond) time.Sleep(200 * time.Millisecond)
envTopic1Ct1 := protocol.NewEnvelope(tests.CreateWakuMessage("ct1", 0), 0, TOPIC+"1") envTopic1Ct1 := protocol.NewEnvelope(tests.CreateWakuMessage("ct1", 0), 0, PUBSUB_TOPIC+"1")
wg.Add(1) wg.Add(1)
go func() { go func() {
defer wg.Done() defer wg.Done()
@ -164,7 +164,7 @@ func TestSubscriptionsNotify(t *testing.T) {
go failOnReceive(ctx, 2) go failOnReceive(ctx, 2)
time.Sleep(200 * time.Millisecond) time.Sleep(200 * time.Millisecond)
envTopic1Ct2 := protocol.NewEnvelope(tests.CreateWakuMessage("ct2", 0), 0, TOPIC+"1") envTopic1Ct2 := protocol.NewEnvelope(tests.CreateWakuMessage("ct2", 0), 0, PUBSUB_TOPIC+"1")
wg.Add(1) wg.Add(1)
go func() { go func() {
defer wg.Done() defer wg.Done()
@ -193,7 +193,7 @@ func TestSubscriptionsNotify(t *testing.T) {
go failOnReceive(ctx, 2) go failOnReceive(ctx, 2)
time.Sleep(200 * time.Millisecond) time.Sleep(200 * time.Millisecond)
envTopic1Ct1_2 := protocol.NewEnvelope(tests.CreateWakuMessage("ct1", 1), 1, TOPIC+"1") envTopic1Ct1_2 := protocol.NewEnvelope(tests.CreateWakuMessage("ct1", 1), 1, PUBSUB_TOPIC+"1")
wg.Add(1) wg.Add(1)
go func() { go func() {