Repository: hyperledger/firefly
Pull request #902 (2022-07-14 16:26:12)

Commonize blockchain code
This PR introduces a new package, `common`, with the hope it can reduce the amount of code re-use across blockchain plugin implementations. Functionality for blockchain callbacks, processing `BatchPin` events, and event stream unpacking has been moved so far. 





Patch:
diff --git a/internal/blockchain/common/common.go b/internal/blockchain/common/common.go
new file mode 100644
index 000000000..836c75a43
--- /dev/null
+++ b/internal/blockchain/common/common.go
@@ -0,0 +1,152 @@
+// Copyright © 2022 Kaleido, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package common
+
+import (
+	"context"
+	"encoding/hex"
+	"strings"
+
+	"github.com/hyperledger/firefly-common/pkg/fftypes"
+	"github.com/hyperledger/firefly-common/pkg/i18n"
+	"github.com/hyperledger/firefly-common/pkg/log"
+	"github.com/hyperledger/firefly/internal/coremsgs"
+	"github.com/hyperledger/firefly/pkg/blockchain"
+	"github.com/hyperledger/firefly/pkg/core"
+)
+
+type BlockchainCallbacks interface {
+	SetHandler(namespace string, handler blockchain.Callbacks)
+	SetOperationalHandler(namespace string, handler core.OperationCallbacks)
+
+	OperationUpdate(ctx context.Context, plugin blockchain.Plugin, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject)
+	BatchPinComplete(ctx context.Context, batch *blockchain.BatchPin, signingKey *core.VerifierRef) error
+	BlockchainNetworkAction(ctx context.Context, namespace, action string, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef) error
+	BlockchainEvent(ctx context.Context, namespace string, event *blockchain.EventWithSubscription) error
+}
+
+type callbacks struct {
+	handlers   map[string]blockchain.Callbacks
+	opHandlers map[string]core.OperationCallbacks
+}
+
+func NewBlockchainCallbacks() BlockchainCallbacks {
+	return &callbacks{
+		handlers:   make(map[string]blockchain.Callbacks),
+		opHandlers: make(map[string]core.OperationCallbacks),
+	}
+}
+
+func (cb *callbacks) SetHandler(namespace string, handler blockchain.Callbacks) {
+	cb.handlers[namespace] = handler
+}
+
+func (cb *callbacks) SetOperationalHandler(namespace string, handler core.OperationCallbacks) {
+	cb.opHandlers[namespace] = handler
+}
+
+func (cb *callbacks) OperationUpdate(ctx context.Context, plugin blockchain.Plugin, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
+	namespace, _, _ := core.ParseNamespacedOpID(ctx, nsOpID)
+	if handler, ok := cb.opHandlers[namespace]; ok {
+		handler.OperationUpdate(plugin, nsOpID, status, blockchainTXID, errorMessage, opOutput)
+		return
+	}
+	log.L(ctx).Errorf("No handler found for blockchain operation '%s'", nsOpID)
+}
+
+func (cb *callbacks) BatchPinComplete(ctx context.Context, batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
+	if handler, ok := cb.handlers[batch.Namespace]; ok {
+		return handler.BatchPinComplete(batch, signingKey)
+	}
+	log.L(ctx).Errorf("No handler found for blockchain batch pin on namespace '%s'", batch.Namespace)
+	return nil
+}
+
+func (cb *callbacks) BlockchainNetworkAction(ctx context.Context, namespace, action string, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef) error {
+	if namespace == "" {
+		// V1 networks don't populate namespace, so deliver the event to every handler
+		for _, handler := range cb.handlers {
+			if err := handler.BlockchainNetworkAction(action, location, event, signingKey); err != nil {
+				return err
+			}
+		}
+	} else {
+		if handler, ok := cb.handlers[namespace]; ok {
+			return handler.BlockchainNetworkAction(action, location, event, signingKey)
+		}
+		log.L(ctx).Errorf("No handler found for blockchain network action on namespace '%s'", namespace)
+	}
+	return nil
+}
+
+func (cb *callbacks) BlockchainEvent(ctx context.Context, namespace string, event *blockchain.EventWithSubscription) error {
+	if namespace == "" {
+		// Older token subscriptions don't populate namespace, so deliver the event to every handler
+		for _, cb := range cb.handlers {
+			// Send the event to all handlers and let them match it to a contract listener
+			if err := cb.BlockchainEvent(event); err != nil {
+				return err
+			}
+		}
+	} else {
+		if handler, ok := cb.handlers[namespace]; ok {
+			return handler.BlockchainEvent(event)
+		}
+		log.L(ctx).Errorf("No handler found for blockchain event on namespace '%s'", namespace)
+	}
+	return nil
+}
+
+func BuildBatchPin(ctx context.Context, namespace string, event *blockchain.Event, sUUIDs string, sBatchHash string, sContexts []string, sPayloadRef string) (batch *blockchain.BatchPin, err error) {
+	hexUUIDs, err := hex.DecodeString(strings.TrimPrefix(sUUIDs, "0x"))
+	if err != nil || len(hexUUIDs) != 32 {
+		log.L(ctx).Errorf("BatchPin event is not valid - bad uuids (%s): %s", sUUIDs, err)
+		return nil, i18n.NewError(ctx, coremsgs.MsgInvalidBatchPinEvent, "bad uuids", sUUIDs, err)
+	}
+	var txnID fftypes.UUID
+	copy(txnID[:], hexUUIDs[0:16])
+	var batchID fftypes.UUID
+	copy(batchID[:], hexUUIDs[16:32])
+
+	var batchHash fftypes.Bytes32
+	err = batchHash.UnmarshalText([]byte(sBatchHash))
+	if err != nil {
+		log.L(ctx).Errorf("BatchPin event is not valid - bad batchHash (%s): %s", sBatchHash, err)
+		return nil, err
+	}
+
+	contexts := make([]*fftypes.Bytes32, len(sContexts))
+	for i, sHash := range sContexts {
+		var hash fftypes.Bytes32
+		err = hash.UnmarshalText([]byte(sHash))
+		if err != nil {
+			log.L(ctx).Errorf("BatchPin event is not valid - bad pin %d (%s): %s", i, sHash, err)
+			return nil, err
+		}
+		contexts[i] = &hash
+	}
+
+	return &blockchain.BatchPin{
+		Namespace:       namespace,
+		TransactionID:   &txnID,
+		BatchID:         &batchID,
+		BatchHash:       &batchHash,
+		BatchPayloadRef: sPayloadRef,
+		Contexts:        contexts,
+		Event:           *event,
+	}, nil
+}
diff --git a/internal/blockchain/common/eventstream.go b/internal/blockchain/common/eventstream.go
new file mode 100644
index 000000000..ec0f09a6c
--- /dev/null
+++ b/internal/blockchain/common/eventstream.go
@@ -0,0 +1,29 @@
+// Copyright © 2022 Kaleido, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package common
+
+import "strings"
+
+func GetNamespaceFromSubName(subName string) string {
+	var parts = strings.Split(subName, "-")
+	// Subscription names post version 1.1 are in the format `ff-sub-<namespace>-<listener ID>`
+	if len(parts) != 4 {
+		// Assume older subscription and return empty string
+		return ""
+	}
+	return parts[2]
+}
diff --git a/internal/blockchain/ethereum/ethereum.go b/internal/blockchain/ethereum/ethereum.go
index 7b40a43d0..c529a7ee1 100644
--- a/internal/blockchain/ethereum/ethereum.go
+++ b/internal/blockchain/ethereum/ethereum.go
@@ -35,6 +35,7 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/wsclient"
 	"github.com/hyperledger/firefly-signer/pkg/abi"
 	"github.com/hyperledger/firefly-signer/pkg/ffi2abi"
