zipper

README

Zipper

It's a module receiving two input streams and aggregate them by key.

Usage

Usage of module is described by the example.

To import module in your code write following line:

import "github.com/dipdup-net/indexer-sdk/pkg/modules/zippper"

Zipper module implements interface Module. So you can use it like any other module. For example:

// create zip module
zip := zipper.NewModule[int]()

// start zip module
zip.Start(ctx)

// your code is here

// close zip module
if err := zip.Close(); err != nil {
    log.Panic(err)
}

Zipper is the generic structure. Type parameter Key is comparable constraint

type Module[Key comparable] struct {
	firstInput  *modules.Input
	secondInput *modules.Input

	output *modules.Output

	firstStream  map[Key]Zippable[Key]
	secondStream map[Key]Zippable[Key]

	zipFunc ZipFunction[Key]

	mx *sync.RWMutex
	wg *sync.WaitGroup
}

Module has 2 constructors:

// creates zip module with default ZipFunction
func NewModule[Key comparable]() *Module[Key]

// creates zip module with custom ZipFunction
func NewModuleWithFunc[Key comparable](f ZipFunction[Key]) (*Module[Key], error)

ZipFunction is the function which can be used to redeclare key comparing rules. The function has to return nil if it can't zip structures. It has following declaration:

type ZipFunction[Type comparable] func(x Zippable[Type], y Zippable[Type]) *Result[Type]

Default zip function declares like that:

func defaultZip[Type comparable](x Zippable[Type], y Zippable[Type]) *Result[Type] {
	if x.Key() != y.Key() {
		return nil
	}
	return &Result[Type]{
		First:  x,
		Second: y,
		Key:    x.Key(),
	}
}

Input

Inputs receives types realizing Zippable interface.

type Zippable[Type comparable] interface {
	Key() Type
}

Zippable requires realization of one method Key which returns key. Received key will be used for zipping data streams. For example:

zip := zipper.NewModule[int]()

type ZipData struct {
	key   int
	value string
}

func (z ZipData) Key() int {
	return z.key
}

In the example zip module created with Key type int. That's why to use the module you should realize method Key on data structure which returns int.

Package contains declared constants of inputs name:

FirstInputName  = "first"
SecondInputName = "second"

Output

When module zipped data stream it sends Result structure to output.

type Result[Type comparable] struct {
	Key    Type
	First  any
	Second any
}

It contains key which was used to zip. Also it contains data of first and second stream types.

Package contains declared constant of output name:

OutputName = "output"

Documentation

Index

• Constants
• Variables
• type Module
  • func NewModule[Key comparable]() *Module[Key]
  • func NewModuleWithFunc[Key comparable](f ZipFunction[Key]) (*Module[Key], error)
  • func (m *Module[Key]) AttachTo(outputModule modules.Module, outputName, inputName string) error
  • func (m *Module[Key]) Close() error
  • func (m *Module[Key]) Input(name string) (*modules.Input, error)
  • func (m *Module[Key]) MustInput(name string) *modules.Input
  • func (m *Module[Key]) MustOutput(name string) *modules.Output
  • func (*Module[Key]) Name() string
  • func (m *Module[Key]) Output(name string) (*modules.Output, error)
  • func (m *Module[Key]) Start(ctx context.Context)
• type Result
• type ZipFunction
• type Zippable

Constants

const (
	FirstInputName  = "first"
	SecondInputName = "second"

	OutputName = "output"

	ModuleName = "zipper"
)

predefined names

Variables

var (
	ErrNilZipFunc = errors.New("nil zip function")
)

errors

Types

type Module

type Module[Key comparable] struct {
	// contains filtered or unexported fields
}

Module - zip module

func NewModule

func NewModule[Key comparable]() *Module[Key]

NewModule - creates zip module

func NewModuleWithFunc

func NewModuleWithFunc[Key comparable](f ZipFunction[Key]) (*Module[Key], error)

NewModuleWithFunc - creates zip module with custom zip function

func (*Module[Key]) AttachTo

func (m *Module[Key]) AttachTo(outputModule modules.Module, outputName, inputName string) error

AttachTo - attach input to output with name

func (*Module[Key]) Close

func (m *Module[Key]) Close() error

Close - gracefully stops module

func (*Module[Key]) Input

func (m *Module[Key]) Input(name string) (*modules.Input, error)

Input - returns input by name

func (*Module[Key]) MustInput

func (m *Module[Key]) MustInput(name string) *modules.Input

MustInput - returns input by name

func (*Module[Key]) MustOutput

func (m *Module[Key]) MustOutput(name string) *modules.Output

MustOutput - returns output by name

func (*Module[Key]) Name

func (*Module[Key]) Name() string

Name - returns module name

func (*Module[Key]) Output

func (m *Module[Key]) Output(name string) (*modules.Output, error)

Output - returns output by name

func (*Module[Key]) Start

func (m *Module[Key]) Start(ctx context.Context)

Start - starts module

type Result

type Result[Type comparable] struct {
	Key    Type
	First  any
	Second any
}

Result - data structure is result of zip operation. It has two entity

type ZipFunction

type ZipFunction[Type comparable] func(x Zippable[Type], y Zippable[Type]) *Result[Type]

ZipFunction - function to check can we zip two entity and if we can zip it. If function returns nil then it can't zip 2 entities

type Zippable

type Zippable[Type comparable] interface {
	Key() Type
}

Zippable - interface of data which can be zipped