Added implementation of stopper module.

Author: vvuwei

go.mod

@@ -2,8 +2,8 @@ module github.com/dipdup-net/indexer-sdk
 
 go 1.20
 
-
 require (
+	github.com/dipdup-io/workerpool v0.0.4
 	github.com/dipdup-net/go-lib v0.3.0
 	github.com/ethereum/go-ethereum v1.12.0
 	github.com/go-testfixtures/testfixtures/v3 v3.9.0

go.sum

@@ -46,6 +46,8 @@ github.com/decred/dcrd/crypto/blake256 v1.0.1 h1:7PltbUIQB7u/FfZ39+DGa/ShuMyJ5il
 github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 h1:8UrgZ3GkP4i/CLijOJx79Yu+etlyjdBU4sfcs2WYQMs=
 github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0/go.mod h1:v57UDF4pDQJcEfFUCRop3lJL149eHGSe9Jvczhzjo/0=
 github.com/denisenkom/go-mssqldb v0.12.3 h1:pBSGx9Tq67pBOTLmxNuirNTeB8Vjmf886Kx+8Y+8shw=
+github.com/dipdup-io/workerpool v0.0.4 h1:m58fuFY3VIPRc+trWpjw2Lsm4FvIgtjP/4VRe79r+/s=
+github.com/dipdup-io/workerpool v0.0.4/go.mod h1:m6YMqx7M+fORTyabHD/auKymBRpbDax0t1aPZ1i7GZA=
 github.com/dipdup-net/go-lib v0.3.0 h1:56OImCLIHyLL4Da7UI5H2xyTP65i2FUXSO6Gyxll460=
 github.com/dipdup-net/go-lib v0.3.0/go.mod h1:oBDOSsM/F8fEnmuDnaJ6QA/cHH4lne49ASbsh8WXDe4=
 github.com/docker/distribution v2.8.2+incompatible h1:T3de5rq0dB1j30rp0sA2rER+m322EBzniBPB6ZIzuh8=

pkg/modules/stopper/stopper.go

@@ -0,0 +1,94 @@
+package stopper
+
+import (
+	"context"
+	"github.com/dipdup-io/workerpool"
+	"github.com/dipdup-net/indexer-sdk/pkg/modules"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+)
+
+const (
+	InputName = "signal"
+)
+
+// Module - cancels context of all application if get signal.
+//
+//	                |----------------|
+//	                |                |
+//	-- struct{} ->  |     MODULE     |
+//	                |                |
+//	                |----------------|
+type Module struct {
+	input *modules.Input
+	stop  context.CancelFunc
+	log   zerolog.Logger
+	g     workerpool.Group
+}
+
+func NewModule(cancelFunc context.CancelFunc) Module {
+	m := Module{
+		input: modules.NewInput(InputName),
+		stop:  cancelFunc,
+		g:     workerpool.NewGroup(),
+	}
+	m.log = log.With().Str("module", m.Name()).Logger()
+
+	return m
+}
+
+func (*Module) Name() string {
+	return "stopper"
+}
+
+// Start -
+func (s *Module) Start(ctx context.Context) {
+	s.g.GoCtx(ctx, s.listen)
+}
+
+func (s *Module) listen(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-s.input.Listen():
+			log.Info().Msg("stop signal received")
+			if s.stop != nil {
+				log.Info().Msg("cancelling context...")
+				s.stop()
+				return
+			}
+		}
+	}
+}
+
+// Close -
+func (s *Module) Close() error {
+	s.g.Wait()
+	return s.input.Close()
+}
+
+// Output -
+func (*Module) Output(name string) (*modules.Output, error) {
+	return nil, errors.Wrap(modules.ErrUnknownOutput, name)
+}
+
+// Input -
+func (s *Module) Input(name string) (*modules.Input, error) {
+	if name != InputName {
+		return nil, errors.Wrap(modules.ErrUnknownInput, name)
+	}
+	return s.input, nil
+}
+
+// AttachTo -
+func (s *Module) AttachTo(name string, input *modules.Input) error {
+	output, err := s.Output(name)
+	if err != nil {
+		return err
+	}
+
+	output.Attach(input)
+	return nil
+}