feat: add eventbus subscriber name

gfanton · gfanton · commit 7ab1b8048fcb · 2023-05-04T09:13:24.000+02:00
Signed-off-by: gfanton &lt;8671905+gfanton@users.noreply.github.com&gt;
diff --git a/accesscontroller/orbitdb/accesscontroller_orbitdb.go b/accesscontroller/orbitdb/accesscontroller_orbitdb.go
@@ -16,6 +16,7 @@ import (
 
 	cid "github.com/ipfs/go-cid"
 	"github.com/libp2p/go-libp2p/core/event"
+	"github.com/libp2p/go-libp2p/p2p/host/eventbus"
 	"go.uber.org/zap"
 )
 
@@ -217,7 +218,7 @@ func (o *orbitDBAccessController) Load(ctx context.Context, address string) erro
 		new(stores.EventWrite),
 		new(stores.EventReady),
 		new(stores.EventReplicated),
-	})
+	}, eventbus.Name("odb/load"))
 	if err != nil {
 		return fmt.Errorf("unable subscribe to store events: %w", err)
 	}
diff --git a/baseorbitdb/orbitdb.go b/baseorbitdb/orbitdb.go
@@ -813,7 +813,8 @@ func (o *orbitDB) EventBus() event.Bus {
 }
 
 func (o *orbitDB) monitorDirectChannel(ctx context.Context, bus event.Bus) error {
-	sub, err := bus.Subscribe(new(iface.EventPubSubPayload), eventbus.BufSize(128))
+	sub, err := bus.Subscribe(new(iface.EventPubSubPayload),
+		eventbus.BufSize(128), eventbus.Name("odb/monitor-direct-channel"))
 	if err != nil {
 		return fmt.Errorf("unable to init pubsub subscriber: %w", err)
 	}
diff --git a/stores/basestore/base_store.go b/stores/basestore/base_store.go
@@ -268,7 +268,8 @@ func (b *BaseStore) InitBaseStore(ipfs coreapi.CoreAPI, identity *identityprovid
 
 	b.options = options
 
-	sub, err := b.replicator.EventBus().Subscribe(replicator.Events, eventbus.BufSize(128))
+	sub, err := b.replicator.EventBus().Subscribe(replicator.Events,
+		eventbus.BufSize(128), eventbus.Name("odb/base-store-main-loop"))
 	if err != nil {
 		return fmt.Errorf("unable to subscribe to replicator events: %w", err)
 	}
@@ -1008,7 +1009,7 @@ func (b *BaseStore) replicate() error {
 }
 
 func (b *BaseStore) storeListener(topic iface.PubSubTopic) error {
-	sub, err := b.EventBus().Subscribe(new(stores.EventWrite))
+	sub, err := b.EventBus().Subscribe(new(stores.EventWrite), eventbus.Name("odb/store-listener"))
 	if err != nil {
 		return fmt.Errorf("unable to init event bus: %w", err)
 	}

Original file line number	Diff line number	Diff line change
`@@ -16,6 +16,7 @@ import (`
`16`	`16`
`17`	`17`	`cid "github.com/ipfs/go-cid"`
`18`	`18`	`"github.com/libp2p/go-libp2p/core/event"`
	`19`	`+ "github.com/libp2p/go-libp2p/p2p/host/eventbus"`
`19`	`20`	`"go.uber.org/zap"`
`20`	`21`	`)`
`21`	`22`
`@@ -217,7 +218,7 @@ func (o *orbitDBAccessController) Load(ctx context.Context, address string) erro`
`217`	`218`	`new(stores.EventWrite),`
`218`	`219`	`new(stores.EventReady),`
`219`	`220`	`new(stores.EventReplicated),`
`220`		`- })`
	`221`	`+ }, eventbus.Name("odb/load"))`
`221`	`222`	`if err != nil {`
`222`	`223`	`return fmt.Errorf("unable subscribe to store events: %w", err)`
`223`	`224`	`}`
Original file line number	Diff line number	Diff line change
`@@ -813,7 +813,8 @@ func (o *orbitDB) EventBus() event.Bus {`
`813`	`813`	`}`
`814`	`814`
`815`	`815`	`func (o *orbitDB) monitorDirectChannel(ctx context.Context, bus event.Bus) error {`
`816`		`- sub, err := bus.Subscribe(new(iface.EventPubSubPayload), eventbus.BufSize(128))`
	`816`	`+ sub, err := bus.Subscribe(new(iface.EventPubSubPayload),`
	`817`	`+ eventbus.BufSize(128), eventbus.Name("odb/monitor-direct-channel"))`
`817`	`818`	`if err != nil {`
`818`	`819`	`return fmt.Errorf("unable to init pubsub subscriber: %w", err)`
`819`	`820`	`}`
Original file line number	Diff line number	Diff line change
`@@ -268,7 +268,8 @@ func (b BaseStore) InitBaseStore(ipfs coreapi.CoreAPI, identity identityprovid`
`268`	`268`
`269`	`269`	`b.options = options`
`270`	`270`
`271`		`- sub, err := b.replicator.EventBus().Subscribe(replicator.Events, eventbus.BufSize(128))`
	`271`	`+ sub, err := b.replicator.EventBus().Subscribe(replicator.Events,`
	`272`	`+ eventbus.BufSize(128), eventbus.Name("odb/base-store-main-loop"))`
`272`	`273`	`if err != nil {`
`273`	`274`	`return fmt.Errorf("unable to subscribe to replicator events: %w", err)`
`274`	`275`	`}`
`@@ -1008,7 +1009,7 @@ func (b *BaseStore) replicate() error {`
`1008`	`1009`	`}`
`1009`	`1010`
`1010`	`1011`	`func (b *BaseStore) storeListener(topic iface.PubSubTopic) error {`
`1011`		`- sub, err := b.EventBus().Subscribe(new(stores.EventWrite))`
	`1012`	`+ sub, err := b.EventBus().Subscribe(new(stores.EventWrite), eventbus.Name("odb/store-listener"))`
`1012`	`1013`	`if err != nil {`
`1013`	`1014`	`return fmt.Errorf("unable to init event bus: %w", err)`
`1014`	`1015`	`}`