Move event plugin init to namespace manager

internal/blockchain/ethereum/ethereum.go

@@ -81,17 +81,17 @@ type Ethereum struct {
 }
 
 type callbacks struct {
-	listeners []blockchain.Callbacks
+	handlers []blockchain.Callbacks
 }
 
 func (cb *callbacks) BlockchainOpUpdate(plugin blockchain.Plugin, nsOpID string, txState blockchain.TransactionStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
-	for _, cb := range cb.listeners {
+	for _, cb := range cb.handlers {
 		cb.BlockchainOpUpdate(plugin, nsOpID, txState, blockchainTXID, errorMessage, opOutput)
 	}
 }
 
 func (cb *callbacks) BatchPinComplete(batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
-	for _, cb := range cb.listeners {
+	for _, cb := range cb.handlers {
 		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.listeners {
+	for _, cb := range cb.handlers {
 		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.listeners {
+	for _, cb := range cb.handlers {
 		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) RegisterListener(listener blockchain.Callbacks) {
-	e.callbacks.listeners = append(e.callbacks.listeners, listener)
+func (e *Ethereum) SetHandler(handler blockchain.Callbacks) {
+	e.callbacks.handlers = append(e.callbacks.handlers, handler)
 }
 
 func (e *Ethereum) Start() (err error) {

internal/blockchain/ethereum/ethereum_test.go

@@ -1096,7 +1096,7 @@ func TestHandleMessageBatchPinOK(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
-		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
 		wsconn:    wsm,
 	}
 
@@ -1576,7 +1576,7 @@ func TestHandleBadPayloadsAndThenReceiptFailure(t *testing.T) {
 	}`)
 
 	em := &blockchainmocks.Callbacks{}
-	e.RegisterListener(em)
+	e.SetHandler(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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-b5b97a4e-a317-4053-6400-1474650efcb5"
 

internal/blockchain/fabric/fabric.go

@@ -72,17 +72,17 @@ type Fabric struct {
 }
 
 type callbacks struct {
-	listeners []blockchain.Callbacks
+	handlers []blockchain.Callbacks
 }
 
 func (cb *callbacks) BlockchainOpUpdate(plugin blockchain.Plugin, nsOpID string, txState blockchain.TransactionStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
-	for _, cb := range cb.listeners {
+	for _, cb := range cb.handlers {
 		cb.BlockchainOpUpdate(plugin, nsOpID, txState, blockchainTXID, errorMessage, opOutput)
 	}
 }
 
 func (cb *callbacks) BatchPinComplete(batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
-	for _, cb := range cb.listeners {
+	for _, cb := range cb.handlers {
 		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.listeners {
+	for _, cb := range cb.handlers {
 		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.listeners {
+	for _, cb := range cb.handlers {
 		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) RegisterListener(listener blockchain.Callbacks) {
-	f.callbacks.listeners = append(f.callbacks.listeners, listener)
+func (f *Fabric) SetHandler(handler blockchain.Callbacks) {
+	f.callbacks.handlers = append(f.callbacks.handlers, handler)
 }
 
 func (f *Fabric) Start() (err error) {

internal/blockchain/fabric/fabric_test.go

@@ -943,7 +943,7 @@ func TestHandleMessageBatchPinOK(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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.RegisterListener(em)
+	e.SetHandler(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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
 		wsconn:    wsm,
 	}
 
@@ -1336,7 +1336,7 @@ func TestHandleReceiptNoRequestID(t *testing.T) {
 	e := &Fabric{
 		ctx:       context.Background(),
 		topic:     "topic1",
-		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
 		wsconn:    wsm,
 	}
 
@@ -1353,7 +1353,7 @@ func TestHandleReceiptFailedTx(t *testing.T) {
 	e := &Fabric{
 		ctx:       context.Background(),
 		topic:     "topic1",
-		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
 		wsconn:    wsm,
 	}
 
@@ -1545,7 +1545,7 @@ func TestHandleMessageContractEvent(t *testing.T) {
 
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
-		callbacks: callbacks{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []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{listeners: []blockchain.Callbacks{em}},
+		callbacks: callbacks{handlers: []blockchain.Callbacks{em}},
 	}
 	e.fireflyContract.subscription = "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
 

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) RegisterListener(listener database.Callbacks) {
-	psql.SQLCommon.RegisterListener(listener)
+func (psql *Postgres) SetHandler(handler database.Callbacks) {
+	psql.SQLCommon.SetHandler(handler)
 }
 
 func (psql *Postgres) Name() string {

internal/database/postgres/postgres_test.go

@@ -29,7 +29,7 @@ import (
 
 func TestPostgresProvider(t *testing.T) {
 	psql := &Postgres{}
-	psql.RegisterListener(&databasemocks.Callbacks{})
+	psql.SetHandler(&databasemocks.Callbacks{})
 	config := config.RootSection("unittest")
 	psql.InitConfig(config)
 	config.Set(sqlcommon.SQLConfDatasourceURL, "!bad connection")

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.RegisterListener(callbacks)
+	s.SetHandler(callbacks)
 	apiID := fftypes.NewUUID()
 	api := &core.ContractAPI{
 		ID:        apiID,

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.RegisterListener(mp.callbacks)
+	mp.SetHandler(mp.callbacks)
 	return mp, mp.mdb
 }
 

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.RegisterListener(tp.callbacks)
+	tp.SetHandler(tp.callbacks)
 
 	return tp, func() {
 		tp.Close()