feat: remove eventBus usage on persistSeqNo change

Author: erayarslan

couchbase/rollback_mitigation.go

@@ -12,8 +12,6 @@ import (
 
 	"golang.org/x/sync/errgroup"
 
-	"github.com/asaskevich/EventBus"
-
 	"github.com/couchbase/gocbcore/v10"
 
 	"github.com/Trendyol/go-dcp/wrapper"
@@ -22,7 +20,6 @@ import (
 
 	"github.com/Trendyol/go-dcp/config"
 
-	"github.com/Trendyol/go-dcp/helpers"
 	"github.com/Trendyol/go-dcp/logger"
 )
 
@@ -62,20 +59,20 @@ func (v *vbUUIDAndSeqNo) IsOutdated(last *gocbcore.ObserveVbResult) bool {
 }
 
 type rollbackMitigation struct {
-	bus                EventBus.Bus
-	client             Client
-	vbUUIDMap          *wrapper.ConcurrentSwissMap[uint16, gocbcore.VbUUID]
-	configSnapshot     *gocbcore.ConfigSnapshot
-	persistedSeqNos    *wrapper.ConcurrentSwissMap[uint16, []*vbUUIDAndSeqNo]
-	observeCount       *atomic.Uint32
-	config             *config.Dcp
-	observeTimer       *time.Ticker
-	observeCloseCh     chan struct{}
-	observeCloseDoneCh chan struct{}
-	vbIds              []uint16
-	activeGroupID      int
-	configWatchRunning bool
-	closed             bool
+	client                 Client
+	observeTimer           *time.Ticker
+	configSnapshot         *gocbcore.ConfigSnapshot
+	persistedSeqNos        *wrapper.ConcurrentSwissMap[uint16, []*vbUUIDAndSeqNo]
+	observeCount           *atomic.Uint32
+	config                 *config.Dcp
+	vbUUIDMap              *wrapper.ConcurrentSwissMap[uint16, gocbcore.VbUUID]
+	observeCloseCh         chan struct{}
+	observeCloseDoneCh     chan struct{}
+	persistSeqNoDispatcher models.PersistSeqNoDispatcher
+	vbIds                  []uint16
+	activeGroupID          int
+	configWatchRunning     bool
+	closed                 bool
 }
 
 func (r *rollbackMitigation) getRevEpochAndID(snapshot *gocbcore.ConfigSnapshot) (int64, int64) {
@@ -340,7 +337,7 @@ func (r *rollbackMitigation) observe(vbID uint16, replica int, groupID int, vbUU
 				replicas[replica].SetSeqNo(result.PersistSeqNo)
 				replicas[replica].SetVbUUID(result.VbUUID)
 
-				r.bus.Publish(helpers.PersistSeqNoChangedBusEventName, models.PersistSeqNo{
+				r.persistSeqNoDispatcher(&models.PersistSeqNo{
 					VbID:  vbID,
 					SeqNo: r.getMinSeqNo(vbID),
 				})
@@ -440,14 +437,18 @@ func (r *rollbackMitigation) Stop() {
 	logger.Log.Info("rollback mitigation stopped")
 }
 
-func NewRollbackMitigation(client Client, config *config.Dcp, vbIds []uint16, bus EventBus.Bus) RollbackMitigation {
+func NewRollbackMitigation(client Client,
+	config *config.Dcp,
+	vbIds []uint16,
+	persistSeqNoDispatcher models.PersistSeqNoDispatcher,
+) RollbackMitigation {
 	return &rollbackMitigation{
-		client:             client,
-		config:             config,
-		vbIds:              vbIds,
-		bus:                bus,
-		observeCount:       &atomic.Uint32{},
-		observeCloseCh:     make(chan struct{}, 1),
-		observeCloseDoneCh: make(chan struct{}, 1),
+		client:                 client,
+		config:                 config,
+		vbIds:                  vbIds,
+		observeCount:           &atomic.Uint32{},
+		observeCloseCh:         make(chan struct{}, 1),
+		observeCloseDoneCh:     make(chan struct{}, 1),
+		persistSeqNoDispatcher: persistSeqNoDispatcher,
 	}
 }

dcp.go

@@ -125,8 +125,7 @@ func (s *dcp) Start() {
 
 	s.stream = stream.NewStream(
 		s.client, s.metadata, s.config, s.version, s.bucketInfo, s.vBucketDiscovery,
-		s.consumer, collectionIDs, s.stopCh, s.bus, s.eventHandler,
-		tc,
+		s.consumer, collectionIDs, s.stopCh, s.eventHandler, tc,
 	)
 
 	if s.config.LeaderElection.Enabled {

helpers/constants.go

@@ -6,8 +6,7 @@ const (
 	Prefix    string = "_connector:" + Name + ":"
 	TxnPrefix string = "_txn:"
 
-	MembershipChangedBusEventName   string = "membershipChanged"
-	PersistSeqNoChangedBusEventName string = "persistSeqNoChanged"
+	MembershipChangedBusEventName string = "membershipChanged"
 
 	JSONFlags   uint32 = 50333696
 	MaxIntValue uint64 = 0xffffffffffffffff

models/listeners.go

@@ -22,9 +22,10 @@ type DcpStreamEndContext struct {
 }
 
 type (
-	Listener      func(*ListenerContext)
-	ListenerCh    chan ListenerArgs
-	ListenerEndCh chan DcpStreamEndContext
+	Listener               func(*ListenerContext)
+	ListenerCh             chan ListenerArgs
+	ListenerEndCh          chan DcpStreamEndContext
+	PersistSeqNoDispatcher func(persistSeqNo *PersistSeqNo)
 )
 
 type Consumer interface {

stream/stream.go

@@ -13,8 +13,6 @@ import (
 
 	"github.com/Trendyol/go-dcp/membership"
 
-	"github.com/asaskevich/EventBus"
-
 	"github.com/couchbase/gocbcore/v10"
 
 	"github.com/Trendyol/go-dcp/wrapper"
@@ -56,7 +54,6 @@ type stream struct {
 	checkpoint                   Checkpoint
 	rollbackMitigation           couchbase.RollbackMitigation
 	vBucketDiscovery             VBucketDiscovery
-	bus                          EventBus.Bus
 	eventHandler                 models.EventHandler
 	config                       *config.Dcp
 	metric                       *Metric
@@ -239,7 +236,7 @@ func (s *stream) Open() {
 			logger.Log.Info("rollback mitigation is disabled for ephemeral bucket")
 			s.config.RollbackMitigation.Disabled = true
 		} else {
-			s.rollbackMitigation = couchbase.NewRollbackMitigation(s.client, s.config, vbIDs, s.bus)
+			s.rollbackMitigation = couchbase.NewRollbackMitigation(s.client, s.config, vbIDs, s.dispatchPersistSeqNo)
 			s.rollbackMitigation.Start()
 		}
 	}
@@ -329,7 +326,7 @@ func (s *stream) Save() {
 	s.checkpoint.Save()
 }
 
-func (s *stream) dispatchPersistSeqNo(persistSeqNo models.PersistSeqNo) {
+func (s *stream) dispatchPersistSeqNo(persistSeqNo *models.PersistSeqNo) {
 	if s.observers != nil {
 		if observer, ok := s.observers.Load(persistSeqNo.VbID); ok {
 			observer.SetPersistSeqNo(persistSeqNo.SeqNo)
@@ -482,7 +479,6 @@ func NewStream(client couchbase.Client,
 	consumer models.Consumer,
 	collectionIDs map[uint32]string,
 	stopCh chan struct{},
-	bus EventBus.Bus,
 	eventHandler models.EventHandler,
 	tc *tracing.TracerComponent,
 ) Stream {
@@ -497,7 +493,6 @@ func NewStream(client couchbase.Client,
 		finishStreamWithCloseCh:    make(chan struct{}, 1),
 		finishStreamWithEndEventCh: make(chan struct{}, 1),
 		stopCh:                     stopCh,
-		bus:                        bus,
 		eventHandler:               eventHandler,
 		metric:                     &Metric{},
 		tracerComponent:            tc,
@@ -510,12 +505,5 @@ func NewStream(client couchbase.Client,
 		}
 	}
 
-	// there is a single topic for persistSeqNo events related to all vBuckets
-	// the listener is responsible for dispatching the event to the correct observer
-	if err := bus.Subscribe(helpers.PersistSeqNoChangedBusEventName, stream.dispatchPersistSeqNo); err != nil {
-		logger.Log.Error("cannot subscribe to persistSeqNoChanged event, err: %v", err)
-		panic(err)
-	}
-
 	return stream
 }