Update split SDK to 6.0.2 to fix sync bug (#17060)

* Update split SDK to 6.0.2 to fix sync bug

* Vendor and tidy

vendor/github.com/splitio/go-split-commons/v3/push/borrowed.go

@@ -0,0 +1,13 @@
+package push
+
+// Borrowed synchronizer interface to break circular dependencies
+type synchronizerInterface interface {
+	SyncAll(requestNoCache bool) error
+	SynchronizeSplits(till *int64, requestNoCache bool) error
+	LocalKill(splitName string, defaultTreatment string, changeNumber int64)
+	SynchronizeSegment(segmentName string, till *int64, requestNoCache bool) error
+	StartPeriodicFetching()
+	StopPeriodicFetching()
+	StartPeriodicDataRecording()
+	StopPeriodicDataRecording()
+}

vendor/github.com/splitio/go-split-commons/v3/push/constants.go

@@ -0,0 +1,14 @@
+package push
+
+const (
+	workerStatusIdle = iota
+	workerStatusRunning
+	workerStatusShuttingDown
+)
+
+const (
+	pushManagerStatusIdle = iota
+	pushManagerStatusInitializing
+	pushManagerStatusRunning
+	pushManagerStatusShuttingDown
+)

vendor/github.com/splitio/go-split-commons/v3/push/manager.go

@@ -0,0 +1,237 @@
+package push
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/splitio/go-split-commons/v3/conf"
+	"github.com/splitio/go-split-commons/v3/dtos"
+	"github.com/splitio/go-split-commons/v3/service"
+	"github.com/splitio/go-split-commons/v3/service/api/sse"
+	"github.com/splitio/go-toolkit/v4/common"
+	"github.com/splitio/go-toolkit/v4/logging"
+	"github.com/splitio/go-toolkit/v4/struct/traits/lifecycle"
+)
+
+// Status update contants that will be propagated to the push manager's user
+const (
+	StatusUp = iota
+	StatusDown
+	StatusRetryableError
+	StatusNonRetryableError
+)
+
+// ErrAlreadyRunning is the error to be returned when .Start() is called on an already running instance
+var ErrAlreadyRunning = errors.New("push manager already running")
+
+// ErrNotRunning is the error to be returned when .Stop() is called on a non-running instance
+var ErrNotRunning = errors.New("push manager not running")
+
+// Manager interface contains public methods for push manager
+type Manager interface {
+	Start() error
+	Stop() error
+	StopWorkers()
+	StartWorkers()
+}
+
+// ManagerImpl implements the manager interface
+type ManagerImpl struct {
+	parser            NotificationParser
+	sseClient         sse.StreamingClient
+	authAPI           service.AuthClient
+	processor         Processor
+	statusTracker     StatusTracker
+	feedback          FeedbackLoop
+	nextRefresh       *time.Timer
+	refreshTokenMutex sync.Mutex
+	/*
+		running           *gtSync.AtomicBool
+		status            int32
+		shutdownWaiter    chan struct{}
+	*/
+	lifecycle lifecycle.Manager
+	logger    logging.LoggerInterface
+}
+
+// FeedbackLoop is a type alias for the type of chan that must be supplied for push status tobe propagated
+type FeedbackLoop = chan<- int64
+
+// NewManager constructs a new push manager
+func NewManager(
+	logger logging.LoggerInterface,
+	synchronizer synchronizerInterface,
+	cfg *conf.AdvancedConfig,
+	feedbackLoop chan<- int64,
+	authAPI service.AuthClient,
+) (*ManagerImpl, error) {
+
+	processor, err := NewProcessor(cfg.SplitUpdateQueueSize, cfg.SegmentUpdateQueueSize, synchronizer, logger)
+	if err != nil {
+		return nil, fmt.Errorf("error instantiating processor: %w", err)
+	}
+
+	statusTracker := NewStatusTracker(logger)
+	parser := &NotificationParserImpl{
+		logger:            logger,
+		onSplitUpdate:     processor.ProcessSplitChangeUpdate,
+		onSplitKill:       processor.ProcessSplitKillUpdate,
+		onSegmentUpdate:   processor.ProcessSegmentChangeUpdate,
+		onControlUpdate:   statusTracker.HandleControl,
+		onOccupancyMesage: statusTracker.HandleOccupancy,
+		onAblyError:       statusTracker.HandleAblyError,
+	}
+
+	manager := &ManagerImpl{
+		authAPI:       authAPI,
+		sseClient:     sse.NewStreamingClient(cfg, logger),
+		statusTracker: statusTracker,
+		feedback:      feedbackLoop,
+		processor:     processor,
+		parser:        parser,
+		logger:        logger,
+	}
+	manager.lifecycle.Setup()
+	return manager, nil
+}
+
+// Start initiates the authentication flow and if successful initiates a connection
+func (m *ManagerImpl) Start() error {
+	if !m.lifecycle.BeginInitialization() {
+		return ErrAlreadyRunning
+	}
+	m.triggerConnectionFlow()
+	return nil
+}
+
+// Stop method stops the sse client and it's status monitoring goroutine
+func (m *ManagerImpl) Stop() error {
+	if !m.lifecycle.BeginShutdown() {
+		return ErrNotRunning
+	}
+	m.statusTracker.NotifySSEShutdownExpected()
+	m.withRefreshTokenLock(func() {
+		if m.nextRefresh != nil {
+			m.nextRefresh.Stop()
+		}
+	})
+	m.StopWorkers()
+	m.sseClient.StopStreaming()
+	m.lifecycle.AwaitShutdownComplete()
+	return nil
+}
+
+// StartWorkers start the splits & segments workers
+func (m *ManagerImpl) StartWorkers() {
+	m.processor.StartWorkers()
+}
+
+// StopWorkers stops the splits & segments workers
+func (m *ManagerImpl) StopWorkers() {
+	m.processor.StopWorkers()
+}
+
+func (m *ManagerImpl) performAuthentication() (*dtos.Token, *int64) {
+	token, err := m.authAPI.Authenticate()
+	if err != nil {
+		if errType, ok := err.(dtos.HTTPError); ok {
+			if errType.Code >= http.StatusInternalServerError {
+				m.logger.Error(fmt.Sprintf("Error authenticating: %s", err.Error()))
+				return nil, common.Int64Ref(StatusRetryableError)
+			}
+			return nil, common.Int64Ref(StatusNonRetryableError) // 400, 401, etc
+		}
+		// Not an HTTP eerror, most likely a tcp/bad connection. Should retry
+		return nil, common.Int64Ref(StatusRetryableError)
+	}
+	if !token.PushEnabled {
+		return nil, common.Int64Ref(StatusNonRetryableError)
+	}
+	return token, nil
+}
+
+func (m *ManagerImpl) eventHandler(e sse.IncomingMessage) {
+	newStatus, err := m.parser.ParseAndForward(e)
+	if newStatus != nil {
+		m.feedback <- *newStatus
+	} else if err != nil {
+		m.logger.Error("error parsing message: ", err)
+		m.logger.Debug("failed message: ", e)
+		m.feedback <- StatusRetryableError
+	}
+}
+
+func (m *ManagerImpl) triggerConnectionFlow() {
+	token, status := m.performAuthentication()
+	if status != nil {
+		m.lifecycle.AbnormalShutdown()
+		defer m.lifecycle.ShutdownComplete()
+		m.feedback <- *status
+		return
+	}
+
+	tokenList, err := token.ChannelList()
+	if err != nil {
+		m.logger.Error("error parsing channel list: ", err)
+		m.lifecycle.AbnormalShutdown()
+		defer m.lifecycle.ShutdownComplete()
+		m.feedback <- StatusRetryableError
+		return
+	}
+
+	m.statusTracker.Reset()
+	sseStatus := make(chan int, 100)
+	m.sseClient.ConnectStreaming(token.Token, sseStatus, tokenList, m.eventHandler)
+	go func() {
+		defer m.lifecycle.ShutdownComplete()
+		if !m.lifecycle.InitializationComplete() {
+			return
+		}
+		for {
+			message := <-sseStatus
+			switch message {
+			case sse.StatusFirstEventOk:
+				when, err := token.CalculateNextTokenExpiration()
+				if err != nil || when <= 0 {
+					m.logger.Warning("Failed to calculate next token expiration time. Defaulting to 50 minutes")
+					when = 50 * time.Minute
+				}
+				m.withRefreshTokenLock(func() {
+					m.nextRefresh = time.AfterFunc(when, func() {
+						m.logger.Info("Refreshing SSE auth token.")
+						m.Stop()
+						m.Start()
+					})
+				})
+				m.feedback <- StatusUp
+			case sse.StatusConnectionFailed:
+				m.lifecycle.AbnormalShutdown()
+				m.logger.Error("SSE Connection failed")
+				m.feedback <- StatusRetryableError
+				return
+			case sse.StatusDisconnected:
+				m.logger.Debug("propagating sse disconnection event")
+				status := m.statusTracker.HandleDisconnection()
+				if status != nil { // connection ended unexpectedly
+					m.lifecycle.AbnormalShutdown()
+					m.feedback <- *status
+				}
+				return
+			case sse.StatusUnderlyingClientInUse:
+				m.lifecycle.AbnormalShutdown()
+				m.logger.Error("unexpected error in streaming. Switching to polling")
+				m.feedback <- StatusNonRetryableError
+				return
+			}
+		}
+	}()
+}
+
+func (m *ManagerImpl) withRefreshTokenLock(f func()) {
+	m.refreshTokenMutex.Lock()
+	defer m.refreshTokenMutex.Unlock()
+	f()
+}

