Repository: hyperledger/firefly
Pull request #873 (2022-06-22 14:12:36)

Use docker buildx for multiarch builds
This PR switches to the `docker buildx` command for building images for multiple CPU architectures. Currently, the build jobs are set up to build `linux/amd64` and `linux/arm64`. This is set in the `Makefile`. Because building a Docker image for a non-native CPU architecture is very slow (because it's emulated), mutiarch builds are not the default when running `make docker`. Instead a new command, `make docker-multiarch` has been added, which is used by our GitHub Actions.

> **NOTE:** From my testing `docker buildx` seems to return status 0, despite the fact that it may have failed to push to a repository. This can result in false positives for our GitHub Actions. I looked into this and it seems to be an open issue with `buildx` https://github.com/docker/buildx/issues/732


Patch:
diff --git a/.github/workflows/docker_main.yml b/.github/workflows/docker_main.yml
index 8a0263f4d..fac90c03e 100644
--- a/.github/workflows/docker_main.yml
+++ b/.github/workflows/docker_main.yml
@@ -13,6 +13,18 @@ jobs:
         with:
           fetch-depth: 0
 
+      - name: Set up Docker Buildx
+        id: buildx
+        uses: docker/setup-buildx-action@v2
+      
+      - name: Set up qemu emulators
+        run: |
+          docker run --rm --privileged tonistiigi/binfmt --install all
+
+      - name: Docker login
+        run: |
+          echo ${{ secrets.GITHUB_TOKEN }} | docker login ghcr.io -u $GITHUB_ACTOR --password-stdin
+
       - name: Set build tag
         id: build_tag_generator
         run: |
@@ -28,18 +40,7 @@ jobs:
             --label commit=$GITHUB_SHA \
             --label build_date=$(date -u +"%Y-%m-%dT%H:%M:%SZ") \
             --label tag=${{ steps.build_tag_generator.outputs.BUILD_TAG }} \
-            --tag ghcr.io/hyperledger/firefly:${{ steps.build_tag_generator.outputs.BUILD_TAG }}" \
-            docker
-
-      - name: Tag release
-        run: docker tag ghcr.io/hyperledger/firefly:${{ steps.build_tag_generator.outputs.BUILD_TAG }} ghcr.io/hyperledger/firefly:head
-
-      - name: Push docker image
-        run: |
-          echo ${{ secrets.GITHUB_TOKEN }} | docker login ghcr.io -u $GITHUB_ACTOR --password-stdin
-          docker push ghcr.io/hyperledger/firefly:${{ steps.build_tag_generator.outputs.BUILD_TAG }}
-
-      - name: Push head tag
-        run: |
-          echo ${{ secrets.GITHUB_TOKEN }} | docker login ghcr.io -u $GITHUB_ACTOR --password-stdin
-          docker push ghcr.io/hyperledger/firefly:head
+            --tag ghcr.io/hyperledger/firefly:${{ steps.build_tag_generator.outputs.BUILD_TAG }} \
+            --tag ghcr.io/hyperledger/firefly:head \
+            --push" \
+            docker-multiarch
\ No newline at end of file
diff --git a/.github/workflows/docker_release.yml b/.github/workflows/docker_release.yml
index ca85c396f..b708187fe 100644
--- a/.github/workflows/docker_release.yml
+++ b/.github/workflows/docker_release.yml
@@ -12,67 +12,44 @@ jobs:
         with:
           fetch-depth: 0
 
-      - name: Build
-        run: |
-          make DOCKER_ARGS="\
-            --build-arg BUILD_VERSION=${GITHUB_REF##*/} \
-            --build-arg GIT_REF=$GITHUB_SHA \
-            --label commit=$GITHUB_SHA \
-            --label build_date=$(date -u +"%Y-%m-%dT%H:%M:%SZ") \
-            --label tag=${GITHUB_REF##*/} \
-            --tag ghcr.io/hyperledger/firefly:${GITHUB_REF##*/} \
-            --tag ghcr.io/hyperledger/firefly:head" \
-            docker
-
-      - name: Push docker image
-        run: |
-          echo ${{ secrets.GITHUB_TOKEN }} | docker login ghcr.io -u $GITHUB_ACTOR --password-stdin
-          docker push ghcr.io/hyperledger/firefly:${GITHUB_REF##*/}
-
-      - name: Push head tag
-        run: |
-          echo ${{ secrets.GITHUB_TOKEN }} | docker login ghcr.io -u $GITHUB_ACTOR --password-stdin
-          docker push ghcr.io/hyperledger/firefly:head
-
-      - name: Tag latest release
-        if: github.event.action == 'released'
-        run: docker tag ghcr.io/hyperledger/firefly:${GITHUB_REF##*/} ghcr.io/hyperledger/firefly:latest
-
-      - name: Push latest tag
+      - name: Set up Docker Buildx
+        id: buildx
+        uses: docker/setup-buildx-action@v2
+      
+      - name: Set up qemu emulators
+        run: |
+          docker run --rm --privileged tonistiigi/binfmt --install all
+      
+      - name: Set latest tag
         if: github.event.action == 'released'
         run: |
-          echo ${{ secrets.GITHUB_TOKEN }} | docker login ghcr.io -u $GITHUB_ACTOR --password-stdin
-          docker push ghcr.io/hyperledger/firefly:latest
+          echo "DOCKER_TAGS=${{ env.DOCKER_TAGS }} --tag ghcr.io/hyperledger/firefly:latest" >> $GITHUB_ENV
 
-      - name: Tag alpha release
+      - name: Set alpha tag
         if: github.event.action == 'prereleased' && contains(github.ref, 'alpha')
         run: |
-          docker tag ghcr.io/hyperledger/firefly:${GITHUB_REF##*/} ghcr.io/hyperledger/firefly:alpha
+            echo "DOCKER_TAGS=${{ env.DOCKER_TAGS }} --tag ghcr.io/hyperledger/firefly:alpha" >> $GITHUB_ENV
 
-      - name: Push alpha tag
-        if: github.event.action == 'prereleased' && contains(github.ref, 'alpha')
-        run: |
-          echo ${{ secrets.GITHUB_TOKEN }} | docker login ghcr.io -u $GITHUB_ACTOR --password-stdin
-          docker push ghcr.io/hyperledger/firefly:alpha
-
-      - name: Tag beta release
-        if: github.event.action == 'prereleased' && contains(github.ref, 'beta')
-        run: |
-          docker tag ghcr.io/hyperledger/firefly:${GITHUB_REF##*/} ghcr.io/hyperledger/firefly:beta
-
-      - name: Push beta tag
+      - name: Set beta tag
         if: github.event.action == 'prereleased' && contains(github.ref, 'beta')
         run: |
-          echo ${{ secrets.GITHUB_TOKEN }} | docker login ghcr.io -u $GITHUB_ACTOR --password-stdin
-          docker push ghcr.io/hyperledger/firefly:beta
+            echo "DOCKER_TAGS=${{ env.DOCKER_TAGS }} --tag ghcr.io/hyperledger/firefly:beta" >> $GITHUB_ENV
 
-      - name: Tag rc release
+      - name: Set rc tag
         if: github.event.action == 'prereleased' && contains(github.ref, 'rc')
         run: |
-          docker tag ghcr.io/hyperledger/firefly:${GITHUB_REF##*/} ghcr.io/hyperledger/firefly:rc
+            echo "DOCKER_TAGS=${{ env.DOCKER_TAGS }} --tag ghcr.io/hyperledger/firefly:rc" >> $GITHUB_ENV
 
-      - name: Push rc tag
-        if: github.event.action == 'prereleased' && contains(github.ref, 'rc')
+      - name: Build
         run: |
-          echo ${{ secrets.GITHUB_TOKEN }} | docker login ghcr.io -u $GITHUB_ACTOR --password-stdin
-          docker push ghcr.io/hyperledger/firefly:rc
\ No newline at end of file
+          make DOCKER_ARGS="\
+            --build-arg BUILD_VERSION=${GITHUB_REF##*/} \
+            --build-arg GIT_REF=$GITHUB_SHA \
+            --label commit=$GITHUB_SHA \
+            --label build_date=$(date -u +"%Y-%m-%dT%H:%M:%SZ") \
+            --label tag=${GITHUB_REF##*/} \
+            --tag ghcr.io/hyperledger/firefly:${GITHUB_REF##*/} \
+            --tag ghcr.io/hyperledger/firefly:head \
+            ${{ env.DOCKER_TAGS }}" \
+            --push
+           docker-multiarch
diff --git a/Makefile b/Makefile
index f420d2448..c72b0a130 100644
--- a/Makefile
+++ b/Makefile
@@ -99,5 +99,7 @@ manifest:
 		./manifestgen.sh
 docker:
 		./docker_build.sh $(DOCKER_ARGS)
+docker-multiarch:
+		./docker_build.sh --platform linux/amd64,linux/arm64 $(DOCKER_ARGS) 
 docs: .ALWAYS
 		cd docs && bundle install && bundle exec jekyll build && bundle exec htmlproofer --disable-external --allow-hash-href --assume-extension ./_site --url-swap '^/firefly/:/' --url-ignore /127.0.0.1/,/localhost/
\ No newline at end of file
diff --git a/docker_build.sh b/docker_build.sh
index 45b65f542..11fead542 100755
--- a/docker_build.sh
+++ b/docker_build.sh
@@ -36,7 +36,8 @@ echo BASE_TAG=$BASE_TAG
 echo UI_TAG=$UI_TAG
 echo UI_RELEASE=$UI_RELEASE
 
-docker build \
+docker buildx create --name firefly --use
+docker buildx build \
     -t hyperledger/firefly \
     --build-arg FIREFLY_BUILDER_TAG=$FIREFLY_BUILDER_TAG \
     --build-arg FABRIC_BUILDER_TAG=$FABRIC_BUILDER_TAG \
@@ -46,4 +47,5 @@ docker build \
     --build-arg UI_TAG=$UI_TAG \
     --build-arg UI_RELEASE=$UI_RELEASE \
     $@ \
-    .
\ No newline at end of file
+    .
+docker buildx rm firefly
\ No newline at end of file
diff --git a/internal/blockchain/ethereum/ethereum.go b/internal/blockchain/ethereum/ethereum.go
index 4474b9dba..74389618a 100644
--- a/internal/blockchain/ethereum/ethereum.go
+++ b/internal/blockchain/ethereum/ethereum.go
@@ -81,17 +81,17 @@ type Ethereum struct {
 }
 
 type callbacks struct {
-	handlers []blockchain.Callbacks
+	listeners []blockchain.Callbacks
 }
 
 func (cb *callbacks) BlockchainOpUpdate(plugin blockchain.Plugin, nsOpID string, txState blockchain.TransactionStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		cb.BlockchainOpUpdate(plugin, nsOpID, txState, blockchainTXID, errorMessage, opOutput)
 	}
 }
 
 func (cb *callbacks) BatchPinComplete(batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		if err := cb.BatchPinComplete(batch, signingKey); err != nil {
 			return err
 		}
@@ -100,7 +100,7 @@ func (cb *callbacks) BatchPinComplete(batch *blockchain.BatchPin, signingKey *co
 }
 
 func (cb *callbacks) BlockchainNetworkAction(action string, event *blockchain.Event, signingKey *core.VerifierRef) error {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		if err := cb.BlockchainNetworkAction(action, event, signingKey); err != nil {
 			return err
 		}
@@ -109,7 +109,7 @@ func (cb *callbacks) BlockchainNetworkAction(action string, event *blockchain.Ev
 }
 
 func (cb *callbacks) BlockchainEvent(event *blockchain.EventWithSubscription) error {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		if err := cb.BlockchainEvent(event); err != nil {
 			return err
 		}
@@ -247,8 +247,8 @@ func (e *Ethereum) Init(ctx context.Context, config config.Section, metrics metr
 	return nil
 }
 
-func (e *Ethereum) SetHandler(handler blockchain.Callbacks) {
-	e.callbacks.handlers = append(e.callbacks.handlers, handler)
+func (e *Ethereum) RegisterListener(listener blockchain.Callbacks) {
+	e.callbacks.listeners = append(e.callbacks.listeners, listener)
 }
 
 func (e *Ethereum) Start() (err error) {
diff --git a/internal/blockchain/fabric/fabric.go b/internal/blockchain/fabric/fabric.go
index 353e40812..efea17ce1 100644
--- a/internal/blockchain/fabric/fabric.go
+++ b/internal/blockchain/fabric/fabric.go
@@ -72,17 +72,17 @@ type Fabric struct {
 }
 
 type callbacks struct {
-	handlers []blockchain.Callbacks
+	listeners []blockchain.Callbacks
 }
 
 func (cb *callbacks) BlockchainOpUpdate(plugin blockchain.Plugin, nsOpID string, txState blockchain.TransactionStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		cb.BlockchainOpUpdate(plugin, nsOpID, txState, blockchainTXID, errorMessage, opOutput)
 	}
 }
 
 func (cb *callbacks) BatchPinComplete(batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		if err := cb.BatchPinComplete(batch, signingKey); err != nil {
 			return err
 		}
@@ -91,7 +91,7 @@ func (cb *callbacks) BatchPinComplete(batch *blockchain.BatchPin, signingKey *co
 }
 
 func (cb *callbacks) BlockchainNetworkAction(action string, event *blockchain.Event, signingKey *core.VerifierRef) error {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		if err := cb.BlockchainNetworkAction(action, event, signingKey); err != nil {
 			return err
 		}
@@ -100,7 +100,7 @@ func (cb *callbacks) BlockchainNetworkAction(action string, event *blockchain.Ev
 }
 
 func (cb *callbacks) BlockchainEvent(event *blockchain.EventWithSubscription) error {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		if err := cb.BlockchainEvent(event); err != nil {
 			return err
 		}
@@ -322,8 +322,8 @@ func (f *Fabric) TerminateContract(ctx context.Context, contracts *core.FireFlyC
 	return f.ConfigureContract(ctx, contracts)
 }
 
-func (f *Fabric) SetHandler(handler blockchain.Callbacks) {
-	f.callbacks.handlers = append(f.callbacks.handlers, handler)
+func (f *Fabric) RegisterListener(listener blockchain.Callbacks) {
+	f.callbacks.listeners = append(f.callbacks.listeners, listener)
 }
 
 func (f *Fabric) Start() (err error) {
diff --git a/internal/database/postgres/postgres.go b/internal/database/postgres/postgres.go
index 0277d1301..a047ccd85 100644
--- a/internal/database/postgres/postgres.go
+++ b/internal/database/postgres/postgres.go
@@ -42,8 +42,8 @@ func (psql *Postgres) Init(ctx context.Context, config config.Section) error {
 	return psql.SQLCommon.Init(ctx, psql, config, capabilities)
 }
 
-func (psql *Postgres) SetHandler(handler database.Callbacks) {
-	psql.SQLCommon.SetHandler(handler)
+func (psql *Postgres) RegisterListener(listener database.Callbacks) {
+	psql.SQLCommon.RegisterListener(listener)
 }
 
 func (psql *Postgres) Name() string {
diff --git a/internal/database/sqlcommon/sqlcommon.go b/internal/database/sqlcommon/sqlcommon.go
index 311748277..770be6973 100644
--- a/internal/database/sqlcommon/sqlcommon.go
+++ b/internal/database/sqlcommon/sqlcommon.go
@@ -45,35 +45,35 @@ type SQLCommon struct {
 }
 
 type callbacks struct {
-	handlers []database.Callbacks
+	listeners []database.Callbacks
 }
 
 func (cb *callbacks) OrderedUUIDCollectionNSEvent(resType database.OrderedUUIDCollectionNS, eventType core.ChangeEventType, ns string, id *fftypes.UUID, sequence int64) {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		cb.OrderedUUIDCollectionNSEvent(resType, eventType, ns, id, sequence)
 	}
 }
 
 func (cb *callbacks) OrderedCollectionNSEvent(resType database.OrderedCollectionNS, eventType core.ChangeEventType, ns string, sequence int64) {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		cb.OrderedCollectionNSEvent(resType, eventType, ns, sequence)
 	}
 }
 
 func (cb *callbacks) UUIDCollectionNSEvent(resType database.UUIDCollectionNS, eventType core.ChangeEventType, ns string, id *fftypes.UUID) {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		cb.UUIDCollectionNSEvent(resType, eventType, ns, id)
 	}
 }
 
 func (cb *callbacks) UUIDCollectionEvent(resType database.UUIDCollection, eventType core.ChangeEventType, id *fftypes.UUID) {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		cb.UUIDCollectionEvent(resType, eventType, id)
 	}
 }
 
 func (cb *callbacks) HashCollectionNSEvent(resType database.HashCollectionNS, eventType core.ChangeEventType, ns string, hash *fftypes.Bytes32) {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		cb.HashCollectionNSEvent(resType, eventType, ns, hash)
 	}
 }
@@ -130,8 +130,8 @@ func (s *SQLCommon) Init(ctx context.Context, provider Provider, config config.S
 	return nil
 }
 
-func (s *SQLCommon) SetHandler(handler database.Callbacks) {
-	s.callbacks.handlers = append(s.callbacks.handlers, handler)
+func (s *SQLCommon) RegisterListener(listener database.Callbacks) {
+	s.callbacks.listeners = append(s.callbacks.listeners, listener)
 }
 
 func (s *SQLCommon) Capabilities() *database.Capabilities { return s.capabilities }
diff --git a/internal/database/sqlite3/sqlite3.go b/internal/database/sqlite3/sqlite3.go
index a8402cb90..d0eafb3d4 100644
--- a/internal/database/sqlite3/sqlite3.go
+++ b/internal/database/sqlite3/sqlite3.go
@@ -58,8 +58,8 @@ func (sqlite *SQLite3) Init(ctx context.Context, config config.Section) error {
 	return sqlite.SQLCommon.Init(ctx, sqlite, config, capabilities)
 }
 
-func (sqlite *SQLite3) SetHandler(handler database.Callbacks) {
-	sqlite.SQLCommon.SetHandler(handler)
+func (sqlite *SQLite3) RegisterListener(listener database.Callbacks) {
+	sqlite.SQLCommon.RegisterListener(listener)
 }
 
 func (sqlite *SQLite3) Name() string {
diff --git a/internal/dataexchange/ffdx/ffdx.go b/internal/dataexchange/ffdx/ffdx.go
index e4359d4e0..eed9b14b7 100644
--- a/internal/dataexchange/ffdx/ffdx.go
+++ b/internal/dataexchange/ffdx/ffdx.go
@@ -50,11 +50,11 @@ type FFDX struct {
 }
 
 type callbacks struct {
-	handlers []dataexchange.Callbacks
+	listeners []dataexchange.Callbacks
 }
 
 func (cb *callbacks) DXEvent(event dataexchange.DXEvent) {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		cb.DXEvent(event)
 	}
 }
@@ -148,8 +148,8 @@ func (h *FFDX) SetNodes(nodes []fftypes.JSONObject) {
 	h.nodes = nodes
 }
 
-func (h *FFDX) SetHandler(handler dataexchange.Callbacks) {
-	h.callbacks.handlers = append(h.callbacks.handlers, handler)
+func (h *FFDX) RegisterListener(listener dataexchange.Callbacks) {
+	h.callbacks.listeners = append(h.callbacks.listeners, listener)
 }
 
 func (h *FFDX) Start() error {
diff --git a/internal/events/eifactory/factory.go b/internal/events/eifactory/factory.go
index 635765b47..01f46ecdf 100644
--- a/internal/events/eifactory/factory.go
+++ b/internal/events/eifactory/factory.go
@@ -19,6 +19,7 @@ package eifactory
 import (
 	"context"
 
+	"github.com/hyperledger/firefly-common/pkg/config"
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/events/system"
@@ -41,6 +42,12 @@ func init() {
 	}
 }
 
+func InitConfig(config config.Section) {
+	for _, plugin := range plugins {
+		plugin.InitConfig(config.SubSection(plugin.Name()))
+	}
+}
+
 func GetPlugin(ctx context.Context, pluginType string) (events.Plugin, error) {
 	plugin, ok := pluginsByName[pluginType]
 	if !ok {
diff --git a/internal/events/event_manager.go b/internal/events/event_manager.go
index 0ac33010d..93cc81ecd 100644
--- a/internal/events/event_manager.go
+++ b/internal/events/event_manager.go
@@ -46,7 +46,6 @@ import (
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/hyperledger/firefly/pkg/dataexchange"
-	"github.com/hyperledger/firefly/pkg/events"
 	"github.com/hyperledger/firefly/pkg/sharedstorage"
 	"github.com/hyperledger/firefly/pkg/tokens"
 	"github.com/karlseguin/ccache"
@@ -113,7 +112,7 @@ type eventManager struct {
 	chainListenerCacheTTL time.Duration
 }
 
-func NewEventManager(ctx context.Context, ns string, ni sysmessaging.LocalNodeInfo, si sharedstorage.Plugin, di database.Plugin, bi blockchain.Plugin, im identity.Manager, dh definitions.DefinitionHandler, dm data.Manager, bm broadcast.Manager, pm privatemessaging.Manager, am assets.Manager, sd shareddownload.Manager, mm metrics.Manager, txHelper txcommon.Helper, transports map[string]events.Plugin) (EventManager, error) {
+func NewEventManager(ctx context.Context, ns string, ni sysmessaging.LocalNodeInfo, si sharedstorage.Plugin, di database.Plugin, bi blockchain.Plugin, im identity.Manager, dh definitions.DefinitionHandler, dm data.Manager, bm broadcast.Manager, pm privatemessaging.Manager, am assets.Manager, sd shareddownload.Manager, mm metrics.Manager, txHelper txcommon.Helper) (EventManager, error) {
 	if ni == nil || si == nil || di == nil || bi == nil || im == nil || dh == nil || dm == nil || bm == nil || pm == nil || am == nil {
 		return nil, i18n.NewError(ctx, coremsgs.MsgInitializationNilDepError, "EventManager")
 	}
@@ -151,7 +150,7 @@ func NewEventManager(ctx context.Context, ns string, ni sysmessaging.LocalNodeIn
 	em.blobReceiver = newBlobReceiver(ctx, em.aggregator)
 
 	var err error
-	if em.subManager, err = newSubscriptionManager(ctx, ns, di, dm, newEventNotifier, bm, pm, txHelper, transports); err != nil {
+	if em.subManager, err = newSubscriptionManager(ctx, di, dm, newEventNotifier, bm, pm, txHelper); err != nil {
 		return nil, err
 	}
 
@@ -253,9 +252,9 @@ func (em *eventManager) GetPlugins() []*core.NodeStatusPlugin {
 	eventsArray := make([]*core.NodeStatusPlugin, 0)
 	plugins := em.subManager.transports
 
-	for name := range plugins {
+	for _, plugin := range plugins {
 		eventsArray = append(eventsArray, &core.NodeStatusPlugin{
-			PluginType: name,
+			PluginType: plugin.Name(),
 		})
 	}
 
diff --git a/internal/events/subscription_manager.go b/internal/events/subscription_manager.go
index 15b6f462d..07860ed9c 100644
--- a/internal/events/subscription_manager.go
+++ b/internal/events/subscription_manager.go
@@ -30,6 +30,8 @@ import (
 	"github.com/hyperledger/firefly/internal/coreconfig"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/data"
+	"github.com/hyperledger/firefly/internal/events/eifactory"
+	"github.com/hyperledger/firefly/internal/events/system"
 	"github.com/hyperledger/firefly/internal/privatemessaging"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/core"
@@ -73,7 +75,6 @@ type connection struct {
 
 type subscriptionManager struct {
 	ctx                       context.Context
-	namespace                 string
 	database                  database.Plugin
 	data                      data.Manager
 	txHelper                  txcommon.Helper
@@ -91,14 +92,13 @@ type subscriptionManager struct {
 	retry                     retry.Retry
 }
 
-func newSubscriptionManager(ctx context.Context, ns string, di database.Plugin, dm data.Manager, en *eventNotifier, bm broadcast.Manager, pm privatemessaging.Manager, txHelper txcommon.Helper, transports map[string]events.Plugin) (*subscriptionManager, error) {
+func newSubscriptionManager(ctx context.Context, di database.Plugin, dm data.Manager, en *eventNotifier, bm broadcast.Manager, pm privatemessaging.Manager, txHelper txcommon.Helper) (*subscriptionManager, error) {
 	ctx, cancelCtx := context.WithCancel(ctx)
 	sm := &subscriptionManager{
 		ctx:                       ctx,
-		namespace:                 ns,
 		database:                  di,
 		data:                      dm,
-		transports:                transports,
+		transports:                make(map[string]events.Plugin),
 		connections:               make(map[string]*connection),
 		durableSubs:               make(map[fftypes.UUID]*subscription),
 		newOrUpdatedSubscriptions: make(chan *fftypes.UUID),
@@ -116,13 +116,42 @@ func newSubscriptionManager(ctx context.Context, ns string, di database.Plugin,
 		},
 	}
 
-	for _, ei := range sm.transports {
-		if err := ei.SetHandler(sm.namespace, &boundCallbacks{sm: sm, ei: ei}); err != nil {
-			return nil, err
+	err := sm.loadTransports()
+	if err == nil {
+		err = sm.initTransports()
+	}
+	return sm, err
+}
+
+func (sm *subscriptionManager) loadTransports() error {
+	var err error
+	enabledTransports := config.GetStringSlice(coreconfig.EventTransportsEnabled)
+	uniqueTransports := make(map[string]bool)
+	for _, transport := range enabledTransports {
+		uniqueTransports[transport] = true
+	}
+	// Cannot disable the internal listener
+	uniqueTransports[system.SystemEventsTransport] = true
+	for transport := range uniqueTransports {
+		sm.transports[transport], err = eifactory.GetPlugin(sm.ctx, transport)
+		if err != nil {
+			return err
 		}
 	}
+	return nil
+}
 
-	return sm, nil
+func (sm *subscriptionManager) initTransports() error {
+	var err error
+	for _, ei := range sm.transports {
+		config := config.RootSection("events").SubSection(ei.Name())
+		ei.InitConfig(config)
+		err = ei.Init(sm.ctx, config, &boundCallbacks{sm: sm, ei: ei})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 func (sm *subscriptionManager) start() error {
diff --git a/internal/events/system/events.go b/internal/events/system/events.go
index 50f241d8d..9aa863a3b 100644
--- a/internal/events/system/events.go
+++ b/internal/events/system/events.go
@@ -35,7 +35,7 @@ const (
 type Events struct {
 	ctx          context.Context
 	capabilities *events.Capabilities
-	callbacks    map[string]events.Callbacks
+	callbacks    events.Callbacks
 	mux          sync.Mutex
 	listeners    map[string][]EventListener
 	connID       string
@@ -46,22 +46,17 @@ type EventListener func(event *core.EventDelivery) error
 
 func (se *Events) Name() string { return SystemEventsTransport }
 
-func (se *Events) Init(ctx context.Context, config config.Section) (err error) {
+func (se *Events) Init(ctx context.Context, config config.Section, callbacks events.Callbacks) (err error) {
 	*se = Events{
 		ctx:          ctx,
 		capabilities: &events.Capabilities{},
-		callbacks:    make(map[string]events.Callbacks),
+		callbacks:    callbacks,
 		listeners:    make(map[string][]EventListener),
 		readAhead:    uint16(config.GetInt(SystemEventsConfReadAhead)),
 		connID:       fftypes.ShortID(),
 	}
-	return nil
-}
-
-func (se *Events) SetHandler(namespace string, handler events.Callbacks) error {
-	se.callbacks[namespace] = handler
 	// We have a single logical connection, that matches all subscriptions
-	return handler.RegisterConnection(se.connID, func(sr core.SubscriptionRef) bool { return true })
+	return callbacks.RegisterConnection(se.connID, func(sr core.SubscriptionRef) bool { return true })
 }
 
 func (se *Events) Capabilities() *events.Capabilities {
@@ -75,21 +70,19 @@ func (se *Events) ValidateOptions(options *core.SubscriptionOptions) error {
 func (se *Events) AddListener(ns string, el EventListener) error {
 	no := false
 	newest := core.SubOptsFirstEventNewest
-	if cb, ok := se.callbacks[ns]; ok {
-		err := cb.EphemeralSubscription(se.connID, ns, &core.SubscriptionFilter{ /* all events */ }, &core.SubscriptionOptions{
-			SubscriptionCoreOptions: core.SubscriptionCoreOptions{
-				WithData:   &no,
-				ReadAhead:  &se.readAhead,
-				FirstEvent: &newest,
-			},
-		})
-		if err != nil {
-			return err
-		}
-		se.mux.Lock()
-		se.listeners[ns] = append(se.listeners[ns], el)
-		se.mux.Unlock()
+	err := se.callbacks.EphemeralSubscription(se.connID, ns, &core.SubscriptionFilter{ /* all events */ }, &core.SubscriptionOptions{
+		SubscriptionCoreOptions: core.SubscriptionCoreOptions{
+			WithData:   &no,
+			ReadAhead:  &se.readAhead,
+			FirstEvent: &newest,
+		},
+	})
+	if err != nil {
+		return err
 	}
+	se.mux.Lock()
+	se.listeners[ns] = append(se.listeners[ns], el)
+	se.mux.Unlock()
 	return nil
 }
 
@@ -105,13 +98,11 @@ func (se *Events) DeliveryRequest(connID string, sub *core.Subscription, event *
 			}
 		}
 	}
-	if cb, ok := se.callbacks[sub.Namespace]; ok {
-		cb.DeliveryResponse(connID, &core.EventDeliveryResponse{
-			ID:           event.ID,
-			Rejected:     false,
-			Subscription: event.Subscription,
-			Reply:        nil,
-		})
-	}
+	se.callbacks.DeliveryResponse(connID, &core.EventDeliveryResponse{
+		ID:           event.ID,
+		Rejected:     false,
+		Subscription: event.Subscription,
+		Reply:        nil,
+	})
 	return nil
 }
diff --git a/internal/events/webhooks/webhooks.go b/internal/events/webhooks/webhooks.go
index 0cb284af6..127f091c3 100644
--- a/internal/events/webhooks/webhooks.go
+++ b/internal/events/webhooks/webhooks.go
@@ -40,7 +40,7 @@ import (
 type WebHooks struct {
 	ctx          context.Context
 	capabilities *events.Capabilities
-	callbacks    map[string]events.Callbacks
+	callbacks    events.Callbacks
 	client       *resty.Client
 	connID       string
 }
@@ -62,21 +62,16 @@ type whResponse struct {
 
 func (wh *WebHooks) Name() string { return "webhooks" }
 
-func (wh *WebHooks) Init(ctx context.Context, config config.Section) (err error) {
+func (wh *WebHooks) Init(ctx context.Context, config config.Section, callbacks events.Callbacks) (err error) {
 	*wh = WebHooks{
 		ctx:          ctx,
 		capabilities: &events.Capabilities{},
-		callbacks:    make(map[string]events.Callbacks),
+		callbacks:    callbacks,
 		client:       ffresty.New(ctx, config),
 		connID:       fftypes.ShortID(),
 	}
-	return nil
-}
-
-func (wh *WebHooks) SetHandler(namespace string, handler events.Callbacks) error {
-	wh.callbacks[namespace] = handler
 	// We have a single logical connection, that matches all subscriptions
-	return handler.RegisterConnection(wh.connID, func(sr core.SubscriptionRef) bool { return true })
+	return callbacks.RegisterConnection(wh.connID, func(sr core.SubscriptionRef) bool { return true })
 }
 
 func (wh *WebHooks) Capabilities() *events.Capabilities {
@@ -291,28 +286,26 @@ func (wh *WebHooks) doDelivery(connID string, reply bool, sub *core.Subscription
 		if req != nil && req.replyTx != "" {
 			txType = fftypes.FFEnum(strings.ToLower(req.replyTx))
 		}
-		if cb, ok := wh.callbacks[sub.Namespace]; ok {
-			cb.DeliveryResponse(connID, &core.EventDeliveryResponse{
-				ID:           event.ID,
-				Rejected:     false,
-				Subscription: event.Subscription,
-				Reply: &core.MessageInOut{
-					Message: core.Message{
-						Header: core.MessageHeader{
-							CID:    event.Message.Header.ID,
-							Group:  event.Message.Header.Group,
-							Type:   event.Message.Header.Type,
-							Topics: event.Message.Header.Topics,
-							Tag:    sub.Options.TransportOptions().GetString("replytag"),
-							TxType: txType,
-						},
-					},
-					InlineData: core.InlineData{
-						{Value: fftypes.JSONAnyPtrBytes(b)},
+		wh.callbacks.DeliveryResponse(connID, &core.EventDeliveryResponse{
+			ID:           event.ID,
+			Rejected:     false,
+			Subscription: event.Subscription,
+			Reply: &core.MessageInOut{
+				Message: core.Message{
+					Header: core.MessageHeader{
+						CID:    event.Message.Header.ID,
+						Group:  event.Message.Header.Group,
+						Type:   event.Message.Header.Type,
+						Topics: event.Message.Header.Topics,
+						Tag:    sub.Options.TransportOptions().GetString("replytag"),
+						TxType: txType,
 					},
 				},
-			})
-		}
+				InlineData: core.InlineData{
+					{Value: fftypes.JSONAnyPtrBytes(b)},
+				},
+			},
+		})
 	}
 	return nil
 }
@@ -329,13 +322,11 @@ func (wh *WebHooks) DeliveryRequest(connID string, sub *core.Subscription, event
 		// avoid loops - and there's no way for us to detect here if a user has configured correctly
 		// to avoid a loop.
 		log.L(wh.ctx).Debugf("Webhook subscription with reply enabled called with reply event '%s'", event.ID)
-		if cb, ok := wh.callbacks[sub.Namespace]; ok {
-			cb.DeliveryResponse(connID, &core.EventDeliveryResponse{
-				ID:           event.ID,
-				Rejected:     false,
-				Subscription: event.Subscription,
-			})
-		}
+		wh.callbacks.DeliveryResponse(connID, &core.EventDeliveryResponse{
+			ID:           event.ID,
+			Rejected:     false,
+			Subscription: event.Subscription,
+		})
 		return nil
 	}
 
diff --git a/internal/events/websockets/websockets.go b/internal/events/websockets/websockets.go
index 8b3d0ecc4..08fb57089 100644
--- a/internal/events/websockets/websockets.go
+++ b/internal/events/websockets/websockets.go
@@ -33,7 +33,7 @@ import (
 type WebSockets struct {
 	ctx          context.Context
 	capabilities *events.Capabilities
-	callbacks    map[string]events.Callbacks
+	callbacks    events.Callbacks
 	connections  map[string]*websocketConnection
 	connMux      sync.Mutex
 	upgrader     websocket.Upgrader
@@ -41,12 +41,12 @@ type WebSockets struct {
 
 func (ws *WebSockets) Name() string { return "websockets" }
 
-func (ws *WebSockets) Init(ctx context.Context, config config.Section) error {
+func (ws *WebSockets) Init(ctx context.Context, config config.Section, callbacks events.Callbacks) error {
 	*ws = WebSockets{
 		ctx:          ctx,
 		connections:  make(map[string]*websocketConnection),
 		capabilities: &events.Capabilities{},
-		callbacks:    make(map[string]events.Callbacks),
+		callbacks:    callbacks,
 		upgrader: websocket.Upgrader{
 			ReadBufferSize:  int(config.GetByteSize(ReadBufferSize)),
 			WriteBufferSize: int(config.GetByteSize(WriteBufferSize)),
@@ -59,11 +59,6 @@ func (ws *WebSockets) Init(ctx context.Context, config config.Section) error {
 	return nil
 }
 
-func (ws *WebSockets) SetHandler(namespace string, handler events.Callbacks) error {
-	ws.callbacks[namespace] = handler
-	return nil
-}
-
 func (ws *WebSockets) Capabilities() *events.Capabilities {
 	return ws.capabilities
 }
@@ -104,25 +99,20 @@ func (ws *WebSockets) ServeHTTP(res http.ResponseWriter, req *http.Request) {
 }
 
 func (ws *WebSockets) ack(connID string, inflight *core.EventDeliveryResponse) {
-	if cb, ok := ws.callbacks[inflight.Subscription.Namespace]; ok {
-		cb.DeliveryResponse(connID, inflight)
-	}
+	ws.callbacks.DeliveryResponse(connID, inflight)
 }
 
 func (ws *WebSockets) start(wc *websocketConnection, start *core.WSStart) error {
 	if start.Namespace == "" || (!start.Ephemeral && start.Name == "") {
 		return i18n.NewError(ws.ctx, coremsgs.MsgWSInvalidStartAction)
 	}
-	if cb, ok := ws.callbacks[start.Namespace]; ok {
-		if start.Ephemeral {
-			return cb.EphemeralSubscription(wc.connID, start.Namespace, &start.Filter, &start.Options)
-		}
-		// We can have multiple subscriptions on a single connection
-		return cb.RegisterConnection(wc.connID, func(sr core.SubscriptionRef) bool {
-			return wc.durableSubMatcher(sr)
-		})
+	if start.Ephemeral {
+		return ws.callbacks.EphemeralSubscription(wc.connID, start.Namespace, &start.Filter, &start.Options)
 	}
-	return i18n.NewError(ws.ctx, coremsgs.MsgNamespaceNotExist)
+	// We can have multiple subscriptions on a single
+	return ws.callbacks.RegisterConnection(wc.connID, func(sr core.SubscriptionRef) bool {
+		return wc.durableSubMatcher(sr)
+	})
 }
 
 func (ws *WebSockets) connClosed(connID string) {
@@ -130,9 +120,7 @@ func (ws *WebSockets) connClosed(connID string) {
 	delete(ws.connections, connID)
 	ws.connMux.Unlock()
 	// Drop lock before calling back
-	for _, cb := range ws.callbacks {
-		cb.ConnectionClosed(connID)
-	}
+	ws.callbacks.ConnectionClosed(connID)
 }
 
 func (ws *WebSockets) WaitClosed() {
diff --git a/internal/identity/tbd/tbd.go b/internal/identity/tbd/tbd.go
index 1fb5aac02..197d9269c 100644
--- a/internal/identity/tbd/tbd.go
+++ b/internal/identity/tbd/tbd.go
@@ -37,7 +37,7 @@ func (tbd *TBD) Init(ctx context.Context, config config.Section) (err error) {
 	return nil
 }
 
-func (tbd *TBD) SetHandler(handler identity.Callbacks) {
+func (tbd *TBD) RegisterListener(listener identity.Callbacks) {
 }
 
 func (tbd *TBD) Start() error {
diff --git a/internal/namespace/manager.go b/internal/namespace/manager.go
index 5081d157e..91038c327 100644
--- a/internal/namespace/manager.go
+++ b/internal/namespace/manager.go
@@ -28,8 +28,6 @@ import (
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/database/difactory"
 	"github.com/hyperledger/firefly/internal/dataexchange/dxfactory"
-	"github.com/hyperledger/firefly/internal/events/eifactory"
-	"github.com/hyperledger/firefly/internal/events/system"
 	"github.com/hyperledger/firefly/internal/identity/iifactory"
 	"github.com/hyperledger/firefly/internal/metrics"
 	"github.com/hyperledger/firefly/internal/orchestrator"
@@ -40,7 +38,6 @@ import (
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/hyperledger/firefly/pkg/dataexchange"
-	"github.com/hyperledger/firefly/pkg/events"
 	"github.com/hyperledger/firefly/pkg/identity"
 	"github.com/hyperledger/firefly/pkg/sharedstorage"
 	"github.com/hyperledger/firefly/pkg/tokens"
@@ -93,7 +90,6 @@ type namespaceManager struct {
 		sharedstorage map[string]sharedStoragePlugin
 		dataexchange  map[string]dataExchangePlugin
 		tokens        map[string]tokensPlugin
-		events        map[string]eventsPlugin
 	}
 	metricsEnabled bool
 	metrics        metrics.Manager
@@ -131,11 +127,6 @@ type identityPlugin struct {
 	plugin identity.Plugin
 }
 
-type eventsPlugin struct {
-	config config.Section
-	plugin events.Plugin
-}
-
 func NewNamespaceManager(withDefaults bool) Manager {
 	nm := &namespaceManager{
 		namespaces:     make(map[string]*namespace),
@@ -283,13 +274,6 @@ func (nm *namespaceManager) loadPlugins(ctx context.Context) (err error) {
 		}
 	}
 
-	if nm.plugins.events == nil {
-		nm.plugins.events, err = nm.getEventPlugins(ctx)
-		if err != nil {
-			return err
-		}
-	}
-
 	return nil
 }
 
@@ -552,7 +536,7 @@ func (nm *namespaceManager) initPlugins(ctx context.Context) (err error) {
 		if err = entry.plugin.Init(ctx, entry.config); err != nil {
 			return err
 		}
-		entry.plugin.SetHandler(nm)
+		entry.plugin.RegisterListener(nm)
 	}
 	for _, entry := range nm.plugins.blockchain {
 		if err = entry.plugin.Init(ctx, entry.config, nm.metrics); err != nil {
@@ -574,11 +558,6 @@ func (nm *namespaceManager) initPlugins(ctx context.Context) (err error) {
 			return err
 		}
 	}
-	for _, entry := range nm.plugins.events {
-		if err = entry.plugin.Init(nm.ctx, entry.config); err != nil {
-			return err
-		}
-	}
 	return nil
 }
 
@@ -682,11 +661,6 @@ func (nm *namespaceManager) loadNamespace(ctx context.Context, name string, inde
 		return nil, err
 	}
 
-	p.Events = make(map[string]events.Plugin, len(nm.plugins.events))
-	for name, entry := range nm.plugins.events {
-		p.Events[name] = entry.plugin
-	}
-
 	return &namespace{
 		description: conf.GetString(coreconfig.NamespaceDescription),
 		config:      config,
@@ -857,29 +831,3 @@ func (nm *namespaceManager) ResolveOperationByNamespacedID(ctx context.Context,
 	}
 	return or.Operations().ResolveOperationByID(ctx, ns, u, op)
 }
-
-func (nm *namespaceManager) getEventPlugins(ctx context.Context) (plugins map[string]eventsPlugin, err error) {
-	plugins = make(map[string]eventsPlugin)
-	enabledTransports := config.GetStringSlice(coreconfig.EventTransportsEnabled)
-	uniqueTransports := make(map[string]bool)
-	for _, transport := range enabledTransports {
-		uniqueTransports[transport] = true
-	}
-	// Cannot disable the internal listener
-	uniqueTransports[system.SystemEventsTransport] = true
-	for transport := range uniqueTransports {
-		plugin, err := eifactory.GetPlugin(ctx, transport)
-		if err != nil {
-			return nil, err
-		}
-
-		name := plugin.Name()
-		section := config.RootSection("events").SubSection(name)
-		plugin.InitConfig(section)
-		plugins[name] = eventsPlugin{
-			config: section,
-			plugin: plugin,
-		}
-	}
-	return plugins, err
-}
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index c535abd5b..4b00e0996 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -43,7 +43,6 @@ import (
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/hyperledger/firefly/pkg/dataexchange"
-	eventsplugin "github.com/hyperledger/firefly/pkg/events"
 	idplugin "github.com/hyperledger/firefly/pkg/identity"
 	"github.com/hyperledger/firefly/pkg/sharedstorage"
 	"github.com/hyperledger/firefly/pkg/tokens"
@@ -152,7 +151,6 @@ type Plugins struct {
 	DataExchange  DataExchangePlugin
 	Database      DatabasePlugin
 	Tokens        []TokensPlugin
-	Events        map[string]eventsplugin.Plugin
 }
 
 type Config struct {
@@ -353,8 +351,8 @@ func (or *orchestrator) Operations() operations.Manager {
 }
 
 func (or *orchestrator) initPlugins(ctx context.Context) (err error) {
-	or.plugins.Database.Plugin.SetHandler(or)
-	or.plugins.Blockchain.Plugin.SetHandler(&or.bc)
+	or.plugins.Database.Plugin.RegisterListener(or)
+	or.plugins.Blockchain.Plugin.RegisterListener(&or.bc)
 
 	fb := database.IdentityQueryFactory.NewFilter(ctx)
 	nodes, _, err := or.database().GetIdentities(ctx, or.namespace, fb.And(
@@ -368,12 +366,12 @@ func (or *orchestrator) initPlugins(ctx context.Context) (err error) {
 		nodeInfo[i] = node.Profile
 	}
 	or.plugins.DataExchange.Plugin.SetNodes(nodeInfo)
-	or.plugins.DataExchange.Plugin.SetHandler(&or.bc)
+	or.plugins.DataExchange.Plugin.RegisterListener(&or.bc)
 
-	or.plugins.SharedStorage.Plugin.SetHandler(&or.bc)
+	or.plugins.SharedStorage.Plugin.RegisterListener(&or.bc)
 
 	for _, token := range or.plugins.Tokens {
-		if err := token.Plugin.SetHandler(or.namespace, &or.bc); err != nil {
+		if err := token.Plugin.RegisterListener(or.namespace, &or.bc); err != nil {
 			return err
 		}
 	}
@@ -463,7 +461,7 @@ func (or *orchestrator) initComponents(ctx context.Context) (err error) {
 	}
 
 	if or.events == nil {
-		or.events, err = events.NewEventManager(ctx, or.namespace, or, or.sharedstorage(), or.database(), or.blockchain(), or.identity, or.definitions, or.data, or.broadcast, or.messaging, or.assets, or.sharedDownload, or.metrics, or.txHelper, or.plugins.Events)
+		or.events, err = events.NewEventManager(ctx, or.namespace, or, or.sharedstorage(), or.database(), or.blockchain(), or.identity, or.definitions, or.data, or.broadcast, or.messaging, or.assets, or.sharedDownload, or.metrics, or.txHelper)
 		if err != nil {
 			return err
 		}
diff --git a/internal/sharedstorage/ipfs/ipfs.go b/internal/sharedstorage/ipfs/ipfs.go
index a7a994a9e..a7d878ca4 100644
--- a/internal/sharedstorage/ipfs/ipfs.go
+++ b/internal/sharedstorage/ipfs/ipfs.go
@@ -67,7 +67,7 @@ func (i *IPFS) Init(ctx context.Context, config config.Section) error {
 	return nil
 }
 
-func (i *IPFS) SetHandler(handler sharedstorage.Callbacks) {
+func (i *IPFS) RegisterListener(listener sharedstorage.Callbacks) {
 }
 
 func (i *IPFS) Capabilities() *sharedstorage.Capabilities {
diff --git a/internal/tokens/fftokens/fftokens.go b/internal/tokens/fftokens/fftokens.go
index a068061cf..96a31f108 100644
--- a/internal/tokens/fftokens/fftokens.go
+++ b/internal/tokens/fftokens/fftokens.go
@@ -43,11 +43,11 @@ type FFTokens struct {
 }
 
 type callbacks struct {
-	handlers map[string]tokens.Callbacks
+	listeners map[string]tokens.Callbacks
 }
 
 func (cb *callbacks) TokenOpUpdate(plugin tokens.Plugin, nsOpID string, txState core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
-	for _, cb := range cb.handlers {
+	for _, cb := range cb.listeners {
 		cb.TokenOpUpdate(plugin, nsOpID, txState, blockchainTXID, errorMessage, opOutput)
 	}
 }
@@ -56,13 +56,13 @@ func (cb *callbacks) TokenPoolCreated(namespace string, plugin tokens.Plugin, po
 	if namespace == "" {
 		// Older token subscriptions don't populate namespace, so deliver the event to every listener and let them filter
 		// TODO: deprecate this path
-		for _, cb := range cb.handlers {
+		for _, cb := range cb.listeners {
 			if err := cb.TokenPoolCreated(plugin, pool); err != nil {
 				return err
 			}
 		}
 	} else {
-		if listener, ok := cb.handlers[namespace]; ok {
+		if listener, ok := cb.listeners[namespace]; ok {
 			return listener.TokenPoolCreated(plugin, pool)
 		}
 	}
@@ -73,13 +73,13 @@ func (cb *callbacks) TokensTransferred(namespace string, plugin tokens.Plugin, t
 	if namespace == "" {
 		// Older token subscriptions don't populate namespace, so deliver the event to every listener and let them filter
 		// TODO: deprecate this path
-		for _, cb := range cb.handlers {
+		for _, cb := range cb.listeners {
 			if err := cb.TokensTransferred(plugin, transfer); err != nil {
 				return err
 			}
 		}
 	} else {
-		if listener, ok := cb.handlers[namespace]; ok {
+		if listener, ok := cb.listeners[namespace]; ok {
 			return listener.TokensTransferred(plugin, transfer)
 		}
 	}
@@ -90,13 +90,13 @@ func (cb *callbacks) TokensApproved(namespace string, plugin tokens.Plugin, appr
 	if namespace == "" {
 		// Older token subscriptions don't populate namespace, so deliver the event to every listener and let them filter
 		// TODO: deprecate this path
-		for _, cb := range cb.handlers {
+		for _, cb := range cb.listeners {
 			if err := cb.TokensApproved(plugin, approval); err != nil {
 				return err
 			}
 		}
 	} else {
-		if listener, ok := cb.handlers[namespace]; ok {
+		if listener, ok := cb.listeners[namespace]; ok {
 			return listener.TokensApproved(plugin, approval)
 		}
 	}
@@ -203,7 +203,7 @@ func (ft *FFTokens) Init(ctx context.Context, name string, config config.Section
 	ft.ctx = log.WithLogField(ctx, "proto", "fftokens")
 	ft.configuredName = name
 	ft.capabilities = &tokens.Capabilities{}
-	ft.callbacks.handlers = make(map[string]tokens.Callbacks)
+	ft.callbacks.listeners = make(map[string]tokens.Callbacks)
 
 	if config.GetString(ffresty.HTTPConfigURL) == "" {
 		return i18n.NewError(ctx, coremsgs.MsgMissingPluginConfig, "url", "tokens.fftokens")
@@ -225,8 +225,8 @@ func (ft *FFTokens) Init(ctx context.Context, name string, config config.Section
 	return nil
 }
 
-func (ft *FFTokens) SetHandler(namespace string, handler tokens.Callbacks) error {
-	ft.callbacks.handlers[namespace] = handler
+func (ft *FFTokens) RegisterListener(namespace string, listener tokens.Callbacks) error {
+	ft.callbacks.listeners[namespace] = listener
 
 	res, err := ft.client.R().SetContext(ft.ctx).
 		SetBody(&tokenInit{
diff --git a/mocks/blockchainmocks/plugin.go b/mocks/blockchainmocks/plugin.go
index c2e599dd4..11eeebac0 100644
--- a/mocks/blockchainmocks/plugin.go
+++ b/mocks/blockchainmocks/plugin.go
@@ -268,9 +268,9 @@ func (_m *Plugin) QueryContract(ctx context.Context, location *fftypes.JSONAny,
 	return r0, r1
 }
 
-// SetHandler provides a mock function with given fields: handler
-func (_m *Plugin) SetHandler(handler blockchain.Callbacks) {
-	_m.Called(handler)
+// RegisterListener provides a mock function with given fields: listener
+func (_m *Plugin) RegisterListener(listener blockchain.Callbacks) {
+	_m.Called(listener)
 }
 
 // Start provides a mock function with given fields:
diff --git a/mocks/databasemocks/plugin.go b/mocks/databasemocks/plugin.go
index a70229605..aa4c6bde9 100644
--- a/mocks/databasemocks/plugin.go
+++ b/mocks/databasemocks/plugin.go
@@ -2368,6 +2368,11 @@ func (_m *Plugin) Name() string {
 	return r0
 }
 
+// RegisterListener provides a mock function with given fields: listener
+func (_m *Plugin) RegisterListener(listener database.Callbacks) {
+	_m.Called(listener)
+}
+
 // ReplaceMessage provides a mock function with given fields: ctx, message
 func (_m *Plugin) ReplaceMessage(ctx context.Context, message *core.Message) error {
 	ret := _m.Called(ctx, message)
@@ -2410,11 +2415,6 @@ func (_m *Plugin) RunAsGroup(ctx context.Context, fn func(context.Context) error
 	return r0
 }
 
-// SetHandler provides a mock function with given fields: handler
-func (_m *Plugin) SetHandler(handler database.Callbacks) {
-	_m.Called(handler)
-}
-
 // UpdateBatch provides a mock function with given fields: ctx, id, update
 func (_m *Plugin) UpdateBatch(ctx context.Context, id *fftypes.UUID, update database.Update) error {
 	ret := _m.Called(ctx, id, update)
diff --git a/mocks/dataexchangemocks/plugin.go b/mocks/dataexchangemocks/plugin.go
index 9b1c406c3..7f8268aa0 100644
--- a/mocks/dataexchangemocks/plugin.go
+++ b/mocks/dataexchangemocks/plugin.go
@@ -160,6 +160,11 @@ func (_m *Plugin) Name() string {
 	return r0
 }
 
+// RegisterListener provides a mock function with given fields: listener
+func (_m *Plugin) RegisterListener(listener dataexchange.Callbacks) {
+	_m.Called(listener)
+}
+
 // SendMessage provides a mock function with given fields: ctx, nsOpID, peerID, data
 func (_m *Plugin) SendMessage(ctx context.Context, nsOpID string, peerID string, data []byte) error {
 	ret := _m.Called(ctx, nsOpID, peerID, data)
@@ -174,11 +179,6 @@ func (_m *Plugin) SendMessage(ctx context.Context, nsOpID string, peerID string,
 	return r0
 }
 
-// SetHandler provides a mock function with given fields: handler
-func (_m *Plugin) SetHandler(handler dataexchange.Callbacks) {
-	_m.Called(handler)
-}
-
 // SetNodes provides a mock function with given fields: nodes
 func (_m *Plugin) SetNodes(nodes []fftypes.JSONObject) {
 	_m.Called(nodes)
diff --git a/mocks/eventsmocks/plugin.go b/mocks/eventsmocks/plugin.go
index 792f069f7..f387b3fb9 100644
--- a/mocks/eventsmocks/plugin.go
+++ b/mocks/eventsmocks/plugin.go
@@ -49,13 +49,13 @@ func (_m *Plugin) DeliveryRequest(connID string, sub *core.Subscription, event *
 	return r0
 }
 
-// Init provides a mock function with given fields: ctx, _a1
-func (_m *Plugin) Init(ctx context.Context, _a1 config.Section) error {
-	ret := _m.Called(ctx, _a1)
+// Init provides a mock function with given fields: ctx, _a1, callbacks
+func (_m *Plugin) Init(ctx context.Context, _a1 config.Section, callbacks events.Callbacks) error {
+	ret := _m.Called(ctx, _a1, callbacks)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, config.Section) error); ok {
-		r0 = rf(ctx, _a1)
+	if rf, ok := ret.Get(0).(func(context.Context, config.Section, events.Callbacks) error); ok {
+		r0 = rf(ctx, _a1, callbacks)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -82,20 +82,6 @@ func (_m *Plugin) Name() string {
 	return r0
 }
 
-// SetHandler provides a mock function with given fields: namespace, handler
-func (_m *Plugin) SetHandler(namespace string, handler events.Callbacks) error {
-	ret := _m.Called(namespace, handler)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(string, events.Callbacks) error); ok {
-		r0 = rf(namespace, handler)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
 // ValidateOptions provides a mock function with given fields: options
 func (_m *Plugin) ValidateOptions(options *core.SubscriptionOptions) error {
 	ret := _m.Called(options)
diff --git a/mocks/identitymocks/plugin.go b/mocks/identitymocks/plugin.go
index 1cf2899ed..8cb0c62fe 100644
--- a/mocks/identitymocks/plugin.go
+++ b/mocks/identitymocks/plugin.go
@@ -66,9 +66,9 @@ func (_m *Plugin) Name() string {
 	return r0
 }
 
-// SetHandler provides a mock function with given fields: handler
-func (_m *Plugin) SetHandler(handler identity.Callbacks) {
-	_m.Called(handler)
+// RegisterListener provides a mock function with given fields: listener
+func (_m *Plugin) RegisterListener(listener identity.Callbacks) {
+	_m.Called(listener)
 }
 
 // Start provides a mock function with given fields:
diff --git a/mocks/sharedstoragemocks/plugin.go b/mocks/sharedstoragemocks/plugin.go
index e53897766..2e1f13529 100644
--- a/mocks/sharedstoragemocks/plugin.go
+++ b/mocks/sharedstoragemocks/plugin.go
@@ -91,9 +91,9 @@ func (_m *Plugin) Name() string {
 	return r0
 }
 
-// SetHandler provides a mock function with given fields: handler
-func (_m *Plugin) SetHandler(handler sharedstorage.Callbacks) {
-	_m.Called(handler)
+// RegisterListener provides a mock function with given fields: listener
+func (_m *Plugin) RegisterListener(listener sharedstorage.Callbacks) {
+	_m.Called(listener)
 }
 
 // UploadData provides a mock function with given fields: ctx, data
diff --git a/mocks/tokenmocks/plugin.go b/mocks/tokenmocks/plugin.go
index a0049ea1a..a1a3d3548 100644
--- a/mocks/tokenmocks/plugin.go
+++ b/mocks/tokenmocks/plugin.go
@@ -138,13 +138,13 @@ func (_m *Plugin) Name() string {
 	return r0
 }
 
-// SetHandler provides a mock function with given fields: namespace, handler
-func (_m *Plugin) SetHandler(namespace string, handler tokens.Callbacks) error {
-	ret := _m.Called(namespace, handler)
+// RegisterListener provides a mock function with given fields: namespace, listener
+func (_m *Plugin) RegisterListener(namespace string, listener tokens.Callbacks) error {
+	ret := _m.Called(namespace, listener)
 
 	var r0 error
 	if rf, ok := ret.Get(0).(func(string, tokens.Callbacks) error); ok {
-		r0 = rf(namespace, handler)
+		r0 = rf(namespace, listener)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/pkg/blockchain/plugin.go b/pkg/blockchain/plugin.go
index 84ce9caaf..4ae497d9b 100644
--- a/pkg/blockchain/plugin.go
+++ b/pkg/blockchain/plugin.go
@@ -35,8 +35,8 @@ type Plugin interface {
 	// Init initializes the plugin, with configuration
 	Init(ctx context.Context, config config.Section, metrics metrics.Manager) error
 
-	// SetHandler registers a handler to receive callbacks
-	SetHandler(handler Callbacks)
+	// RegisterListener registers a listener to receive callbacks
+	RegisterListener(listener Callbacks)
 
 	// ConfigureContract initializes the subscription to the FireFly contract
 	// - Checks the provided contract info against the plugin's configuration, and updates it as needed
diff --git a/pkg/database/plugin.go b/pkg/database/plugin.go
index 4da1fc2a1..61ed48fc4 100644
--- a/pkg/database/plugin.go
+++ b/pkg/database/plugin.go
@@ -53,8 +53,8 @@ type Plugin interface {
 	// Init initializes the plugin, with configuration
 	Init(ctx context.Context, config config.Section) error
 
-	// SetHandler registers a handler to receive callbacks
-	SetHandler(handler Callbacks)
+	// RegisterListener registers a listener to receive callbacks
+	RegisterListener(listener Callbacks)
 
 	// Capabilities returns capabilities - not called until after Init
 	Capabilities() *Capabilities
diff --git a/pkg/dataexchange/plugin.go b/pkg/dataexchange/plugin.go
index 7a5746752..3a6db6e4f 100644
--- a/pkg/dataexchange/plugin.go
+++ b/pkg/dataexchange/plugin.go
@@ -66,8 +66,8 @@ type Plugin interface {
 	// SetNodes initializes the known nodes from the database
 	SetNodes(nodes []fftypes.JSONObject)
 
-	// SetHandler registers a handler to receive callbacks
-	SetHandler(handler Callbacks)
+	// RegisterListener registers a listener to receive callbacks
+	RegisterListener(listener Callbacks)
 
 	// Data exchange interface must not deliver any events until start is called
 	Start() error
diff --git a/pkg/events/plugin.go b/pkg/events/plugin.go
index abd2a99a7..83fa3b953 100644
--- a/pkg/events/plugin.go
+++ b/pkg/events/plugin.go
@@ -33,10 +33,8 @@ type Plugin interface {
 	InitConfig(config config.Section)
 
 	// Init initializes the plugin, with configuration
-	Init(ctx context.Context, config config.Section) error
-
-	// SetHandler registers a handler to receive callbacks
-	SetHandler(namespace string, handler Callbacks) error
+	// Returns the supported featureset of the interface
+	Init(ctx context.Context, config config.Section, callbacks Callbacks) error
 
 	// Capabilities returns capabilities - not called until after Init
 	Capabilities() *Capabilities
diff --git a/pkg/identity/plugin.go b/pkg/identity/plugin.go
index cfa43a0a5..59c9f5248 100644
--- a/pkg/identity/plugin.go
+++ b/pkg/identity/plugin.go
@@ -33,8 +33,8 @@ type Plugin interface {
 	// Init initializes the plugin, with configuration
 	Init(ctx context.Context, config config.Section) error
 
-	// SetHandler registers a handler to receive callbacks
-	SetHandler(handler Callbacks)
+	// RegisterListener registers a listener to receive callbacks
+	RegisterListener(listener Callbacks)
 
 	// Blockchain interface must not deliver any events until start is called
 	Start() error
diff --git a/pkg/sharedstorage/plugin.go b/pkg/sharedstorage/plugin.go
index ff8fcbe49..4c6585512 100644
--- a/pkg/sharedstorage/plugin.go
+++ b/pkg/sharedstorage/plugin.go
@@ -34,8 +34,8 @@ type Plugin interface {
 	// Init initializes the plugin, with configuration
 	Init(ctx context.Context, config config.Section) error
 
-	// SetHandler registers a handler to receive callbacks
-	SetHandler(handler Callbacks)
+	// RegisterListener registers a listener to receive callbacks
+	RegisterListener(listener Callbacks)
 
 	// Capabilities returns capabilities - not called until after Init
 	Capabilities() *Capabilities
diff --git a/pkg/tokens/plugin.go b/pkg/tokens/plugin.go
index 309266a72..8be3b561e 100644
--- a/pkg/tokens/plugin.go
+++ b/pkg/tokens/plugin.go
@@ -35,8 +35,8 @@ type Plugin interface {
 	// Init initializes the plugin, with configuration
 	Init(ctx context.Context, name string, config config.Section) error
 
-	// SetHandler registers a handler to receive callbacks
-	SetHandler(namespace string, handler Callbacks) error
+	// RegisterListener registers a listener to receive callbacks
+	RegisterListener(namespace string, listener Callbacks) error
 
 	// Blockchain interface must not deliver any events until start is called
 	Start() error


Test patch:
diff --git a/internal/blockchain/ethereum/ethereum_test.go b/internal/blockchain/ethereum/ethereum_test.go
index 7622918bf..df6e2d754 100644
--- a/internal/blockchain/ethereum/ethereum_test.go
+++ b/internal/blockchain/ethereum/ethereum_test.go
@@ -1096,7 +1096,7 @@ func TestHandleMessageBatchPinOK(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-b5b97a4e-a317-4053-6400-1474650efcb5"
 	e.fireflyContract.networkVersion = 1
@@ -1219,7 +1219,7 @@ func TestHandleMessageEmptyPayloadRef(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-b5b97a4e-a317-4053-6400-1474650efcb5"
 	e.fireflyContract.networkVersion = 1
@@ -1281,7 +1281,7 @@ func TestHandleMessageBatchPinExit(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-b5b97a4e-a317-4053-6400-1474650efcb5"
 	e.fireflyContract.networkVersion = 1
@@ -1506,7 +1506,7 @@ func TestHandleReceiptTXSuccess(t *testing.T) {
 	e := &Ethereum{
 		ctx:       context.Background(),
 		topic:     "topic1",
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 		wsconn:    wsm,
 	}
 
@@ -1576,7 +1576,7 @@ func TestHandleBadPayloadsAndThenReceiptFailure(t *testing.T) {
 	}`)
 
 	em := &blockchainmocks.Callbacks{}
-	e.SetHandler(em)
+	e.RegisterListener(em)
 	txsu := em.On("BlockchainOpUpdate",
 		e,
 		"ns1:"+operationID.String(),
@@ -1821,7 +1821,7 @@ func TestHandleMessageContractEvent(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-b5b97a4e-a317-4053-6400-1474650efcb5"
 
@@ -1883,7 +1883,7 @@ func TestHandleMessageContractEventError(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-b5b97a4e-a317-4053-6400-1474650efcb5"
 
@@ -3075,7 +3075,7 @@ func TestHandleNetworkAction(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-b5b97a4e-a317-4053-6400-1474650efcb5"
 
@@ -3121,7 +3121,7 @@ func TestHandleNetworkActionFail(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-b5b97a4e-a317-4053-6400-1474650efcb5"
 
diff --git a/internal/blockchain/fabric/fabric_test.go b/internal/blockchain/fabric/fabric_test.go
index 8808cad7f..e8ea554b6 100644
--- a/internal/blockchain/fabric/fabric_test.go
+++ b/internal/blockchain/fabric/fabric_test.go
@@ -943,7 +943,7 @@ func TestHandleMessageBatchPinOK(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
 
@@ -990,7 +990,7 @@ func TestHandleMessageEmptyPayloadRef(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
 
@@ -1037,7 +1037,7 @@ func TestHandleMessageBatchPinExit(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
 
@@ -1070,7 +1070,7 @@ func TestHandleMessageBatchPinEmpty(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
 
@@ -1096,7 +1096,7 @@ func TestHandleMessageUnknownEventName(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
 
@@ -1111,7 +1111,7 @@ func TestHandleMessageUnknownEventName(t *testing.T) {
 func TestHandleMessageBatchPinBadBatchHash(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
 	data := []byte(`[{
@@ -1133,7 +1133,7 @@ func TestHandleMessageBatchPinBadBatchHash(t *testing.T) {
 func TestHandleMessageBatchPinBadPin(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
 	data := []byte(`[{
@@ -1155,7 +1155,7 @@ func TestHandleMessageBatchPinBadPin(t *testing.T) {
 func TestHandleMessageBatchPinBadPayloadEncoding(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
 	data := []byte(`[{
@@ -1177,7 +1177,7 @@ func TestHandleMessageBatchPinBadPayloadEncoding(t *testing.T) {
 func TestHandleMessageBatchPinBadPayloadUUIDs(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
 	data := []byte(`[{
@@ -1199,7 +1199,7 @@ func TestHandleMessageBatchPinBadPayloadUUIDs(t *testing.T) {
 func TestHandleMessageBatchBadJSON(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	err := e.handleMessageBatch(context.Background(), []interface{}{10, 20})
 	assert.NoError(t, err)
@@ -1270,7 +1270,7 @@ func TestEventLoopUnexpectedMessage(t *testing.T) {
 		"requestPayload": "{\"from\":\"0x91d2b4381a4cd5c7c0f27565a7d4b829844c8635\",\"gas\":0,\"gasPrice\":0,\"headers\":{\"id\":\"6fb94fff-81d3-4094-567d-e031b1871694\",\"type\":\"SendTransaction\"},\"method\":{\"inputs\":[{\"internalType\":\"bytes32\",\"name\":\"txnId\",\"type\":\"bytes32\"},{\"internalType\":\"bytes32\",\"name\":\"batchId\",\"type\":\"bytes32\"},{\"internalType\":\"bytes32\",\"name\":\"payloadRef\",\"type\":\"bytes32\"}],\"name\":\"broadcastBatch\",\"outputs\":[],\"stateMutability\":\"nonpayable\",\"type\":\"function\"},\"params\":[\"12345\",\"!\",\"!\"],\"to\":\"0xd3266a857285fb75eb7df37353b4a15c8bb828f5\",\"value\":0}"
 	}`)
 	em := &blockchainmocks.Callbacks{}
-	e.SetHandler(em)
+	e.RegisterListener(em)
 	txsu := em.On("BlockchainOpUpdate",
 		e,
 		"ns1:"+operationID.String(),
@@ -1296,7 +1296,7 @@ func TestHandleReceiptTXSuccess(t *testing.T) {
 	e := &Fabric{
 		ctx:       context.Background(),
 		topic:     "topic1",
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 		wsconn:    wsm,
 	}
 
@@ -1336,7 +1336,7 @@ func TestHandleReceiptNoRequestID(t *testing.T) {
 	e := &Fabric{
 		ctx:       context.Background(),
 		topic:     "topic1",
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 		wsconn:    wsm,
 	}
 
@@ -1353,7 +1353,7 @@ func TestHandleReceiptFailedTx(t *testing.T) {
 	e := &Fabric{
 		ctx:       context.Background(),
 		topic:     "topic1",
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 		wsconn:    wsm,
 	}
 
@@ -1545,7 +1545,7 @@ func TestHandleMessageContractEvent(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-b5b97a4e-a317-4053-6400-1474650efcb5"
 
@@ -1603,7 +1603,7 @@ func TestHandleMessageContractEventBadPayload(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-b5b97a4e-a317-4053-6400-1474650efcb5"
 
@@ -1631,7 +1631,7 @@ func TestHandleMessageContractEventError(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-b5b97a4e-a317-4053-6400-1474650efcb5"
 
@@ -2030,7 +2030,7 @@ func TestHandleNetworkAction(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
 
@@ -2068,7 +2068,7 @@ func TestHandleNetworkActionFail(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
+		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
 
diff --git a/internal/database/postgres/postgres_test.go b/internal/database/postgres/postgres_test.go
index dea012436..ffc78365f 100644
--- a/internal/database/postgres/postgres_test.go
+++ b/internal/database/postgres/postgres_test.go
@@ -29,7 +29,7 @@ import (
 
 func TestPostgresProvider(t *testing.T) {
 	psql := &Postgres{}
-	psql.SetHandler(&databasemocks.Callbacks{})
+	psql.RegisterListener(&databasemocks.Callbacks{})
 	config := config.RootSection("unittest")
 	psql.InitConfig(config)
 	config.Set(sqlcommon.SQLConfDatasourceURL, "!bad connection")
diff --git a/internal/database/sqlcommon/contractapis_sql_test.go b/internal/database/sqlcommon/contractapis_sql_test.go
index 154fb681a..9a8868697 100644
--- a/internal/database/sqlcommon/contractapis_sql_test.go
+++ b/internal/database/sqlcommon/contractapis_sql_test.go
@@ -126,7 +126,7 @@ func TestContractAPIDBFailUpdate(t *testing.T) {
 func TestUpsertContractAPIIDMismatch(t *testing.T) {
 	s, db := newMockProvider().init()
 	callbacks := &databasemocks.Callbacks{}
-	s.SetHandler(callbacks)
+	s.RegisterListener(callbacks)
 	apiID := fftypes.NewUUID()
 	api := &core.ContractAPI{
 		ID:        apiID,
diff --git a/internal/database/sqlcommon/provider_mock_test.go b/internal/database/sqlcommon/provider_mock_test.go
index 91a6a2751..9671c1a05 100644
--- a/internal/database/sqlcommon/provider_mock_test.go
+++ b/internal/database/sqlcommon/provider_mock_test.go
@@ -64,7 +64,7 @@ func newMockProvider() *mockProvider {
 // init is a convenience to init for tests that aren't testing init itself
 func (mp *mockProvider) init() (*mockProvider, sqlmock.Sqlmock) {
 	_ = mp.Init(context.Background(), mp, mp.config, mp.capabilities)
-	mp.SetHandler(mp.callbacks)
+	mp.RegisterListener(mp.callbacks)
 	return mp, mp.mdb
 }
 
diff --git a/internal/database/sqlcommon/provider_sqlitego_test.go b/internal/database/sqlcommon/provider_sqlitego_test.go
index 1b6a64360..be0ea3493 100644
--- a/internal/database/sqlcommon/provider_sqlitego_test.go
+++ b/internal/database/sqlcommon/provider_sqlitego_test.go
@@ -65,7 +65,7 @@ func newSQLiteTestProvider(t *testing.T) (*sqliteGoTestProvider, func()) {
 
 	err = tp.Init(context.Background(), tp, tp.config, tp.capabilities)
 	assert.NoError(tp.t, err)
-	tp.SetHandler(tp.callbacks)
+	tp.RegisterListener(tp.callbacks)
 
 	return tp, func() {
 		tp.Close()
diff --git a/internal/database/sqlcommon/tokenpool_sql_test.go b/internal/database/sqlcommon/tokenpool_sql_test.go
index 140f52a6f..db12d4bdf 100644
--- a/internal/database/sqlcommon/tokenpool_sql_test.go
+++ b/internal/database/sqlcommon/tokenpool_sql_test.go
@@ -176,7 +176,7 @@ func TestUpsertTokenPoolFailCommit(t *testing.T) {
 func TestUpsertTokenPoolUpdateIDMismatch(t *testing.T) {
 	s, db := newMockProvider().init()
 	callbacks := &databasemocks.Callbacks{}
-	s.SetHandler(callbacks)
+	s.RegisterListener(callbacks)
 	poolID := fftypes.NewUUID()
 	pool := &core.TokenPool{
 		ID:        poolID,
diff --git a/internal/database/sqlite3/sqlite3_test.go b/internal/database/sqlite3/sqlite3_test.go
index cac1659da..78e5aa0a0 100644
--- a/internal/database/sqlite3/sqlite3_test.go
+++ b/internal/database/sqlite3/sqlite3_test.go
@@ -32,7 +32,7 @@ import (
 
 func TestSQLite3GoProvider(t *testing.T) {
 	sqlite := &SQLite3{}
-	sqlite.SetHandler(&databasemocks.Callbacks{})
+	sqlite.RegisterListener(&databasemocks.Callbacks{})
 	config := config.RootSection("unittest")
 	sqlite.InitConfig(config)
 	config.Set(sqlcommon.SQLConfDatasourceURL, "!wrong://")
diff --git a/internal/dataexchange/ffdx/ffdx_test.go b/internal/dataexchange/ffdx/ffdx_test.go
index 23e788907..25a37476d 100644
--- a/internal/dataexchange/ffdx/ffdx_test.go
+++ b/internal/dataexchange/ffdx/ffdx_test.go
@@ -443,7 +443,7 @@ func TestEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"0"}`, string(msg))
 
 	mcb := &dataexchangemocks.Callbacks{}
-	h.SetHandler(mcb)
+	h.RegisterListener(mcb)
 
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.NamespacedID() == "1" &&
@@ -557,7 +557,7 @@ func TestEventsWithManifest(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"0"}`, string(msg))
 
 	mcb := &dataexchangemocks.Callbacks{}
-	h.SetHandler(mcb)
+	h.RegisterListener(mcb)
 
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.NamespacedID() == "1" &&
@@ -585,7 +585,7 @@ func TestEventLoopReceiveClosed(t *testing.T) {
 	wsm := &wsmocks.WSClient{}
 	h := &FFDX{
 		ctx:       context.Background(),
-		callbacks: callbacks{handlers: []dataexchange.Callbacks{dxc}},
+		callbacks: callbacks{listeners: []dataexchange.Callbacks{dxc}},
 		wsconn:    wsm,
 	}
 	r := make(chan []byte)
@@ -601,7 +601,7 @@ func TestEventLoopSendClosed(t *testing.T) {
 	ctx, cancelCtx := context.WithCancel(context.Background())
 	h := &FFDX{
 		ctx:        ctx,
-		callbacks:  callbacks{handlers: []dataexchange.Callbacks{dxc}},
+		callbacks:  callbacks{listeners: []dataexchange.Callbacks{dxc}},
 		wsconn:     wsm,
 		ackChannel: make(chan *ack, 1),
 	}
@@ -622,7 +622,7 @@ func TestEventLoopClosedContext(t *testing.T) {
 	cancel()
 	h := &FFDX{
 		ctx:       ctx,
-		callbacks: callbacks{handlers: []dataexchange.Callbacks{dxc}},
+		callbacks: callbacks{listeners: []dataexchange.Callbacks{dxc}},
 		wsconn:    wsm,
 	}
 	r := make(chan []byte, 1)
diff --git a/internal/events/event_manager_test.go b/internal/events/event_manager_test.go
index 94b26d92a..2199bdf93 100644
--- a/internal/events/event_manager_test.go
+++ b/internal/events/event_manager_test.go
@@ -42,7 +42,6 @@ import (
 	"github.com/hyperledger/firefly/mocks/txcommonmocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
-	"github.com/hyperledger/firefly/pkg/events"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -81,8 +80,6 @@ func newTestEventManagerCommon(t *testing.T, metrics, dbconcurrency bool) (*even
 	mni := &sysmessagingmocks.LocalNodeInfo{}
 	mdd := &shareddownloadmocks.Manager{}
 	mmi := &metricsmocks.Manager{}
-	mev := &eventsmocks.Plugin{}
-	events := map[string]events.Plugin{"websockets": mev}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
 	mmi.On("IsMetricsEnabled").Return(metrics)
 	if metrics {
@@ -92,9 +89,7 @@ func newTestEventManagerCommon(t *testing.T, metrics, dbconcurrency bool) (*even
 	met.On("Name").Return("ut").Maybe()
 	mbi.On("VerifierType").Return(core.VerifierTypeEthAddress).Maybe()
 	mdi.On("Capabilities").Return(&database.Capabilities{Concurrency: dbconcurrency}).Maybe()
-	mev.On("SetHandler", "ns1", mock.Anything).Return(nil).Maybe()
-	mev.On("ValidateOptions", mock.Anything).Return(nil).Maybe()
-	emi, err := NewEventManager(ctx, "ns1", mni, mpi, mdi, mbi, mim, msh, mdm, mbm, mpm, mam, mdd, mmi, txHelper, events)
+	emi, err := NewEventManager(ctx, "ns1", mni, mpi, mdi, mbi, mim, msh, mdm, mbm, mpm, mam, mdd, mmi, txHelper)
 	em := emi.(*eventManager)
 	em.txHelper = &txcommonmocks.Helper{}
 	mockRunAsGroupPassthrough(mdi)
@@ -129,12 +124,12 @@ func TestStartStop(t *testing.T) {
 }
 
 func TestStartStopBadDependencies(t *testing.T) {
-	_, err := NewEventManager(context.Background(), "", nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
+	_, err := NewEventManager(context.Background(), "", nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
 	assert.Regexp(t, "FF10128", err)
 
 }
 
-func TestStartStopEventListenerFail(t *testing.T) {
+func TestStartStopBadTransports(t *testing.T) {
 	config.Set(coreconfig.EventTransportsEnabled, []string{"wrongun"})
 	defer coreconfig.Reset()
 	mdi := &databasemocks.Plugin{}
@@ -149,14 +144,11 @@ func TestStartStopEventListenerFail(t *testing.T) {
 	mam := &assetmocks.Manager{}
 	msd := &shareddownloadmocks.Manager{}
 	mm := &metricsmocks.Manager{}
-	mev := &eventsmocks.Plugin{}
-	events := map[string]events.Plugin{"websockets": mev}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
-	mdi.On("Capabilities").Return(&database.Capabilities{Concurrency: false})
+	mdi.On("Capabilities").Return(&database.Capabilities{Concurrency: false}).Maybe()
 	mbi.On("VerifierType").Return(core.VerifierTypeEthAddress)
-	mev.On("SetHandler", "ns1", mock.Anything).Return(fmt.Errorf("pop"))
-	_, err := NewEventManager(context.Background(), "ns1", mni, mpi, mdi, mbi, mim, msh, mdm, mbm, mpm, mam, msd, mm, txHelper, events)
-	assert.EqualError(t, err, "pop")
+	_, err := NewEventManager(context.Background(), "ns1", mni, mpi, mdi, mbi, mim, msh, mdm, mbm, mpm, mam, msd, mm, txHelper)
+	assert.Regexp(t, "FF10172", err)
 }
 
 func TestEmitSubscriptionEventsNoops(t *testing.T) {
@@ -406,8 +398,7 @@ func TestAddInternalListener(t *testing.T) {
 
 	conf := config.RootSection("ut.events")
 	ie.InitConfig(conf)
-	ie.Init(em.ctx, conf)
-	ie.SetHandler("ns1", cbs)
+	ie.Init(em.ctx, conf, cbs)
 	em.internalEvents = ie
 	defer cancel()
 	err := em.AddSystemEventListener("ns1", func(event *core.EventDelivery) error { return nil })
@@ -423,6 +414,12 @@ func TestGetPlugins(t *testing.T) {
 		{
 			PluginType: "websockets",
 		},
+		{
+			PluginType: "webhooks",
+		},
+		{
+			PluginType: "system",
+		},
 	}
 
 	assert.ElementsMatch(t, em.GetPlugins(), expectedPlugins)
diff --git a/internal/events/subscription_manager_test.go b/internal/events/subscription_manager_test.go
index e0213f253..19b5e5527 100644
--- a/internal/events/subscription_manager_test.go
+++ b/internal/events/subscription_manager_test.go
@@ -53,7 +53,7 @@ func newTestSubManager(t *testing.T, mei *eventsmocks.Plugin) (*subscriptionMana
 	mei.On("Init", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	mdi.On("GetEvents", mock.Anything, mock.Anything, mock.Anything).Return([]*core.Event{}, nil, nil).Maybe()
 	mdi.On("GetOffset", mock.Anything, mock.Anything, mock.Anything).Return(&core.Offset{RowID: 3333333, Current: 0}, nil).Maybe()
-	sm, err := newSubscriptionManager(ctx, "ns1", mdi, mdm, newEventNotifier(ctx, "ut"), mbm, mpm, txHelper, nil)
+	sm, err := newSubscriptionManager(ctx, mdi, mdm, newEventNotifier(ctx, "ut"), mbm, mpm, txHelper)
 	assert.NoError(t, err)
 	sm.transports = map[string]events.Plugin{
 		"ut": mei,
@@ -116,42 +116,6 @@ func TestRegisterDurableSubscriptions(t *testing.T) {
 	assert.Nil(t, sm.connections["conn2"])
 }
 
-func TestReloadDurableSubscription(t *testing.T) {
-
-	sub1 := fftypes.NewUUID()
-
-	mei := &eventsmocks.Plugin{}
-	sm, cancel := newTestSubManager(t, mei)
-	defer cancel()
-
-	sm.connections["conn1"] = &connection{
-		ei:          mei,
-		id:          "conn1",
-		transport:   "ut",
-		dispatchers: make(map[fftypes.UUID]*eventDispatcher),
-		matcher: func(sr core.SubscriptionRef) bool {
-			return sr.Namespace == "ns1" && sr.Name == "sub1"
-		},
-	}
-
-	mdi := sm.database.(*databasemocks.Plugin)
-	mdi.On("GetSubscriptions", mock.Anything, mock.Anything).Return([]*core.Subscription{
-		{SubscriptionRef: core.SubscriptionRef{
-			ID:        sub1,
-			Namespace: "ns1",
-			Name:      "sub1",
-		}, Transport: "ut"},
-	}, nil, nil)
-	mei.On("ValidateOptions", mock.Anything).Return(nil)
-	err := sm.start()
-	assert.NoError(t, err)
-
-	// Close with active conns
-	sm.close()
-	assert.Nil(t, sm.connections["conn1"])
-	assert.Nil(t, sm.connections["conn2"])
-}
-
 func TestRegisterEphemeralSubscriptions(t *testing.T) {
 	mei := &eventsmocks.Plugin{}
 	sm, cancel := newTestSubManager(t, mei)
@@ -203,6 +167,31 @@ func TestRegisterEphemeralSubscriptionsFail(t *testing.T) {
 
 }
 
+func TestSubManagerBadPlugin(t *testing.T) {
+	mdi := &databasemocks.Plugin{}
+	mdm := &datamocks.Manager{}
+	mbm := &broadcastmocks.Manager{}
+	mpm := &privatemessagingmocks.Manager{}
+	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
+	coreconfig.Reset()
+	config.Set(coreconfig.EventTransportsEnabled, []string{"!unknown!"})
+	_, err := newSubscriptionManager(context.Background(), mdi, mdm, newEventNotifier(context.Background(), "ut"), mbm, mpm, txHelper)
+	assert.Regexp(t, "FF10172", err)
+}
+
+func TestSubManagerTransportInitError(t *testing.T) {
+	mei := &eventsmocks.Plugin{}
+	mei.On("Name").Return("ut")
+	mei.On("InitConfig", mock.Anything).Return()
+	mei.On("Init", mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+
+	sm, cancel := newTestSubManager(t, mei)
+	defer cancel()
+
+	err := sm.initTransports()
+	assert.EqualError(t, err, "pop")
+}
+
 func TestStartSubRestoreFail(t *testing.T) {
 	mei := &eventsmocks.Plugin{}
 	sm, cancel := newTestSubManager(t, mei)
diff --git a/internal/events/system/events_test.go b/internal/events/system/events_test.go
index 65aab44b0..40c5c6670 100644
--- a/internal/events/system/events_test.go
+++ b/internal/events/system/events_test.go
@@ -42,8 +42,7 @@ func newTestEvents(t *testing.T) (se *Events, cancel func()) {
 	ctx, cancelCtx := context.WithCancel(context.Background())
 	config := config.RootSection("ut.events")
 	se.InitConfig(config)
-	se.Init(ctx, config)
-	se.SetHandler("ns1", cbs)
+	se.Init(ctx, config, cbs)
 	assert.Equal(t, "system", se.Name())
 	assert.NotNil(t, se.Capabilities())
 	assert.Nil(t, se.ValidateOptions(&core.SubscriptionOptions{}))
@@ -55,13 +54,7 @@ func TestDeliveryRequestOk(t *testing.T) {
 	se, cancel := newTestEvents(t)
 	defer cancel()
 
-	sub := &core.Subscription{
-		SubscriptionRef: core.SubscriptionRef{
-			Namespace: "ns1",
-		},
-	}
-
-	cbs := se.callbacks["ns1"].(*eventsmocks.Callbacks)
+	cbs := se.callbacks.(*eventsmocks.Callbacks)
 	cbs.On("EphemeralSubscription", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil)
 	cbs.On("DeliveryResponse", se.connID, mock.Anything).Return(nil)
 
@@ -72,7 +65,7 @@ func TestDeliveryRequestOk(t *testing.T) {
 	})
 	assert.NoError(t, err)
 
-	err = se.DeliveryRequest(se.connID, sub, &core.EventDelivery{
+	err = se.DeliveryRequest(se.connID, &core.Subscription{}, &core.EventDelivery{
 		EnrichedEvent: core.EnrichedEvent{
 			Event: core.Event{
 				Namespace: "ns1",
@@ -100,7 +93,7 @@ func TestDeliveryRequestFail(t *testing.T) {
 	se, cancel := newTestEvents(t)
 	defer cancel()
 
-	cbs := se.callbacks["ns1"].(*eventsmocks.Callbacks)
+	cbs := se.callbacks.(*eventsmocks.Callbacks)
 	cbs.On("EphemeralSubscription", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil)
 
 	err := se.AddListener("ns1", func(event *core.EventDelivery) error {
@@ -124,7 +117,7 @@ func TestAddListenerFail(t *testing.T) {
 	se, cancel := newTestEvents(t)
 	defer cancel()
 
-	cbs := se.callbacks["ns1"].(*eventsmocks.Callbacks)
+	cbs := se.callbacks.(*eventsmocks.Callbacks)
 	cbs.On("EphemeralSubscription", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
 
 	err := se.AddListener("ns1", func(event *core.EventDelivery) error { return nil })
diff --git a/internal/events/webhooks/webhooks_test.go b/internal/events/webhooks/webhooks_test.go
index 2a70c4383..1193eeee9 100644
--- a/internal/events/webhooks/webhooks_test.go
+++ b/internal/events/webhooks/webhooks_test.go
@@ -47,8 +47,7 @@ func newTestWebHooks(t *testing.T) (wh *WebHooks, cancel func()) {
 	ctx, cancelCtx := context.WithCancel(context.Background())
 	svrConfig := config.RootSection("ut.webhooks")
 	wh.InitConfig(svrConfig)
-	wh.Init(ctx, svrConfig)
-	wh.SetHandler("ns1", cbs)
+	wh.Init(ctx, svrConfig, cbs)
 	assert.Equal(t, "webhooks", wh.Name())
 	assert.NotNil(t, wh.Capabilities())
 	return wh, cancelCtx
@@ -147,9 +146,6 @@ func TestRequestWithBodyReplyEndToEnd(t *testing.T) {
 	msgID := fftypes.NewUUID()
 	groupHash := fftypes.NewRandB32()
 	sub := &core.Subscription{
-		SubscriptionRef: core.SubscriptionRef{
-			Namespace: "ns1",
-		},
 		Options: core.SubscriptionOptions{
 			SubscriptionCoreOptions: core.SubscriptionCoreOptions{
 				WithData: &yes,
@@ -211,7 +207,7 @@ func TestRequestWithBodyReplyEndToEnd(t *testing.T) {
 		}`),
 	}
 
-	mcb := wh.callbacks["ns1"].(*eventsmocks.Callbacks)
+	mcb := wh.callbacks.(*eventsmocks.Callbacks)
 	mcb.On("DeliveryResponse", mock.Anything, mock.MatchedBy(func(response *core.EventDeliveryResponse) bool {
 		assert.Equal(t, *msgID, *response.Reply.Message.Header.CID)
 		assert.Equal(t, *groupHash, *response.Reply.Message.Header.Group)
@@ -257,9 +253,6 @@ func TestRequestWithEmptyStringBodyReplyEndToEnd(t *testing.T) {
 	msgID := fftypes.NewUUID()
 	groupHash := fftypes.NewRandB32()
 	sub := &core.Subscription{
-		SubscriptionRef: core.SubscriptionRef{
-			Namespace: "ns1",
-		},
 		Options: core.SubscriptionOptions{
 			SubscriptionCoreOptions: core.SubscriptionCoreOptions{
 				WithData: &yes,
@@ -321,7 +314,7 @@ func TestRequestWithEmptyStringBodyReplyEndToEnd(t *testing.T) {
 		}`),
 	}
 
-	mcb := wh.callbacks["ns1"].(*eventsmocks.Callbacks)
+	mcb := wh.callbacks.(*eventsmocks.Callbacks)
 	mcb.On("DeliveryResponse", mock.Anything, mock.MatchedBy(func(response *core.EventDeliveryResponse) bool {
 		assert.Equal(t, *msgID, *response.Reply.Message.Header.CID)
 		assert.Equal(t, *groupHash, *response.Reply.Message.Header.Group)
@@ -415,9 +408,6 @@ func TestRequestReplyEmptyData(t *testing.T) {
 
 	yes := true
 	sub := &core.Subscription{
-		SubscriptionRef: core.SubscriptionRef{
-			Namespace: "ns1",
-		},
 		Options: core.SubscriptionOptions{
 			SubscriptionCoreOptions: core.SubscriptionCoreOptions{
 				WithData: &yes,
@@ -444,7 +434,7 @@ func TestRequestReplyEmptyData(t *testing.T) {
 		},
 	}
 
-	mcb := wh.callbacks["ns1"].(*eventsmocks.Callbacks)
+	mcb := wh.callbacks.(*eventsmocks.Callbacks)
 	mcb.On("DeliveryResponse", mock.Anything, mock.MatchedBy(func(response *core.EventDeliveryResponse) bool {
 		assert.Equal(t, *msgID, *response.Reply.Message.Header.CID)
 		assert.Nil(t, response.Reply.Message.Header.Group)
@@ -455,8 +445,6 @@ func TestRequestReplyEmptyData(t *testing.T) {
 	err := wh.DeliveryRequest(mock.Anything, sub, event, core.DataArray{})
 	assert.NoError(t, err)
 	assert.True(t, called)
-
-	mcb.AssertExpectations(t)
 }
 
 func TestRequestReplyBadJSON(t *testing.T) {
@@ -473,11 +461,7 @@ func TestRequestReplyBadJSON(t *testing.T) {
 	server := httptest.NewServer(r)
 	defer server.Close()
 
-	sub := &core.Subscription{
-		SubscriptionRef: core.SubscriptionRef{
-			Namespace: "ns1",
-		},
-	}
+	sub := &core.Subscription{}
 	to := sub.Options.TransportOptions()
 	to["url"] = fmt.Sprintf("http://%s/myapi", server.Listener.Addr())
 	to["reply"] = true
@@ -499,7 +483,7 @@ func TestRequestReplyBadJSON(t *testing.T) {
 		},
 	}
 
-	mcb := wh.callbacks["ns1"].(*eventsmocks.Callbacks)
+	mcb := wh.callbacks.(*eventsmocks.Callbacks)
 	mcb.On("DeliveryResponse", mock.Anything, mock.MatchedBy(func(response *core.EventDeliveryResponse) bool {
 		assert.Equal(t, float64(502), response.Reply.InlineData[0].Value.JSONObject()["status"])
 		assert.Regexp(t, "FF10257", response.Reply.InlineData[0].Value.JSONObject().GetObject("body")["error"])
@@ -508,10 +492,7 @@ func TestRequestReplyBadJSON(t *testing.T) {
 
 	err := wh.DeliveryRequest(mock.Anything, sub, event, core.DataArray{})
 	assert.NoError(t, err)
-
-	mcb.AssertExpectations(t)
 }
-
 func TestRequestReplyDataArrayBadStatusB64(t *testing.T) {
 	wh, cancel := newTestWebHooks(t)
 	defer cancel()
@@ -536,9 +517,6 @@ func TestRequestReplyDataArrayBadStatusB64(t *testing.T) {
 
 	yes := true
 	sub := &core.Subscription{
-		SubscriptionRef: core.SubscriptionRef{
-			Namespace: "ns1",
-		},
 		Options: core.SubscriptionOptions{
 			SubscriptionCoreOptions: core.SubscriptionCoreOptions{
 				WithData: &yes,
@@ -565,7 +543,7 @@ func TestRequestReplyDataArrayBadStatusB64(t *testing.T) {
 		},
 	}
 
-	mcb := wh.callbacks["ns1"].(*eventsmocks.Callbacks)
+	mcb := wh.callbacks.(*eventsmocks.Callbacks)
 	mcb.On("DeliveryResponse", mock.Anything, mock.MatchedBy(func(response *core.EventDeliveryResponse) bool {
 		assert.Equal(t, *msgID, *response.Reply.Message.Header.CID)
 		assert.Nil(t, response.Reply.Message.Header.Group)
@@ -594,11 +572,7 @@ func TestRequestReplyDataArrayError(t *testing.T) {
 	server := httptest.NewServer(r)
 	server.Close()
 
-	sub := &core.Subscription{
-		SubscriptionRef: core.SubscriptionRef{
-			Namespace: "ns1",
-		},
-	}
+	sub := &core.Subscription{}
 	to := sub.Options.TransportOptions()
 	to["url"] = fmt.Sprintf("http://%s/myapi", server.Listener.Addr())
 	to["reply"] = true
@@ -619,7 +593,7 @@ func TestRequestReplyDataArrayError(t *testing.T) {
 		},
 	}
 
-	mcb := wh.callbacks["ns1"].(*eventsmocks.Callbacks)
+	mcb := wh.callbacks.(*eventsmocks.Callbacks)
 	mcb.On("DeliveryResponse", mock.Anything, mock.MatchedBy(func(response *core.EventDeliveryResponse) bool {
 		assert.Equal(t, *msgID, *response.Reply.Message.Header.CID)
 		assert.Nil(t, response.Reply.Message.Header.Group)
@@ -647,11 +621,7 @@ func TestRequestReplyBuildRequestFailFastAsk(t *testing.T) {
 	server := httptest.NewServer(r)
 	server.Close()
 
-	sub := &core.Subscription{
-		SubscriptionRef: core.SubscriptionRef{
-			Namespace: "ns1",
-		},
-	}
+	sub := &core.Subscription{}
 	sub.Options.TransportOptions()["reply"] = true
 	sub.Options.TransportOptions()["fastack"] = true
 	event := &core.EventDelivery{
@@ -672,7 +642,7 @@ func TestRequestReplyBuildRequestFailFastAsk(t *testing.T) {
 	}
 
 	waiter := make(chan struct{})
-	mcb := wh.callbacks["ns1"].(*eventsmocks.Callbacks)
+	mcb := wh.callbacks.(*eventsmocks.Callbacks)
 	dr := mcb.On("DeliveryResponse", mock.Anything, mock.MatchedBy(func(response *core.EventDeliveryResponse) bool {
 		assert.Equal(t, *msgID, *response.Reply.Message.Header.CID)
 		assert.Nil(t, response.Reply.Message.Header.Group)
@@ -701,9 +671,6 @@ func TestDeliveryRequestNilMessage(t *testing.T) {
 
 	yes := true
 	sub := &core.Subscription{
-		SubscriptionRef: core.SubscriptionRef{
-			Namespace: "ns1",
-		},
 		Options: core.SubscriptionOptions{
 			SubscriptionCoreOptions: core.SubscriptionCoreOptions{
 				WithData: &yes,
@@ -732,9 +699,6 @@ func TestDeliveryRequestReplyToReply(t *testing.T) {
 
 	yes := true
 	sub := &core.Subscription{
-		SubscriptionRef: core.SubscriptionRef{
-			Namespace: "ns1",
-		},
 		Options: core.SubscriptionOptions{
 			SubscriptionCoreOptions: core.SubscriptionCoreOptions{
 				WithData: &yes,
@@ -760,13 +724,11 @@ func TestDeliveryRequestReplyToReply(t *testing.T) {
 		},
 	}
 
-	mcb := wh.callbacks["ns1"].(*eventsmocks.Callbacks)
+	mcb := wh.callbacks.(*eventsmocks.Callbacks)
 	mcb.On("DeliveryResponse", mock.Anything, mock.MatchedBy(func(response *core.EventDeliveryResponse) bool {
 		return !response.Rejected // should be accepted as a no-op so we can move on to other events
 	}))
 
 	err := wh.DeliveryRequest(mock.Anything, sub, event, nil)
 	assert.NoError(t, err)
-
-	mcb.AssertExpectations(t)
 }
diff --git a/internal/events/websockets/websockets_test.go b/internal/events/websockets/websockets_test.go
index 5800357d6..ffd7c54d3 100644
--- a/internal/events/websockets/websockets_test.go
+++ b/internal/events/websockets/websockets_test.go
@@ -46,8 +46,7 @@ func newTestWebsockets(t *testing.T, cbs *eventsmocks.Callbacks, queryParams ...
 	ctx, cancelCtx := context.WithCancel(context.Background())
 	svrConfig := config.RootSection("ut.websockets")
 	ws.InitConfig(svrConfig)
-	ws.Init(ctx, svrConfig)
-	ws.SetHandler("ns1", cbs)
+	ws.Init(ctx, svrConfig, cbs)
 	assert.Equal(t, "websockets", ws.Name())
 	assert.NotNil(t, ws.Capabilities())
 	cbs.On("ConnectionClosed", mock.Anything).Return(nil).Maybe()
@@ -182,10 +181,7 @@ func TestStartReceiveAckEphemeral(t *testing.T) {
 		EnrichedEvent: core.EnrichedEvent{
 			Event: core.Event{ID: fftypes.NewUUID()},
 		},
-		Subscription: core.SubscriptionRef{
-			ID:        fftypes.NewUUID(),
-			Namespace: "ns1",
-		},
+		Subscription: core.SubscriptionRef{ID: fftypes.NewUUID()},
 	}, nil)
 
 	b := <-wsc.Receive()
@@ -308,10 +304,7 @@ func TestAutoStartReceiveAckEphemeral(t *testing.T) {
 		EnrichedEvent: core.EnrichedEvent{
 			Event: core.Event{ID: fftypes.NewUUID()},
 		},
-		Subscription: core.SubscriptionRef{
-			ID:        fftypes.NewUUID(),
-			Namespace: "ns1",
-		},
+		Subscription: core.SubscriptionRef{ID: fftypes.NewUUID()},
 	}, nil)
 
 	b := <-wsc.Receive()
@@ -339,19 +332,6 @@ func TestAutoStartBadOptions(t *testing.T) {
 	cbs.AssertExpectations(t)
 }
 
-func TestAutoStartBadNamespace(t *testing.T) {
-	cbs := &eventsmocks.Callbacks{}
-	_, wsc, cancel := newTestWebsockets(t, cbs, "ephemeral", "namespace=ns2")
-	defer cancel()
-
-	b := <-wsc.Receive()
-	var res core.WSError
-	err := json.Unmarshal(b, &res)
-	assert.NoError(t, err)
-	assert.Regexp(t, "FF10187", res.Error)
-	cbs.AssertExpectations(t)
-}
-
 func TestHandleAckWithAutoAck(t *testing.T) {
 	eventUUID := fftypes.NewUUID()
 	wsc := &websocketConnection{
@@ -416,7 +396,7 @@ func TestHandleAckMultipleStartedNoSubSingleMatch(t *testing.T) {
 		ctx: context.Background(),
 		ws: &WebSockets{
 			ctx:       context.Background(),
-			callbacks: map[string]events.Callbacks{"ns1": cbs},
+			callbacks: cbs,
 		},
 		started:      []*websocketStartedSub{{ephemeral: false, name: "name1", namespace: "ns1"}},
 		sendMessages: make(chan interface{}, 1),
@@ -516,7 +496,7 @@ func TestDispatchAutoAck(t *testing.T) {
 		connID: fftypes.NewUUID().String(),
 		ws: &WebSockets{
 			ctx:         context.Background(),
-			callbacks:   map[string]events.Callbacks{"ns1": cbs},
+			callbacks:   cbs,
 			connections: make(map[string]*websocketConnection),
 		},
 		started:      []*websocketStartedSub{{ephemeral: false, name: "name1", namespace: "ns1"}},
diff --git a/internal/namespace/manager_test.go b/internal/namespace/manager_test.go
index 60976473e..4d882ef66 100644
--- a/internal/namespace/manager_test.go
+++ b/internal/namespace/manager_test.go
@@ -34,7 +34,6 @@ import (
 	"github.com/hyperledger/firefly/mocks/blockchainmocks"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/dataexchangemocks"
-	"github.com/hyperledger/firefly/mocks/eventsmocks"
 	"github.com/hyperledger/firefly/mocks/identitymocks"
 	"github.com/hyperledger/firefly/mocks/metricsmocks"
 	"github.com/hyperledger/firefly/mocks/operationmocks"
@@ -58,7 +57,6 @@ type testNamespaceManager struct {
 	mdx *dataexchangemocks.Plugin
 	mps *sharedstoragemocks.Plugin
 	mti *tokenmocks.Plugin
-	mev *eventsmocks.Plugin
 }
 
 func (nm *testNamespaceManager) cleanup(t *testing.T) {
@@ -85,7 +83,6 @@ func newTestNamespaceManager(resetConfig bool) *testNamespaceManager {
 		mdx: &dataexchangemocks.Plugin{},
 		mps: &sharedstoragemocks.Plugin{},
 		mti: &tokenmocks.Plugin{},
-		mev: &eventsmocks.Plugin{},
 		namespaceManager: namespaceManager{
 			ctx:         context.Background(),
 			namespaces:  make(map[string]*namespace),
@@ -110,9 +107,6 @@ func newTestNamespaceManager(resetConfig bool) *testNamespaceManager {
 	nm.plugins.tokens = map[string]tokensPlugin{
 		"erc721": {plugin: nm.mti},
 	}
-	nm.plugins.events = map[string]eventsPlugin{
-		"websockets": {plugin: nm.mev},
-	}
 	nm.namespaceManager.metrics = nm.mmi
 	nm.namespaceManager.adminEvents = nm.mae
 	return nm
@@ -132,12 +126,11 @@ func TestInit(t *testing.T) {
 	nm.utOrchestrator = mo
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", mock.Anything).Return()
+	nm.mdi.On("RegisterListener", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mti.On("Init", mock.Anything, mock.Anything, mock.Anything).Return(nil)
-	nm.mev.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mbi.On("NetworkVersion").Return(2)
 
 	ctx, cancelCtx := context.WithCancel(context.Background())
@@ -170,7 +163,7 @@ func TestInitBlockchainFail(t *testing.T) {
 	nm.utOrchestrator = &orchestratormocks.Orchestrator{}
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", mock.Anything).Return()
+	nm.mdi.On("RegisterListener", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(fmt.Errorf("pop"))
 
 	ctx, cancelCtx := context.WithCancel(context.Background())
@@ -185,7 +178,7 @@ func TestInitDataExchangeFail(t *testing.T) {
 	nm.utOrchestrator = &orchestratormocks.Orchestrator{}
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", mock.Anything).Return()
+	nm.mdi.On("RegisterListener", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
 
@@ -201,7 +194,7 @@ func TestInitSharedStorageFail(t *testing.T) {
 	nm.utOrchestrator = &orchestratormocks.Orchestrator{}
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", mock.Anything).Return()
+	nm.mdi.On("RegisterListener", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
@@ -218,7 +211,7 @@ func TestInitTokensFail(t *testing.T) {
 	nm.utOrchestrator = &orchestratormocks.Orchestrator{}
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", mock.Anything).Return()
+	nm.mdi.On("RegisterListener", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
@@ -229,38 +222,18 @@ func TestInitTokensFail(t *testing.T) {
 	assert.EqualError(t, err, "pop")
 }
 
-func TestInitEventsFail(t *testing.T) {
-	nm := newTestNamespaceManager(true)
-	defer nm.cleanup(t)
-
-	nm.utOrchestrator = &orchestratormocks.Orchestrator{}
-
-	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", mock.Anything).Return()
-	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
-	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mti.On("Init", mock.Anything, mock.Anything, mock.Anything).Return(nil)
-	nm.mev.On("Init", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
-
-	ctx, cancelCtx := context.WithCancel(context.Background())
-	err := nm.Init(ctx, cancelCtx)
-	assert.EqualError(t, err, "pop")
-}
-
 func TestInitOrchestratorFail(t *testing.T) {
 	nm := newTestNamespaceManager(true)
 	defer nm.cleanup(t)
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", mock.Anything).Return()
+	nm.mdi.On("RegisterListener", mock.Anything).Return()
 	nm.mdi.On("GetIdentities", mock.Anything, "default", mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
-	nm.mbi.On("SetHandler", mock.Anything).Return()
+	nm.mbi.On("RegisterListener", mock.Anything).Return()
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mti.On("Init", mock.Anything, mock.Anything, mock.Anything).Return(nil)
-	nm.mev.On("Init", mock.Anything, mock.Anything).Return(nil)
 
 	ctx, cancelCtx := context.WithCancel(context.Background())
 	err := nm.Init(ctx, cancelCtx)
@@ -277,12 +250,11 @@ func TestInitVersion1(t *testing.T) {
 	nm.utOrchestrator = mo
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", mock.Anything).Return()
+	nm.mdi.On("RegisterListener", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mti.On("Init", mock.Anything, mock.Anything, mock.Anything).Return(nil)
-	nm.mev.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mbi.On("NetworkVersion").Return(1)
 
 	ctx, cancelCtx := context.WithCancel(context.Background())
@@ -306,12 +278,11 @@ func TestInitVersion1Fail(t *testing.T) {
 	nm.utOrchestrator = mo
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", mock.Anything).Return()
+	nm.mdi.On("RegisterListener", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mti.On("Init", mock.Anything, mock.Anything, mock.Anything).Return(nil)
-	nm.mev.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mbi.On("NetworkVersion").Return(1)
 
 	ctx, cancelCtx := context.WithCancel(context.Background())
@@ -675,25 +646,6 @@ func TestTokensPluginDuplicate(t *testing.T) {
 	assert.Regexp(t, "FF10395", err)
 }
 
-func TestEventsPluginDefaults(t *testing.T) {
-	nm := newTestNamespaceManager(true)
-	defer nm.cleanup(t)
-	nm.plugins.events = nil
-	plugins, err := nm.getEventPlugins(context.Background())
-	assert.Equal(t, 3, len(plugins))
-	assert.NoError(t, err)
-}
-
-func TestEventsPluginBadType(t *testing.T) {
-	nm := newTestNamespaceManager(true)
-	defer nm.cleanup(t)
-	nm.plugins.events = nil
-	config.Set(coreconfig.EventTransportsEnabled, []string{"!unknown!"})
-	ctx, cancelCtx := context.WithCancel(context.Background())
-	err := nm.Init(ctx, cancelCtx)
-	assert.Error(t, err)
-}
-
 func TestInitBadNamespace(t *testing.T) {
 	nm := newTestNamespaceManager(true)
 	defer nm.cleanup(t)
@@ -701,12 +653,11 @@ func TestInitBadNamespace(t *testing.T) {
 	nm.utOrchestrator = &orchestratormocks.Orchestrator{}
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", mock.Anything).Return()
+	nm.mdi.On("RegisterListener", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mti.On("Init", mock.Anything, mock.Anything, mock.Anything).Return(nil)
-	nm.mev.On("Init", mock.Anything, mock.Anything).Return(nil)
 
 	viper.SetConfigType("yaml")
 	err := viper.ReadConfig(strings.NewReader(`
diff --git a/internal/orchestrator/orchestrator_test.go b/internal/orchestrator/orchestrator_test.go
index f1dd35919..9f1c75193 100644
--- a/internal/orchestrator/orchestrator_test.go
+++ b/internal/orchestrator/orchestrator_test.go
@@ -185,13 +185,13 @@ func TestNewOrchestrator(t *testing.T) {
 func TestInitOK(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
-	or.mdi.On("SetHandler", mock.Anything).Return()
-	or.mbi.On("SetHandler", mock.Anything).Return()
+	or.mdi.On("RegisterListener", mock.Anything).Return()
+	or.mbi.On("RegisterListener", mock.Anything).Return()
 	or.mdi.On("GetIdentities", mock.Anything, "ns", mock.Anything).Return([]*core.Identity{{}}, nil, nil)
-	or.mdx.On("SetHandler", mock.Anything).Return()
+	or.mdx.On("RegisterListener", mock.Anything).Return()
 	or.mdx.On("SetNodes", mock.Anything).Return()
-	or.mps.On("SetHandler", mock.Anything).Return()
-	or.mti.On("SetHandler", "ns", mock.Anything).Return(nil)
+	or.mps.On("RegisterListener", mock.Anything).Return()
+	or.mti.On("RegisterListener", "ns", mock.Anything).Return(nil)
 	err := or.Init(or.ctx, or.cancelCtx)
 	assert.NoError(t, err)
 
@@ -209,13 +209,13 @@ func TestInitOK(t *testing.T) {
 func TestInitTokenListenerFail(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
-	or.mdi.On("SetHandler", mock.Anything).Return()
-	or.mbi.On("SetHandler", mock.Anything).Return()
+	or.mdi.On("RegisterListener", mock.Anything).Return()
+	or.mbi.On("RegisterListener", mock.Anything).Return()
 	or.mdi.On("GetIdentities", mock.Anything, "ns", mock.Anything).Return([]*core.Identity{{}}, nil, nil)
-	or.mdx.On("SetHandler", mock.Anything).Return()
+	or.mdx.On("RegisterListener", mock.Anything).Return()
 	or.mdx.On("SetNodes", mock.Anything).Return()
-	or.mps.On("SetHandler", mock.Anything).Return()
-	or.mti.On("SetHandler", "ns", mock.Anything).Return(fmt.Errorf("pop"))
+	or.mps.On("RegisterListener", mock.Anything).Return()
+	or.mti.On("RegisterListener", "ns", mock.Anything).Return(fmt.Errorf("pop"))
 	err := or.Init(or.ctx, or.cancelCtx)
 	assert.EqualError(t, err, "pop")
 }
@@ -223,8 +223,8 @@ func TestInitTokenListenerFail(t *testing.T) {
 func TestInitDataexchangeNodesFail(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
-	or.mdi.On("SetHandler", mock.Anything).Return()
-	or.mbi.On("SetHandler", mock.Anything).Return()
+	or.mdi.On("RegisterListener", mock.Anything).Return()
+	or.mbi.On("RegisterListener", mock.Anything).Return()
 	or.mdi.On("GetIdentities", mock.Anything, "ns", mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
 	ctx := context.Background()
 	err := or.initPlugins(ctx)
diff --git a/internal/tokens/fftokens/fftokens_test.go b/internal/tokens/fftokens/fftokens_test.go
index 4081672df..452b5c17a 100644
--- a/internal/tokens/fftokens/fftokens_test.go
+++ b/internal/tokens/fftokens/fftokens_test.go
@@ -262,7 +262,7 @@ func TestCreateTokenPoolSynchronous(t *testing.T) {
 		})
 
 	mcb := &tokenmocks.Callbacks{}
-	err := h.SetHandler("ns1", mcb)
+	err := h.RegisterListener("ns1", mcb)
 	assert.NoError(t, err)
 
 	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
@@ -424,7 +424,7 @@ func TestActivateTokenPoolSynchronous(t *testing.T) {
 		})
 
 	mcb := &tokenmocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
+	h.RegisterListener("ns1", mcb)
 	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && p.TX.ID == nil && p.Event.ProtocolID == ""
 	})).Return(nil)
@@ -474,7 +474,7 @@ func TestActivateTokenPoolSynchronousBadResponse(t *testing.T) {
 		})
 
 	mcb := &tokenmocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
+	h.RegisterListener("ns1", mcb)
 	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && p.TX.ID == nil
 	})).Return(nil)
@@ -810,7 +810,7 @@ func TestReceiptEvents(t *testing.T) {
 	assert.NoError(t, err)
 
 	mcb := &tokenmocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
+	h.RegisterListener("ns1", mcb)
 	opID := fftypes.NewUUID()
 
 	// receipt: bad ID - passed through
@@ -854,7 +854,7 @@ func TestPoolEvents(t *testing.T) {
 	assert.NoError(t, err)
 
 	mcb := &tokenmocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
+	h.RegisterListener("ns1", mcb)
 	txID := fftypes.NewUUID()
 
 	// token-pool: missing data
@@ -950,7 +950,7 @@ func TestTransferEvents(t *testing.T) {
 	assert.NoError(t, err)
 
 	mcb := &tokenmocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
+	h.RegisterListener("ns1", mcb)
 	txID := fftypes.NewUUID()
 
 	// token-mint: missing data
@@ -1169,7 +1169,7 @@ func TestApprovalEvents(t *testing.T) {
 	assert.NoError(t, err)
 
 	mcb := &tokenmocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
+	h.RegisterListener("ns1", mcb)
 	txID := fftypes.NewUUID()
 
 	// token-approval: success
diff --git a/test/e2e/run.sh b/test/e2e/run.sh
index 8368e7ccc..54991d949 100755
--- a/test/e2e/run.sh
+++ b/test/e2e/run.sh
@@ -88,7 +88,7 @@ if [ "$CREATE_STACK" == "true" ]; then
 fi
 
 if [ "$BUILD_FIREFLY" == "true" ]; then
-  make -C ../.. docker
+  make -C ../.. DOCKER_ARGS="--load" docker
   checkOk $?
 fi