Move plugin start calls to namespace manager

Avoid starting plugins more than once.

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

Author: awrichar

internal/namespace/manager.go

@@ -220,6 +220,21 @@ func (nm *namespaceManager) Start() error {
 		metrics.Registry()
 	}
 	for _, ns := range nm.namespaces {
+		if ns.plugins.Blockchain.Plugin != nil {
+			if err := ns.plugins.Blockchain.Plugin.Start(); err != nil {
+				return err
+			}
+		}
+		if ns.plugins.DataExchange.Plugin != nil {
+			if err := ns.plugins.DataExchange.Plugin.Start(); err != nil {
+				return err
+			}
+		}
+		for _, plugin := range ns.plugins.Tokens {
+			if err := plugin.Plugin.Start(); err != nil {
+				return err
+			}
+		}
 		if err := ns.orchestrator.Start(); err != nil {
 			return err
 		}

internal/namespace/manager_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/hyperledger/firefly/internal/database/difactory"
 	"github.com/hyperledger/firefly/internal/dataexchange/dxfactory"
 	"github.com/hyperledger/firefly/internal/identity/iifactory"
+	"github.com/hyperledger/firefly/internal/orchestrator"
 	"github.com/hyperledger/firefly/internal/sharedstorage/ssfactory"
 	"github.com/hyperledger/firefly/internal/tokens/tifactory"
 	"github.com/hyperledger/firefly/mocks/blockchainmocks"
@@ -1126,7 +1127,70 @@ func TestStart(t *testing.T) {
 	mo.AssertExpectations(t)
 }
 
-func TestStartFail(t *testing.T) {
+func TestStartBlockchainFail(t *testing.T) {
+	nm := newTestNamespaceManager(true)
+	defer nm.cleanup(t)
+
+	nm.namespaces = map[string]*namespace{
+		"ns": {
+			plugins: orchestrator.Plugins{
+				Blockchain: orchestrator.BlockchainPlugin{
+					Plugin: nm.mbi,
+				},
+			},
+		},
+	}
+
+	nm.mbi.On("Start").Return(fmt.Errorf("pop"))
+
+	err := nm.Start()
+	assert.EqualError(t, err, "pop")
+
+}
+
+func TestStartDataExchangeFail(t *testing.T) {
+	nm := newTestNamespaceManager(true)
+	defer nm.cleanup(t)
+
+	nm.namespaces = map[string]*namespace{
+		"ns": {
+			plugins: orchestrator.Plugins{
+				DataExchange: orchestrator.DataExchangePlugin{
+					Plugin: nm.mdx,
+				},
+			},
+		},
+	}
+
+	nm.mdx.On("Start").Return(fmt.Errorf("pop"))
+
+	err := nm.Start()
+	assert.EqualError(t, err, "pop")
+
+}
+
+func TestStartTokensFail(t *testing.T) {
+	nm := newTestNamespaceManager(true)
+	defer nm.cleanup(t)
+
+	nm.namespaces = map[string]*namespace{
+		"ns": {
+			plugins: orchestrator.Plugins{
+				Tokens: []orchestrator.TokensPlugin{{
+					Plugin: nm.mti,
+				}},
+			},
+		},
+	}
+
+	nm.mti.On("Start").Return(fmt.Errorf("pop"))
+
+	err := nm.Start()
+	assert.EqualError(t, err, "pop")
+
+}
+
+func TestStartOrchestratorFail(t *testing.T) {
 	nm := newTestNamespaceManager(true)
 	defer nm.cleanup(t)
 

internal/orchestrator/orchestrator.go

@@ -262,29 +262,16 @@ func (or *orchestrator) Start() (err error) {
 		if err == nil {
 			err = or.broadcast.Start()
 		}
-		if err == nil {
-			err = or.messaging.Start()
-		}
 		if err == nil {
 			err = or.sharedDownload.Start()
 		}
 	}
-	if err == nil {
-		err = or.blockchain().Start()
-	}
 	if err == nil {
 		err = or.events.Start()
 	}
 	if err == nil {
 		err = or.operations.Start()
 	}
-	if err == nil {
-		for _, el := range or.tokens() {
-			if err = el.Start(); err != nil {
-				break
-			}
-		}
-	}
 	or.started = true
 	return err
 }

internal/orchestrator/orchestrator_test.go

@@ -379,41 +379,6 @@ func TestStartBatchFail(t *testing.T) {
 	assert.EqualError(t, err, "pop")
 }
 
-func TestStartTokensFail(t *testing.T) {
-	coreconfig.Reset()
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-	or.mdi.On("GetNamespace", mock.Anything, "ns").Return(nil, nil)
-	or.mdi.On("UpsertNamespace", mock.Anything, mock.Anything, true).Return(nil)
-	or.mmp.On("ConfigureContract", mock.Anything, &core.FireFlyContracts{}).Return(nil)
-	or.mbi.On("Start").Return(nil)
-	or.mba.On("Start").Return(nil)
-	or.mem.On("Start").Return(nil)
-	or.mbm.On("Start").Return(nil)
-	or.mpm.On("Start").Return(nil)
-	or.msd.On("Start").Return(nil)
-	or.mom.On("Start").Return(nil)
-	or.mti.On("Start").Return(fmt.Errorf("pop"))
-	err := or.Start()
-	assert.EqualError(t, err, "pop")
-}
-
-func TestStartBlockchainsFail(t *testing.T) {
-	coreconfig.Reset()
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-	or.mdi.On("GetNamespace", mock.Anything, "ns").Return(nil, nil)
-	or.mdi.On("UpsertNamespace", mock.Anything, mock.Anything, true).Return(nil)
-	or.mmp.On("ConfigureContract", mock.Anything, &core.FireFlyContracts{}).Return(nil)
-	or.mbm.On("Start").Return(nil)
-	or.mpm.On("Start").Return(nil)
-	or.mba.On("Start").Return(nil)
-	or.msd.On("Start").Return(nil)
-	or.mbi.On("Start").Return(fmt.Errorf("pop"))
-	err := or.Start()
-	assert.EqualError(t, err, "pop")
-}
-
 func TestStartBlockchainsConfigureFail(t *testing.T) {
 	coreconfig.Reset()
 	or := newTestOrchestrator()
@@ -431,12 +396,9 @@ func TestStartStopOk(t *testing.T) {
 	or.mdi.On("GetNamespace", mock.Anything, "ns").Return(nil, nil)
 	or.mdi.On("UpsertNamespace", mock.Anything, mock.Anything, true).Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything, &core.FireFlyContracts{}).Return(nil)
-	or.mbi.On("Start").Return(nil)
 	or.mba.On("Start").Return(nil)
 	or.mem.On("Start").Return(nil)
 	or.mbm.On("Start").Return(nil)
-	or.mpm.On("Start").Return(nil)
-	or.mti.On("Start").Return(nil)
 	or.msd.On("Start").Return(nil)
 	or.mom.On("Start").Return(nil)
 	or.mba.On("WaitStop").Return(nil)

internal/privatemessaging/privatemessaging.go

@@ -49,7 +49,6 @@ type Manager interface {
 	core.Named
 	GroupManager
 
-	Start() error
 	NewMessage(msg *core.MessageInOut) sysmessaging.MessageSender
 	SendMessage(ctx context.Context, in *core.MessageInOut, waitConfirm bool) (out *core.Message, err error)
 	RequestReply(ctx context.Context, request *core.MessageInOut) (reply *core.MessageInOut, err error)
@@ -161,10 +160,6 @@ func (pm *privateMessaging) Name() string {
 	return "PrivateMessaging"
 }
 
-func (pm *privateMessaging) Start() error {
-	return pm.exchange.Start()
-}
-
 func (pm *privateMessaging) dispatchPinnedBatch(ctx context.Context, state *batch.DispatchState) error {
 	err := pm.dispatchBatchCommon(ctx, state)
 	if err != nil {

internal/privatemessaging/privatemessaging_test.go

@@ -535,14 +535,3 @@ func TestTransferBlobsOpInsertFail(t *testing.T) {
 
 	mdi.AssertExpectations(t)
 }
-
-func TestStart(t *testing.T) {
-	pm, cancel := newTestPrivateMessaging(t)
-	defer cancel()
-
-	mdx := pm.exchange.(*dataexchangemocks.Plugin)
-	mdx.On("Start").Return(nil)
-
-	err := pm.Start()
-	assert.NoError(t, err)
-}