vendor/github.com/splitio/go-split-commons/v3/push/parser.go

@@ -0,0 +1,399 @@
+package push
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/splitio/go-split-commons/v3/service/api/sse"
+
+	"github.com/splitio/go-toolkit/v4/logging"
+)
+
+// SSE event type constants
+const (
+	SSEEventTypeSync    = "sync"
+	SSEEventTypeMessage = "message"
+	SSEEventTypeError   = "error"
+)
+
+// Message type constants
+const (
+	MessageTypeUpdate = iota
+	MessageTypeControl
+	MessageTypeOccupancy
+)
+
+// Update type constants
+const (
+	UpdateTypeSplitChange   = "SPLIT_UPDATE"
+	UpdateTypeSplitKill     = "SPLIT_KILL"
+	UpdateTypeSegmentChange = "SEGMENT_UPDATE"
+	UpdateTypeContol        = "CONTROL"
+)
+
+// Control type constants
+const (
+	ControlTypeStreamingEnabled  = "STREAMING_ENABLED"
+	ControlTypeStreamingPaused   = "STREAMING_PAUSED"
+	ControlTypeStreamingDisabled = "STREAMING_DISABLED"
+)
+
+const (
+	occupancuName   = "[meta]occupancy"
+	occupancyPrefix = "[?occupancy=metrics.publishers]"
+)
+
+// ErrEmptyEvent indicates an event without message and event fields
+var ErrEmptyEvent = errors.New("empty incoming event")
+
+// NotificationParser interface
+type NotificationParser interface {
+	ParseAndForward(sse.IncomingMessage) (*int64, error)
+}
+
+// NotificationParserImpl implementas the NotificationParser interface
+type NotificationParserImpl struct {
+	logger            logging.LoggerInterface
+	onSplitUpdate     func(*SplitChangeUpdate) error
+	onSplitKill       func(*SplitKillUpdate) error
+	onSegmentUpdate   func(*SegmentChangeUpdate) error
+	onControlUpdate   func(*ControlUpdate) *int64
+	onOccupancyMesage func(*OccupancyMessage) *int64
+	onAblyError       func(*AblyError) *int64
+}
+
+// ParseAndForward accepts an incoming RAW event and returns a properly parsed & typed event
+func (p *NotificationParserImpl) ParseAndForward(raw sse.IncomingMessage) (*int64, error) {
+
+	if raw.Event() == "" {
+		if raw.ID() == "" {
+			return nil, ErrEmptyEvent
+		}
+		// If it has ID its a sync event, which we're not using not. Ignore.
+		p.logger.Debug("Ignoring sync event")
+		return nil, nil
+	}
+
+	data := genericData{}
+	err := json.Unmarshal([]byte(raw.Data()), &data)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing JSON: %w", err)
+	}
+
+	switch raw.Event() {
+	case SSEEventTypeError:
+		return p.parseError(&data)
+	case SSEEventTypeMessage:
+		return p.parseMessage(&data)
+	}
+
+	return nil, nil
+
+}
+
+func (p *NotificationParserImpl) parseError(data *genericData) (*int64, error) {
+	return p.onAblyError(&AblyError{
+		code:       data.Code,
+		statusCode: data.StatusCode,
+		message:    data.Message,
+		href:       data.Href,
+		timestamp:  data.Timestamp,
+	}), nil
+}
+
+func (p *NotificationParserImpl) parseMessage(data *genericData) (*int64, error) {
+	var nested genericMessageData
+	err := json.Unmarshal([]byte(data.Data), &nested)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing message nested json data: %w", err)
+	}
+
+	if data.Name == occupancuName {
+		return p.onOccupancyMesage(&OccupancyMessage{
+			BaseMessage: BaseMessage{
+				timestamp: data.Timestamp,
+				channel:   data.Channel,
+			},
+			publishers: nested.Metrics.Publishers,
+		}), nil
+	}
+
+	return p.parseUpdate(data, &nested)
+}
+
+func (p *NotificationParserImpl) parseUpdate(data *genericData, nested *genericMessageData) (*int64, error) {
+	if data == nil || nested == nil {
+		return nil, errors.New("parseUpdate: data cannot be nil")
+	}
+
+	base := BaseUpdate{
+		BaseMessage:  BaseMessage{timestamp: data.Timestamp, channel: data.Channel},
+		changeNumber: nested.ChangeNumber,
+	}
+
+	switch nested.Type {
+	case UpdateTypeSplitChange:
+		return nil, p.onSplitUpdate(&SplitChangeUpdate{BaseUpdate: base})
+	case UpdateTypeSplitKill:
+		return nil, p.onSplitKill(&SplitKillUpdate{BaseUpdate: base, splitName: nested.SplitName, defaultTreatment: nested.DefaultTreatment})
+	case UpdateTypeSegmentChange:
+		return nil, p.onSegmentUpdate(&SegmentChangeUpdate{BaseUpdate: base, segmentName: nested.SegmentName})
+	case UpdateTypeContol:
+		return p.onControlUpdate(&ControlUpdate{BaseMessage: base.BaseMessage, controlType: nested.ControlType}), nil
+	default:
+		// TODO: log full event in debug mode
+		return nil, fmt.Errorf("invalid update type: %s", nested.Type)
+	}
+}
+
+// Event basic interface
+type Event interface {
+	fmt.Stringer
+	EventType() string
+	Timestamp() int64
+}
+
+// SSESyncEvent represents an SSE Sync event with only id (used for resuming connections)
+type SSESyncEvent struct {
+	id        string
+	timestamp int64
+}
+
+// EventType always returns SSEEventTypeSync for SSESyncEvents
+func (e *SSESyncEvent) EventType() string { return SSEEventTypeSync }
+
+// Timestamp returns the timestamp of the event parsing
+func (e *SSESyncEvent) Timestamp() int64 { return e.timestamp }
+
+// String returns the string represenation of the event
+func (e *SSESyncEvent) String() string {
+	return fmt.Sprintf("SSESync(id=%s,timestamp=%d)", e.id, e.timestamp)
+}
+
+// AblyError struct
+type AblyError struct {
+	code       int
+	statusCode int
+	message    string
+	href       string
+	timestamp  int64
+}
+
+// EventType always returns SSEEventTypeError for AblyError
+func (a *AblyError) EventType() string { return SSEEventTypeError }
+
+// Code returns the error code
+func (a *AblyError) Code() int { return a.code }
+
+// StatusCode returns the status code
+func (a *AblyError) StatusCode() int { return a.statusCode }
+
+// Message returns the error message
+func (a *AblyError) Message() string { return a.message }
+
+// Href returns the documentation link
+func (a *AblyError) Href() string { return a.href }
+
+// Timestamp returns the error timestamp
+func (a *AblyError) Timestamp() int64 { return a.timestamp }
+
+// IsRetryable returns whether the error is recoverable via a push subsystem restart
+func (a *AblyError) IsRetryable() bool { return a.code >= 40140 && a.code <= 40149 }
+
+// String returns the string representation of the ably error
+func (a *AblyError) String() string {
+	return fmt.Sprintf("AblyError(code=%d,statusCode=%d,message=%s,timestamp=%d,isRetryable=%t)",
+		a.code, a.statusCode, a.message, a.timestamp, a.IsRetryable())
+}
+
+// Message basic interface
+type Message interface {
+	Event
+	MessageType() int64
+	Channel() string
+}
+
+// BaseMessage contains the basic message-specific fields and methods
+type BaseMessage struct {
+	timestamp int64
+	channel   string
+}
+
+// EventType always returns SSEEventTypeMessage for BaseMessage and embedding types
+func (m *BaseMessage) EventType() string { return SSEEventTypeMessage }
+
+// Timestamp returns the timestamp of the message reception
+func (m *BaseMessage) Timestamp() int64 { return m.timestamp }
+
+// Channel returns which channel the message was received in
+func (m *BaseMessage) Channel() string { return m.channel }
+
+// OccupancyMessage contains fields & methods related to ocupancy messages
+type OccupancyMessage struct {
+	BaseMessage
+	publishers int64
+}
+
+// MessageType always returns MessageTypeOccupancy for Occupancy messages
+func (o *OccupancyMessage) MessageType() int64 { return MessageTypeOccupancy }
+
+// ChannelWithoutPrefix returns the original channel namem without the metadata prefix
+func (o *OccupancyMessage) ChannelWithoutPrefix() string {
+	return strings.Replace(o.Channel(), occupancyPrefix, "", 1)
+}
+
+// Publishers returbs the amount of publishers in the current channel
+func (o *OccupancyMessage) Publishers() int64 {
+	return o.publishers
+}
+
+// Strings returns the string representation of an occupancy message
+func (o *OccupancyMessage) String() string {
+	return fmt.Sprintf("Occupancy(channel=%s,publishers=%d,timestamp=%d)",
+		o.Channel(), o.publishers, o.Timestamp())
+}
+
+// Update basic interface
+type Update interface {
+	Message
+	UpdateType() string
+	ChangeNumber() int64
+}
+
+// BaseUpdate contains fields & methods related to update-based messages
+type BaseUpdate struct {
+	BaseMessage
+	changeNumber int64
+}
+
+// MessageType alwats returns MessageType for Update messages
+func (b *BaseUpdate) MessageType() int64 { return MessageTypeUpdate }
+
+// ChangeNumber returns the changeNumber of the update
+func (b *BaseUpdate) ChangeNumber() int64 { return b.changeNumber }
+
+// SplitChangeUpdate represents a SplitChange notification generated in the split servers
+type SplitChangeUpdate struct {
+	BaseUpdate
+}
+
+// UpdateType always returns UpdateTypeSplitChange for SplitKillUpdate messages
+func (u *SplitChangeUpdate) UpdateType() string { return UpdateTypeSplitChange }
+
+// String returns the String representation of a split change notification
+func (u *SplitChangeUpdate) String() string {
+	return fmt.Sprintf("SplitChange(channel=%s,changeNumber=%d,timestamp=%d)",
+		u.Channel(), u.ChangeNumber(), u.Timestamp())
+}
+
+// SplitKillUpdate represents a SplitKill notification generated in the split servers
+type SplitKillUpdate struct {
+	BaseUpdate
+	splitName        string
+	defaultTreatment string
+}
+
+// UpdateType always returns UpdateTypeSplitKill for SplitKillUpdate messages
+func (u *SplitKillUpdate) UpdateType() string { return UpdateTypeSplitKill }
+
+// SplitName returns the name of the killed split
+func (u *SplitKillUpdate) SplitName() string { return u.splitName }
+
+// DefaultTreatment returns the last default treatment seen in the split servers for this split
+func (u *SplitKillUpdate) DefaultTreatment() string { return u.defaultTreatment }
+
+// ToSplitChangeUpdate Maps this kill notification to a split change one
+func (u *SplitKillUpdate) ToSplitChangeUpdate() *SplitChangeUpdate {
+	return &SplitChangeUpdate{BaseUpdate: u.BaseUpdate}
+}
+
+// String returns the string representation of this update
+func (u *SplitKillUpdate) String() string {
+	return fmt.Sprintf("SplitKill(channel=%s,changeNumber=%d,splitName=%s,defaultTreatment=%s,timestamp=%d)",
+		u.Channel(), u.ChangeNumber(), u.SplitName(), u.DefaultTreatment(), u.Timestamp())
+}
+
+// SegmentChangeUpdate represents a segment change notification generated in the split servers.
+type SegmentChangeUpdate struct {
+	BaseUpdate
+	segmentName string
+}
+
+// UpdateType is always UpdateTypeSegmentChange for Segmet Updates
+func (u *SegmentChangeUpdate) UpdateType() string { return UpdateTypeSegmentChange }
+
+// SegmentName returns the name of the updated segment
+func (u *SegmentChangeUpdate) SegmentName() string { return u.segmentName }
+
+// String returns the string representation of a segment update notification
+func (u *SegmentChangeUpdate) String() string {
+	return fmt.Sprintf("SegmentChange(channel=%s,changeNumber=%d,segmentName=%s,timestamp=%d",
+		u.Channel(), u.ChangeNumber(), u.segmentName, u.Timestamp())
+}
+
+// ControlUpdate represents a control notification generated by the split push subsystem
+type ControlUpdate struct {
+	BaseMessage
+	controlType string
+}
+
+// MessageType always returns MessageTypeControl for Control messages
+func (u *ControlUpdate) MessageType() int64 { return MessageTypeControl }
+
+// ControlType returns the type of control notification received
+func (u *ControlUpdate) ControlType() string { return u.controlType }
+
+// String returns a string representation of this notification
+func (u *ControlUpdate) String() string {
+	return fmt.Sprintf("Control(channel=%s,type=%s,timestamp=%d)",
+		u.Channel(), u.controlType, u.Timestamp())
+}
+
+type genericData struct {
+
+	// Error associated data
+	Code       int    `json:"code"`
+	StatusCode int    `json:"statusCode"`
+	Message    string `json:"message"`
+	Href       string `json:"href"`
+
+	ClientID  string `json:"clientId"`
+	ID        string `json:"id"`
+	Name      string `json:"name"`
+	Timestamp int64  `json:"timestamp"`
+	Encoding  string `json:"encoding"`
+	Channel   string `json:"channel"`
+	Data      string `json:"data"`
+
+	//"id":"tO4rXGE4CX:0:0","timestamp":1612897630627,"encoding":"json","channel":"[?occupancy=metrics.publishers]control_sec","data":"{\"metrics\":{\"publishers\":0}}","name":"[meta]occupancy"}
+
+}
+
+type metrics struct {
+	Publishers int64 `json:"publishers"`
+}
+
+type genericMessageData struct {
+	Metrics          metrics `json:"metrics"`
+	Type             string  `json:"type"`
+	ChangeNumber     int64   `json:"changeNumber"`
+	SplitName        string  `json:"splitName"`
+	DefaultTreatment string  `json:"defaultTreatment"`
+	SegmentName      string  `json:"segmentName"`
+	ControlType      string  `json:"controlType"`
+
+	// {\"type\":\"SPLIT_UPDATE\",\"changeNumber\":1612909342671}"}
+}
+
+// Compile-type assertions of interface requirements
+var _ Event = &AblyError{}
+var _ Message = &OccupancyMessage{}
+var _ Message = &SplitChangeUpdate{}
+var _ Message = &SplitKillUpdate{}
+var _ Message = &SegmentChangeUpdate{}
+var _ Message = &ControlUpdate{}
+var _ Update = &SplitChangeUpdate{}
+var _ Update = &SplitKillUpdate{}
+var _ Update = &SegmentChangeUpdate{}

