Adds BaseModule

.run/examples_cron.run.xml

@@ -0,0 +1,21 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="examples/cron" type="GoApplicationRunConfiguration" factoryName="Go Application">
+    <module name="indexer-sdk" />
+    <working_directory value="$PROJECT_DIR$/examples/cron" />
+    <EXTENSION ID="net.ashald.envfile">
+      <option name="IS_ENABLED" value="false" />
+      <option name="IS_SUBST" value="false" />
+      <option name="IS_PATH_MACRO_SUPPORTED" value="false" />
+      <option name="IS_IGNORE_MISSING_FILES" value="false" />
+      <option name="IS_ENABLE_EXPERIMENTAL_INTEGRATIONS" value="false" />
+      <ENTRIES>
+        <ENTRY IS_ENABLED="true" PARSER="runconfig" IS_EXECUTABLE="false" />
+      </ENTRIES>
+    </EXTENSION>
+    <kind value="DIRECTORY" />
+    <package value="github.com/dipdup-net/indexer-sdk/examples/cron" />
+    <directory value="$PROJECT_DIR$/examples/cron" />
+    <filePath value="$PROJECT_DIR$/examples/cron/main.go" />
+    <method v="2" />
+  </configuration>
+</component>

.run/examples_grpc.run.xml

@@ -0,0 +1,21 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="examples/grpc" type="GoApplicationRunConfiguration" factoryName="Go Application">
+    <module name="indexer-sdk" />
+    <working_directory value="$PROJECT_DIR$/examples/grpc" />
+    <EXTENSION ID="net.ashald.envfile">
+      <option name="IS_ENABLED" value="false" />
+      <option name="IS_SUBST" value="false" />
+      <option name="IS_PATH_MACRO_SUPPORTED" value="false" />
+      <option name="IS_IGNORE_MISSING_FILES" value="false" />
+      <option name="IS_ENABLE_EXPERIMENTAL_INTEGRATIONS" value="false" />
+      <ENTRIES>
+        <ENTRY IS_ENABLED="true" PARSER="runconfig" IS_EXECUTABLE="false" />
+      </ENTRIES>
+    </EXTENSION>
+    <kind value="DIRECTORY" />
+    <package value="github.com/dipdup-net/indexer-sdk/examples/grpc" />
+    <directory value="$PROJECT_DIR$/examples/grpc" />
+    <filePath value="$PROJECT_DIR$/examples/grpc/main.go" />
+    <method v="2" />
+  </configuration>
+</component>

.run/examples_zipper.run.xml

@@ -0,0 +1,21 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="examples/zipper" type="GoApplicationRunConfiguration" factoryName="Go Application">
+    <module name="indexer-sdk" />
+    <working_directory value="$PROJECT_DIR$/examples/zipper" />
+    <EXTENSION ID="net.ashald.envfile">
+      <option name="IS_ENABLED" value="false" />
+      <option name="IS_SUBST" value="false" />
+      <option name="IS_PATH_MACRO_SUPPORTED" value="false" />
+      <option name="IS_IGNORE_MISSING_FILES" value="false" />
+      <option name="IS_ENABLE_EXPERIMENTAL_INTEGRATIONS" value="false" />
+      <ENTRIES>
+        <ENTRY IS_ENABLED="true" PARSER="runconfig" IS_EXECUTABLE="false" />
+      </ENTRIES>
+    </EXTENSION>
+    <kind value="DIRECTORY" />
+    <package value="github.com/dipdup-net/indexer-sdk/examples/zipper/examples/zipper" />
+    <directory value="$PROJECT_DIR$/examples/zipper" />
+    <filePath value="$PROJECT_DIR$/examples/zipper/main.go" />
+    <method v="2" />
+  </configuration>
+</component>

examples/cron/custom.go

