Refactor poller into a service

Author: eweziyi

plugins/turing/runner/experiment_runner.go

@@ -257,6 +257,9 @@ func (er *experimentRunner) startBackgroundServices(
 			}
 		}()
 	}
+	if er.appContext.PollerService != nil {
+		er.appContext.PollerService.Start()
+	}
 }
 
 func (er *experimentRunner) getRequestParams(

treatment-service/appcontext/appcontext.go

@@ -24,6 +24,7 @@ type AppContext struct {
 
 	AssignedTreatmentLogger *monitoring.AssignedTreatmentLogger
 	LocalStorage            *models.LocalStorage
+	PollerService           *services.PollerService
 }
 
 func NewAppContext(cfg *config.Config) (*AppContext, error) {
@@ -122,6 +123,11 @@ func NewAppContext(cfg *config.Config) (*AppContext, error) {
 		return nil, err
 	}
 
+	var pollerService *services.PollerService
+	if cfg.ManagementServicePollerConfig.Enabled {
+		pollerService = services.NewPollerService(cfg.ManagementServicePollerConfig, localStorage)
+	}
+
 	appContext := &AppContext{
 		ExperimentService:       experimentSvc,
 		MetricService:           metricService,
@@ -131,6 +137,7 @@ func NewAppContext(cfg *config.Config) (*AppContext, error) {
 		AssignedTreatmentLogger: logger,
 		MessageQueueService:     messageQueueService,
 		LocalStorage:            localStorage,
+		PollerService:           pollerService,
 	}
 
 	return appContext, nil

treatment-service/server/server.go

@@ -33,8 +33,6 @@ type Server struct {
 	subscribe bool
 	// cleanup captures all the actions to be executed on server shut down
 	cleanup []func()
-	// poller captures the poller instance
-	poller *Poller
 }
 
 // NewServer creates and configures an APIServer serving all application routes.
@@ -108,11 +106,6 @@ func NewServer(configFiles []string) (*Server, error) {
 		subscribe = true
 	}
 
-	var poller *Poller
-	if cfg.ManagementServicePollerConfig.Enabled {
-		poller = NewPoller(cfg.ManagementServicePollerConfig, appCtx.LocalStorage)
-	}
-
 	srv := http.Server{
 		Addr:    cfg.ListenAddress(),
 		Handler: mux,
@@ -123,7 +116,6 @@ func NewServer(configFiles []string) (*Server, error) {
 		appContext: appCtx,
 		subscribe:  subscribe,
 		cleanup:    cleanup,
-		poller:     poller,
 	}, nil
 }
 
@@ -141,11 +133,6 @@ func (srv *Server) Start() {
 	}()
 	log.Printf("Listening on %s\n", srv.Addr)
 
-	if srv.poller != nil {
-		log.Println("Starting poller...")
-		srv.poller.Start()
-	}
-
 	stop := make(chan os.Signal, 1)
 	signal.Notify(stop, os.Interrupt)
 
@@ -193,5 +180,9 @@ func (srv *Server) startBackgroundService(errChannel chan error) context.CancelF
 		}
 	}()
 
+	if srv.appContext.PollerService != nil {
+		srv.appContext.PollerService.Start()
+	}
+
 	return cancel
 }

treatment-service/server/poller.go renamed to treatment-service/services/poller_service.go

@@ -1,4 +1,4 @@
-package server
+package services
 
 import (
 	"log"
@@ -8,24 +8,25 @@ import (
 	"github.com/caraml-dev/xp/treatment-service/models"
 )
 
-type Poller struct {
+type PollerService struct {
 	pollerConfig config.ManagementServicePollerConfig
 	localStorage *models.LocalStorage
 	stopChannel  chan struct{}
 }
 
-// NewPoller creates a new Poller instance with the given configuration and local storage.
+// NewPollerService creates a new PollerService instance with the given configuration and local storage.
 // pollerConfig: configuration for the poller
 // localStorage: local storage to be used by the poller
-func NewPoller(pollerConfig config.ManagementServicePollerConfig, localStorage *models.LocalStorage) *Poller {
-	return &Poller{
+func NewPollerService(pollerConfig config.ManagementServicePollerConfig, localStorage *models.LocalStorage) *PollerService {
+	return &PollerService{
 		pollerConfig: pollerConfig,
 		localStorage: localStorage,
 		stopChannel:  make(chan struct{}),
 	}
 }
 
-func (p *Poller) Start() {
+func (p *PollerService) Start() {
+	log.Println("Starting management service poller service...")
 	pollInterval := time.Duration(p.pollerConfig.PollIntervalSeconds) * time.Second
 	ticker := time.NewTicker(pollInterval)
 	go func() {
@@ -46,11 +47,11 @@ func (p *Poller) Start() {
 	}()
 }
 
-func (p *Poller) Stop() {
+func (p *PollerService) Stop() {
 	close(p.stopChannel)
 }
 
-func (p *Poller) Refresh() error {
+func (p *PollerService) Refresh() error {
 	err := p.localStorage.Init()
 	return err
 }