MM-28859 Add feature flag managment system using split.io and remove viper. (#15954)

* Add feature flag managment system using split.io and remove viper. * Fixing tests. * Attempt to fix postgres tests. * Fix watch filepath for advanced logging. * Review fixes. * Some error wrapping. * Remove unessisary store interface. * Desanitize SplitKey * Simplify. * Review feedback. * Rename split mlog adatper to split logger. * fsInner * Style. * Restore oldcfg test. * Downgrading non-actionable feature flag errors to warnings. Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
2020-10-29 15:54:39 -07:00
--- a/vendor/github.com/splitio/go-split-commons/v2/push/dtos.go
+++ b/vendor/github.com/splitio/go-split-commons/v2/push/dtos.go
@@ -0,0 +1,27 @@
+package push
+
+// IncomingEvent struct to process every kind of notification that comes from streaming
+type IncomingEvent struct {
+	id         *string
+	timestamp  *int64
+	encoding   *string
+	data       *string
+	name       *string
+	clientID   *string
+	event      string
+	channel    *string
+	message    *string
+	code       *int
+	statusCode *int
+	href       *string
+}
+
+// Metrics dto
+type Metrics struct {
+	Publishers int `json:"publishers"`
+}
+
+// Occupancy dto
+type Occupancy struct {
+	Data Metrics `json:"metrics"`
+}
--- a/vendor/github.com/splitio/go-split-commons/v2/push/eventhandler.go
+++ b/vendor/github.com/splitio/go-split-commons/v2/push/eventhandler.go
@@ -0,0 +1,89 @@
+package push
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/splitio/go-split-commons/v2/dtos"
+	"github.com/splitio/go-toolkit/v3/logging"
+)
+
+// EventHandler struct
+type EventHandler struct {
+	keeper    *Keeper
+	parser    *NotificationParser
+	processor *Processor
+	logger    logging.LoggerInterface
+}
+
+// NewEventHandler builds new EventHandler
+func NewEventHandler(keeper *Keeper, parser *NotificationParser, processor *Processor, logger logging.LoggerInterface) *EventHandler {
+	return &EventHandler{
+		keeper:    keeper,
+		parser:    parser,
+		processor: processor,
+		logger:    logger,
+	}
+}
+
+func (e *EventHandler) wrapOccupancy(incomingEvent IncomingEvent) *Occupancy {
+	if incomingEvent.data == nil {
+		return nil
+	}
+
+	var occupancy *Occupancy
+	err := json.Unmarshal([]byte(*incomingEvent.data), &occupancy)
+	if err != nil {
+		return nil
+	}
+
+	return occupancy
+}
+
+func (e *EventHandler) wrapUpdateEvent(incomingEvent IncomingEvent) *dtos.IncomingNotification {
+	if incomingEvent.data == nil {
+		return nil
+	}
+	var incomingNotification *dtos.IncomingNotification
+	err := json.Unmarshal([]byte(*incomingEvent.data), &incomingNotification)
+	if err != nil {
+		e.logger.Error("cannot parse data as IncomingNotification type")
+		return nil
+	}
+	incomingNotification.Channel = *incomingEvent.channel
+	return incomingNotification
+}
+
+// HandleIncomingMessage handles incoming message from streaming
+func (e *EventHandler) HandleIncomingMessage(event map[string]interface{}) {
+	incomingEvent := e.parser.Parse(event)
+	switch incomingEvent.event {
+	case update:
+		e.logger.Debug("Update event received")
+		incomingNotification := e.wrapUpdateEvent(incomingEvent)
+		if incomingNotification == nil {
+			e.logger.Debug("Skipping incoming notification...")
+			return
+		}
+		e.logger.Debug("Incoming Notification:", incomingNotification)
+		err := e.processor.Process(*incomingNotification)
+		if err != nil {
+			e.logger.Debug("Could not process notification", err.Error())
+			return
+		}
+	case occupancy:
+		e.logger.Debug("Presence event received")
+		occupancy := e.wrapOccupancy(incomingEvent)
+		if occupancy == nil || incomingEvent.channel == nil {
+			e.logger.Debug("Skipping occupancy...")
+			return
+		}
+		e.keeper.UpdateManagers(*incomingEvent.channel, occupancy.Data.Publishers)
+		return
+	case errorType: // TODO: Update this when logic is fully defined
+		e.logger.Error(fmt.Sprintf("Error received: %+v", incomingEvent))
+	default:
+		e.logger.Debug(fmt.Sprintf("Unexpected incomingEvent: %+v", incomingEvent))
+		e.logger.Error("Unexpected type of event received")
+	}
+}
--- a/vendor/github.com/splitio/go-split-commons/v2/push/interface.go
+++ b/vendor/github.com/splitio/go-split-commons/v2/push/interface.go
@@ -0,0 +1,10 @@
+package push
+
+// Manager interface for Push Manager
+type Manager interface {
+	Start()
+	Stop()
+	StartWorkers()
+	StopWorkers()
+	IsRunning() bool
+}
--- a/vendor/github.com/splitio/go-split-commons/v2/push/keeper.go
+++ b/vendor/github.com/splitio/go-split-commons/v2/push/keeper.go
@@ -0,0 +1,98 @@
+package push
+
+import (
+	"strings"
+	"sync"
+)
+
+const (
+	// PublisherNotPresent there are no publishers sending data
+	PublisherNotPresent = iota
+	// PublisherAvailable there are publishers running
+	PublisherAvailable
+)
+
+const (
+	prefix = "[?occupancy=metrics.publishers]"
+)
+
+// last struct for storing the last notification
+type last struct {
+	manager   string
+	timestamp int64
+	mutex     *sync.RWMutex
+}
+
+// Keeper struct
+type Keeper struct {
+	managers     map[string]int
+	activeRegion string
+	last         last
+	publishers   chan<- int
+	mutex        *sync.RWMutex
+}
+
+// NewKeeper creates new keeper
+func NewKeeper(publishers chan int) *Keeper {
+	last := last{
+		mutex: &sync.RWMutex{},
+	}
+	return &Keeper{
+		managers:     make(map[string]int),
+		activeRegion: "us-east-1",
+		mutex:        &sync.RWMutex{},
+		publishers:   publishers,
+		last:         last,
+	}
+}
+
+func (k *Keeper) cleanManagerPrefix(manager string) string {
+	return strings.Replace(manager, prefix, "", -1)
+}
+
+// Publishers returns the quantity of publishers for a particular manager
+func (k *Keeper) Publishers(manager string) *int {
+	k.mutex.RLock()
+	defer k.mutex.RUnlock()
+	publisher, ok := k.managers[manager]
+	if ok {
+		return &publisher
+	}
+	return nil
+}
+
+// UpdateManagers updates current manager count
+func (k *Keeper) UpdateManagers(manager string, publishers int) {
+	parsedManager := k.cleanManagerPrefix(manager)
+	k.mutex.Lock()
+	defer k.mutex.Unlock()
+	k.managers[parsedManager] = publishers
+
+	isAvailable := false
+	for _, publishers := range k.managers {
+		if publishers > 0 {
+			isAvailable = true
+			break
+		}
+	}
+	if !isAvailable {
+		k.publishers <- PublisherNotPresent
+		return
+	}
+	k.publishers <- PublisherAvailable
+}
+
+// LastNotification return the latest notification saved
+func (k *Keeper) LastNotification() (string, int64) {
+	k.last.mutex.RLock()
+	defer k.last.mutex.RUnlock()
+	return k.last.manager, k.last.timestamp
+}
+
+// UpdateLastNotification updates last message received
+func (k *Keeper) UpdateLastNotification(manager string, timestamp int64) {
+	k.last.mutex.Lock()
+	defer k.last.mutex.Unlock()
+	k.last.manager = k.cleanManagerPrefix(manager)
+	k.last.timestamp = timestamp
+}
--- a/vendor/github.com/splitio/go-split-commons/v2/push/parser.go
+++ b/vendor/github.com/splitio/go-split-commons/v2/push/parser.go
@@ -0,0 +1,64 @@
+package push
+
+import (
+	"github.com/splitio/go-toolkit/v3/common"
+	"github.com/splitio/go-toolkit/v3/logging"
+)
+
+const (
+	update    = "update"
+	errorType = "error"
+	occupancy = "[meta]occupancy"
+)
+
+// NotificationParser struct
+type NotificationParser struct {
+	logger logging.LoggerInterface
+}
+
+// NewNotificationParser creates notifcation parser
+func NewNotificationParser(logger logging.LoggerInterface) *NotificationParser {
+	return &NotificationParser{
+		logger: logger,
+	}
+}
+
+// Parse parses incoming event from streaming
+func (n *NotificationParser) Parse(event map[string]interface{}) IncomingEvent {
+	incomingEvent := IncomingEvent{
+		id:       common.AsStringOrNil(event["id"]),
+		encoding: common.AsStringOrNil(event["encoding"]),
+		data:     common.AsStringOrNil(event["data"]),
+		name:     common.AsStringOrNil(event["name"]),
+		clientID: common.AsStringOrNil(event["clientId"]),
+		channel:  common.AsStringOrNil(event["channel"]),
+		message:  common.AsStringOrNil(event["message"]),
+		href:     common.AsStringOrNil(event["href"]),
+	}
+
+	timestamp := common.AsFloat64OrNil(event["timestamp"])
+	if timestamp != nil {
+		incomingEvent.timestamp = common.Int64Ref(int64(*timestamp))
+	}
+	code := common.AsFloat64OrNil(event["code"])
+	if code != nil {
+		incomingEvent.code = common.IntRef(int(*code))
+	}
+	statusCode := common.AsFloat64OrNil(event["statusCode"])
+	if statusCode != nil {
+		incomingEvent.statusCode = common.IntRef(int(*statusCode))
+	}
+
+	if incomingEvent.code != nil && incomingEvent.statusCode != nil {
+		incomingEvent.event = errorType
+		return incomingEvent
+	}
+
+	if incomingEvent.name != nil && *incomingEvent.name == occupancy {
+		incomingEvent.event = occupancy
+		return incomingEvent
+	}
+
+	incomingEvent.event = update
+	return incomingEvent
+}
--- a/vendor/github.com/splitio/go-split-commons/v2/push/processor.go
+++ b/vendor/github.com/splitio/go-split-commons/v2/push/processor.go
@@ -0,0 +1,112 @@
+package push
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/splitio/go-split-commons/v2/dtos"
+	"github.com/splitio/go-split-commons/v2/storage"
+	"github.com/splitio/go-toolkit/v3/logging"
+)
+
+const (
+	segmentQueueCheck     = 5000
+	splitQueueCheck       = 5000
+	streamingPausedType   = "STREAMING_PAUSED"
+	streamingResumedType  = "STREAMING_RESUMED"
+	streamingDisabledType = "STREAMING_DISABLED"
+)
+
+const (
+	// StreamingPaused The SDK should stop processing incoming UPDATE-type events
+	streamingPaused = iota
+	// StreamingResumed The SDK should resume processing UPDATE-type events (if not already)
+	streamingResumed
+	// StreamingDisabled The SDK should disable streaming completely and don’t try to reconnect until the SDK is re-instantiated
+	streamingDisabled
+)
+
+// Processor struct for notification processor
+type Processor struct {
+	segmentQueue  chan dtos.SegmentChangeNotification
+	splitQueue    chan dtos.SplitChangeNotification
+	splitStorage  storage.SplitStorageProducer
+	controlStatus chan<- int
+	logger        logging.LoggerInterface
+}
+
+// NewProcessor creates new processor
+func NewProcessor(segmentQueue chan dtos.SegmentChangeNotification, splitQueue chan dtos.SplitChangeNotification, splitStorage storage.SplitStorageProducer, logger logging.LoggerInterface, controlStatus chan int) (*Processor, error) {
+	if cap(segmentQueue) < segmentQueueCheck {
+		return nil, errors.New("Small size of segmentQueue")
+	}
+	if cap(splitQueue) < splitQueueCheck {
+		return nil, errors.New("Small size of splitQueue")
+	}
+	if cap(controlStatus) < 1 {
+		return nil, errors.New("Small size for control chan")
+	}
+
+	return &Processor{
+		segmentQueue:  segmentQueue,
+		splitQueue:    splitQueue,
+		splitStorage:  splitStorage,
+		controlStatus: controlStatus,
+		logger:        logger,
+	}, nil
+}
+
+// Process takes an incoming notification and generates appropriate notifications for it.
+func (p *Processor) Process(i dtos.IncomingNotification) error {
+	switch i.Type {
+	case dtos.SplitUpdate:
+		if i.ChangeNumber == nil {
+			return errors.New("ChangeNumber could not be nil, discarded")
+		}
+		splitUpdate := dtos.NewSplitChangeNotification(i.Channel, *i.ChangeNumber)
+		p.splitQueue <- splitUpdate
+	case dtos.SegmentUpdate:
+		if i.ChangeNumber == nil {
+			return errors.New("ChangeNumber could not be nil, discarded")
+		}
+		if i.SegmentName == nil {
+			return errors.New("SegmentName could not be nil, discarded")
+		}
+		segmentUpdate := dtos.NewSegmentChangeNotification(i.Channel, *i.ChangeNumber, *i.SegmentName)
+		p.segmentQueue <- segmentUpdate
+	case dtos.SplitKill:
+		if i.ChangeNumber == nil {
+			return errors.New("ChangeNumber could not be nil, discarded")
+		}
+		if i.SplitName == nil {
+			return errors.New("SplitName could not be nil, discarded")
+		}
+		if i.DefaultTreatment == nil {
+			return errors.New("DefaultTreatment could not be nil, discarded")
+		}
+		splitUpdate := dtos.NewSplitChangeNotification(i.Channel, *i.ChangeNumber)
+		p.splitStorage.KillLocally(*i.SplitName, *i.DefaultTreatment, *i.ChangeNumber)
+		p.splitQueue <- splitUpdate
+	case dtos.Control:
+		if i.ControlType == nil {
+			return errors.New("ControlType could not be nil, discarded")
+		}
+		control := dtos.NewControlNotification(i.Channel, *i.ControlType)
+		switch control.ControlType {
+		case streamingDisabledType:
+			p.logger.Debug("Received notification for disabling streaming")
+			p.controlStatus <- streamingDisabled
+		case streamingPausedType:
+			p.logger.Debug("Received notification for pausing streaming")
+			p.controlStatus <- streamingPaused
+		case streamingResumedType:
+			p.logger.Debug("Received notification for resuming streaming")
+			p.controlStatus <- streamingResumed
+		default:
+			p.logger.Debug(fmt.Sprintf("%s Unexpected type of Control Notification", control.ControlType))
+		}
+	default:
+		return fmt.Errorf("Unknown IncomingNotification type: %T", i)
+	}
+	return nil
+}
--- a/vendor/github.com/splitio/go-split-commons/v2/push/pushmanager.go
+++ b/vendor/github.com/splitio/go-split-commons/v2/push/pushmanager.go
@@ -0,0 +1,370 @@
+package push
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"sync/atomic"
+	"time"
+
+	"github.com/splitio/go-split-commons/v2/conf"
+	"github.com/splitio/go-split-commons/v2/dtos"
+	"github.com/splitio/go-split-commons/v2/service"
+	"github.com/splitio/go-split-commons/v2/service/api/sse"
+	"github.com/splitio/go-split-commons/v2/storage"
+	"github.com/splitio/go-toolkit/v3/common"
+	"github.com/splitio/go-toolkit/v3/logging"
+	sseStatus "github.com/splitio/go-toolkit/v3/sse"
+)
+
+const (
+	resetTimer = 120
+	maxPeriod  = 30 * time.Minute
+)
+
+const (
+	// Ready represents ready
+	Ready = iota
+	// PushIsDown there are no publishers for streaming
+	PushIsDown
+	// PushIsUp there are publishers presents
+	PushIsUp
+	// BackoffAuth backoff is running for authentication
+	BackoffAuth
+	// BackoffSSE backoff is running for connecting to stream
+	BackoffSSE
+	// TokenExpiration flag to restart push services
+	TokenExpiration
+	// StreamingPaused flag for pausing streaming
+	StreamingPaused
+	// StreamingResumed flag for resuming streaming
+	StreamingResumed
+	// StreamingDisabled flag for disabling streaming
+	StreamingDisabled
+	// Reconnect flag to reconnect
+	Reconnect
+	// NonRetriableError represents an error that will force switching to polling
+	NonRetriableError
+)
+
+// PushManager struct for managing push services
+type PushManager struct {
+	authClient             service.AuthClient
+	sseClient              *sse.StreamingClient
+	segmentWorker          *SegmentUpdateWorker
+	splitWorker            *SplitUpdateWorker
+	eventHandler           *EventHandler
+	managerStatus          chan<- int
+	streamingStatus        chan int
+	publishers             chan int
+	logger                 logging.LoggerInterface
+	cancelAuthBackoff      chan struct{}
+	cancelSSEBackoff       chan struct{}
+	cancelTokenExpiration  chan struct{}
+	cancelStreamingWatcher chan struct{}
+	control                chan int
+	status                 atomic.Value
+}
+
+// NewPushManager creates new PushManager
+func NewPushManager(
+	logger logging.LoggerInterface,
+	synchronizeSegmentHandler func(segmentName string, till *int64) error,
+	synchronizeSplitsHandler func(till *int64) error,
+	splitStorage storage.SplitStorage,
+	config *conf.AdvancedConfig,
+	managerStatus chan int,
+	authClient service.AuthClient,
+) (Manager, error) {
+	splitQueue := make(chan dtos.SplitChangeNotification, config.SplitUpdateQueueSize)
+	segmentQueue := make(chan dtos.SegmentChangeNotification, config.SegmentUpdateQueueSize)
+	control := make(chan int, 1)
+	processor, err := NewProcessor(segmentQueue, splitQueue, splitStorage, logger, control)
+	if err != nil {
+		return nil, err
+	}
+	parser := NewNotificationParser(logger)
+	if parser == nil {
+		return nil, errors.New("Could not instantiate NotificationParser")
+	}
+	publishers := make(chan int, 1000)
+	keeper := NewKeeper(publishers)
+	if keeper == nil {
+		return nil, errors.New("Could not instantiate Keeper")
+	}
+	eventHandler := NewEventHandler(keeper, parser, processor, logger)
+	segmentWorker, err := NewSegmentUpdateWorker(segmentQueue, synchronizeSegmentHandler, logger)
+	if err != nil {
+		return nil, err
+	}
+	splitWorker, err := NewSplitUpdateWorker(splitQueue, synchronizeSplitsHandler, logger)
+	if err != nil {
+		return nil, err
+	}
+
+	streamingStatus := make(chan int, 1000)
+	status := atomic.Value{}
+	status.Store(Ready)
+	return &PushManager{
+		authClient:             authClient,
+		sseClient:              sse.NewStreamingClient(config, streamingStatus, logger),
+		segmentWorker:          segmentWorker,
+		splitWorker:            splitWorker,
+		managerStatus:          managerStatus,
+		streamingStatus:        streamingStatus,
+		eventHandler:           eventHandler,
+		publishers:             publishers,
+		logger:                 logger,
+		cancelAuthBackoff:      make(chan struct{}, 1),
+		cancelSSEBackoff:       make(chan struct{}, 1),
+		cancelTokenExpiration:  make(chan struct{}, 1),
+		cancelStreamingWatcher: make(chan struct{}, 1),
+		control:                control,
+		status:                 status,
+	}, nil
+}
+
+func (p *PushManager) cancelStreaming() {
+	p.logger.Error("Error, switching to polling")
+	p.managerStatus <- NonRetriableError
+}
+
+func (p *PushManager) performAuthentication(errResult chan error) *dtos.Token {
+	select {
+	case <-p.cancelAuthBackoff:
+		// Discarding previous msg
+	default:
+	}
+	tokenResult := make(chan *dtos.Token, 1)
+	cancelAuthBackoff := common.WithBackoffCancelling(1*time.Second, maxPeriod, func() bool {
+		token, err := p.authClient.Authenticate()
+		if err != nil {
+			errType, ok := err.(dtos.HTTPError)
+			if ok && errType.Code >= http.StatusInternalServerError {
+				p.managerStatus <- BackoffAuth
+				return false // It will continue retrying
+			}
+			errResult <- errors.New("Error authenticating")
+			return true
+		}
+		tokenResult <- token
+		return true // Result is OK, Stopping Here, no more backoff
+	})
+	defer cancelAuthBackoff()
+	select {
+	case token := <-tokenResult:
+		if !token.PushEnabled {
+			return nil
+		}
+		return token
+	case err := <-errResult:
+		p.logger.Error(err.Error())
+		return nil
+	case <-p.cancelAuthBackoff:
+		return nil
+	}
+}
+
+func (p *PushManager) connectToStreaming(errResult chan error, token string, channels []string) error {
+	select {
+	case <-p.cancelSSEBackoff:
+		// Discarding previous msg
+	default:
+	}
+	sseResult := make(chan struct{}, 1)
+	cancelSSEBackoff := common.WithBackoffCancelling(1*time.Second, maxPeriod, func() bool {
+		p.sseClient.ConnectStreaming(token, channels, p.eventHandler.HandleIncomingMessage)
+		status := <-p.streamingStatus
+		switch status {
+		case sseStatus.OK:
+			sseResult <- struct{}{}
+			return true
+		case sseStatus.ErrorInternal:
+			p.managerStatus <- BackoffSSE
+			return false // It will continue retrying
+		default:
+			errResult <- errors.New("Error connecting streaming")
+			return true
+		}
+	})
+	defer cancelSSEBackoff()
+	select {
+	case <-sseResult:
+		return nil
+	case err := <-errResult:
+		p.logger.Error(err.Error())
+		return err
+	case <-p.cancelSSEBackoff:
+		return nil
+	}
+}
+
+func (p *PushManager) fetchStreamingToken(errResult chan error) (string, []string, error) {
+	token := p.performAuthentication(errResult)
+	if token == nil {
+		return "", []string{}, errors.New("Could not perform authentication")
+	}
+	channels, err := token.ChannelList()
+	if err != nil {
+		return "", []string{}, errors.New("Could not perform authentication")
+	}
+	nextTokenExpiration, err := token.CalculateNextTokenExpiration()
+	if err != nil {
+		return "", []string{}, errors.New("Could not perform authentication")
+	}
+	go func() {
+		// Create timeout timer for calculating next token expiration
+		idleDuration := nextTokenExpiration
+		tokenExpirationTimer := time.NewTimer(idleDuration)
+		defer tokenExpirationTimer.Stop()
+
+		select {
+		case <-tokenExpirationTimer.C: // Timedout
+			p.logger.Info("Token expired")
+			p.managerStatus <- TokenExpiration
+			return
+		case <-p.cancelTokenExpiration:
+			return
+		}
+	}()
+	return token.Token, channels, nil
+}
+
+func (p *PushManager) streamingStatusWatcher() {
+	for {
+		select {
+		case status := <-p.streamingStatus: // Streaming SSE Status
+			switch status {
+			case sseStatus.ErrorKeepAlive: // On ConnectionTimedOut -> Reconnect
+				fallthrough
+			case sseStatus.ErrorInternal: // On Error >= 500 -> Reconnect
+				fallthrough
+			case sseStatus.ErrorReadingStream: // On IOF -> Reconnect
+				p.managerStatus <- Reconnect
+			default: // Whatever other errors -> Send Error to disconnect
+				p.cancelStreaming()
+			}
+		case publisherStatus := <-p.publishers: // Publisher Available/Not Available
+			switch publisherStatus {
+			case PublisherNotPresent:
+				if p.status.Load().(int) != StreamingPaused {
+					p.managerStatus <- PushIsDown
+				}
+			case PublisherAvailable:
+				if p.status.Load().(int) != StreamingPaused {
+					p.managerStatus <- PushIsUp
+				}
+			default:
+				p.logger.Debug(fmt.Sprintf("Unexpected publisher status received %d", publisherStatus))
+			}
+		case controlStatus := <-p.control:
+			switch controlStatus {
+			case streamingPaused:
+				p.logger.Debug("Received Pause Streaming Notification")
+				if p.status.Load().(int) != StreamingPaused {
+					p.logger.Info("Sending Pause Streaming")
+					p.status.Store(StreamingPaused)
+					p.managerStatus <- PushIsDown
+				}
+			case streamingResumed:
+				p.logger.Debug("Received Resume Streaming Notification")
+				if p.status.Load().(int) == StreamingPaused {
+					p.status.Store(StreamingResumed)
+					publishersAvailable := p.eventHandler.keeper.Publishers("control_pri")
+					if publishersAvailable != nil && *publishersAvailable > 0 {
+						p.logger.Info("Sending Resume Streaming")
+						p.managerStatus <- PushIsUp
+					}
+				}
+			case streamingDisabled:
+				p.logger.Info("Received Streaming Disabled Notification")
+				p.managerStatus <- StreamingDisabled
+			default:
+				p.logger.Debug(fmt.Sprintf("Unexpected control status received %d", controlStatus))
+			}
+		case <-p.cancelStreamingWatcher: // Stopping Watcher
+			return
+		}
+	}
+}
+
+func (p *PushManager) drainStatus() {
+	select {
+	case <-p.cancelStreamingWatcher: // Discarding previous msg
+	default:
+	}
+	select {
+	case <-p.cancelTokenExpiration: // Discarding previous token expiration
+	default:
+	}
+}
+
+// Start push services
+func (p *PushManager) Start() {
+	if p.IsRunning() {
+		p.logger.Info("PushManager is already running, skipping Start")
+		return
+	}
+	p.drainStatus()
+
+	// errResult listener for fetching token and connecting to SSE
+	errResult := make(chan error, 1)
+	token, channels, err := p.fetchStreamingToken(errResult)
+	if err != nil {
+		p.cancelStreaming()
+		return
+	}
+	err = p.connectToStreaming(errResult, token, channels)
+	if err != nil {
+		p.cancelStreaming()
+		return
+	}
+
+	// Everything is good, starting workers
+	p.splitWorker.Start()
+	p.segmentWorker.Start()
+
+	// Sending Ready
+	p.managerStatus <- Ready
+
+	// Starting streaming status watcher, it will listen 1) errors in SSE, 2) publishers changes, 3) stop
+	go p.streamingStatusWatcher()
+}
+
+// Stop push services
+func (p *PushManager) Stop() {
+	p.logger.Info("Stopping Push Services")
+	p.cancelAuthBackoff <- struct{}{}
+	p.cancelSSEBackoff <- struct{}{}
+	p.cancelTokenExpiration <- struct{}{}
+	p.cancelStreamingWatcher <- struct{}{}
+	if p.sseClient.IsRunning() {
+		p.sseClient.StopStreaming(true)
+	}
+	p.StopWorkers()
+}
+
+// IsRunning returns true if the services are running
+func (p *PushManager) IsRunning() bool {
+	return p.sseClient.IsRunning() || p.splitWorker.IsRunning() || p.segmentWorker.IsRunning()
+}
+
+// StopWorkers stops workers
+func (p *PushManager) StopWorkers() {
+	if p.splitWorker.IsRunning() {
+		p.splitWorker.Stop()
+	}
+	if p.segmentWorker.IsRunning() {
+		p.segmentWorker.Stop()
+	}
+}
+
+// StartWorkers starts workers
+func (p *PushManager) StartWorkers() {
+	if !p.splitWorker.IsRunning() {
+		p.splitWorker.Start()
+	}
+	if !p.segmentWorker.IsRunning() {
+		p.segmentWorker.Start()
+	}
+}
--- a/vendor/github.com/splitio/go-split-commons/v2/push/segment.go
+++ b/vendor/github.com/splitio/go-split-commons/v2/push/segment.go
@@ -0,0 +1,76 @@
+package push
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+
+	"github.com/splitio/go-split-commons/v2/dtos"
+	"github.com/splitio/go-toolkit/v3/logging"
+)
+
+// SegmentUpdateWorker struct
+type SegmentUpdateWorker struct {
+	activeGoroutines *sync.WaitGroup
+	segmentQueue     chan dtos.SegmentChangeNotification
+	handler          func(segmentName string, till *int64) error
+	logger           logging.LoggerInterface
+	stop             chan struct{}
+	running          atomic.Value
+}
+
+// NewSegmentUpdateWorker creates SegmentUpdateWorker
+func NewSegmentUpdateWorker(segmentQueue chan dtos.SegmentChangeNotification, handler func(segmentName string, till *int64) error, logger logging.LoggerInterface) (*SegmentUpdateWorker, error) {
+	if cap(segmentQueue) < 5000 {
+		return nil, errors.New("")
+	}
+	running := atomic.Value{}
+	running.Store(false)
+
+	return &SegmentUpdateWorker{
+		segmentQueue: segmentQueue,
+		handler:      handler,
+		logger:       logger,
+		stop:         make(chan struct{}, 1),
+		running:      running,
+	}, nil
+}
+
+// Start starts worker
+func (s *SegmentUpdateWorker) Start() {
+	s.logger.Debug("Started SegmentUpdateWorker")
+	if s.IsRunning() {
+		s.logger.Debug("Segment worker is already running")
+		return
+	}
+	s.running.Store(true)
+	go func() {
+		for {
+			select {
+			case segmentUpdate := <-s.segmentQueue:
+				s.logger.Debug("Received Segment update and proceding to perform fetch")
+				s.logger.Debug(fmt.Sprintf("SegmentName: %s\nChangeNumber: %d", segmentUpdate.SegmentName, &segmentUpdate.ChangeNumber))
+				err := s.handler(segmentUpdate.SegmentName, &segmentUpdate.ChangeNumber)
+				if err != nil {
+					s.logger.Error(err)
+				}
+			case <-s.stop:
+				return
+			}
+		}
+	}()
+}
+
+// Stop stops worker
+func (s *SegmentUpdateWorker) Stop() {
+	if s.IsRunning() {
+		s.stop <- struct{}{}
+		s.running.Store(false)
+	}
+}
+
+// IsRunning indicates if worker is running or not
+func (s *SegmentUpdateWorker) IsRunning() bool {
+	return s.running.Load().(bool)
+}
--- a/vendor/github.com/splitio/go-split-commons/v2/push/split.go
+++ b/vendor/github.com/splitio/go-split-commons/v2/push/split.go
@@ -0,0 +1,77 @@
+package push
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+
+	"github.com/splitio/go-split-commons/v2/dtos"
+	"github.com/splitio/go-toolkit/v3/logging"
+)
+
+// SplitUpdateWorker struct
+type SplitUpdateWorker struct {
+	activeGoroutines *sync.WaitGroup
+	splitQueue       chan dtos.SplitChangeNotification
+	handler          func(till *int64) error
+	logger           logging.LoggerInterface
+	stop             chan struct{}
+	running          atomic.Value
+}
+
+// NewSplitUpdateWorker creates SplitUpdateWorker
+func NewSplitUpdateWorker(splitQueue chan dtos.SplitChangeNotification, handler func(till *int64) error, logger logging.LoggerInterface) (*SplitUpdateWorker, error) {
+	if cap(splitQueue) < 5000 {
+		return nil, errors.New("")
+	}
+	running := atomic.Value{}
+	running.Store(false)
+
+	return &SplitUpdateWorker{
+		activeGoroutines: &sync.WaitGroup{},
+		splitQueue:       splitQueue,
+		handler:          handler,
+		logger:           logger,
+		running:          running,
+		stop:             make(chan struct{}, 1),
+	}, nil
+}
+
+// Start starts worker
+func (s *SplitUpdateWorker) Start() {
+	s.logger.Debug("Started SplitUpdateWorker")
+	if s.IsRunning() {
+		s.logger.Info("Split worker is already running")
+		return
+	}
+	s.running.Store(true)
+	go func() {
+		for {
+			select {
+			case splitUpdate := <-s.splitQueue:
+				s.logger.Debug("Received Split update and proceding to perform fetch")
+				s.logger.Debug(fmt.Sprintf("ChangeNumber: %d", splitUpdate.ChangeNumber))
+				err := s.handler(&splitUpdate.ChangeNumber)
+				if err != nil {
+					s.logger.Error(err)
+				}
+			case <-s.stop:
+				return
+			}
+		}
+	}()
+}
+
+// Stop stops worker
+func (s *SplitUpdateWorker) Stop() {
+	if s.IsRunning() {
+		s.stop <- struct{}{}
+		s.running.Store(false)
+	}
+}
+
+// IsRunning indicates if worker is running or not
+func (s *SplitUpdateWorker) IsRunning() bool {
+	return s.running.Load().(bool)
+}