@@ -9,6 +9,8 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
+var _ modules.Module = (*CustomModule)(nil)
+
 // CustomModule -
 type CustomModule struct {
 	everySecond     *modules.Input
@@ -50,8 +52,22 @@ func (m *CustomModule) Output(name string) (*modules.Output, error) {
 }
 
 // AttachTo -
-func (m *CustomModule) AttachTo(name string, input *modules.Input) error {
-	return errors.Wrap(modules.ErrUnknownOutput, name)
+func (m *CustomModule) AttachTo(outputM modules.Module, outputName, inputName string) error {
+	output, err := outputM.Output(outputName)
+	if err != nil {
+		return nil
+	}
+
+	switch inputName {
+	case "every_second":
+		output.Attach(m.everySecond)
+		return nil
+	case "every_five_second":
+		output.Attach(m.everyFiveSecond)
+		return nil
+	default:
+		return errors.Wrap(modules.ErrUnknownInput, inputName)
+	}
 }
 
 func (m *CustomModule) listen(ctx context.Context) {

examples/grpc/client.go

@@ -23,6 +23,8 @@ type Client struct {
 	wg     *sync.WaitGroup
 }
 
+var _ modules.Module = (*Client)(nil)
+
 // NewClient -
 func NewClient(server string) *Client {
 	return &Client{
@@ -113,12 +115,18 @@ func (client *Client) Output(name string) (*modules.Output, error) {
 }
 
 // AttachTo -
-func (client *Client) AttachTo(name string, input *modules.Input) error {
-	output, err := client.Output(name)
+func (client *Client) AttachTo(outputModule modules.Module, outputName, inputName string) error {
+	outputChannel, err := outputModule.Output(outputName)
+	if err != nil {
+		return err
+	}
+
+	input, err := client.Input(inputName)
 	if err != nil {
 		return err
 	}
-	output.Attach(input)
+
+	outputChannel.Attach(input)
 	return nil
 }
 

examples/grpc/custom.go

@@ -45,8 +45,19 @@ func (m *CustomModule) Output(name string) (*modules.Output, error) {
 }
 
 // AttachTo -
-func (m *CustomModule) AttachTo(name string, input *modules.Input) error {
-	return errors.Wrap(modules.ErrUnknownOutput, name)
+func (m *CustomModule) AttachTo(outputModule modules.Module, outputName, inputName string) error {
+	outputChannel, err := outputModule.Output(outputName)
+	if err != nil {
+		return err
+	}
+
+	input, err := m.Input(inputName)
+	if err != nil {
+		return err
+	}
+
+	outputChannel.Attach(input)
+	return nil
 }
 
 func (m *CustomModule) listen(ctx context.Context) {

examples/zipper/custom.go

@@ -43,12 +43,18 @@ func (m *CustomModule) Output(name string) (*modules.Output, error) {
 }
 
 // AttachTo -
-func (m *CustomModule) AttachTo(name string, input *modules.Input) error {
-	output, err := m.Output(name)
+func (m *CustomModule) AttachTo(outputModule modules.Module, outputName, inputName string) error {
+	outputChannel, err := outputModule.Output(outputName)
 	if err != nil {
 		return err
 	}
-	output.Attach(input)
+
+	input, err := m.Input(inputName)
+	if err != nil {
+		return err
+	}
+
+	outputChannel.Attach(input)
 	return nil
 }
 

examples/zipper/main.go

@@ -15,17 +15,19 @@ func main() {
 	first := NewCustomModule(10, -1, "first")
 	second := NewCustomModule(0, 1, "second")
 
-	if err := modules.Connect(first, zip, "output", zipper.FirstInputName); err != nil {
+	if err := modules.Connect(first, zip, zipper.OutputName, zipper.FirstInputName); err != nil {
 		log.Panic(err)
 	}
-	if err := modules.Connect(second, zip, "output", zipper.SecondInputName); err != nil {
+	if err := modules.Connect(second, zip, zipper.OutputName, zipper.SecondInputName); err != nil {
 		log.Panic(err)
 	}
 
 	fakeInput := modules.NewInput("fake")
-	if err := zip.AttachTo(zipper.OutputName, fakeInput); err != nil {
+	zipOutput, err := zip.Output(zipper.OutputName)
+	if err != nil {
 		log.Panic(err)
 	}
+	zipOutput.Attach(fakeInput)
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 	defer cancel()

pkg/modules/README.md

@@ -97,7 +97,3 @@ gRPC module where realized default client and server. Detailed docs can be found
 ### Cron
 
 Cron module implements cron scheduler. Detailed docs can be found [here](/pkg/modules/cron/).
-
-### Duplicator
-
-Duplicator module repeats signal from one of inputs to all outputs. Detailed docs can be found [here](/pkg/modules/duplicator/).

pkg/modules/baseModule.go

@@ -0,0 +1,81 @@
+package modules
+
+import (
+	"context"
+	"github.com/dipdup-io/workerpool"
+	"github.com/dipdup-net/indexer-sdk/pkg/sync"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+)
+
+var _ Module = (*BaseModule)(nil)
+
+type BaseModule struct {
+	name    string
+	inputs  *sync.Map[string, *Input]
+	outputs *sync.Map[string, *Output]
+	Log     zerolog.Logger
+	G       workerpool.Group
+}
+
+func New(name string) BaseModule {
+	m := BaseModule{
+		name:    name,
+		inputs:  sync.NewMap[string, *Input](),
+		outputs: sync.NewMap[string, *Output](),
+		Log:     log.With().Str("module", name).Logger(),
+		G:       workerpool.NewGroup(),
+	}
+
+	return m
+}
+
+func (m *BaseModule) Name() string {
+	return m.name
+}
+
+func (*BaseModule) Start(_ context.Context) {}
+
+func (*BaseModule) Close() error {
+	return nil
+}
+
+func (m *BaseModule) Input(name string) (*Input, error) {
+	input, ok := m.inputs.Get(name)
+	if !ok {
+		return nil, errors.Wrap(ErrUnknownInput, name)
+	}
+	return input, nil
+}
+
+func (m *BaseModule) CreateInput(name string) {
+	m.inputs.Set(name, NewInput(name))
+}
+
+func (m *BaseModule) Output(name string) (*Output, error) {
+	output, ok := m.outputs.Get(name)
+	if !ok {
+		return nil, errors.Wrap(ErrUnknownOutput, name)
+	}
+	return output, nil
+}
+
+func (m *BaseModule) CreateOutput(name string) {
+	m.outputs.Set(name, NewOutput(name))
+}
+
+func (m *BaseModule) AttachTo(outputModule Module, outputName, inputName string) error {
+	outputChannel, err := outputModule.Output(outputName)
+	if err != nil {
+		return err
+	}
+
+	input, err := m.Input(inputName)
+	if err != nil {
+		return err
+	}
+
+	outputChannel.Attach(input)
+	return nil
+}