Merge pull request #611 from LF-Decentralized-Trust-labs/event-listeners

Event listeners

Author: peterbroadhurst

.dockerignore

@@ -25,7 +25,6 @@ perf/
 
 # Other folders not needed for the Docker build
 example/
-sdk/
 
 # The operator has its own docker build (with its own .dockerignore), so we don't want to rebuild the whole
 # Paladin docker every time you're re-running the operator/test.

common/go/pkg/pldmsgs/en_descriptions.go

@@ -167,6 +167,15 @@ var (
 	TransactionReceiptFiltersDomain                         = pdm("TransactionReceiptFilters.domain", "Only deliver receipts for an individual domain (only valid with type=private)")
 	TransactionReceiptOptionsDomainReceipts                 = pdm("TransactionReceiptOptions.domainReceipts", "When true, a full domain receipt will be generated for each event with complete state data")
 	TransactionReceiptOptionsIncompleteStateReceiptBehavior = pdm("TransactionReceiptOptions.incompleteStateReceiptBehavior", "When set to 'block_contract', if a transaction with incomplete state data is detected then delivery of all receipts on that individual smart contract address will pause until the missing state arrives. Receipts for other contract addresses continue to be delivered")
+	BlockchainEventListenerName                             = pdm("BlockchainEventListener.name", "Unique name for the blockchain event listener")
+	BlockchainEventListenerCreated                          = pdm("BlockchainEventListener.created", "Time the listener was created")
+	BlockchainEventListenerStarted                          = pdm("BlockchainEventListener.started", "If the listener is started - can be set to false to disable delivery server-side")
+	BlockchainEventListenerSources                          = pdm("BlockchainEventListener.sources", "Sources of events")
+	BlockchainEventListenerOptions                          = pdm("BlockchainEventListener.options", "Options for the event listener")
+	BlockchainEventListenerOptionsBatchSize                 = pdm("BlockchainEventListenerOptions.batchSize", "The maximum number of events to deliver in each batch")
+	BlockchainEventListenerOptionsBatchTimeout              = pdm("BlockchainEventListenerOptions.batchTimeout", "The maximum time to wait for a batch to fill before delivering")
+	BlockchainEventListenerSourceABI                        = pdm("BlockchainEventListenerSource.abi", "The ABI containing events to listen for")
+	BlockchainEventListenerSourceAddress                    = pdm("BlockchainEventListenerSource.address", "The address to listen for events from")
 )
 
 // query/query_json.go

core/go/componenttest/component_test.go