vendor/github.com/splitio/go-split-commons/v3/push/processor.go

@@ -0,0 +1,107 @@
+package push
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/splitio/go-toolkit/v4/logging"
+)
+
+const (
+	splitQueueMinSize   = 5000
+	segmentQueueMinSize = 5000
+)
+
+// Processor provides the interface for an update-message processor
+type Processor interface {
+	ProcessSplitChangeUpdate(update *SplitChangeUpdate) error
+	ProcessSplitKillUpdate(update *SplitKillUpdate) error
+	ProcessSegmentChangeUpdate(update *SegmentChangeUpdate) error
+	StartWorkers()
+	StopWorkers()
+}
+
+// ProcessorImpl struct for notification processor
+type ProcessorImpl struct {
+	segmentQueue  chan SegmentChangeUpdate
+	splitQueue    chan SplitChangeUpdate
+	splitWorker   *SplitUpdateWorker
+	segmentWorker *SegmentUpdateWorker
+	synchronizer  synchronizerInterface
+	logger        logging.LoggerInterface
+}
+
+// NewProcessor creates new processor
+func NewProcessor(
+	splitQueueSize int64,
+	segmentQueueSize int64,
+	synchronizer synchronizerInterface,
+	logger logging.LoggerInterface,
+) (*ProcessorImpl, error) {
+	if segmentQueueSize < segmentQueueMinSize {
+		return nil, errors.New("Small size of segmentQueue")
+	}
+	if splitQueueSize < splitQueueMinSize {
+		return nil, errors.New("Small size of splitQueue")
+	}
+
+	splitQueue := make(chan SplitChangeUpdate, splitQueueSize)
+	splitWorker, err := NewSplitUpdateWorker(splitQueue, synchronizer, logger)
+	if err != nil {
+		return nil, fmt.Errorf("error instantiating split worker: %w", err)
+	}
+
+	segmentQueue := make(chan SegmentChangeUpdate, segmentQueueSize)
+	segmentWorker, err := NewSegmentUpdateWorker(segmentQueue, synchronizer, logger)
+	if err != nil {
+		return nil, fmt.Errorf("error instantiating split worker: %w", err)
+	}
+
+	return &ProcessorImpl{
+		splitQueue:    splitQueue,
+		splitWorker:   splitWorker,
+		segmentQueue:  segmentQueue,
+		segmentWorker: segmentWorker,
+		synchronizer:  synchronizer,
+		logger:        logger,
+	}, nil
+}
+
+// ProcessSplitChangeUpdate accepts a split change notifications and schedules a fetch
+func (p *ProcessorImpl) ProcessSplitChangeUpdate(update *SplitChangeUpdate) error {
+	if update == nil {
+		return errors.New("split change update cannot be nil")
+	}
+	p.splitQueue <- *update
+	return nil
+}
+
+// ProcessSplitKillUpdate accepts a split kill notification, issues a local kill and schedules a fetch
+func (p *ProcessorImpl) ProcessSplitKillUpdate(update *SplitKillUpdate) error {
+	if update == nil {
+		return errors.New("split change update cannot be nil")
+	}
+	p.synchronizer.LocalKill(update.SplitName(), update.DefaultTreatment(), update.ChangeNumber())
+	return p.ProcessSplitChangeUpdate(update.ToSplitChangeUpdate())
+}
+
+// ProcessSegmentChangeUpdate accepts a segment change notification and schedules a fetch
+func (p *ProcessorImpl) ProcessSegmentChangeUpdate(update *SegmentChangeUpdate) error {
+	if update == nil {
+		return errors.New("split change update cannot be nil")
+	}
+	p.segmentQueue <- *update
+	return nil
+}
+
+// StartWorkers enables split & segments workers
+func (p *ProcessorImpl) StartWorkers() {
+	p.splitWorker.Start()
+	p.segmentWorker.Start()
+}
+
+// StopWorkers pauses split & segments workers
+func (p *ProcessorImpl) StopWorkers() {
+	p.splitWorker.Stop()
+	p.segmentWorker.Stop()
+}

