[FME-14174] Added atomic int to validate id the sse client it's running or not

service/api/sse/client.go

@@ -3,6 +3,7 @@
 import (
 	"errors"
 	"strings"
+	"sync/atomic"
 
 	"github.com/splitio/go-split-commons/v9/conf"
 	"github.com/splitio/go-split-commons/v9/dtos"
@@ -18,6 +19,12 @@
 	keepAlive = 70
 )
 
+const (
+	StateIdle      int32 = 0  // Client is idle/not started
+	StateRunning   int32 = 1  // Client is running
+	StateDestroyed int32 = -1 // Client is destroyed/stopped
+)
+
 // StreamingClient interface
 type StreamingClient interface {
 	ConnectStreaming(token string, streamingStatus chan int, channelList []string, handleIncomingMessage func(IncomingMessage))
@@ -32,6 +39,7 @@
 	lifecycle lifecycle.Manager
 	metadata  dtos.Metadata
 	clientKey *string
+	state     atomic.Int32 // Atomic state: 0=Idle, 1=Running, -1=Destroyed
 }
 
 // Status constants
@@ -54,15 +62,20 @@
 		metadata:  metadata,
 		clientKey: clientKey,
 	}
+	client.state.Store(StateIdle)
 	client.lifecycle.Setup()
 	return client
 }
 
 // ConnectStreaming connects to streaming
 func (s *StreamingClientImpl) ConnectStreaming(token string, streamingStatus chan int, channelList []string, handleIncomingMessage func(IncomingMessage)) {
-
+	if !s.state.CompareAndSwap(StateIdle, StateRunning) {
+		s.logger.Info("Client is not in idle state (already running or destroyed). Ignoring")
+		return
+	}
 	if !s.lifecycle.BeginInitialization() {
 		s.logger.Info("Connection is already in process/running. Ignoring")
+		s.state.Store(StateIdle) // Reset state since lifecycle check failed
 		return
 	}
 
@@ -72,10 +85,27 @@
 	params["v"] = version
 
 	go func() {
-		defer s.lifecycle.ShutdownComplete()
+		defer func() {
+			s.lifecycle.ShutdownComplete()
+			// Reset to idle if not destroyed
+			if s.state.Load() != StateDestroyed {
+				s.state.Store(StateIdle)
+			}
+		}()
+
+		// Early exit if client was destroyed while goroutine was starting
+		if s.state.Load() <= StateIdle {
+			s.logger.Info("Client state is not valid (destroyed or idle). Exiting goroutine")
+			return
+		}
 		if !s.lifecycle.InitializationComplete() {
 			return
 		}
+		// Final check before starting connection - prevent race if Destroy was called
+		if s.state.Load() != StateRunning {
+			s.logger.Info("Client destroyed before connection started. Exiting goroutine")
+			return
+		}
 		firstEventReceived := gtSync.NewAtomicBool(false)
 		out := s.sseClient.Do(params, api.AddMetadataToHeaders(s.metadata, nil, s.clientKey), func(m IncomingMessage) {
 			if firstEventReceived.TestAndSet() && !m.IsError() {
@@ -113,6 +143,8 @@
 
 // StopStreaming stops streaming
 func (s *StreamingClientImpl) StopStreaming() {
+	// Set atomic state to destroyed immediately to prevent new goroutines
+	s.state.Store(StateDestroyed)
 	if !s.lifecycle.BeginShutdown() {
 		s.logger.Info("SSE client wrapper not running. Ignoring")
 		return
@@ -124,5 +156,5 @@
 
 // IsRunning returns true if the client is running
 func (s *StreamingClientImpl) IsRunning() bool {
-	return s.lifecycle.IsRunning()
+	return s.lifecycle.IsRunning() && s.state.Load() == StateRunning
 }