@@ -30,8 +30,6 @@ import (
 	"github.com/hyperledger/firefly-signer/pkg/abi"
 	"github.com/kaleido-io/paladin/config/pkg/confutil"
 	"github.com/kaleido-io/paladin/core/componenttest/domains"
-	"github.com/kaleido-io/paladin/core/pkg/blockindexer"
-	"github.com/kaleido-io/paladin/core/pkg/persistence"
 
 	"github.com/kaleido-io/paladin/sdk/go/pkg/pldapi"
 	"github.com/kaleido-io/paladin/sdk/go/pkg/pldclient"
@@ -50,35 +48,12 @@ func TestRunSimpleStorageEthTransaction(t *testing.T) {
 	ctx := context.Background()
 	logrus.SetLevel(logrus.DebugLevel)
 
-	instance := newInstanceForComponentTesting(t, deployDomainRegistry(t), nil, nil, nil, false)
-	cm := instance.cm
+	instance := newInstanceForComponentTesting(t, deployDomainRegistry(t), nil, nil, nil, true)
 	c := pldclient.Wrap(instance.client).ReceiptPollingInterval(250 * time.Millisecond)
 
 	build, err := solutils.LoadBuild(ctx, simpleStorageBuildJSON)
 	require.NoError(t, err)
 
-	eventStreamEvents := make(chan *pldapi.EventWithData, 2 /* all the events we exepct */)
-	_, err = cm.BlockIndexer().AddEventStream(ctx, cm.Persistence().NOTX(), &blockindexer.InternalEventStream{
-		Handler: func(ctx context.Context, tx persistence.DBTX, batch *blockindexer.EventDeliveryBatch) error {
-			// With SQLite we cannot hang in here with a DB TX - as there's only one per process.
-			for _, e := range batch.Events {
-				select {
-				case eventStreamEvents <- e:
-				default:
-					assert.Fail(t, "more than expected number of events received")
-				}
-			}
-			return nil
-		},
-		Definition: &blockindexer.EventStream{
-			Name: "unittest",
-			Sources: []blockindexer.EventStreamSource{{
-				ABI: abi.ABI{build.ABI.Events()["Changed"]},
-			}},
-		},
-	})
-	require.NoError(t, err)
-
 	simpleStorage := c.ForABI(ctx, build.ABI).Public().From("key1")
 
 	res := simpleStorage.Clone().
@@ -89,6 +64,58 @@ func TestRunSimpleStorageEthTransaction(t *testing.T) {
 	require.NoError(t, res.Error())
 	contractAddr := res.Receipt().ContractAddress
 
+	// set up the event listener
+	success, err := c.PTX().CreateBlockchainEventListener(ctx, &pldapi.BlockchainEventListener{
+		Name:    "listener1",
+		Started: confutil.P(false),
+		Sources: []pldapi.BlockchainEventListenerSource{{
+			ABI:     abi.ABI{build.ABI.Events()["Changed"]},
+			Address: contractAddr,
+		}},
+	})
+	require.NoError(t, err)
+	require.True(t, success)
+
+	wsClient, err := c.WebSocket(ctx, instance.wsConfig)
+	require.NoError(t, err)
+
+	sub, err := wsClient.PTX().SubscribeBlockchainEvents(ctx, "listener1")
+	require.NoError(t, err)
+
+	eventData := make(chan string)
+
+	go func() {
+		for {
+			subNotification, ok := <-sub.Notifications()
+			if ok {
+				var batch pldapi.TransactionEventBatch
+				_ = json.Unmarshal(subNotification.GetResult(), &batch)
+				for _, e := range batch.Events {
+					eventData <- e.Data.String()
+				}
+			}
+			err = subNotification.Ack(ctx)
+			require.NoError(t, err)
+		}
+	}()
+
+	// pause to make sure that if an event was going to be received, it would have been
+	ticker1 := time.NewTicker(10 * time.Millisecond)
+	defer ticker1.Stop()
+
+	select {
+	case <-eventData:
+		t.FailNow()
+	case <-ticker1.C:
+	}
+
+	success, err = c.PTX().StartBlockchainEventListener(ctx, "listener1")
+	require.NoError(t, err)
+	require.True(t, success)
+
+	data := <-eventData
+	assert.JSONEq(t, `{"x":"11223344"}`, data)
+
 	var getX1 pldtypes.RawJSON
 	err = simpleStorage.Clone().
 		Function("get").
@@ -105,6 +132,9 @@ func TestRunSimpleStorageEthTransaction(t *testing.T) {
 		Send().Wait(5 * time.Second)
 	require.NoError(t, res.Error())
 
+	data = <-eventData
+	assert.JSONEq(t, `{"x":"99887766"}`, data)
+
 	var getX2 pldtypes.RawJSON
 	err = simpleStorage.Clone().
 		Function("get").
@@ -114,12 +144,34 @@ func TestRunSimpleStorageEthTransaction(t *testing.T) {
 	require.NoError(t, err)
 	assert.JSONEq(t, `{"x":"99887766"}`, getX2.Pretty())
 
-	// Expect our event listener to be queued up with two Changed events
-	event1 := <-eventStreamEvents
-	assert.JSONEq(t, `{"x":"11223344"}`, string(event1.Data))
-	event2 := <-eventStreamEvents
-	assert.JSONEq(t, `{"x":"99887766"}`, string(event2.Data))
+	// stop the event listener
+	success, err = c.PTX().StopBlockchainEventListener(ctx, "listener1")
+	require.NoError(t, err)
+	require.True(t, success)
+
+	res = simpleStorage.Clone().
+		Function("set").
+		To(contractAddr).
+		Inputs(`{"_x":1234}`).
+		Send().Wait(5 * time.Second)
+	require.NoError(t, res.Error())
+
+	// pause to make sure that if an event was going to be received, it would have been
+	ticker2 := time.NewTicker(10 * time.Millisecond)
+	defer ticker2.Stop()
+
+	select {
+	case <-eventData:
+		t.FailNow()
+	case <-ticker2.C:
+	}
+
+	success, err = c.PTX().StartBlockchainEventListener(ctx, "listener1")
+	require.NoError(t, err)
+	require.True(t, success)
 
+	data = <-eventData
+	assert.JSONEq(t, `{"x":"1234"}`, data)
 }
 
 func TestUpdatePublicTransaction(t *testing.T) {

core/go/db/migrations/postgres/000019_event_streams_started.down.sql

@@ -0,0 +1,5 @@
+BEGIN;
+
+ALTER TABLE "event_streams" DROP COLUMN "started";
+
+COMMIT;

core/go/db/migrations/postgres/000019_event_streams_started.up.sql

@@ -0,0 +1,5 @@
+BEGIN;
+
+ALTER TABLE "event_streams" ADD "started" BOOLEAN DEFAULT TRUE;
+
+COMMIT;

core/go/db/migrations/sqlite/000019_event_streams_started.down.sql

@@ -0,0 +1 @@
+ALTER TABLE "event_streams" DROP COLUMN "started";

core/go/db/migrations/sqlite/000019_event_streams_started.up.sql

@@ -0,0 +1 @@
+ALTER TABLE "event_streams" ADD "started" BOOLEAN DEFAULT TRUE;

core/go/internal/componentmgr/manager.go

@@ -315,6 +315,9 @@ func (cm *componentManager) startBlockIndexer() (err error) {
 		_, err = cm.blockIndexer.GetBlockListenerHeight(cm.bgCtx)
 		err = cm.wrapIfErr(err, msgs.MsgComponentBlockIndexerStartError)
 	}
+	if err == nil {
+		err = cm.txManager.LoadBlockchainEventListeners()
+	}
 	return err
 }
 

core/go/internal/componentmgr/manager_test.go

@@ -201,6 +201,7 @@ func TestStartOK(t *testing.T) {
 	mockTxManager := componentmocks.NewTXManager(t)
 	mockTxManager.On("Start").Return(nil)
 	mockTxManager.On("Stop").Return()
+	mockTxManager.On("LoadBlockchainEventListeners").Return(nil)
 
 	mockGroupManager := componentmocks.NewGroupManager(t)
 	mockGroupManager.On("Start").Return(nil)

core/go/internal/components/txmgr.go

@@ -77,7 +77,11 @@ type ReceiptReceiver interface {
 	DeliverReceiptBatch(ctx context.Context, batchID uint64, receipts []*pldapi.TransactionReceiptFull) error
 }
 
-type ReceiptReceiverCloser interface {
+type BlockchainEventReceiver interface {
+	DeliverBlockchainEventBatch(ctx context.Context, batchID uuid.UUID, events []*pldapi.EventWithData) error
+}
+
+type ReceiverCloser interface {
 	Close()
 }
 
@@ -118,10 +122,11 @@ type TXManager interface {
 	StartReceiptListener(ctx context.Context, name string) error
 	StopReceiptListener(ctx context.Context, name string) error
 	DeleteReceiptListener(ctx context.Context, name string) error
-	AddReceiptReceiver(ctx context.Context, name string, r ReceiptReceiver) (ReceiptReceiverCloser, error)
+	AddReceiptReceiver(ctx context.Context, name string, r ReceiptReceiver) (ReceiverCloser, error)
 
 	// These functions for use of other components
 
+	LoadBlockchainEventListeners() error
 	NotifyStatesDBChanged(ctx context.Context) // called by state manager after committing DB TXs writing new states that might fill in gaps
 	PrepareInternalPrivateTransaction(ctx context.Context, dbTX persistence.DBTX, tx *pldapi.TransactionInput, submitMode pldapi.SubmitMode) (*ValidatedTransaction, error)
 	UpsertInternalPrivateTxsFinalizeIDs(ctx context.Context, dbTX persistence.DBTX, txis []*ValidatedTransaction) error

core/go/internal/domainmgr/domain.go

@@ -173,8 +173,8 @@ func (d *domain) processDomainConfig(dbTX persistence.DBTX, confRes *prototk.Con
 
 	// Create the event stream
 	d.eventStream, err = d.dm.blockIndexer.AddEventStream(d.ctx, dbTX, &blockindexer.InternalEventStream{
-		Definition: stream,
-		Handler:    d.handleEventBatch,
+		Definition:  stream,
+		HandlerDBTX: d.handleEventBatch,
 	})
 	if err != nil {
 		return nil, err

core/go/internal/domainmgr/event_indexer_test.go

@@ -78,10 +78,10 @@ func TestEventIndexingWithDB(t *testing.T) {
 						Signature:        eventSig_PaladinRegisterSmartContract_V0,
 					},
 					Data: pldtypes.RawJSON(`{
-						"txId": "` + pldtypes.Bytes32UUIDFirst16(deployTX).String() + `",
-						"instance": "` + contractAddr.String() + `",
-						"config": "0xfeedbeef"
-					}`),
+						 "txId": "` + pldtypes.Bytes32UUIDFirst16(deployTX).String() + `",
+						 "instance": "` + contractAddr.String() + `",
+						 "config": "0xfeedbeef"
+					 }`),
 				},
 			},
 		})
@@ -143,8 +143,8 @@ func TestEventIndexingBadEvent(t *testing.T) {
 					Address:           *td.d.registryAddress,
 					SoliditySignature: eventSolSig_PaladinRegisterSmartContract_V0,
 					Data: pldtypes.RawJSON(`{
-						 "config": "cannot parse this"
-					 }`),
+						  "config": "cannot parse this"
+					  }`),
 				},
 			},
 		})
@@ -185,10 +185,10 @@ func TestEventIndexingInsertError(t *testing.T) {
 						Signature:        eventSig_PaladinRegisterSmartContract_V0,
 					},
 					Data: pldtypes.RawJSON(`{
-						"txId": "` + pldtypes.Bytes32UUIDFirst16(deployTX).String() + `",
-						"domain": "` + contractAddr.String() + `",
-						"data": "0xfeedbeef"
-					}`),
+						 "txId": "` + pldtypes.Bytes32UUIDFirst16(deployTX).String() + `",
+						 "domain": "` + contractAddr.String() + `",
+						 "data": "0xfeedbeef"
+					 }`),
 				},
 			},
 		})