FFM-2096 Add Interface for hooking into SSE Events#

Author: jcox250

client/client.go

@@ -184,12 +184,12 @@ func (c *CfClient) streamConnect() {
 		defer c.mux.RUnlock()
 		c.streamConnected = false
 	}
-	conn := stream.NewSSEClient(c.sdkKey, c.token, sseClient, c.config.Cache, c.api, c.config.Logger, streamErr)
+	conn := stream.NewSSEClient(c.sdkKey, c.token, sseClient, c.config.Cache, c.api, c.config.Logger, streamErr, c.config.eventStreamListener)
 
 	// Connect kicks off a goroutine that attempts to establish a stream connection
 	// while this is happening we set streamConnected to true - if any errors happen
 	// in this process streamConnected will be set back to false by the streamErr function
-	conn.Connect(c.environmentID)
+	conn.Connect(c.environmentID, c.sdkKey)
 	c.streamConnected = true
 }
 

client/config.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 
 	"github.com/harness/ff-golang-server-sdk/evaluation"
+	"github.com/harness/ff-golang-server-sdk/stream"
 
 	"github.com/harness/ff-golang-server-sdk/cache"
 	"github.com/harness/ff-golang-server-sdk/logger"
@@ -13,16 +14,17 @@ import (
 )
 
 type config struct {
-	url          string
-	eventsURL    string
-	pullInterval uint // in minutes
-	Cache        cache.Cache
-	Store        storage.Storage
-	Logger       logger.Logger
-	httpClient   *http.Client
-	enableStream bool
-	enableStore  bool
-	target       evaluation.Target
+	url                 string
+	eventsURL           string
+	pullInterval        uint // in minutes
+	Cache               cache.Cache
+	Store               storage.Storage
+	Logger              logger.Logger
+	httpClient          *http.Client
+	enableStream        bool
+	enableStore         bool
+	target              evaluation.Target
+	eventStreamListener stream.EventStreamListener
 }
 
 func newDefaultConfig() *config {

client/options.go

@@ -7,6 +7,7 @@ import (
 	"github.com/harness/ff-golang-server-sdk/evaluation"
 	"github.com/harness/ff-golang-server-sdk/logger"
 	"github.com/harness/ff-golang-server-sdk/storage"
+	"github.com/harness/ff-golang-server-sdk/stream"
 )
 
 // ConfigOption is used as return value for advanced client configuration
@@ -88,3 +89,11 @@ func WithTarget(target evaluation.Target) ConfigOption {
 		config.target = target
 	}
 }
+
+// WithEventStreamListener configures the SDK to forward Events from the Feature
+// Flag server to the passed EventStreamListener
+func WithEventStreamListener(cs stream.EventStreamListener) ConfigOption {
+	return func(config *config) {
+		config.eventStreamListener = cs
+	}
+}

stream/sse.go