vendor/github.com/splitio/go-split-commons/v3/push/segment.go

@@ -0,0 +1,82 @@
+package push
+
+import (
+	"errors"
+	"fmt"
+	"sync/atomic"
+
+	"github.com/splitio/go-toolkit/v4/common"
+	"github.com/splitio/go-toolkit/v4/logging"
+	"github.com/splitio/go-toolkit/v4/struct/traits/lifecycle"
+)
+
+// SegmentUpdateWorker struct
+type SegmentUpdateWorker struct {
+	segmentQueue chan SegmentChangeUpdate
+	sync         synchronizerInterface
+	logger       logging.LoggerInterface
+	lifecycle    lifecycle.Manager
+}
+
+// NewSegmentUpdateWorker creates SegmentUpdateWorker
+func NewSegmentUpdateWorker(
+	segmentQueue chan SegmentChangeUpdate,
+	synchronizer synchronizerInterface,
+	logger logging.LoggerInterface,
+) (*SegmentUpdateWorker, error) {
+	if cap(segmentQueue) < 5000 {
+		return nil, errors.New("")
+	}
+	running := atomic.Value{}
+	running.Store(false)
+
+	worker := &SegmentUpdateWorker{
+		segmentQueue: segmentQueue,
+		sync:         synchronizer,
+		logger:       logger,
+	}
+	worker.lifecycle.Setup()
+	return worker, nil
+}
+
+// Start starts worker
+func (s *SegmentUpdateWorker) Start() {
+	if !s.lifecycle.BeginInitialization() {
+		s.logger.Info("Segment worker is already running")
+		return
+	}
+
+	go func() {
+		if !s.lifecycle.InitializationComplete() {
+			return
+		}
+		defer s.lifecycle.ShutdownComplete()
+		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.sync.SynchronizeSegment(segmentUpdate.SegmentName(), common.Int64Ref(segmentUpdate.ChangeNumber()), true)
+				if err != nil {
+					s.logger.Error(err)
+				}
+			case <-s.lifecycle.ShutdownRequested():
+				return
+			}
+		}
+	}()
+}
+
+// Stop stops worker
+func (s *SegmentUpdateWorker) Stop() {
+	if !s.lifecycle.BeginShutdown() {
+		s.logger.Debug("Split worker not runnning. Ignoring.")
+		return
+	}
+	s.lifecycle.AwaitShutdownComplete()
+}
+
+// IsRunning indicates if worker is running or not
+func (s *SegmentUpdateWorker) IsRunning() bool {
+	return s.lifecycle.IsRunning()
+}

