Repository: hyperledger/firefly
Pull request #999 (2022-08-23 20:59:19)

Do not rely on "topic" to be present in event streams


Patch:
diff --git a/cmd/firefly.go b/cmd/firefly.go
index 2edd25b12..a4df2b212 100644
--- a/cmd/firefly.go
+++ b/cmd/firefly.go
@@ -24,7 +24,6 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
-	"time"
 
 	"github.com/gorilla/mux"
 	"github.com/hyperledger/firefly-common/pkg/config"
@@ -119,6 +118,7 @@ func run() error {
 	}
 
 	// Setup signal handling to cancel the context, which shuts down the API Server
+	errChan := make(chan error)
 	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
 
 	for {
@@ -126,9 +126,7 @@ func run() error {
 		rootCtx, rootCancelCtx := context.WithCancel(ctx)
 		mgr := getRootManager()
 		as := apiserver.NewAPIServer()
-		errChan := make(chan error, 1)
-		ffDone := make(chan struct{})
-		go startFirefly(rootCtx, rootCancelCtx, mgr, as, errChan, ffDone)
+		go startFirefly(rootCtx, rootCancelCtx, mgr, as, errChan)
 		select {
 		case sig := <-sigs:
 			log.L(ctx).Infof("Shutting down due to %s", sig.String())
@@ -138,8 +136,6 @@ func run() error {
 		case <-rootCtx.Done():
 			log.L(ctx).Infof("Restarting due to configuration change")
 			mgr.WaitStop()
-			// Must wait for the server to close before we restart
-			<-ffDone
 			// Re-read the configuration
 			resetConfig()
 			if err := config.ReadConfig(configSuffix, cfgFile); err != nil {
@@ -153,10 +149,9 @@ func run() error {
 	}
 }
 
-func startFirefly(ctx context.Context, cancelCtx context.CancelFunc, mgr namespace.Manager, as apiserver.Server, errChan chan error, ffDone chan struct{}) {
+func startFirefly(ctx context.Context, cancelCtx context.CancelFunc, mgr namespace.Manager, as apiserver.Server, errChan chan error) {
 	var err error
 	// Start debug listener
-	var debugServer *http.Server
 	debugPort := config.GetInt(coreconfig.DebugPort)
 	if debugPort >= 0 {
 		r := mux.NewRouter()
@@ -165,20 +160,12 @@ func startFirefly(ctx context.Context, cancelCtx context.CancelFunc, mgr namespa
 		r.PathPrefix("/debug/pprof/symbol").HandlerFunc(pprof.Symbol)
 		r.PathPrefix("/debug/pprof/trace").HandlerFunc(pprof.Trace)
 		r.PathPrefix("/debug/pprof/").HandlerFunc(pprof.Index)
-		debugServer = &http.Server{Addr: fmt.Sprintf("localhost:%d", debugPort), Handler: r, ReadHeaderTimeout: 30 * time.Second}
 		go func() {
-			_ = debugServer.ListenAndServe()
+			_ = http.ListenAndServe(fmt.Sprintf("localhost:%d", debugPort), r)
 		}()
 		log.L(ctx).Debugf("Debug HTTP endpoint listening on localhost:%d", debugPort)
 	}
 
-	defer func() {
-		if debugServer != nil {
-			_ = debugServer.Close()
-		}
-		close(ffDone)
-	}()
-
 	if err = mgr.Init(ctx, cancelCtx); err != nil {
 		errChan <- err
 		return
diff --git a/internal/blockchain/ethereum/ethereum.go b/internal/blockchain/ethereum/ethereum.go
index 77725f605..c9d09e030 100644
--- a/internal/blockchain/ethereum/ethereum.go
+++ b/internal/blockchain/ethereum/ethereum.go
@@ -768,7 +768,7 @@ func (e *Ethereum) GenerateFFI(ctx context.Context, generationRequest *fftypes.F
 	var input FFIGenerationInput
 	err := json.Unmarshal(generationRequest.Input.Bytes(), &input)
 	if err != nil {
-		return nil, i18n.WrapError(ctx, err, coremsgs.MsgFFIGenerationFailed, "unable to deserialize JSON as ABI")
+		return nil, i18n.NewError(ctx, coremsgs.MsgFFIGenerationFailed, "unable to deserialize JSON as ABI")
 	}
 	if len(*input.ABI) == 0 {
 		return nil, i18n.NewError(ctx, coremsgs.MsgFFIGenerationFailed, "ABI is empty")
diff --git a/internal/blockchain/ethereum/eventstream.go b/internal/blockchain/ethereum/eventstream.go
index a5ad9ed0b..5979f3eee 100644
--- a/internal/blockchain/ethereum/eventstream.go
+++ b/internal/blockchain/ethereum/eventstream.go
@@ -78,46 +78,44 @@ func (s *streamManager) getEventStreams(ctx context.Context) (streams []*eventSt
 	return streams, nil
 }
 
-func (s *streamManager) createEventStream(ctx context.Context, topic string, batchSize, batchTimeout uint) (*eventStream, error) {
-	stream := eventStream{
+func buildEventStream(topic string, batchSize, batchTimeout uint) *eventStream {
+	return &eventStream{
 		Name:           topic,
 		ErrorHandling:  "block",
 		BatchSize:      batchSize,
 		BatchTimeoutMS: batchTimeout,
 		Type:           "websocket",
-		WebSocket:      eventStreamWebsocket{Topic: topic},
-		Timestamps:     true,
+		// Some implementations require a "topic" to be set separately, while others rely only on the name.
+		// We set them to the same thing for cross compatibility.
+		WebSocket:  eventStreamWebsocket{Topic: topic},
+		Timestamps: true,
 	}
+}
+
+func (s *streamManager) createEventStream(ctx context.Context, topic string, batchSize, batchTimeout uint) (*eventStream, error) {
+	stream := buildEventStream(topic, batchSize, batchTimeout)
 	res, err := s.client.R().
 		SetContext(ctx).
-		SetBody(&stream).
-		SetResult(&stream).
+		SetBody(stream).
+		SetResult(stream).
 		Post("/eventstreams")
 	if err != nil || !res.IsSuccess() {
 		return nil, ffresty.WrapRestErr(ctx, res, err, coremsgs.MsgEthconnectRESTErr)
 	}
-	return &stream, nil
+	return stream, nil
 }
 
 func (s *streamManager) updateEventStream(ctx context.Context, topic string, batchSize, batchTimeout uint, eventStreamID string) (*eventStream, error) {
-	stream := eventStream{
-		Name:           topic,
-		ErrorHandling:  "block",
-		BatchSize:      batchSize,
-		BatchTimeoutMS: batchTimeout,
-		Type:           "websocket",
-		WebSocket:      eventStreamWebsocket{Topic: topic},
-		Timestamps:     true,
-	}
+	stream := buildEventStream(topic, batchSize, batchTimeout)
 	res, err := s.client.R().
 		SetContext(ctx).
-		SetBody(&stream).
-		SetResult(&stream).
+		SetBody(stream).
+		SetResult(stream).
 		Patch("/eventstreams/" + eventStreamID)
 	if err != nil || !res.IsSuccess() {
 		return nil, ffresty.WrapRestErr(ctx, res, err, coremsgs.MsgEthconnectRESTErr)
 	}
-	return &stream, nil
+	return stream, nil
 }
 
 func (s *streamManager) ensureEventStream(ctx context.Context, topic string, batchSize, batchTimeout uint) (*eventStream, error) {
@@ -126,7 +124,7 @@ func (s *streamManager) ensureEventStream(ctx context.Context, topic string, bat
 		return nil, err
 	}
 	for _, stream := range existingStreams {
-		if stream.WebSocket.Topic == topic {
+		if stream.Name == topic {
 			stream, err = s.updateEventStream(ctx, topic, batchSize, batchTimeout, stream.ID)
 			if err != nil {
 				return nil, err
diff --git a/internal/blockchain/fabric/eventstream.go b/internal/blockchain/fabric/eventstream.go
index f26028fa8..8b883ec40 100644
--- a/internal/blockchain/fabric/eventstream.go
+++ b/internal/blockchain/fabric/eventstream.go
@@ -83,25 +83,31 @@ func (s *streamManager) getEventStreams(ctx context.Context) (streams []*eventSt
 	return streams, nil
 }
 
-func (s *streamManager) createEventStream(ctx context.Context, topic string, batchSize, batchTimeout uint) (*eventStream, error) {
-	stream := eventStream{
+func buildEventStream(topic string, batchSize, batchTimeout uint) *eventStream {
+	return &eventStream{
 		Name:           topic,
 		ErrorHandling:  "block",
 		BatchSize:      batchSize,
 		BatchTimeoutMS: batchTimeout,
 		Type:           "websocket",
-		WebSocket:      eventStreamWebsocket{Topic: topic},
-		Timestamps:     true,
+		// Some implementations require a "topic" to be set separately, while others rely only on the name.
+		// We set them to the same thing for cross compatibility.
+		WebSocket:  eventStreamWebsocket{Topic: topic},
+		Timestamps: true,
 	}
+}
+
+func (s *streamManager) createEventStream(ctx context.Context, topic string, batchSize, batchTimeout uint) (*eventStream, error) {
+	stream := buildEventStream(topic, batchSize, batchTimeout)
 	res, err := s.client.R().
 		SetContext(ctx).
-		SetBody(&stream).
-		SetResult(&stream).
+		SetBody(stream).
+		SetResult(stream).
 		Post("/eventstreams")
 	if err != nil || !res.IsSuccess() {
 		return nil, ffresty.WrapRestErr(ctx, res, err, coremsgs.MsgFabconnectRESTErr)
 	}
-	return &stream, nil
+	return stream, nil
 }
 
 func (s *streamManager) ensureEventStream(ctx context.Context, topic string, batchSize, batchTimeout uint) (*eventStream, error) {
@@ -110,7 +116,7 @@ func (s *streamManager) ensureEventStream(ctx context.Context, topic string, bat
 		return nil, err
 	}
 	for _, stream := range existingStreams {
-		if stream.WebSocket.Topic == topic {
+		if stream.Name == topic {
 			return stream, nil
 		}
 	}


Test patch:
diff --git a/cmd/firefly_test.go b/cmd/firefly_test.go
index 363005621..167fa389f 100644
--- a/cmd/firefly_test.go
+++ b/cmd/firefly_test.go
@@ -102,9 +102,8 @@ func TestExecOkRestartThenExit(t *testing.T) {
 		initCount++
 		if initCount == 2 {
 			init.ReturnArguments = mock.Arguments{fmt.Errorf("second run")}
-		} else {
-			cancelOrContext()
 		}
+		cancelOrContext()
 	}
 	o.On("Start").Return(nil)
 	ws := o.On("WaitStop")
@@ -152,7 +151,7 @@ func TestAPIServerError(t *testing.T) {
 	as.On("Serve", mock.Anything, o).Return(fmt.Errorf("pop"))
 
 	errChan := make(chan error)
-	go startFirefly(context.Background(), func() {}, o, as, errChan, make(chan struct{}))
+	go startFirefly(context.Background(), func() {}, o, as, errChan)
 	err := <-errChan
 	assert.EqualError(t, err, "pop")
 }
diff --git a/internal/blockchain/ethereum/ethereum_test.go b/internal/blockchain/ethereum/ethereum_test.go
index 6aeba046f..274303c1a 100644
--- a/internal/blockchain/ethereum/ethereum_test.go
+++ b/internal/blockchain/ethereum/ethereum_test.go
@@ -444,7 +444,7 @@ func TestStreamUpdateError(t *testing.T) {
 	defer httpmock.DeactivateAndReset()
 
 	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
-		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", Name: "topic1"}}))
 	httpmock.RegisterResponder("PATCH", "http://localhost:12345/eventstreams/es12345",
 		httpmock.NewStringResponder(500, `pop`))
 
@@ -468,13 +468,13 @@ func TestInitAllExistingStreams(t *testing.T) {
 	defer httpmock.DeactivateAndReset()
 
 	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
-		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", Name: "topic1"}}))
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, []subscription{
 			{ID: "sub12345", Stream: "es12345", Name: "ns1_BatchPin_3078373143373635" /* this is the subname for our combo of instance path and BatchPin */},
 		}))
 	httpmock.RegisterResponder("PATCH", "http://localhost:12345/eventstreams/es12345",
-		httpmock.NewJsonResponderOrPanic(200, &eventStream{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}))
+		httpmock.NewJsonResponderOrPanic(200, &eventStream{ID: "es12345", Name: "topic1"}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 2))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, subscription{}))
@@ -508,13 +508,13 @@ func TestInitAllExistingStreamsV1(t *testing.T) {
 	defer httpmock.DeactivateAndReset()
 
 	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
-		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", Name: "topic1"}}))
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, []subscription{
 			{ID: "sub12345", Stream: "es12345", Name: "BatchPin_3078373143373635" /* this is the subname for our combo of instance path and BatchPin */},
 		}))
 	httpmock.RegisterResponder("PATCH", "http://localhost:12345/eventstreams/es12345",
-		httpmock.NewJsonResponderOrPanic(200, &eventStream{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}))
+		httpmock.NewJsonResponderOrPanic(200, &eventStream{ID: "es12345", Name: "topic1"}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 1))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, subscription{}))
@@ -548,13 +548,13 @@ func TestInitAllExistingStreamsOld(t *testing.T) {
 	defer httpmock.DeactivateAndReset()
 
 	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
-		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", Name: "topic1"}}))
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, []subscription{
 			{ID: "sub12345", Stream: "es12345", Name: "BatchPin"},
 		}))
 	httpmock.RegisterResponder("PATCH", "http://localhost:12345/eventstreams/es12345",
-		httpmock.NewJsonResponderOrPanic(200, &eventStream{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}))
+		httpmock.NewJsonResponderOrPanic(200, &eventStream{ID: "es12345", Name: "topic1"}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 1))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, subscription{}))
@@ -588,13 +588,13 @@ func TestInitAllExistingStreamsInvalidName(t *testing.T) {
 	defer httpmock.DeactivateAndReset()
 
 	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
-		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", Name: "topic1"}}))
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, []subscription{
 			{ID: "sub12345", Stream: "es12345", Name: "BatchPin_3078373143373635" /* this is the subname for our combo of instance path and BatchPin */},
 		}))
 	httpmock.RegisterResponder("PATCH", "http://localhost:12345/eventstreams/es12345",
-		httpmock.NewJsonResponderOrPanic(200, &eventStream{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}))
+		httpmock.NewJsonResponderOrPanic(200, &eventStream{ID: "es12345", Name: "topic1"}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 2))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, subscription{}))
diff --git a/internal/blockchain/fabric/fabric_test.go b/internal/blockchain/fabric/fabric_test.go
index 1e88fda77..a52df957c 100644
--- a/internal/blockchain/fabric/fabric_test.go
+++ b/internal/blockchain/fabric/fabric_test.go
@@ -225,7 +225,7 @@ func TestInitAllExistingStreams(t *testing.T) {
 	defer httpmock.DeactivateAndReset()
 
 	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
-		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", Name: "topic1"}}))
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, []subscription{
 			{ID: "sub12345", Stream: "es12345", Name: "ns1_BatchPin"},
@@ -264,7 +264,7 @@ func TestInitAllExistingStreamsV1(t *testing.T) {
 	defer httpmock.DeactivateAndReset()
 
 	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
-		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", Name: "topic1"}}))
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, []subscription{
 			{ID: "sub12345", Stream: "es12345", Name: "BatchPin"},
@@ -300,7 +300,7 @@ func TestAddFireflySubscriptionQuerySubsFail(t *testing.T) {
 	resetConf(e)
 
 	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
-		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", Name: "topic1"}}))
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(500, "pop"))
 	httpmock.RegisterResponder("POST", fmt.Sprintf("http://localhost:12345/query"),
@@ -334,7 +334,7 @@ func TestAddFireflySubscriptionGetVersionError(t *testing.T) {
 	resetConf(e)
 
 	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
-		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", Name: "topic1"}}))
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, []subscription{
 			{ID: "sub12345", Stream: "es12345", Name: "ns1_BatchPin"},
@@ -373,7 +373,7 @@ func TestAddAndRemoveFireflySubscriptionDeprecatedSubName(t *testing.T) {
 	defer httpmock.DeactivateAndReset()
 
 	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
-		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", Name: "topic1"}}))
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, []subscription{
 			{ID: "sub12345", Stream: "es12345", Name: "BatchPin"},
@@ -416,7 +416,7 @@ func TestAddFireflySubscriptionInvalidSubName(t *testing.T) {
 	defer httpmock.DeactivateAndReset()
 
 	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
-		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", Name: "topic1"}}))
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, []subscription{
 			{ID: "sub12345", Stream: "es12345", Name: "BatchPin"},
@@ -462,7 +462,7 @@ func TestInitNewConfig(t *testing.T) {
 	defer httpmock.DeactivateAndReset()
 
 	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
-		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", Name: "topic1"}}))
 
 	resetConf(e)
 	utFabconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")