Merge pull request #20 from kaleido-io/eventstreams-fix

Fix event stream topic selection

Author: peterbroadhurst

internal/events/eventstream.go

@@ -59,7 +59,6 @@ var esDefaults struct {
 	blockedRetryDelay         fftypes.FFDuration
 	webhookRequestTimeout     fftypes.FFDuration
 	websocketDistributionMode apitypes.DistributionMode
-	topic                     string
 	retry                     *retry.Retry
 }
 
@@ -164,7 +163,7 @@ func (es *eventStream) initAction(startedState *startedStreamState) {
 	case apitypes.EventStreamTypeWebhook:
 		startedState.action = newWebhookAction(ctx, es.spec.Webhook).attemptBatch
 	case apitypes.EventStreamTypeWebSocket:
-		startedState.action = newWebSocketAction(es.wsChannels, es.spec.WebSocket, *es.spec.Name).attemptBatch
+		startedState.action = newWebSocketAction(es.wsChannels, es.spec.WebSocket, *es.spec.WebSocket.Topic).attemptBatch
 	default:
 		// mergeValidateEsConfig always be called previous to this
 		panic(i18n.NewError(ctx, tmmsgs.MsgInvalidStreamType, *es.spec.Type))
@@ -237,7 +236,7 @@ func mergeValidateEsConfig(ctx context.Context, base *apitypes.EventStream, upda
 	changed = apitypes.CheckUpdateEnum(changed, &merged.Type, base.Type, updates.Type, apitypes.EventStreamTypeWebSocket)
 	switch *merged.Type {
 	case apitypes.EventStreamTypeWebSocket:
-		if merged.WebSocket, changed, err = mergeValidateWsConfig(ctx, changed, base.WebSocket, updates.WebSocket); err != nil {
+		if merged.WebSocket, changed, err = mergeValidateWsConfig(ctx, changed, *merged.Name, base.WebSocket, updates.WebSocket); err != nil {
 			return nil, false, err
 		}
 	case apitypes.EventStreamTypeWebhook:

internal/events/eventstream_test.go

@@ -135,7 +135,10 @@ func TestConfigNewDefaultsUpdate(t *testing.T) {
 	InitDefaults()
 
 	es := testESConf(t, `{
-		"name": "test1"
+		"name":  "test1",
+		"websocket": {
+			"topic": "test1"
+		}
 	}`)
 	es, changed, err := mergeValidateEsConfig(context.Background(), nil, es)
 	assert.NoError(t, err)
@@ -157,7 +160,7 @@ func TestConfigNewDefaultsUpdate(t *testing.T) {
 		"type":"websocket",
 		"websocket": {
 			"distributionMode":"load_balance",
-			"topic":""
+			"topic":"test1"
 		}
 	}`, string(b))
 
@@ -297,7 +300,10 @@ func TestInitActionBadAction(t *testing.T) {
 func TestWebSocketEventStreamsE2EMigrationThenStart(t *testing.T) {
 
 	es := newTestEventStream(t, `{
-		"name": "ut_stream"
+		"name":  "ut_stream",
+		"websocket": {
+			"topic": "ut_stream"
+		}
 	}`)
 
 	addr := "0x12345"
@@ -1198,7 +1204,8 @@ func TestWebSocketBroadcastActionCloseDuringCheckpoint(t *testing.T) {
 	es := newTestEventStream(t, `{
 		"name": "ut_stream",
 		"websocket": {
-			"distributionMode": "broadcast"
+			"distributionMode": "broadcast",
+			"topic": "ut_stream"
 		}
 	}`)
 

internal/events/websockets.go

@@ -26,7 +26,7 @@ import (
 	"github.com/hyperledger/firefly-transaction-manager/pkg/apitypes"
 )
 
-func mergeValidateWsConfig(ctx context.Context, changed bool, base *apitypes.WebSocketConfig, updates *apitypes.WebSocketConfig) (*apitypes.WebSocketConfig, bool, error) {
+func mergeValidateWsConfig(ctx context.Context, changed bool, esName string, base *apitypes.WebSocketConfig, updates *apitypes.WebSocketConfig) (*apitypes.WebSocketConfig, bool, error) {
 
 	if base == nil {
 		base = &apitypes.WebSocketConfig{}
@@ -48,7 +48,7 @@ func mergeValidateWsConfig(ctx context.Context, changed bool, base *apitypes.Web
 	}
 
 	// Topic
-	changed = apitypes.CheckUpdateString(changed, &merged.Topic, base.Topic, updates.Topic, esDefaults.topic)
+	changed = apitypes.CheckUpdateString(changed, &merged.Topic, base.Topic, updates.Topic, esName /* default to the ES name */)
 
 	return merged, changed, nil
 }