POS-86: Graceful shutdown of Heimdall deamon, REST and Bridge

bridge/cmd/start.go

@@ -1,6 +1,7 @@
 package cmd
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"os/signal"
@@ -19,6 +20,7 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/tendermint/tendermint/libs/common"
 	httpClient "github.com/tendermint/tendermint/rpc/client"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/maticnetwork/heimdall/helper"
 	"github.com/spf13/viper"
@@ -28,6 +30,105 @@ const (
 	waitDuration = 1 * time.Minute
 )
 
+// StartBridgeWithCtx starts bridge service and is able to shutdow gracefully
+// returns service errors, if any
+func StartBridgeWithCtx(shutdownCtx context.Context) error {
+	var logger = helper.Logger.With("module", "bridge/cmd/")
+
+	// create codec
+	cdc := app.MakeCodec()
+	// queue connector & http client
+	_queueConnector := queue.NewQueueConnector(helper.GetConfig().AmqpURL)
+	_queueConnector.StartWorker()
+
+	_txBroadcaster := broadcaster.NewTxBroadcaster(cdc)
+	_httpClient := httpClient.NewHTTP(helper.GetConfig().TendermintRPCUrl, "/websocket")
+
+	// selected services to start
+	services := []common.Service{}
+	services = append(services,
+		listener.NewListenerService(cdc, _queueConnector, _httpClient),
+		processor.NewProcessorService(cdc, _queueConnector, _httpClient, _txBroadcaster),
+	)
+
+	// Start http client
+	err := _httpClient.Start()
+	if err != nil {
+		logger.Error("Error connecting to server: %v", err)
+		return err
+	}
+
+	// cli context
+	cliCtx := cliContext.NewCLIContext().WithCodec(cdc)
+	cliCtx.BroadcastMode = client.BroadcastAsync
+	cliCtx.TrustNode = true
+
+	// start bridge services only when node fully synced
+	loop := true
+	for loop {
+		select {
+		case <-shutdownCtx.Done():
+			return nil
+		case <-time.After(waitDuration):
+			if !util.IsCatchingUp(cliCtx) {
+				logger.Info("Node up to date, starting bridge services")
+				loop = false
+			} else {
+				logger.Info("Waiting for heimdall to be synced")
+			}
+		}
+	}
+
+	// start services
+	g := new(errgroup.Group)
+	for _, service := range services {
+		// loop variable must be captured
+		srv := service
+		g.Go(func() error {
+			if err := srv.Start(); err != nil {
+				logger.Error("GetStartCmd | serv.Start", "Error", err)
+				return err
+			}
+			<-srv.Quit()
+			return nil
+		})
+	}
+
+	// shutdown phase
+	g.Go(func() error {
+		// wait for interrupt and start shut down
+		<-shutdownCtx.Done()
+
+		logger.Info("Received stop signal - Stopping all services")
+		for _, service := range services {
+			srv := service
+			if srv.IsRunning() {
+				if err := srv.Stop(); err != nil {
+					logger.Error("GetStartCmd | service.Stop", "Error", err)
+					return err
+				}
+			}
+		}
+		// stop http client
+		if err := _httpClient.Stop(); err != nil {
+			logger.Error("GetStartCmd | _httpClient.Stop", "Error", err)
+			return err
+		}
+		// stop db instance
+		util.CloseBridgeDBInstance()
+
+		return nil
+	})
+
+	// wait for all routines to finish and log error
+	if err := g.Wait(); err != nil {
+		logger.Error("Bridge stopped", "err", err)
+		return err
+	}
+
+	return nil
+}
+
 // StartBridge starts bridge service, isStandAlone prevents os.Exit if the bridge started as side service
 func StartBridge(isStandAlone bool) {
 	var logger = helper.Logger.With("module", "bridge/cmd/")
@@ -102,7 +203,7 @@ func StartBridge(isStandAlone bool) {
 		time.Sleep(waitDuration)
 	}
 
-	// strt all processes
+	// start all processes
 	for _, service := range services {
 		go func(serv common.Service) {
 			defer wg.Done()