Match start/stop calls from orchestrator

Signed-off-by: Andrew Richardson <andrew.richardson@kaleido.io>

internal/data/data_manager.go

@@ -51,6 +51,7 @@ type Manager interface {
 	UploadBlob(ctx context.Context, inData *core.DataRefOrValue, blob *ffapi.Multipart, autoMeta bool) (*core.Data, error)
 	DownloadBlob(ctx context.Context, dataID string) (*core.Blob, io.ReadCloser, error)
 	HydrateBatch(ctx context.Context, persistedBatch *core.BatchPersisted) (*core.Batch, error)
+	Start()
 	WaitStop()
 }
 
@@ -123,10 +124,13 @@ func NewDataManager(ctx context.Context, ns core.NamespaceRef, di database.Plugi
 		batchTimeout: config.GetDuration(coreconfig.MessageWriterBatchTimeout),
 		maxInserts:   config.GetInt(coreconfig.MessageWriterBatchMaxInserts),
 	})
-	dm.messageWriter.start()
 	return dm, nil
 }
 
+func (dm *dataManager) Start() {
+	dm.messageWriter.start()
+}
+
 func (dm *dataManager) CheckDatatype(ctx context.Context, datatype *core.Datatype) error {
 	_, err := newJSONValidator(ctx, dm.namespace.LocalName, datatype)
 	return err

internal/data/data_manager_test.go

@@ -44,6 +44,7 @@ func newTestDataManager(t *testing.T) (*dataManager, context.Context, func()) {
 	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
 	dm, err := NewDataManager(ctx, ns, mdi, mdx)
 	assert.NoError(t, err)
+	dm.Start()
 	return dm.(*dataManager), ctx, func() {
 		cancel()
 		dm.WaitStop()

internal/orchestrator/orchestrator.go

@@ -245,6 +245,7 @@ func (or *orchestrator) tokens() map[string]tokens.Plugin {
 }
 
 func (or *orchestrator) Start() (err error) {
+	or.data.Start()
 	if or.config.Multiparty.Enabled {
 		err = or.multiparty.ConfigureContract(or.ctx)
 		if err == nil {
@@ -288,6 +289,10 @@ func (or *orchestrator) WaitStop() {
 		or.sharedDownload.WaitStop()
 		or.sharedDownload = nil
 	}
+	if or.events != nil {
+		or.events.WaitStop()
+		or.events = nil
+	}
 	if or.operations != nil {
 		or.operations.WaitStop()
 		or.operations = nil

internal/orchestrator/orchestrator_test.go

@@ -367,6 +367,7 @@ func TestStartBatchFail(t *testing.T) {
 	coreconfig.Reset()
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
+	or.mdm.On("Start").Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything).Return(nil)
 	or.mba.On("Start").Return(fmt.Errorf("pop"))
 	err := or.Start()
@@ -377,6 +378,7 @@ func TestStartBlockchainsConfigureFail(t *testing.T) {
 	coreconfig.Reset()
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
+	or.mdm.On("Start").Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything).Return(fmt.Errorf("pop"))
 	err := or.Start()
 	assert.EqualError(t, err, "pop")
@@ -387,6 +389,7 @@ func TestStartStopOk(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
 	or.mmp.On("ConfigureContract", mock.Anything).Return(nil)
+	or.mdm.On("Start").Return(nil)
 	or.mba.On("Start").Return(nil)
 	or.mem.On("Start").Return(nil)
 	or.mbm.On("Start").Return(nil)
@@ -397,6 +400,7 @@ func TestStartStopOk(t *testing.T) {
 	or.mdm.On("WaitStop").Return(nil)
 	or.msd.On("WaitStop").Return(nil)
 	or.mom.On("WaitStop").Return(nil)
+	or.mem.On("WaitStop").Return(nil)
 	err := or.Start()
 	assert.NoError(t, err)
 	or.WaitStop()