vendor/github.com/splitio/go-split-commons/v3/push/split.go

@@ -0,0 +1,80 @@
+package push
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/splitio/go-toolkit/v4/common"
+	"github.com/splitio/go-toolkit/v4/logging"
+	"github.com/splitio/go-toolkit/v4/struct/traits/lifecycle"
+)
+
+// SplitUpdateWorker struct
+type SplitUpdateWorker struct {
+	splitQueue chan SplitChangeUpdate
+	sync       synchronizerInterface
+	logger     logging.LoggerInterface
+	lifecycle  lifecycle.Manager
+}
+
+// NewSplitUpdateWorker creates SplitUpdateWorker
+func NewSplitUpdateWorker(
+	splitQueue chan SplitChangeUpdate,
+	synchronizer synchronizerInterface,
+	logger logging.LoggerInterface,
+) (*SplitUpdateWorker, error) {
+	if cap(splitQueue) < 5000 {
+		return nil, errors.New("")
+	}
+
+	worker := &SplitUpdateWorker{
+		splitQueue: splitQueue,
+		sync:       synchronizer,
+		logger:     logger,
+	}
+	worker.lifecycle.Setup()
+	return worker, nil
+}
+
+// Start starts worker
+func (s *SplitUpdateWorker) Start() {
+	if !s.lifecycle.BeginInitialization() {
+		s.logger.Info("Split worker is already running")
+		return
+	}
+
+	s.logger.Debug("Started SplitUpdateWorker")
+	go func() {
+		defer s.lifecycle.ShutdownComplete()
+		if !s.lifecycle.InitializationComplete() {
+			return
+		}
+		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.sync.SynchronizeSplits(common.Int64Ref(splitUpdate.ChangeNumber()), true)
+				if err != nil {
+					s.logger.Error(err)
+				}
+			case <-s.lifecycle.ShutdownRequested():
+				return
+			}
+		}
+	}()
+}
+
+// Stop stops worker
+func (s *SplitUpdateWorker) Stop() {
+	if !s.lifecycle.BeginShutdown() {
+		s.logger.Debug("Split worker not runnning. Ignoring.")
+		return
+	}
+	s.lifecycle.AwaitShutdownComplete()
+}
+
+// IsRunning indicates if worker is running or not
+func (s *SplitUpdateWorker) IsRunning() bool {
+	return s.lifecycle.IsRunning()
+}

