Introducing a pluggable error handler (#13)

* Introducing a pluggable error handler

* Allowing the user to provide a FatalErrorHandler with streamingclient.Config

Co-authored-by: Chrusty <>

pkg/streaming-client/config.go

@@ -17,11 +17,12 @@ const (
 
 // Config defines parameters for the client:
 type Config struct {
-	LogLevel      logrus.Level // Logging level (default is "info")
-	SDKKey        string       // SDK key (copied from the UI), in the format "{namedCache}/environmentID/APIKey"
-	ServerAddress string       // FeatureHub API endpoint
-	WaitForData   bool         // New() will block until some data has arrived
-	client        interfaces.Client
+	LogLevel          logrus.Level      // Logging level (default is "info")
+	SDKKey            string            // SDK key (copied from the UI), in the format "{namedCache}/environmentID/APIKey"
+	ServerAddress     string            // FeatureHub API endpoint
+	WaitForData       bool              // New() will block until some data has arrived
+	client            interfaces.Client // A FeatureHub client implementation
+	fatalErrorHandler *ErrorFunc        // A user-provided handler func for fatal asynchronous errors
 }
 
 // NewConfig returns a configured Config:
@@ -43,7 +44,11 @@ func (c *Config) Connect() (*Config, error) {
 	}
 
 	// Start the client:
-	client.Start()
+	if c.fatalErrorHandler != nil {
+		client.WithFatalErrorHandler(*c.fatalErrorHandler).Start()
+	} else {
+		client.Start()
+	}
 
 	c.client = client
 	return c, nil
@@ -95,6 +100,12 @@ func (c *Config) WithContext(context *models.Context) *ClientWithContext {
 	}
 }
 
+// WithFatalErrorHandler configures an error handler which will be called for asynchronous fatal errors:
+func (c *Config) WithFatalErrorHandler(fatalErrorFunc ErrorFunc) *Config {
+	c.fatalErrorHandler = &fatalErrorFunc
+	return c
+}
+
 // WithLogLevel adds a logLevel to the config:
 func (c *Config) WithLogLevel(logLevel logrus.Level) *Config {
 	c.LogLevel = logLevel

pkg/streaming-client/streaming_client.go

@@ -12,12 +12,16 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+// ErrorFunc is called when asynchronous errors are encountered:
+type ErrorFunc func(error, string, map[string]interface{})
+
 // StreamingClient implements the client interface by by subscribing to server-side events:
 type StreamingClient struct {
 	analyticsCollectors []interfaces.AnalyticsCollector
 	analyticsMutex      sync.Mutex
 	apiClient           *eventsource.Stream
 	config              *Config
+	fatalErrorHandler   ErrorFunc
 	features            map[string]*models.FeatureState
 	featuresMutex       sync.Mutex
 	featuresURL         string
@@ -60,6 +64,9 @@ func NewStreamingClient(config *Config) (*StreamingClient, error) {
 		notifiers: make(notifiers),
 	}
 
+	// Use the default fatalErrorFunc to handle fatal errors:
+	client.WithFatalErrorHandler(client.fatalErrorFunc)
+
 	// Report that we're starting:
 	logger.WithField("server_address", client.config.ServerAddress).Info("Subscribing to FeatureHub server")
 
@@ -81,6 +88,11 @@ func NewStreamingClient(config *Config) (*StreamingClient, error) {
 	return client, nil
 }
 
+// FatalErrorFunc is called when an unrecoverable asynchronous error is encountered:
+func (c *StreamingClient) fatalErrorFunc(err error, message string, details map[string]interface{}) {
+	c.logger.WithError(err).WithFields(details).Fatal(message)
+}
+
 // ReadinessListener defines a callback function which will be triggered once the client has received data for the first time:
 func (c *StreamingClient) ReadinessListener(callbackFunc func()) {
 	c.readinessListener = callbackFunc
@@ -110,6 +122,12 @@ func (c *StreamingClient) WithContext(context *models.Context) *ClientWithContex
 	}
 }
 
+// WithFatalErrorHandler configures an error handler which will be called for asynchronous fatal errors:
+func (c *StreamingClient) WithFatalErrorHandler(fatalErrorFunc ErrorFunc) *StreamingClient {
+	c.fatalErrorHandler = fatalErrorFunc
+	return c
+}
+
 // isReady triggers various notifications that the client is ready to serve data:
 func (c *StreamingClient) isReady() {
 

pkg/streaming-client/streaming_client_handlers.go

@@ -42,7 +42,11 @@ func (c *StreamingClient) handleEvents() {
 
 		// Failures (from the FeatureHub server):
 		case models.FHFailure:
-			c.logger.WithError(&errors.ErrFromAPI{}).WithField("event", event.Event()).WithField("message", event.Data()).Fatal("Failure from FeatureHub server")
+			details := map[string]interface{}{
+				"event":   event.Event(),
+				"message": event.Data(),
+			}
+			c.fatalErrorHandler(&errors.ErrFromAPI{}, "Failure from FeatureHub server", details)
 
 		// One specific feature (replaces the previous version):
 		case models.FHFeature:
@@ -64,7 +68,12 @@ func (c *StreamingClient) handleSSEError(event eventsource.Event) {
 	if c.hasData {
 		c.logger.WithError(&errors.ErrFromAPI{}).WithField("event", event.Event()).WithField("message", event.Data()).Error("Error from API client")
 	} else {
-		c.logger.WithError(&errors.ErrFromAPI{}).WithField("event", event.Event()).WithField("message", event.Data()).Fatal("Error from API client")
+		// Use the fatailErrorFunc for this one:
+		details := map[string]interface{}{
+			"event":   event.Event(),
+			"message": event.Data(),
+		}
+		c.fatalErrorFunc(&errors.ErrFromAPI{}, "Error from API client", details)
 	}
 }