Add manager function and fixed bug

cmd/sidecar/start.go

@@ -104,11 +104,7 @@ func httpPProf(port int64) {
 	go func() {
 		addr := fmt.Sprintf("localhost:%d", port)
 		fmt.Printf("Pprof on localhost:%d\n\n", port)
-		err := http.ListenAndServe(addr, nil)
-		if err != nil {
-			fmt.Println(err)
-			panic(err)
-		}
+		tool.Asset(http.ListenAndServe(addr, nil))
 	}()
 }
 

internal/app/sidecar.go

@@ -64,10 +64,9 @@ func NewSidecar(repoRoot string, config *repo.Config) (internal.Launcher, error)
 	// TODO hub client
 	// privateKey DH
 	cryptor := txcrypto.NewCryptor(nil, appchainMgr, privateKey)
-	clientPort := appchain.NewPorts(clients, cryptor, logger)
-	r.Adds(clientPort)
-	rule, err := rulemgr.New(store, loggers.Logger(loggers.RuleMgr))
-	tool.Asset(err)
+	managerPort := appchain.NewManager(store, clients, cryptor, logger)
+	r.Adds(managerPort.Ports())
+	rule := rulemgr.New(store, loggers.Logger(loggers.RuleMgr))
 	mg, err := manager.NewManager(addr.String(), pm, appchainMgr, rule, loggers.Logger(loggers.Manager))
 	tool.Asset(err)
 	apiServer := api.NewServer(config, r, loggers.Logger(loggers.ApiServer))

internal/appchain/appchain.go

