feat: qol improvements

api/api.go

@@ -38,6 +38,7 @@ type api struct {
 	app              *fiber.App
 	config           *dcp.Dcp
 	registerer       *metric.Registerer
+	membershipInfo   *membership.Model
 	bus              EventBus.Bus
 }
 
@@ -96,10 +97,18 @@ func (s *api) info(c *fiber.Ctx) error {
 	}
 	logger.Log.Debug("new info arrived for member: %v/%v", req.MemberNumber, req.TotalMembers)
 
-	s.bus.Publish(helpers.MembershipChangedBusEventName, &membership.Model{
+	var newInfo = &membership.Model{
 		MemberNumber: req.MemberNumber,
 		TotalMembers: req.TotalMembers,
-	})
+	}
+
+	if newInfo.IsChanged(s.membershipInfo) {
+		s.membershipInfo = newInfo
+
+		logger.Log.Debug("new info arrived for member: %v/%v", newInfo.MemberNumber, newInfo.TotalMembers)
+
+		s.bus.Publish(helpers.MembershipChangedBusEventName, newInfo)
+	}
 
 	return c.SendString("OK")
 }

couchbase/membership.go

@@ -266,10 +266,16 @@ func (h *cbMembership) rebalance(instances []Instance) {
 		logger.Log.Error("error while rebalance, self = %v, err: %v", string(h.id), err)
 		panic(err)
 	} else {
-		h.bus.Publish(helpers.MembershipChangedBusEventName, &membership.Model{
+		var newInfo = &membership.Model{
 			MemberNumber: selfOrder,
 			TotalMembers: len(instances),
-		})
+		}
+
+		if newInfo.IsChanged(h.info) {
+			logger.Log.Debug("new info arrived for member: %v/%v", newInfo.MemberNumber, newInfo.TotalMembers)
+
+			h.bus.Publish(helpers.MembershipChangedBusEventName, newInfo)
+		}
 
 		h.lastActiveInstances = instances
 	}

dcp.go

@@ -84,14 +84,8 @@ func (s *dcp) SetEventHandler(eventHandler models.EventHandler) {
 	s.eventHandler = eventHandler
 }
 
-func (s *dcp) membershipChangedListener(m *membership.Model) {
-	currentInfo := &membership.Model{
-		MemberNumber: s.vBucketDiscovery.GetMetric().MemberNumber,
-		TotalMembers: s.vBucketDiscovery.GetMetric().TotalMembers,
-	}
-	if s.stream.IsOpen() && currentInfo.IsChanged(m) {
-		s.stream.Rebalance()
-	}
+func (s *dcp) membershipChangedListener(_ *membership.Model) {
+	s.stream.Rebalance()
 }
 
 //nolint:funlen

example/config_dynamic.yml

@@ -1,7 +1,7 @@
 hosts:
   - localhost:8091
 username: user
-password: password
+password: 123456
 bucketName: dcp-test
 dcp:
   group:

example/config_k8s_default.yml

@@ -1,7 +1,7 @@
 hosts:
   - couchbase.couchbase.svc.cluster.local:8091
 username: user
-password: password
+password: 123456
 bucketName: dcp-test
 dcp:
   group:

example/config_k8s_leader_election.yml

@@ -1,7 +1,7 @@
 hosts:
   - couchbase.couchbase.svc.cluster.local:8091
 username: user
-password: password
+password: 123456
 bucketName: dcp-test
 dcp:
   group:

membership/dynamic_membership.go

@@ -16,6 +16,7 @@ func (d *dynamicMembership) GetInfo() *Model {
 	if d.info != nil {
 		return d.info
 	}
+	logger.Log.Info("dynamic membership waiting first request")
 	return <-d.infoChan
 }
 

membership/membership.go

@@ -1,5 +1,7 @@
 package membership
 
+import "github.com/Trendyol/go-dcp/logger"
+
 type Membership interface {
 	GetInfo() *Model
 	Close()
@@ -23,5 +25,10 @@ func (s *Model) IsChanged(other *Model) bool {
 		return true
 	}
 
-	return s.MemberNumber != other.MemberNumber || s.TotalMembers != other.TotalMembers
+	var res = s.MemberNumber != other.MemberNumber || s.TotalMembers != other.TotalMembers
+	if !res {
+		logger.Log.Info("membership info not changed")
+	}
+
+	return res
 }