adds stopper module (#6)

Author: vvuwei

.gitattributes

@@ -0,0 +1,2 @@
+go.sum merge=union
+**/mock/* linguist-generated=true

.gitignore

@@ -4,6 +4,7 @@
 *.dll
 *.so
 *.dylib
+.DS_Store
 
 # Test binary, built with `go test -c`
 *.test

Makefile

@@ -1,6 +1,9 @@
 -include .env
 export $(shell sed 's/=.*//' .env)
 
+init:
+	chmod +x init.dev.sh && ./init.dev.sh
+
 lint:
 	golangci-lint run
 
@@ -21,4 +24,6 @@ generate:
 		-p github.com/dipdup-io/uniswap
 
 test:
-	go test ./...
+	go test ./...
+
+.PHONY: init lint generate test

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=

init.dev.sh

@@ -0,0 +1,7 @@
+#!/bin/sh
+
+# VCS config
+git config --local core.attributesfile ./.gitattributes
+
+# Install third tools
+#go install go.uber.org/mock/mockgen@main

pkg/modules/cron/README.md

@@ -1,12 +1,12 @@
 # Cron
 
-Cron is the module which implements a cron specific parser and notifications about scheduler events.
+Cron is the module that implements a cron specific parser and notifications about scheduler events.
 
 ## Usage
 
-Usage of cron module is described by the [example](/examples/cron/).
+The example [describes usage of cron module](/examples/cron/).
 
-To import module in your code write following line:
+To import module in your code, write the following line:
 
 ```go
 import "github.com/dipdup-net/indexer-sdk/pkg/modules/cron"
@@ -33,7 +33,9 @@ if err := cronModule.Close(); err != nil {
 
 ## Config
 
-Default yaml config of cron module contains only one field `jobs`. It's a map of job names to cron pattern. Job names is used like subscription id in inner-message communication.
+Default yaml config of cron module contains only one field `jobs`.
+It's a map of job names to a cron pattern.
+Job names are used like subscription id in inner-message communication.
 
 ```yaml
 cron:
@@ -46,7 +48,9 @@ cron:
 
 ## Output
 
-Module sends to its outputs empty struct which notifies all connected modules about scheduled event. Each job of cron module has own output with names pointed in configuration file. So if your module should execute some work on `every_second` scheduled events from example you should connect it:
+Module sends to its outputs empty struct which notifies all connected modules about scheduled event.
+Each job of cron module has its own output with names pointed in the configuration file.
+So if your module should execute some work on `every_second` scheduled events from example you should connect it:
 
 ```go
 // with helper function
@@ -62,7 +66,7 @@ if err := cronModule.AttachTo("every_second", customModule.everySecond); err !=
 }
 ```
 
-Example of handling message from cron's outputs:
+Example of a handling message from cron's outputs:
 
 ```go
 for {

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
+}

pkg/modules/stopper/stopper_test.go

@@ -0,0 +1,32 @@
+package stopper
+
+import (
+	"context"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestStopperCallsStop(t *testing.T) {
+	stopWasCalled := false
+
+	var stopFunc context.CancelFunc = func() {
+		stopWasCalled = true
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	stopperModule := NewModule(stopFunc)
+	stopperModule.Start(ctx)
+
+	stopperInput, err := stopperModule.Input(InputName)
+	assert.NoError(t, err)
+
+	// Act: send stop signal to stopper
+	stopperInput.Push(struct{}{})
+
+	err = stopperModule.Close()
+	assert.NoError(t, err)
+
+	assert.True(t, stopWasCalled, "stop was never called")
+}