Merge pull request #1050 from kaleido-io/fftm-acks

Handle FFTM new style acks with batchNumber

Author: peterbroadhurst

internal/blockchain/ethereum/ethereum.go

@@ -77,8 +77,9 @@ type queryOutput struct {
 }
 
 type ethWSCommandPayload struct {
-	Type  string `json:"type"`
-	Topic string `json:"topic,omitempty"`
+	Type        string `json:"type"`
+	Topic       string `json:"topic,omitempty"`
+	BatchNumber int64  `json:"batchNumber,omitempty"`
 }
 
 type ethError struct {
@@ -438,7 +439,6 @@ func (e *Ethereum) eventLoop() {
 	defer close(e.closed)
 	l := log.L(e.ctx).WithField("role", "event-loop")
 	ctx := log.WithLogger(e.ctx, l)
-	ack, _ := json.Marshal(map[string]string{"type": "ack", "topic": e.topic})
 	for {
 		select {
 		case <-ctx.Done():
@@ -461,10 +461,32 @@ func (e *Ethereum) eventLoop() {
 			case []interface{}:
 				err = e.handleMessageBatch(ctx, msgTyped)
 				if err == nil {
+					ack, _ := json.Marshal(&ethWSCommandPayload{
+						Type:  "ack",
+						Topic: e.topic,
+					})
 					err = e.wsconn.Send(ctx, ack)
 				}
 			case map[string]interface{}:
-				e.handleReceipt(ctx, fftypes.JSONObject(msgTyped))
+				isBatch := false
+				if batchNumber, ok := msgTyped["batchNumber"].(float64); ok {
+					if events, ok := msgTyped["events"].([]interface{}); ok {
+						// FFTM delivery with a batch number to use in the ack
+						isBatch = true
+						err = e.handleMessageBatch(ctx, events)
+						if err == nil {
+							ack, _ := json.Marshal(&ethWSCommandPayload{
+								Type:        "ack",
+								Topic:       e.topic,
+								BatchNumber: int64(batchNumber),
+							})
+							err = e.wsconn.Send(ctx, ack)
+						}
+					}
+				}
+				if !isBatch {
+					e.handleReceipt(ctx, fftypes.JSONObject(msgTyped))
+				}
 			default:
 				l.Errorf("Message unexpected: %+v", msgTyped)
 				continue

internal/blockchain/ethereum/ethereum_test.go

@@ -162,6 +162,68 @@ func TestInitMissingTopic(t *testing.T) {
 	assert.Regexp(t, "FF10138.*topic", err)
 }
 
+func TestInitAndStartWithEthConnect(t *testing.T) {
+
+	log.SetLevel("trace")
+	e, cancel := newTestEthereum()
+	defer cancel()
+
+	toServer, fromServer, wsURL, done := wsclient.NewTestWSServer(nil)
+	defer done()
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	u, _ := url.Parse(wsURL)
+	u.Scheme = "http"
+	httpURL := u.String()
+
+	httpmock.RegisterResponder("GET", fmt.Sprintf("%s/eventstreams", httpURL),
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{}))
+	httpmock.RegisterResponder("POST", fmt.Sprintf("%s/eventstreams", httpURL),
+		httpmock.NewJsonResponderOrPanic(200, eventStream{ID: "es12345"}))
+
+	resetConf(e)
+	utEthconnectConf.Set(ffresty.HTTPConfigURL, httpURL)
+	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utEthconnectConf.Set(EthconnectConfigInstanceDeprecated, "/instances/0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
+	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
+	utFFTMConf.Set(ffresty.HTTPConfigURL, "http://ethc.example.com:12345")
+
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
+	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics, cmi)
+	assert.NoError(t, err)
+	assert.NotNil(t, e.fftmClient)
+
+	assert.Equal(t, "ethereum", e.Name())
+	assert.Equal(t, core.VerifierTypeEthAddress, e.VerifierType())
+
+	assert.NoError(t, err)
+
+	assert.Equal(t, 2, httpmock.GetTotalCallCount())
+	assert.Equal(t, "es12345", e.streamID)
+	assert.NotNil(t, e.Capabilities())
+
+	err = e.Start()
+	assert.NoError(t, err)
+
+	startupMessage := <-toServer
+	assert.Equal(t, `{"type":"listen","topic":"topic1"}`, startupMessage)
+	startupMessage = <-toServer
+	assert.Equal(t, `{"type":"listenreplies"}`, startupMessage)
+	fromServer <- `[]` // empty batch, will be ignored, but acked
+	reply := <-toServer
+	assert.Equal(t, `{"type":"ack","topic":"topic1"}`, reply)
+
+	// Bad data will be ignored
+	fromServer <- `!json`
+	fromServer <- `{"not": "a reply"}`
+	fromServer <- `42`
+
+}
+
 func TestInitAndStartWithFFTM(t *testing.T) {
 
 	log.SetLevel("trace")
@@ -213,9 +275,9 @@ func TestInitAndStartWithFFTM(t *testing.T) {
 	assert.Equal(t, `{"type":"listen","topic":"topic1"}`, startupMessage)
 	startupMessage = <-toServer
 	assert.Equal(t, `{"type":"listenreplies"}`, startupMessage)
-	fromServer <- `[]` // empty batch, will be ignored, but acked
+	fromServer <- `{"batchNumber":12345,"events":[]}` // empty batch, will be ignored, but acked
 	reply := <-toServer
-	assert.Equal(t, `{"topic":"topic1","type":"ack"}`, reply)
+	assert.Equal(t, `{"type":"ack","topic":"topic1","batchNumber":12345}`, reply)
 
 	// Bad data will be ignored
 	fromServer <- `!json`

manifest.json

@@ -6,8 +6,8 @@
   },
   "evmconnect": {
     "image": "ghcr.io/hyperledger/firefly-evmconnect",
-    "tag": "v1.1.6",
-    "sha": "1fc4bbf662477e96701684446f7b4639bb1cfa1c39ceb2bd0a422877f32578cd"
+    "tag": "v1.1.7",
+    "sha": "a7d1282c75a6a19d399829de1b2868187c41489e55c4ddd2c669131776b7c1f6"
   },
   "fabconnect": {
     "image": "ghcr.io/hyperledger/firefly-fabconnect",
@@ -21,13 +21,13 @@
   },
   "tokens-erc1155": {
     "image": "ghcr.io/hyperledger/firefly-tokens-erc1155",
-    "tag": "v1.1.2",
-    "sha": "552e65b971709f605fb07d98c5a8899f5f567b717c783eeb0282136dc961c6d4"
+    "tag": "v1.1.3",
+    "sha": "48dd255d84cf9ce3682a8b6a3379dd173ecaf48fdf1777383ac40705778ee725"
   },
   "tokens-erc20-erc721": {
     "image": "ghcr.io/hyperledger/firefly-tokens-erc20-erc721",
-    "tag": "v1.1.1",
-    "sha": "2e04f9be8ed0b72ac0754ff8bb9ce52c4e1d79b45716fd2677e5bf1e193697f3"
+    "tag": "v1.1.2",
+    "sha": "e1612497060206137d43bf9bdbc39907f1835a05d6c2dcb20a130c13224c68c9"
   },
   "signer": {
     "image": "ghcr.io/hyperledger/firefly-signer",