breaksync

Documentation

Index

• Variables
• type Rupture
  • func NewRupture(name string, extract func(string) string) *Rupture
• type Source
  • func NewChanSource[T any](input <-chan T, keyer func(T) string, readRupture *Rupture) *Source[T]
  • func NewSliceSource[T any](arr []T, keyer func(T) string, readRupture *Rupture) *Source[T]
  • func NewSource[T any](reader func() (T, error), keyer func(T) string, readRupture *Rupture) *Source[T]
  • func (s *Source[T]) ComputeSynchro(key string)
  • func (s *Source[T]) Current() any
  • func (s *Source[T]) CurrentKey() string
  • func (s *Source[T]) IsSynchronized() bool
  • func (s *Source[T]) NextKey() string
  • func (s *Source[T]) Read() error
  • func (s *Source[T]) ReadRupture() *Rupture
• type SyncSource
• type Synchronization
  • func NewSynchronization() *Synchronization
  • func (s *Synchronization) AddSources(sources ...SyncSource) *Synchronization
  • func (s *Synchronization) Run(business func(uint64, []any) error) (err error)

Variables

var Identity = func(a string) string {
	return a
}

Identity consider key as rupture value

Types

type Rupture

type Rupture struct {
	// contains filtered or unexported fields
}

Rupture in a break/sync algorithm

func NewRupture

func NewRupture(name string, extract func(string) string) *Rupture

NewRupture creates and initialize Rupture

type Source

type Source[T any] struct {
	// contains filtered or unexported fields
}

Source of data in a break/sync algorithm

func NewChanSource

func NewChanSource[T any](input <-chan T, keyer func(T) string, readRupture *Rupture) *Source[T]

NewChanSource is a source from a chan, read sequentially

func NewSliceSource

func NewSliceSource[T any](arr []T, keyer func(T) string, readRupture *Rupture) *Source[T]

NewSliceSource is a source from a slice, read sequentially

func NewSource

func NewSource[T any](reader func() (T, error), keyer func(T) string, readRupture *Rupture) *Source[T]

NewSource creates and initialize Source

func (*Source[T]) ComputeSynchro

func (s *Source[T]) ComputeSynchro(key string)

func (*Source[T]) Current

func (s *Source[T]) Current() any

func (*Source[T]) CurrentKey

func (s *Source[T]) CurrentKey() string

func (*Source[T]) IsSynchronized

func (s *Source[T]) IsSynchronized() bool

func (*Source[T]) NextKey

func (s *Source[T]) NextKey() string

func (*Source[T]) Read

func (s *Source[T]) Read() error

func (*Source[T]) ReadRupture

func (s *Source[T]) ReadRupture() *Rupture

type SyncSource

type SyncSource interface {
	ReadRupture() *Rupture
	Current() any
	CurrentKey() string
	NextKey() string
	IsSynchronized() bool
	ComputeSynchro(string)
	Read() error
}

SyncSource behavior interface

type Synchronization

type Synchronization struct {
	// contains filtered or unexported fields
}

Synchronization in a break/sync algorithm

func NewSynchronization

func NewSynchronization() *Synchronization

NewSynchronization creates and initializes Synchronization

func (*Synchronization) AddSources

func (s *Synchronization) AddSources(sources ...SyncSource) *Synchronization

AddSources adds given source

func (*Synchronization) Run

func (s *Synchronization) Run(business func(uint64, []any) error) (err error)

Run start break/sync algorithm