vendor/github.com/splitio/go-split-commons/v3/push/statustracker.go

@@ -0,0 +1,158 @@
+package push
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/splitio/go-toolkit/v4/common"
+	"github.com/splitio/go-toolkit/v4/logging"
+)
+
+// StatusTracker keeps track of the status of the push subsystem and generates appropriate status change notifications.
+type StatusTracker interface {
+	HandleOccupancy(*OccupancyMessage) *int64
+	HandleControl(*ControlUpdate) *int64
+	HandleAblyError(*AblyError) *int64
+	HandleDisconnection() *int64
+	NotifySSEShutdownExpected()
+	Reset()
+}
+
+// StatusTrackerImpl is a concrete implementation of the StatusTracker interface
+type StatusTrackerImpl struct {
+	logger                 logging.LoggerInterface
+	mutex                  sync.Mutex
+	occupancy              map[string]int64
+	lastControlTimestamp   int64
+	lastOccupancyTimestamp int64
+	lastControlMessage     string
+	lastStatusPropagated   int64
+	shutdownExpected       bool
+}
+
+// NotifySSEShutdownExpected should be called when we are forcefully closing the SSE client
+func (p *StatusTrackerImpl) NotifySSEShutdownExpected() {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+	p.shutdownExpected = true
+}
+
+// Reset should be called on initialization and when the a new connection is being established (to start from scratch)
+func (p *StatusTrackerImpl) Reset() {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+	p.occupancy = map[string]int64{"control_pri": 2, "control_sec": 2}
+	p.lastControlMessage = ControlTypeStreamingEnabled
+	p.lastStatusPropagated = StatusUp
+	p.shutdownExpected = false
+}
+
+// HandleOccupancy should be called for every occupancy notification received
+func (p *StatusTrackerImpl) HandleOccupancy(message *OccupancyMessage) (newStatus *int64) {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+	if p.shutdownExpected {
+		return nil // we don't care about occupancy if we're disconnecting
+	}
+
+	channel := message.ChannelWithoutPrefix()
+	if _, ok := p.occupancy[channel]; !ok {
+		p.logger.Warning(fmt.Sprintf("received occupancy on non-registered channel '%s'. Ignoring", channel))
+		return nil
+	}
+
+	p.lastOccupancyTimestamp = message.Timestamp()
+	p.occupancy[channel] = message.Publishers()
+	return p.updateStatus()
+}
+
+// HandleAblyError should be called whenever an ably error is received
+func (p *StatusTrackerImpl) HandleAblyError(errorEvent *AblyError) (newStatus *int64) {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+	if p.shutdownExpected {
+		return nil // we don't care about occupancy if we're disconnecting
+	}
+
+	// Regardless of whether the error is retryable or not, we're going to close the connection
+	p.shutdownExpected = true
+
+	if errorEvent.IsRetryable() {
+		p.logger.Info("Received retryable error message. Restarting SSE connection with backoff")
+		return p.propagateStatus(StatusRetryableError)
+	}
+
+	p.logger.Info("Received non-retryable error message. Disabling streaming")
+	return p.propagateStatus(StatusNonRetryableError)
+}
+
+// HandleControl should be called whenever a control notification is received
+func (p *StatusTrackerImpl) HandleControl(controlUpdate *ControlUpdate) *int64 {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+	if p.shutdownExpected {
+		return nil // we don't care about occupancy if we're disconnecting
+	}
+
+	if p.lastControlTimestamp > controlUpdate.timestamp {
+		p.logger.Warning("Received an old control update. Ignoring")
+		return nil
+	}
+
+	p.lastControlMessage = controlUpdate.controlType
+	p.lastControlTimestamp = controlUpdate.timestamp
+	return p.updateStatus()
+}
+
+// HandleDisconnection should be called whenver the SSE client gets disconnected
+func (p *StatusTrackerImpl) HandleDisconnection() *int64 {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+	if !p.shutdownExpected {
+		return p.propagateStatus(StatusRetryableError)
+	}
+	return nil
+}
+
+// NewStatusTracker returns a new StatusTracker
+func NewStatusTracker(logger logging.LoggerInterface) *StatusTrackerImpl {
+	tracker := &StatusTrackerImpl{logger: logger}
+	tracker.Reset()
+	return tracker
+}
+
+func (p *StatusTrackerImpl) occupancyOk() bool {
+	for _, v := range p.occupancy {
+		if v > 0 {
+			return true
+		}
+	}
+	return false
+}
+
+func (p *StatusTrackerImpl) updateStatus() *int64 {
+	if p.lastStatusPropagated == StatusUp {
+		if !p.occupancyOk() || p.lastControlMessage == ControlTypeStreamingPaused {
+			return p.propagateStatus(StatusDown)
+		}
+		if p.lastControlMessage == ControlTypeStreamingDisabled {
+			return p.propagateStatus(StatusNonRetryableError)
+		}
+	}
+	if p.lastStatusPropagated == StatusDown {
+		if p.occupancyOk() && p.lastControlMessage == ControlTypeStreamingEnabled {
+			return p.propagateStatus(StatusUp)
+		}
+		if p.lastControlMessage == ControlTypeStreamingDisabled {
+			return p.propagateStatus(StatusNonRetryableError)
+		}
+	}
+	return nil
+}
+
+func (p *StatusTrackerImpl) propagateStatus(newStatus int64) *int64 {
+	p.lastStatusPropagated = newStatus
+	return common.Int64Ref(newStatus)
+}
+
+var _ StatusTracker = &StatusTrackerImpl{}