Move DX events interface to async with callback

Makefile

@@ -43,6 +43,7 @@ $(eval $(call makemock, pkg/events,                Callbacks,          eventsmoc
 $(eval $(call makemock, pkg/identity,              Plugin,             identitymocks))
 $(eval $(call makemock, pkg/identity,              Callbacks,          identitymocks))
 $(eval $(call makemock, pkg/dataexchange,          Plugin,             dataexchangemocks))
+$(eval $(call makemock, pkg/dataexchange,          DXEvent,            dataexchangemocks))
 $(eval $(call makemock, pkg/dataexchange,          Callbacks,          dataexchangemocks))
 $(eval $(call makemock, pkg/tokens,                Plugin,             tokenmocks))
 $(eval $(call makemock, pkg/tokens,                Callbacks,          tokenmocks))

docs/architecture/ping_pong_txflow.md

@@ -50,7 +50,7 @@ This is deliberately a simple flow, and all kinds of additional layers might wel
 
 ## Broadcast Public Description of Binary Data Asset (Member 1)
 
-- Upload BLOB of the actual data
+- Upload Blob of the actual data
   - Returns a hash of the payload
 - Upload JSON containing the public index data 
   - Includes the hash of the full payload

docs/gettingstarted/broadcast_data.md

@@ -24,7 +24,7 @@ nav_order: 4
   - Can be sent in-line, uploaded in advanced, or received from other parties
   - Can include smaller JSON payloads suitable for database storage
     - These can be verified against a `datatype`
-  - Can include references to large (multi megabyte/gigabyte) BLOB data
+  - Can include references to large (multi megabyte/gigabyte) Blob data
 - Sequenced via the blockchain
   - The blockchain does not contain any data, just a hash pin
 - Batched for efficiency
@@ -125,7 +125,7 @@ Here we make two API calls.
 
 2. Broadcast a message referring to that data
 
-- The BLOB attachment gets published to shared storage
+- The Blob attachment gets published to shared storage
   - This happens the first time a broadcast happens on a data attachment
 - A pin goes to the blockchain
 - The metadata goes into a batch with the message
@@ -144,7 +144,7 @@ curl -sLo - https://github.com/hyperledger/firefly/raw/main/docs/firefly_logo.pn
    http://localhost:5000/api/v1/namespaces/default/data
 ```
 
-### Example data response from BLOB upload
+### Example data response from Blob upload
 
 Status: `200 OK` - your data is uploaded to your local FireFly node
 

docs/gettingstarted/private_send.md

@@ -24,7 +24,7 @@ nav_order: 4
   - Can be sent in-line, uploaded in advanced, or received from other parties
   - Can include smaller JSON payloads suitable for database storage
     - These can be verified against a `datatype`
-  - Can include references to large (multi megabyte/gigabyte) BLOB data
+  - Can include references to large (multi megabyte/gigabyte) Blob data
 - A `group` specifies who has visibility to the data
   - The author must be included in the group - auto-added if omitted
   - Can be specified in-line in the message by listing recipients directly
@@ -210,7 +210,7 @@ Here we make two API calls.
 
 2. Privately send a message referring to that data
 
-- The BLOB is sent privately to each party
+- The Blob is sent privately to each party
 - A pin goes to the blockchain
 - The metadata goes into a batch with the message
 
@@ -228,7 +228,7 @@ curl -sLo - https://github.com/hyperledger/firefly/raw/main/docs/firefly_logo.pn
    http://localhost:5000/api/v1/namespaces/default/data
 ```
 
-### Example data response from BLOB upload
+### Example data response from Blob upload
 
 Status: `200 OK` - your data is uploaded to your local FireFly node
 

docs/gettingstarted/query_messages.md

@@ -34,7 +34,7 @@ This builds on the APIs to query and filter messages, described below
 ### Example 1: Query confirmed messages
 
 These are the messages ready to be processed in your application.
-All data associated with the message (including BLOB attachments) is available,
+All data associated with the message (including Blob attachments) is available,
 and if they are sequenced by the blockchain, then those blockchain transactions
 are complete.
 

internal/apiserver/route_get_data_blob.go

@@ -42,7 +42,7 @@ var getDataBlob = &oapispec.Route{
 	JSONOutputValue: func() interface{} { return []byte{} },
 	JSONOutputCodes: []int{http.StatusOK},
 	JSONHandler: func(r *oapispec.APIRequest) (output interface{}, err error) {
-		blob, reader, err := getOr(r.Ctx).Data().DownloadBLOB(r.Ctx, r.PP["ns"], r.PP["dataid"])
+		blob, reader, err := getOr(r.Ctx).Data().DownloadBlob(r.Ctx, r.PP["ns"], r.PP["dataid"])
 		if err == nil {
 			r.ResponseHeaders.Set(fftypes.HTTPHeadersBlobHashSHA256, blob.Hash.String())
 			if blob.Size > 0 {

internal/apiserver/route_get_data_blob_test.go

@@ -37,7 +37,7 @@ func TestGetDataBlob(t *testing.T) {
 	res := httptest.NewRecorder()
 
 	blobHash := fftypes.NewRandB32()
-	mdm.On("DownloadBLOB", mock.Anything, "mynamespace", "abcd1234").
+	mdm.On("DownloadBlob", mock.Anything, "mynamespace", "abcd1234").
 		Return(&fftypes.Blob{
 			Hash: blobHash,
 			Size: 12345,

internal/apiserver/route_post_data.go

@@ -74,7 +74,7 @@ var postData = &oapispec.Route{
 			}
 			data.Value = fftypes.JSONAnyPtr(metadata)
 		}
-		output, err = getOr(r.Ctx).Data().UploadBLOB(r.Ctx, r.PP["ns"], data, r.Part, strings.EqualFold(r.FP["autometa"], "true"))
+		output, err = getOr(r.Ctx).Data().UploadBlob(r.Ctx, r.PP["ns"], data, r.Part, strings.EqualFold(r.FP["autometa"], "true"))
 		return output, err
 	},
 }

internal/apiserver/route_post_data_test.go

@@ -67,7 +67,7 @@ func TestPostDataBinary(t *testing.T) {
 
 	res := httptest.NewRecorder()
 
-	mdm.On("UploadBLOB", mock.Anything, "ns1", mock.AnythingOfType("*fftypes.DataRefOrValue"), mock.AnythingOfType("*fftypes.Multipart"), false).
+	mdm.On("UploadBlob", mock.Anything, "ns1", mock.AnythingOfType("*fftypes.DataRefOrValue"), mock.AnythingOfType("*fftypes.Multipart"), false).
 		Return(&fftypes.Data{}, nil)
 	r.ServeHTTP(res, req)
 
@@ -107,7 +107,7 @@ func TestPostDataBinaryObjAutoMeta(t *testing.T) {
 
 	res := httptest.NewRecorder()
 
-	mdm.On("UploadBLOB", mock.Anything, "ns1", mock.MatchedBy(func(d *fftypes.DataRefOrValue) bool {
+	mdm.On("UploadBlob", mock.Anything, "ns1", mock.MatchedBy(func(d *fftypes.DataRefOrValue) bool {
 		assert.Equal(t, `{"filename":"anything"}`, string(*d.Value))
 		assert.Equal(t, fftypes.ValidatorTypeJSON, d.Validator)
 		assert.Equal(t, "fileinfo", d.Datatype.Name)
@@ -141,7 +141,7 @@ func TestPostDataBinaryStringMetadata(t *testing.T) {
 
 	res := httptest.NewRecorder()
 
-	mdm.On("UploadBLOB", mock.Anything, "ns1", mock.MatchedBy(func(d *fftypes.DataRefOrValue) bool {
+	mdm.On("UploadBlob", mock.Anything, "ns1", mock.MatchedBy(func(d *fftypes.DataRefOrValue) bool {
 		assert.Equal(t, `"string metadata"`, string(*d.Value))
 		assert.Equal(t, "", string(d.Validator))
 		assert.Nil(t, d.Datatype)
@@ -174,7 +174,7 @@ func TestPostDataTrailingMetadata(t *testing.T) {
 
 	res := httptest.NewRecorder()
 
-	mdm.On("UploadBLOB", mock.Anything, "ns1", mock.Anything, mock.AnythingOfType("*fftypes.Multipart"), false).
+	mdm.On("UploadBlob", mock.Anything, "ns1", mock.Anything, mock.AnythingOfType("*fftypes.Multipart"), false).
 		Return(&fftypes.Data{}, nil)
 	r.ServeHTTP(res, req)
 
@@ -201,7 +201,7 @@ func TestPostDataBinaryMissing(t *testing.T) {
 
 	res := httptest.NewRecorder()
 
-	mdm.On("UploadBLOB", mock.Anything, "ns1", mock.AnythingOfType("*fftypes.DataRefOrValue"), mock.AnythingOfType("*fftypes.Multipart"), false).
+	mdm.On("UploadBlob", mock.Anything, "ns1", mock.AnythingOfType("*fftypes.DataRefOrValue"), mock.AnythingOfType("*fftypes.Multipart"), false).
 		Return(&fftypes.Data{}, nil)
 	r.ServeHTTP(res, req)
 
@@ -221,7 +221,7 @@ func TestPostDataBadForm(t *testing.T) {
 
 	res := httptest.NewRecorder()
 
-	mdm.On("UploadBLOB", mock.Anything, "ns1", mock.AnythingOfType("*fftypes.DataRefOrValue"), mock.AnythingOfType("*fftypes.Multipart"), false).
+	mdm.On("UploadBlob", mock.Anything, "ns1", mock.AnythingOfType("*fftypes.DataRefOrValue"), mock.AnythingOfType("*fftypes.Multipart"), false).
 		Return(&fftypes.Data{}, nil)
 	r.ServeHTTP(res, req)
 

internal/blockchain/ethereum/ethereum.go

@@ -389,7 +389,7 @@ func (e *Ethereum) handleContractEvent(ctx context.Context, msgJSON fftypes.JSON
 	return e.callbacks.BlockchainEvent(event)
 }
 
-func (e *Ethereum) handleReceipt(ctx context.Context, reply fftypes.JSONObject) error {
+func (e *Ethereum) handleReceipt(ctx context.Context, reply fftypes.JSONObject) {
 	l := log.L(ctx)
 
 	headers := reply.GetObject("headers")
@@ -399,19 +399,19 @@ func (e *Ethereum) handleReceipt(ctx context.Context, reply fftypes.JSONObject)
 	message := reply.GetString("errorMessage")
 	if requestID == "" || replyType == "" {
 		l.Errorf("Reply cannot be processed - missing fields: %+v", reply)
-		return nil // Swallow this and move on
+		return
 	}
 	operationID, err := fftypes.ParseUUID(ctx, requestID)
 	if err != nil {
 		l.Errorf("Reply cannot be processed - bad ID: %+v", reply)
-		return nil // Swallow this and move on
+		return
 	}
 	updateType := fftypes.OpStatusSucceeded
 	if replyType != "TransactionSuccess" {
 		updateType = fftypes.OpStatusFailed
 	}
 	l.Infof("Ethconnect '%s' reply: request=%s tx=%s message=%s", replyType, requestID, txHash, message)
-	return e.callbacks.BlockchainOpUpdate(e, operationID, updateType, txHash, message, reply)
+	e.callbacks.BlockchainOpUpdate(e, operationID, updateType, txHash, message, reply)
 }
 
 func (e *Ethereum) buildEventLocationString(msgJSON fftypes.JSONObject) string {
@@ -483,7 +483,7 @@ func (e *Ethereum) eventLoop() {
 					err = e.wsconn.Send(ctx, ack)
 				}
 			case map[string]interface{}:
-				err = e.handleReceipt(ctx, fftypes.JSONObject(msgTyped))
+				e.handleReceipt(ctx, fftypes.JSONObject(msgTyped))
 			default:
 				l.Errorf("Message unexpected: %+v", msgTyped)
 				continue

internal/blockchain/ethereum/ethereum_test.go

@@ -1067,6 +1067,7 @@ func TestEventLoopContextCancelled(t *testing.T) {
 	wsm.On("Close").Return()
 	e.closed = make(chan struct{})
 	e.eventLoop() // we're simply looking for it exiting
+	wsm.AssertExpectations(t)
 }
 
 func TestEventLoopReceiveClosed(t *testing.T) {
@@ -1079,19 +1080,24 @@ func TestEventLoopReceiveClosed(t *testing.T) {
 	wsm.On("Close").Return()
 	e.closed = make(chan struct{})
 	e.eventLoop() // we're simply looking for it exiting
+	wsm.AssertExpectations(t)
 }
 
 func TestEventLoopSendClosed(t *testing.T) {
 	e, cancel := newTestEthereum()
-	cancel()
+	s := make(chan []byte, 1)
+	s <- []byte(`[]`)
 	r := make(chan []byte)
 	wsm := e.wsconn.(*wsmocks.WSClient)
-	close(r)
-	wsm.On("Receive").Return((<-chan []byte)(r))
-	wsm.On("Send", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+	wsm.On("Receive").Return((<-chan []byte)(s))
+	wsm.On("Send", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
+		go cancel()
+		close(r)
+	}).Return(fmt.Errorf("pop"))
 	wsm.On("Close").Return()
 	e.closed = make(chan struct{})
 	e.eventLoop() // we're simply looking for it exiting
+	wsm.AssertExpectations(t)
 }
 
 func TestHandleReceiptTXSuccess(t *testing.T) {
@@ -1139,8 +1145,7 @@ func TestHandleReceiptTXSuccess(t *testing.T) {
 
 	err := json.Unmarshal(data.Bytes(), &reply)
 	assert.NoError(t, err)
-	err = e.handleReceipt(context.Background(), reply)
-	assert.NoError(t, err)
+	e.handleReceipt(context.Background(), reply)
 
 }
 
@@ -1189,7 +1194,7 @@ func TestHandleBadPayloadsAndThenReceiptFailure(t *testing.T) {
 	<-done
 }
 
-func TestHandleReceiptNoRequestID(t *testing.T) {
+func TestHandleMsgBatchBadDAta(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	wsm := &wsmocks.WSClient{}
 	e := &Ethereum{
@@ -1203,8 +1208,7 @@ func TestHandleReceiptNoRequestID(t *testing.T) {
 	data := fftypes.JSONAnyPtr(`{}`)
 	err := json.Unmarshal(data.Bytes(), &reply)
 	assert.NoError(t, err)
-	err = e.handleReceipt(context.Background(), reply)
-	assert.NoError(t, err)
+	e.handleReceipt(context.Background(), reply)
 }
 
 func TestHandleReceiptBadRequestID(t *testing.T) {
@@ -1221,8 +1225,7 @@ func TestHandleReceiptBadRequestID(t *testing.T) {
 	data := fftypes.JSONAnyPtr(`{"headers":{"requestId":"1","type":"TransactionSuccess"}}`)
 	err := json.Unmarshal(data.Bytes(), &reply)
 	assert.NoError(t, err)
-	err = e.handleReceipt(context.Background(), reply)
-	assert.NoError(t, err)
+	e.handleReceipt(context.Background(), reply)
 }
 
 func TestFormatNil(t *testing.T) {

internal/blockchain/fabric/fabric.go

@@ -370,7 +370,7 @@ func (f *Fabric) handleContractEvent(ctx context.Context, msgJSON fftypes.JSONOb
 	return f.callbacks.BlockchainEvent(event)
 }
 
-func (f *Fabric) handleReceipt(ctx context.Context, reply fftypes.JSONObject) error {
+func (f *Fabric) handleReceipt(ctx context.Context, reply fftypes.JSONObject) {
 	l := log.L(ctx)
 
 	headers := reply.GetObject("headers")
@@ -380,19 +380,19 @@ func (f *Fabric) handleReceipt(ctx context.Context, reply fftypes.JSONObject) er
 	message := reply.GetString("errorMessage")
 	if requestID == "" || replyType == "" {
 		l.Errorf("Reply cannot be processed: %+v", reply)
-		return nil // Swallow this and move on
+		return
 	}
 	operationID, err := fftypes.ParseUUID(ctx, requestID)
 	if err != nil {
 		l.Errorf("Reply cannot be processed - bad ID: %+v", reply)
-		return nil // Swallow this and move on
+		return
 	}
 	updateType := fftypes.OpStatusSucceeded
 	if replyType != "TransactionSuccess" {
 		updateType = fftypes.OpStatusFailed
 	}
 	l.Infof("Fabconnect '%s' reply tx=%s (request=%s) %s", replyType, txHash, requestID, message)
-	return f.callbacks.BlockchainOpUpdate(f, operationID, updateType, txHash, message, reply)
+	f.callbacks.BlockchainOpUpdate(f, operationID, updateType, txHash, message, reply)
 }
 
 func (f *Fabric) handleMessageBatch(ctx context.Context, messages []interface{}) error {
@@ -460,7 +460,7 @@ func (f *Fabric) eventLoop() {
 					err = f.wsconn.Send(ctx, ack)
 				}
 			case map[string]interface{}:
-				err = f.handleReceipt(ctx, fftypes.JSONObject(msgTyped))
+				f.handleReceipt(ctx, fftypes.JSONObject(msgTyped))
 			default:
 				l.Errorf("Message unexpected: %+v", msgTyped)
 				continue