@@ -16,7 +16,6 @@ import (
 	"gitlab.33.cn/link33/sidecar/internal/txcrypto"
 	"gitlab.33.cn/link33/sidecar/model/pb"
 	"gitlab.33.cn/link33/sidecar/pkg/plugins"
-	"gitlab.33.cn/link33/sidecar/tool"
 )
 
 type AppChain interface {
@@ -35,7 +34,7 @@ type appChain struct {
 	msgCh   chan *pb.Message
 }
 
-func NewPort(client plugins.Client, cryptor txcrypto.Cryptor, logger logrus.FieldLogger) (port.Port, error) {
+func NewPort(client plugins.Client, cryptor txcrypto.Cryptor, logger logrus.FieldLogger) port.Port {
 	ctx, cancel := context.WithCancel(context.Background())
 	return &appChain{
 		client:  client,
@@ -45,14 +44,13 @@ func NewPort(client plugins.Client, cryptor txcrypto.Cryptor, logger logrus.Fiel
 		ctx:     ctx,
 		cancel:  cancel,
 		msgCh:   make(chan *pb.Message, port.Capacity),
-	}, nil
+	}
 }
 
 func NewPorts(clients []plugins.Client, cryptor txcrypto.Cryptor, logger logrus.FieldLogger) []port.Port {
 	var ps []port.Port
 	for _, c := range clients {
-		p, err := NewPort(c, cryptor, logger)
-		tool.Asset(err)
+		p := NewPort(c, cryptor, logger)
 		ps = append(ps, p)
 	}
 

internal/appchain/manager.go

+package appchain
+
+import (
+	"context"
+
+	"github.com/meshplus/bitxhub-kit/storage"
+	"github.com/sirupsen/logrus"
+
+	"gitlab.33.cn/link33/sidecar/internal/port"
+	"gitlab.33.cn/link33/sidecar/internal/txcrypto"
+	"gitlab.33.cn/link33/sidecar/pkg/plugins"
+	"gitlab.33.cn/link33/sidecar/tool"
+)
+
+type Manager struct {
+	ctx     context.Context
+	g       *tool.Group
+	clients []plugins.Client
+	store   storage.Storage
+	cryptor txcrypto.Cryptor
+	logger  logrus.FieldLogger
+}
+
+func NewManager(store storage.Storage, clients []plugins.Client, cryptor txcrypto.Cryptor, logger logrus.FieldLogger) *Manager {
+	g, ctx := tool.WithContext(context.Background())
+	return &Manager{
+		ctx:     ctx,
+		g:       g,
+		clients: clients,
+		store:   store,
+		cryptor: cryptor,
+		logger:  logger,
+	}
+}
+
+func (m *Manager) Start() error {
+	for _, client := range m.clients {
+		m.g.Go(func() error {
+			return client.Start()
+		})
+	}
+	return m.g.Wait()
+}
+
+func (m *Manager) Stop() error {
+	for _, client := range m.clients {
+		m.g.Go(func() error {
+			return client.Stop()
+		})
+	}
+	return m.g.Wait()
+}
+
+func (m *Manager) Ports() []port.Port {
+	return NewPorts(m.clients, m.cryptor, m.logger)
+}

internal/peermgr/swarm.go

@@ -94,10 +94,7 @@ func New(config *repo.Config, router router.Router, nodePrivKey crypto.PrivateKe
 }
 
 func (swarm *Swarm) add(p port.Port) {
-	err := swarm.router.Add(p)
-	if err != nil {
-		swarm.logger.Error(err)
-	}
+	swarm.router.Add(p)
 }
 
 func (swarm *Swarm) Start() error {

internal/router/interface.go

@@ -10,9 +10,9 @@ import (
 type Router interface {
 	internal.Launcher
 
-	Add(p port.Port) error
+	Add(p port.Port)
 
-	Adds(p []port.Port) error
+	Adds(p []port.Port)
 
 	Route(ibtp *pb.Message) error
 

internal/router/router.go

@@ -56,7 +56,7 @@ func (r *router) Stop() error {
 }
 
 // TODO
-func (r *router) Add(p port.Port) error {
+func (r *router) Add(p port.Port) {
 	r.portMap.Add(p)
 	go func() {
 		c := p.ListenIBTPX()
@@ -83,14 +83,12 @@ func (r *router) Add(p port.Port) error {
 			}
 		}
 	}()
-	return nil
 }
 
-func (r *router) Adds(p []port.Port) error {
+func (r *router) Adds(p []port.Port) {
 	for _, pt := range p {
 		r.Add(pt)
 	}
-	return nil
 }
 
 func (r *router) Load(key string) (value port.Port, ok bool) {

internal/rulemgr/rule_mgr.go

@@ -25,7 +25,7 @@ type RuleMgr struct {
 	logger logrus.FieldLogger
 }
 
-func New(storage storage.Storage, logger logrus.FieldLogger) (*RuleMgr, error) {
+func New(storage storage.Storage, logger logrus.FieldLogger) *RuleMgr {
 	ledger := &CodeLedger{
 		storage: storage,
 	}
@@ -36,10 +36,9 @@ func New(storage storage.Storage, logger logrus.FieldLogger) (*RuleMgr, error) {
 		Ve:     ve,
 		logger: logger,
 	}
-	return rm, nil
+	return rm
 }
 
 func (rm *RuleMgr) Validate(address, from string, proof, payload []byte, validators string) (bool, error) {
-	ok, err := rm.Ve.Validate(address, from, proof, payload, validators)
-	return ok, err
+	return rm.Ve.Validate(address, from, proof, payload, validators)
 }

pkg/plugins/grpc.go

@@ -165,6 +165,9 @@ func (g *GRPCClient) Stop() error {
 	if err != nil {
 		return err
 	}
+	if !g.kernel.Exited() {
+		g.kernel.Kill()
+	}
 	g.doneContext.Done()
 	return nil
 }
@@ -332,14 +335,6 @@ func (g *GRPCClient) Type() string {
 	return response.Type
 }
 
-func (g *GRPCClient) Kill() {
-	g.kernel.Kill()
-}
-
-func (g *GRPCClient) Exited() bool {
-	return g.kernel.Exited()
-}
-
 func (g *GRPCClient) Bind(kern Kernel) {
 	g.kernel = kern
 }

pkg/plugins/interface.go

@@ -19,7 +19,6 @@ type Launcher interface {
 //go:generate mockgen -destination mock_client/mock_client.go -package mock_client -source interface.go
 type Client interface { // 业务实现委托接口。
 	Launcher
-	Kernel
 	Bind(kern Kernel)
 
 	// Initialize initialize plugin client

pkg/plugins/start.go

@@ -99,20 +99,19 @@ func CreateClients(appchainConfigs []repo.Appchain, extra []byte) []Client {
 		// Request the plugin
 		raw, err := rpcClient.Dispense(appchainConfig.Plugin)
 		tool.Asset(err)
-		var appchain Client
+		var client Client
 		switch raw.(type) {
 		case *GRPCClient:
-			appchain = raw.(*GRPCClient)
+			client = raw.(*GRPCClient)
 		default:
 			tool.Asset(fmt.Errorf("unsupported kernel type"))
 		}
 
 		// initialize our kernel plugin
-		err = appchain.Initialize(pluginConfigPath, appchainConfig.DID, extra)
+		err = client.Initialize(pluginConfigPath, appchainConfig.DID, extra)
 		tool.Asset(err)
-		appchain.Bind(kernel)
-		tool.Asset(appchain.Start())
-		clients = append(clients, appchain)
+		client.Bind(kernel)
+		clients = append(clients, client)
 	}
 
 	return clients