@@ -3,6 +3,7 @@ package stream
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/harness/ff-golang-server-sdk/cache"
@@ -17,11 +18,12 @@ import (
 
 // SSEClient is Server Send Event object
 type SSEClient struct {
-	api           rest.ClientWithResponsesInterface
-	client        *sse.Client
-	cache         cache.Cache
-	logger        logger.Logger
-	onStreamError func()
+	api                 rest.ClientWithResponsesInterface
+	client              *sse.Client
+	cache               cache.Cache
+	logger              logger.Logger
+	onStreamError       func()
+	eventStreamListener EventStreamListener
 }
 
 var json = jsoniter.ConfigCompatibleWithStandardLibrary
@@ -35,24 +37,26 @@ func NewSSEClient(
 	api rest.ClientWithResponsesInterface,
 	logger logger.Logger,
 	onStreamError func(),
+	eventStreamListener EventStreamListener,
 ) *SSEClient {
 	client.Headers["Authorization"] = fmt.Sprintf("Bearer %s", token)
 	client.Headers["API-Key"] = apiKey
 	client.OnDisconnect(func(client *sse.Client) {
 		onStreamError()
 	})
 	sseClient := &SSEClient{
-		client:        client,
-		cache:         cache,
-		api:           api,
-		logger:        logger,
-		onStreamError: onStreamError,
+		client:              client,
+		cache:               cache,
+		api:                 api,
+		logger:              logger,
+		onStreamError:       onStreamError,
+		eventStreamListener: eventStreamListener,
 	}
 	return sseClient
 }
 
 // Connect will subscribe to SSE stream
-func (c *SSEClient) Connect(environment string) {
+func (c *SSEClient) Connect(environment string, apiKey string) {
 	c.logger.Infof("Start subscribing to Stream")
 	// don't use the default exponentialBackoff strategy - we have our own disconnect logic
 	// of polling the service then re-establishing a new stream once we can connect
@@ -62,6 +66,8 @@ func (c *SSEClient) Connect(environment string) {
 		err := c.client.Subscribe("*", func(msg *sse.Event) {
 			c.logger.Infof("Event received: %s", msg.Data)
 
+			wg := &sync.WaitGroup{}
+
 			cfMsg := Message{}
 			if len(msg.Data) > 0 {
 				err := json.Unmarshal(msg.Data, &cfMsg)
@@ -76,16 +82,25 @@ func (c *SSEClient) Connect(environment string) {
 					// and subscribe to that event
 					switch cfMsg.Event {
 					case dto.SseDeleteEvent:
+						wg.Add(1)
+
 						go func(identifier string) {
+							defer wg.Done()
+
 							c.cache.Remove(dto.Key{
 								Type: dto.KeyFeature,
 								Name: identifier,
 							})
 						}(cfMsg.Identifier)
+
 					case dto.SsePatchEvent, dto.SseCreateEvent:
 						fallthrough
 					default:
+						wg.Add(1)
+
 						go func(env, identifier string) {
+							defer wg.Done()
+
 							ctx, cancel := context.WithTimeout(context.Background(), time.Second*60)
 							defer cancel()
 							response, err := c.api.GetFeatureConfigByIdentifierWithResponse(ctx, env, identifier)
@@ -101,20 +116,30 @@ func (c *SSEClient) Connect(environment string) {
 							}
 						}(environment, cfMsg.Identifier)
 					}
+
 				case dto.KeySegment:
 					// need open client spec change
 					switch cfMsg.Event {
 					case dto.SseDeleteEvent:
+						wg.Add(1)
+
 						go func(identifier string) {
+							defer wg.Done()
+
 							c.cache.Remove(dto.Key{
 								Type: dto.KeySegment,
 								Name: identifier,
 							})
 						}(cfMsg.Identifier)
+
 					case dto.SsePatchEvent, dto.SseCreateEvent:
 						fallthrough
 					default:
+						wg.Add(1)
+
 						go func(env, identifier string) {
+							defer wg.Done()
+
 							ctx, cancel := context.WithTimeout(context.Background(), time.Second*60)
 							defer cancel()
 							response, err := c.api.GetSegmentByIdentifierWithResponse(ctx, env, identifier)
@@ -131,6 +156,19 @@ func (c *SSEClient) Connect(environment string) {
 						}(environment, cfMsg.Identifier)
 					}
 				}
+
+				if c.eventStreamListener != nil {
+					sendWithTimeout := func() error {
+						ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+						defer cancel()
+						return c.eventStreamListener.Pub(ctx, Event{APIKey: apiKey, Environment: environment, Event: msg})
+					}
+
+					wg.Wait()
+					if err := sendWithTimeout(); err != nil {
+						c.logger.Errorf("error while forwarding SSE Event to change stream: %s", err)
+					}
+				}
 			}
 		})
 		if err != nil {

stream/stream.go

@@ -1,7 +1,30 @@
 package stream
 
+import (
+	"context"
+
+	"github.com/r3labs/sse"
+)
+
 // Connection is simple interface for streams
 type Connection interface {
 	Connect(environment string) error
 	OnDisconnect(func() error) error
 }
+
+// EventStreamListener provides a way to hook in to the SSE Events that the SDK
+// recieves from the FeatureFlags server and forward them on to another type.
+type EventStreamListener interface {
+	// Pub publishes an event from the SDK to your Listener
+	Pub(ctx context.Context, event Event) error
+}
+
+// Event defines the structure of an event that gets sent to a EventStreamListener
+type Event struct {
+	// APIKey is the SDKs API Key
+	APIKey string
+	// Environment is the ID of the environment that the event occured for
+	Environment string
+	// Event is the SSEEvent that was sent from the FeatureFlags server to the SDK
+	Event *sse.Event
+}