feat: improve health check

couchbase/healthcheck.go

@@ -0,0 +1,44 @@
+package couchbase
+
+import (
+	"time"
+
+	"github.com/Trendyol/go-dcp/config"
+	"github.com/Trendyol/go-dcp/logger"
+)
+
+type HealthCheck interface {
+	Start(ch chan struct{})
+	Stop()
+}
+
+type healthCheck struct {
+	ticker *time.Ticker
+	config *config.HealthCheck
+	client Client
+}
+
+func (h *healthCheck) Start(ch chan struct{}) {
+	h.ticker = time.NewTicker(h.config.Interval)
+
+	go func() {
+		for range h.ticker.C {
+			if _, err := h.client.Ping(); err != nil {
+				logger.Log.Error("health check failed: %v", err)
+				h.ticker.Stop()
+				ch <- struct{}{}
+				break
+			}
+		}
+	}()
+}
+
+func (h *healthCheck) Stop() {
+	h.ticker.Stop()
+}
+
+func NewHealthCheck(client Client) HealthCheck {
+	return &healthCheck{
+		client: client,
+	}
+}

dcp.go

@@ -7,7 +7,6 @@ import (
 	"os/signal"
 	"reflect"
 	"syscall"
-	"time"
 
 	"github.com/asaskevich/EventBus"
 
@@ -57,7 +56,7 @@ type dcp struct {
 	apiShutdown       chan struct{}
 	healCheckFailedCh chan struct{}
 	config            *config.Dcp
-	healthCheckTicker *time.Ticker
+	healthCheck       couchbase.HealthCheck
 	listener          models.Listener
 	readyCh           chan struct{}
 	cancelCh          chan os.Signal
@@ -66,25 +65,6 @@ type dcp struct {
 	closeWithCancel   bool
 }
 
-func (s *dcp) startHealthCheck() {
-	s.healthCheckTicker = time.NewTicker(s.config.HealthCheck.Interval)
-
-	go func() {
-		for range s.healthCheckTicker.C {
-			if _, err := s.client.Ping(); err != nil {
-				logger.Log.Error("health check failed: %v", err)
-				s.healthCheckTicker.Stop()
-				s.healCheckFailedCh <- struct{}{}
-				break
-			}
-		}
-	}()
-}
-
-func (s *dcp) stopHealthCheck() {
-	s.healthCheckTicker.Stop()
-}
-
 func (s *dcp) SetMetadata(metadata metadata.Metadata) {
 	s.metadata = metadata
 }
@@ -162,7 +142,8 @@ func (s *dcp) Start() {
 	signal.Notify(s.cancelCh, syscall.SIGTERM, syscall.SIGINT, syscall.SIGABRT, syscall.SIGQUIT)
 
 	if !s.config.HealthCheck.Disabled {
-		s.startHealthCheck()
+		s.healthCheck = couchbase.NewHealthCheck(s.client)
+		s.healthCheck.Start(s.healCheckFailedCh)
 	}
 
 	logger.Log.Info("dcp stream started")
@@ -183,7 +164,7 @@ func (s *dcp) WaitUntilReady() chan struct{} {
 
 func (s *dcp) Close() {
 	if !s.config.HealthCheck.Disabled {
-		s.stopHealthCheck()
+		s.healthCheck.Stop()
 	}
 	s.vBucketDiscovery.Close()