Handle FFTM new style acks with batchNumber

Signed-off-by: Peter Broadhurst <[email protected]>

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`