breaksync

Documentation

Index

• Variables
• type Rupture
  • func NewRupture(name string, extract func([]byte) []byte) *Rupture
• type Source
  • func NewChanSource[T any](input <-chan T, keyer func(T) []byte, readRupture *Rupture) *Source[T]
  • func NewSliceSource[T any](arr []T, keyer func(T) []byte, readRupture *Rupture) *Source[T]
  • func NewSource[T any](reader func() (T, error), keyer func(T) []byte, readRupture *Rupture) *Source[T]
  • func (s *Source[T]) ComputeSynchro(key []byte)
  • func (s *Source[T]) Current() any
  • func (s *Source[T]) CurrentKey() []byte
  • func (s *Source[T]) IsSynchronized() bool
  • func (s *Source[T]) NextKey() []byte
  • 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) []byte {
	return []byte(a)
}

var RuptureIdentity = func(a []byte) []byte {
	return a
}

Types

type Rupture

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

func NewRupture

func NewRupture(name string, extract func([]byte) []byte) *Rupture

type Source

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

func NewChanSource

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

func NewSliceSource

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

func NewSource

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

func (*Source[T]) ComputeSynchro

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

func (*Source[T]) Current

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

func (*Source[T]) CurrentKey

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

func (*Source[T]) IsSynchronized

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

func (*Source[T]) NextKey

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

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() []byte
	NextKey() []byte
	IsSynchronized() bool
	ComputeSynchro([]byte)
	Read() error
}

type Synchronization

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

func NewSynchronization

func NewSynchronization() *Synchronization

func (*Synchronization) AddSources

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

func (*Synchronization) Run

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