+	"github.com/hyperledger/firefly/internal/blockchain/common"
 	"github.com/hyperledger/firefly/internal/coreconfig"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/metrics"
@@ -53,7 +54,7 @@ type Ethereum struct {
 	prefixShort     string
 	prefixLong      string
 	capabilities    *blockchain.Capabilities
-	callbacks       callbacks
+	callbacks       common.BlockchainCallbacks
 	client          *resty.Client
 	fftmClient      *resty.Client
 	streams         *streamManager
@@ -73,63 +74,6 @@ type subscriptionInfo struct {
 	version   int
 }
 
-type callbacks struct {
-	handlers   map[string]blockchain.Callbacks
-	opHandlers map[string]core.OperationCallbacks
-}
-
-func (cb *callbacks) OperationUpdate(ctx context.Context, plugin blockchain.Plugin, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
-	namespace, _, _ := core.ParseNamespacedOpID(ctx, nsOpID)
-	if handler, ok := cb.opHandlers[namespace]; ok {
-		handler.OperationUpdate(plugin, nsOpID, status, blockchainTXID, errorMessage, opOutput)
-		return
-	}
-	log.L(ctx).Errorf("No handler found for blockchain operation '%s'", nsOpID)
-}
-
-func (cb *callbacks) BatchPinComplete(ctx context.Context, batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
-	if handler, ok := cb.handlers[batch.Namespace]; ok {
-		return handler.BatchPinComplete(batch, signingKey)
-	}
-	log.L(ctx).Errorf("No handler found for blockchain batch pin on namespace '%s'", batch.Namespace)
-	return nil
-}
-
-func (cb *callbacks) BlockchainNetworkAction(ctx context.Context, namespace, action string, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef) error {
-	if namespace == "" {
-		// V1 networks don't populate namespace, so deliver the event to every handler
-		for _, handler := range cb.handlers {
-			if err := handler.BlockchainNetworkAction(action, location, event, signingKey); err != nil {
-				return err
-			}
-		}
-	} else {
-		if handler, ok := cb.handlers[namespace]; ok {
-			return handler.BlockchainNetworkAction(action, location, event, signingKey)
-		}
-		log.L(ctx).Errorf("No handler found for blockchain network action on namespace '%s'", namespace)
-	}
-	return nil
-}
-
-func (cb *callbacks) BlockchainEvent(ctx context.Context, namespace string, event *blockchain.EventWithSubscription) error {
-	if namespace == "" {
-		// Older token subscriptions don't populate namespace, so deliver the event to every handler
-		for _, cb := range cb.handlers {
-			// Send the event to all handlers and let them match it to a contract listener
-			if err := cb.BlockchainEvent(event); err != nil {
-				return err
-			}
-		}
-	} else {
-		if handler, ok := cb.handlers[namespace]; ok {
-			return handler.BlockchainEvent(event)
-		}
-		log.L(ctx).Errorf("No handler found for blockchain event on namespace '%s'", namespace)
-	}
-	return nil
-}
-
 type eventStreamWebsocket struct {
 	Topic string `json:"topic"`
 }
@@ -187,8 +131,7 @@ func (e *Ethereum) Init(ctx context.Context, conf config.Section, metrics metric
 	e.ctx = log.WithLogField(ctx, "proto", "ethereum")
 	e.metrics = metrics
 	e.capabilities = &blockchain.Capabilities{}
-	e.callbacks.handlers = make(map[string]blockchain.Callbacks)
-	e.callbacks.opHandlers = make(map[string]core.OperationCallbacks)
+	e.callbacks = common.NewBlockchainCallbacks()
 
 	if addressResolverConf.GetString(AddressResolverURLTemplate) != "" {
 		if e.addressResolver, err = newAddressResolver(ctx, addressResolverConf); err != nil {
@@ -242,11 +185,11 @@ func (e *Ethereum) Init(ctx context.Context, conf config.Section, metrics metric
 }
 
 func (e *Ethereum) SetHandler(namespace string, handler blockchain.Callbacks) {
-	e.callbacks.handlers[namespace] = handler
+	e.callbacks.SetHandler(namespace, handler)
 }
 
 func (e *Ethereum) SetOperationHandler(namespace string, handler core.OperationCallbacks) {
-	e.callbacks.opHandlers[namespace] = handler
+	e.callbacks.SetOperationalHandler(namespace, handler)
 }
 
 func (e *Ethereum) Start() (err error) {
@@ -414,44 +357,11 @@ func (e *Ethereum) handleBatchPinEvent(ctx context.Context, location *fftypes.JS
 		namespace = subInfo.namespace
 	}
 
-	hexUUIDs, err := hex.DecodeString(strings.TrimPrefix(sUUIDs, "0x"))
-	if err != nil || len(hexUUIDs) != 32 {
-		log.L(ctx).Errorf("BatchPin event is not valid - bad uuids (%s): %+v", err, msgJSON)
-		return nil // move on
-	}
-	var txnID fftypes.UUID
-	copy(txnID[:], hexUUIDs[0:16])
-	var batchID fftypes.UUID
-	copy(batchID[:], hexUUIDs[16:32])
-
-	var batchHash fftypes.Bytes32
-	err = batchHash.UnmarshalText([]byte(sBatchHash))
+	batch, err := common.BuildBatchPin(ctx, namespace, event, sUUIDs, sBatchHash, sContexts, sPayloadRef)
 	if err != nil {
-		log.L(ctx).Errorf("BatchPin event is not valid - bad batchHash (%s): %+v", err, msgJSON)
 		return nil // move on
 	}
 
-	contexts := make([]*fftypes.Bytes32, len(sContexts))
-	for i, sHash := range sContexts {
-		var hash fftypes.Bytes32
-		err = hash.UnmarshalText([]byte(sHash))
-		if err != nil {
-			log.L(ctx).Errorf("BatchPin event is not valid - bad pin %d (%s): %+v", i, err, msgJSON)
-			return nil // move on
-		}
-		contexts[i] = &hash
-	}
-
-	batch := &blockchain.BatchPin{
-		Namespace:       namespace,
-		TransactionID:   &txnID,
-		BatchID:         &batchID,
-		BatchHash:       &batchHash,
-		BatchPayloadRef: sPayloadRef,
-		Contexts:        contexts,
-		Event:           *event,
-	}
-
 	// If there's an error dispatching the event, we must return the error and shutdown
 	return e.callbacks.BatchPinComplete(ctx, batch, verifier)
 }
@@ -462,7 +372,7 @@ func (e *Ethereum) handleContractEvent(ctx context.Context, msgJSON fftypes.JSON
 		return err
 	}
 
-	namespace := e.streams.getNamespaceFromSubName(subName)
+	namespace := common.GetNamespaceFromSubName(subName)
 	event := e.parseBlockchainEvent(ctx, msgJSON)
 	if event != nil {
 		err = e.callbacks.BlockchainEvent(ctx, namespace, &blockchain.EventWithSubscription{
diff --git a/internal/blockchain/ethereum/eventstream.go b/internal/blockchain/ethereum/eventstream.go
index 7439fc293..e4aaa9db2 100644
--- a/internal/blockchain/ethereum/eventstream.go
+++ b/internal/blockchain/ethereum/eventstream.go
@@ -21,7 +21,6 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"fmt"
-	"strings"
 	"time"
 
 	"github.com/go-resty/resty/v2"
@@ -254,13 +253,3 @@ func (s *streamManager) ensureFireFlySubscription(ctx context.Context, namespace
 	log.L(ctx).Infof("%s subscription: %s", abi.Name, sub.ID)
 	return sub, subNS, nil
 }
-
-func (s *streamManager) getNamespaceFromSubName(subName string) string {
-	var parts = strings.Split(subName, "-")
-	// Subscription names post version 1.1 are in the format `ff-sub-<namespace>-<listener ID>`
-	if len(parts) != 4 {
-		// Assume older subscription and return empty string
-		return ""
-	}
-	return parts[2]
-}
diff --git a/internal/blockchain/fabric/eventstream.go b/internal/blockchain/fabric/eventstream.go
index e4f0e3d9b..e538524c6 100644
--- a/internal/blockchain/fabric/eventstream.go
+++ b/internal/blockchain/fabric/eventstream.go
@@ -19,7 +19,6 @@ package fabric
 import (
 	"context"
 	"fmt"
-	"strings"
 	"time"
 
 	"github.com/go-resty/resty/v2"
@@ -222,13 +221,3 @@ func (s *streamManager) ensureFireFlySubscription(ctx context.Context, namespace
 	log.L(ctx).Infof("%s subscription: %s", event, sub.ID)
 	return sub, subNS, nil
 }
-
-func (s *streamManager) getNamespaceFromSubName(subName string) string {
-	var parts = strings.Split(subName, "-")
-	// Subscription names post version 1.1 are in the format `ff-sub-<namespace>-<listener ID>`
-	if len(parts) != 4 {
-		// Assume older subscription and return empty string
-		return ""
-	}
-	return parts[2]
-}
diff --git a/internal/blockchain/fabric/fabric.go b/internal/blockchain/fabric/fabric.go
index 9805f2bb5..d275c6e30 100644
--- a/internal/blockchain/fabric/fabric.go
+++ b/internal/blockchain/fabric/fabric.go
@@ -33,6 +33,7 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly-common/pkg/wsclient"
+	"github.com/hyperledger/firefly/internal/blockchain/common"
 	"github.com/hyperledger/firefly/internal/coreconfig"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/metrics"
@@ -53,7 +54,7 @@ type Fabric struct {
 	prefixShort    string
 	prefixLong     string
 	capabilities   *blockchain.Capabilities
-	callbacks      callbacks
+	callbacks      common.BlockchainCallbacks
 	client         *resty.Client
 	streams        *streamManager
 	streamID       string
@@ -73,63 +74,6 @@ type subscriptionInfo struct {
 	version   int
 }
 
-type callbacks struct {
-	handlers   map[string]blockchain.Callbacks
-	opHandlers map[string]core.OperationCallbacks
-}
-
-func (cb *callbacks) OperationUpdate(ctx context.Context, plugin blockchain.Plugin, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
-	namespace, _, _ := core.ParseNamespacedOpID(ctx, nsOpID)
-	if handler, ok := cb.opHandlers[namespace]; ok {
-		handler.OperationUpdate(plugin, nsOpID, status, blockchainTXID, errorMessage, opOutput)
-	} else {
-		log.L(ctx).Errorf("No handler found for blockchain operation '%s'", nsOpID)
-	}
-}
-
-func (cb *callbacks) BatchPinComplete(ctx context.Context, batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
-	if handler, ok := cb.handlers[batch.Namespace]; ok {
-		return handler.BatchPinComplete(batch, signingKey)
-	}
-	log.L(ctx).Errorf("No handler found for blockchain batch pin on namespace '%s'", batch.Namespace)
-	return nil
-}
-
-func (cb *callbacks) BlockchainNetworkAction(ctx context.Context, namespace, action string, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef) error {
-	if namespace == "" {
-		// Older networks don't populate namespace, so deliver the event to every handler
-		for _, handler := range cb.handlers {
-			if err := handler.BlockchainNetworkAction(action, location, event, signingKey); err != nil {
-				return err
-			}
-		}
-	} else {
-		if handler, ok := cb.handlers[namespace]; ok {
-			return handler.BlockchainNetworkAction(action, location, event, signingKey)
-		}
-		log.L(ctx).Errorf("No handler found for blockchain network action on namespace '%s'", namespace)
-	}
-	return nil
-}
-
-func (cb *callbacks) BlockchainEvent(ctx context.Context, namespace string, event *blockchain.EventWithSubscription) error {
-	if namespace == "" {
-		// Older token subscriptions don't populate namespace, so deliver the event to every handler
-		for _, cb := range cb.handlers {
-			// Send the event to all handlers and let them match it to a contract listener
-			if err := cb.BlockchainEvent(event); err != nil {
-				return err
-			}
-		}
-	} else {
-		if handler, ok := cb.handlers[namespace]; ok {
-			return handler.BlockchainEvent(event)
-		}
-		log.L(ctx).Errorf("No handler found for blockchain event on namespace '%s'", namespace)
-	}
-	return nil
-}
-
 type eventStreamWebsocket struct {
 	Topic string `json:"topic"`
 }
@@ -257,8 +201,7 @@ func (f *Fabric) Init(ctx context.Context, conf config.Section, metrics metrics.
 	f.idCache = make(map[string]*fabIdentity)
 	f.metrics = metrics
 	f.capabilities = &blockchain.Capabilities{}
-	f.callbacks.handlers = make(map[string]blockchain.Callbacks)
-	f.callbacks.opHandlers = make(map[string]core.OperationCallbacks)
+	f.callbacks = common.NewBlockchainCallbacks()
 
 	if fabconnectConf.GetString(ffresty.HTTPConfigURL) == "" {
 		return i18n.NewError(ctx, coremsgs.MsgMissingPluginConfig, "url", "blockchain.fabric.fabconnect")
@@ -305,11 +248,11 @@ func (f *Fabric) Init(ctx context.Context, conf config.Section, metrics metrics.
 }
 
 func (f *Fabric) SetHandler(namespace string, handler blockchain.Callbacks) {
-	f.callbacks.handlers[namespace] = handler
+	f.callbacks.SetHandler(namespace, handler)
 }
 
 func (f *Fabric) SetOperationHandler(namespace string, handler core.OperationCallbacks) {
-	f.callbacks.opHandlers[namespace] = handler
+	f.callbacks.SetOperationalHandler(namespace, handler)
 }
 
 func (f *Fabric) Start() (err error) {
@@ -421,44 +364,11 @@ func (f *Fabric) handleBatchPinEvent(ctx context.Context, location *fftypes.JSON
 		namespace = subInfo.namespace
 	}
 
-	hexUUIDs, err := hex.DecodeString(strings.TrimPrefix(sUUIDs, "0x"))
-	if err != nil || len(hexUUIDs) != 32 {
-		log.L(ctx).Errorf("BatchPin event is not valid - bad uuids (%s): %s", sUUIDs, err)
-		return nil // move on
-	}
-	var txnID fftypes.UUID
-	copy(txnID[:], hexUUIDs[0:16])
-	var batchID fftypes.UUID
-	copy(batchID[:], hexUUIDs[16:32])
-
-	var batchHash fftypes.Bytes32
-	err = batchHash.UnmarshalText([]byte(sBatchHash))
+	batch, err := common.BuildBatchPin(ctx, namespace, event, sUUIDs, sBatchHash, sContexts, sPayloadRef)
 	if err != nil {
-		log.L(ctx).Errorf("BatchPin event is not valid - bad batchHash (%s): %s", sBatchHash, err)
 		return nil // move on
 	}
 
-	contexts := make([]*fftypes.Bytes32, len(sContexts))
-	for i, sHash := range sContexts {
-		var hash fftypes.Bytes32
-		err = hash.UnmarshalText([]byte(sHash))
-		if err != nil {
-			log.L(ctx).Errorf("BatchPin event is not valid - bad pin %d (%s): %s", i, sHash, err)
-			return nil // move on
-		}
-		contexts[i] = &hash
-	}
-
-	batch := &blockchain.BatchPin{
-		Namespace:       namespace,
-		TransactionID:   &txnID,
-		BatchID:         &batchID,
-		BatchHash:       &batchHash,
-		BatchPayloadRef: sPayloadRef,
-		Contexts:        contexts,
-		Event:           *event,
-	}
-
 	// If there's an error dispatching the event, we must return the error and shutdown
 	return f.callbacks.BatchPinComplete(ctx, batch, verifier)
 }
@@ -472,7 +382,7 @@ func (f *Fabric) handleContractEvent(ctx context.Context, msgJSON fftypes.JSONOb
 	if err != nil {
 		return err
 	}
-	namespace := f.streams.getNamespaceFromSubName(subName)
+	namespace := common.GetNamespaceFromSubName(subName)
 	event := f.parseBlockchainEvent(ctx, msgJSON)
 	if event == nil {
 		return nil // move on
diff --git a/internal/coremsgs/en_error_messages.go b/internal/coremsgs/en_error_messages.go
index cf7993bd3..c67896efe 100644
--- a/internal/coremsgs/en_error_messages.go
+++ b/internal/coremsgs/en_error_messages.go
@@ -257,4 +257,5 @@ var (
 	MsgMessagesNotSupported               = ffe("FF10415", "Messages are not supported in this namespace", 400)
 	MsgInvalidSubscriptionForNetwork      = ffe("FF10416", "Subscription name '%s' is invalid according to multiparty network rules in effect (network version=%d)")
 	MsgBlockchainNotConfigured            = ffe("FF10417", "No blockchain plugin configured")
+	MsgInvalidBatchPinEvent               = ffe("FF10418", "BatchPin event is not valid - %s (%s): %s")
 )


Test patch:
diff --git a/internal/blockchain/ethereum/ethereum_test.go b/internal/blockchain/ethereum/ethereum_test.go
index 101d3d63b..87a580f8a 100644
--- a/internal/blockchain/ethereum/ethereum_test.go
+++ b/internal/blockchain/ethereum/ethereum_test.go
@@ -31,6 +31,7 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly-common/pkg/wsclient"
+	"github.com/hyperledger/firefly/internal/blockchain/common"
 	"github.com/hyperledger/firefly/internal/coreconfig"
 	"github.com/hyperledger/firefly/mocks/blockchainmocks"
 	"github.com/hyperledger/firefly/mocks/coremocks"
@@ -93,8 +94,7 @@ func newTestEthereum() (*Ethereum, func()) {
 		metrics:     mm,
 		cache:       ccache.New(ccache.Configure().MaxSize(100)),
 	}
-	e.callbacks.handlers = make(map[string]blockchain.Callbacks)
-	e.callbacks.opHandlers = make(map[string]core.OperationCallbacks)
+	e.callbacks = common.NewBlockchainCallbacks()
 	return e, func() {
 		cancel()
 		if e.closed != nil {
@@ -979,7 +979,7 @@ func TestHandleMessageBatchPinOK(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{handlers: make(map[string]blockchain.Callbacks)},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
 	e.SetHandler("ns1", em)
 
@@ -1070,8 +1070,9 @@ func TestHandleMessageBatchPinV2(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
@@ -1145,9 +1146,11 @@ func TestHandleMessageBatchPinMissingAddress(t *testing.T) {
 ]`)
 
 	em := &blockchainmocks.Callbacks{}
+
 	e := &Ethereum{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
@@ -1231,9 +1234,11 @@ func TestHandleMessageEmptyPayloadRef(t *testing.T) {
 ]`)
 
 	em := &blockchainmocks.Callbacks{}
+
 	e := &Ethereum{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
@@ -1297,9 +1302,12 @@ func TestHandleMessageBatchPinExit(t *testing.T) {
 	}
 
 	em := &blockchainmocks.Callbacks{}
+
 	e := &Ethereum{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
+
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1555,9 +1563,10 @@ func TestHandleReceiptTXSuccess(t *testing.T) {
 	e := &Ethereum{
 		ctx:       context.Background(),
 		topic:     "topic1",
-		callbacks: callbacks{opHandlers: map[string]core.OperationCallbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 		wsconn:    wsm,
 	}
+	e.SetOperationHandler("ns1", em)
 
 	var reply fftypes.JSONObject
 	operationID := fftypes.NewUUID()
@@ -1879,7 +1888,7 @@ func TestHandleMessageContractEventOldSubscription(t *testing.T) {
 			ID: "sub2", Stream: "es12345", Name: "ff-sub-1132312312312",
 		}))
 
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1954,7 +1963,8 @@ func TestHandleMessageContractEventErrorOldSubscription(t *testing.T) {
 			ID: "sub2", Stream: "es12345", Name: "ff-sub-1132312312312",
 		}))
 
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}}
+	e.callbacks = common.NewBlockchainCallbacks()
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -2016,7 +2026,8 @@ func TestHandleMessageContractEventWithNamespace(t *testing.T) {
 			ID: "sub2", Stream: "es12345", Name: "ff-sub-ns1-1132312312312",
 		}))
 
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}}
+	e.callbacks = common.NewBlockchainCallbacks()
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -2090,7 +2101,7 @@ func TestHandleMessageContractEventNoNamespaceHandlers(t *testing.T) {
 			ID: "sub2", Stream: "es12345", Name: "ff-sub-ns1-1132312312312",
 		}))
 
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns2": em}}
+	e.SetHandler("ns2", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -2140,7 +2151,8 @@ func TestHandleMessageContractEventSubNameError(t *testing.T) {
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions/sub2",
 		httpmock.NewJsonResponderOrPanic(500, ethError{Error: "pop"}))
 
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}}
+	e.callbacks = common.NewBlockchainCallbacks()
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -2188,7 +2200,7 @@ func TestHandleMessageContractEventError(t *testing.T) {
 			ID: "sub2", Stream: "es12345", Name: "ff-sub-ns1-1132312312312",
 		}))
 
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -2824,8 +2836,9 @@ func TestHandleNetworkAction(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -2874,8 +2887,9 @@ func TestHandleNetworkActionV2(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -2924,8 +2938,9 @@ func TestHandleNetworkActionFail(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
diff --git a/internal/blockchain/fabric/fabric_test.go b/internal/blockchain/fabric/fabric_test.go
index 97672a109..a7d92698c 100644
--- a/internal/blockchain/fabric/fabric_test.go
+++ b/internal/blockchain/fabric/fabric_test.go
@@ -32,6 +32,7 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly-common/pkg/wsclient"
+	"github.com/hyperledger/firefly/internal/blockchain/common"
 	"github.com/hyperledger/firefly/internal/coreconfig"
 	"github.com/hyperledger/firefly/mocks/blockchainmocks"
 	"github.com/hyperledger/firefly/mocks/coremocks"
@@ -67,7 +68,7 @@ func newTestFabric() (*Fabric, func()) {
 		wsconn:         wsm,
 		cache:          ccache.New(ccache.Configure().MaxSize(100)),
 	}
-	e.callbacks.handlers = make(map[string]blockchain.Callbacks)
+	e.callbacks = common.NewBlockchainCallbacks()
 	return e, func() {
 		cancel()
 		if e.closed != nil {
@@ -982,8 +983,9 @@ func TestHandleMessageBatchPinOK(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1045,8 +1047,9 @@ func TestHandleMessageBatchPinV2(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1106,8 +1109,9 @@ func TestHandleMessageBatchPinMissingChaincodeID(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1138,8 +1142,9 @@ func TestHandleMessageEmptyPayloadRef(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1190,8 +1195,9 @@ func TestHandleMessageBatchPinExit(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1228,8 +1234,9 @@ func TestHandleMessageBatchPinEmpty(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1259,8 +1266,9 @@ func TestHandleMessageUnknownEventName(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1279,8 +1287,9 @@ func TestHandleMessageUnknownEventName(t *testing.T) {
 func TestHandleMessageBatchPinBadBatchHash(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1307,8 +1316,9 @@ func TestHandleMessageBatchPinBadBatchHash(t *testing.T) {
 func TestHandleMessageBatchPinBadPin(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1335,8 +1345,9 @@ func TestHandleMessageBatchPinBadPin(t *testing.T) {
 func TestHandleMessageBatchPinBadPayloadEncoding(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1363,8 +1374,9 @@ func TestHandleMessageBatchPinBadPayloadEncoding(t *testing.T) {
 func TestHandleMessageBatchPinBadPayloadUUIDs(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1391,8 +1403,9 @@ func TestHandleMessageBatchPinBadPayloadUUIDs(t *testing.T) {
 func TestHandleMessageBatchBadJSON(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	err := e.handleMessageBatch(context.Background(), []interface{}{10, 20})
 	assert.NoError(t, err)
 	assert.Equal(t, 0, len(em.Calls))
@@ -1488,7 +1501,7 @@ func TestHandleReceiptTXSuccess(t *testing.T) {
 	e := &Fabric{
 		ctx:       context.Background(),
 		topic:     "topic1",
-		callbacks: callbacks{opHandlers: make(map[string]core.OperationCallbacks)},
+		callbacks: common.NewBlockchainCallbacks(),
 		wsconn:    wsm,
 	}
 	e.SetOperationHandler("ns1", em)
@@ -1530,9 +1543,10 @@ func TestHandleReceiptNoRequestID(t *testing.T) {
 	e := &Fabric{
 		ctx:       context.Background(),
 		topic:     "topic1",
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 		wsconn:    wsm,
 	}
+	e.SetHandler("ns1", em)
 
 	var reply fftypes.JSONObject
 	data := []byte(`{}`)
@@ -1547,9 +1561,10 @@ func TestHandleReceiptFailedTx(t *testing.T) {
 	e := &Fabric{
 		ctx:       context.Background(),
 		topic:     "topic1",
-		callbacks: callbacks{opHandlers: map[string]core.OperationCallbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 		wsconn:    wsm,
 	}
+	e.SetOperationHandler("ns1", em)
 
 	var reply fftypes.JSONObject
 	operationID := fftypes.NewUUID()
@@ -1751,7 +1766,8 @@ func TestHandleMessageContractEventOldSubscription(t *testing.T) {
 		}))
 
 	e.streams = newTestStreamManager(e.client, e.signer)
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}}
+	e.callbacks = common.NewBlockchainCallbacks()
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1835,7 +1851,8 @@ func TestHandleMessageContractEventNamespacedHandlers(t *testing.T) {
 		}))
 
 	e.streams = newTestStreamManager(e.client, e.signer)
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}}
+	e.callbacks = common.NewBlockchainCallbacks()
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1908,7 +1925,8 @@ func TestHandleMessageContractEventNoNamespacedHandlers(t *testing.T) {
 		}))
 
 	e.streams = newTestStreamManager(e.client, e.signer)
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns2": em}}
+	e.callbacks = common.NewBlockchainCallbacks()
+	e.SetHandler("ns2", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -1956,7 +1974,8 @@ func TestHandleMessageContractEventNoPayload(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e.streams = newTestStreamManager(e.client, e.signer)
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}}
+	e.callbacks = common.NewBlockchainCallbacks()
+	e.SetHandler("ns1", em)
 
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
@@ -1987,8 +2006,9 @@ func TestHandleMessageContractEventBadPayload(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-cb37cc07-e873-4f58-44ab-55add6bba320"] = subscriptionInfo{
 		namespace: "ns1",
@@ -2030,7 +2050,8 @@ func TestHandleMessageContractOldSubError(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e.streams = newTestStreamManager(e.client, e.signer)
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}}
+	e.callbacks = common.NewBlockchainCallbacks()
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -2074,7 +2095,8 @@ func TestHandleMessageContractEventError(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e.streams = newTestStreamManager(e.client, e.signer)
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}}
+	e.callbacks = common.NewBlockchainCallbacks()
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -2116,7 +2138,8 @@ func TestHandleMessageContractGetSubError(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e.streams = newTestStreamManager(e.client, e.signer)
-	e.callbacks = callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}}
+	e.callbacks = common.NewBlockchainCallbacks()
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
 		namespace: "ns1",
@@ -2492,8 +2515,9 @@ func TestHandleNetworkAction(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
@@ -2536,8 +2560,9 @@ func TestHandleNetworkActionV2(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
@@ -2580,8 +2605,9 @@ func TestHandleNetworkActionFail(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: map[string]blockchain.Callbacks{"ns1": em}},
+		callbacks: common.NewBlockchainCallbacks(),
 	}
+	e.SetHandler("ns1", em)
 	e.subs = map[string]subscriptionInfo{}
 	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
 		namespace: "ns1",