syncz

Documentation

Index

• type EventCallback
• type Mutex
  • func NewMutex() Mutex
  • func (m Mutex) Lock(ctx context.Context) bool
  • func (m Mutex) TryLock() bool
  • func (m Mutex) Unlock()
• type StripedValue
  • func NewStripedValueInit[V any](n int, newV func() V) StripedValue[V]
  • func (v *StripedValue[V]) GetPointer(x int64) *V
• type Subscriptions
  • func (s *Subscriptions[E]) Dispatch(ctx context.Context, e E)
  • func (s *Subscriptions[E]) On(ctx context.Context, cb EventCallback[E])
• type WorkSource
• type Worker
• type WorkerFactory
• type WorkerFunc
  • func (wf WorkerFunc) Run(ctx context.Context)
• type WorkerHolder
  • func NewComparableWorkerHolder[C comparable](factory func(C) Worker) *WorkerHolder[C]
  • func NewProtoWorkerHolder[C proto.Message](factory func(C) Worker) *WorkerHolder[C]
  • func NewWorkerHolder[C any](factory func(C) Worker, isEqual func(config1, config2 C) bool) *WorkerHolder[C]
  • func (w *WorkerHolder[C]) ApplyConfig(ctx context.Context, config C) bool
  • func (w *WorkerHolder[C]) StopAndWait()
• type WorkerManager
  • func NewWorkerManager[ID comparable, C any](log *zap.Logger, fld func(ID) zap.Field, workerFactory WorkerFactory[ID, C], ...) *WorkerManager[ID, C]
  • func (m *WorkerManager[ID, C]) ApplyConfiguration(sources []WorkSource[ID, C]) error
  • func (m *WorkerManager[ID, C]) StopAllWorkers()

Types

type EventCallback

type EventCallback[E any] func(ctx context.Context, e E)

type Mutex

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

Mutex is a non-reentrant (like sync.Mutex) mutex that (unlike sync.Mutex) allows to acquire the mutex with a possibility to abort the attempt early if a context signals done.

A buffered channel of size 1 is used as the mutex. Think of it as of a box - the party that has put something into it has acquired the mutex. To unlock it, remove the contents from the box, so that someone else can use it. An empty box is created in the NewMutex() constructor.

TryLock, Lock, and Unlock provide memory access ordering guarantees by piggybacking on channel's "happens before" guarantees. See https://golang.org/ref/mem

func NewMutex

func NewMutex() Mutex

func (Mutex) Lock

func (m Mutex) Lock(ctx context.Context) bool

func (Mutex) TryLock

func (m Mutex) TryLock() bool

func (Mutex) Unlock

func (m Mutex) Unlock()

type StripedValue

type StripedValue[V any] struct {
	// Stripes holds the stripes.
	Stripes []V
	// contains filtered or unexported fields
}

StripedValue is a value that is partitioned into 2^n stripes. See https://github.com/google/guava/wiki/StripedExplained for a similar idea applied to mutexes.

Number of stripes is a power of two to make it possible to use fast bit operations instead of slow division remainder.

func NewStripedValueInit

func NewStripedValueInit[V any](n int, newV func() V) StripedValue[V]

NewStripedValueInit constructs a new striped value. Each stripe has a value provided by a constructor function.

It's not a pointer because it doesn't contain any non-pointer mutable fields.

func (*StripedValue[V]) GetPointer

func (v *StripedValue[V]) GetPointer(x int64) *V

GetPointer retrieves the pointer to the stripe for x.

type Subscriptions

type Subscriptions[E any] struct {
	// contains filtered or unexported fields
}

func (*Subscriptions[E]) Dispatch

func (s *Subscriptions[E]) Dispatch(ctx context.Context, e E)

Dispatch dispatches the given event to all added subscriptions.

func (*Subscriptions[E]) On

func (s *Subscriptions[E]) On(ctx context.Context, cb EventCallback[E])

type WorkSource

type WorkSource[ID comparable, C any] struct {
	ID            ID
	Configuration C
}

type Worker

type Worker interface {
	Run(context.Context)
}

type WorkerFactory

type WorkerFactory[ID comparable, C any] interface {
	New(WorkSource[ID, C]) Worker
}

type WorkerFunc

type WorkerFunc func(context.Context)

func (WorkerFunc) Run

func (wf WorkerFunc) Run(ctx context.Context)

type WorkerHolder

type WorkerHolder[C any] struct {
	// contains filtered or unexported fields
}

WorkerHolder holds a worker and restarts it when configuration changes.

func NewComparableWorkerHolder

func NewComparableWorkerHolder[C comparable](factory func(C) Worker) *WorkerHolder[C]

func NewProtoWorkerHolder

func NewProtoWorkerHolder[C proto.Message](factory func(C) Worker) *WorkerHolder[C]

func NewWorkerHolder

func NewWorkerHolder[C any](factory func(C) Worker, isEqual func(config1, config2 C) bool) *WorkerHolder[C]

func (*WorkerHolder[C]) ApplyConfig

func (w *WorkerHolder[C]) ApplyConfig(ctx context.Context, config C) bool

ApplyConfig ensures a worker is running with the provided or equal config.

This method starts a worker if it's not running already. If it is running and the config is not equal then the worker is stopped, a new worker is started then with the new config.

func (*WorkerHolder[C]) StopAndWait

func (w *WorkerHolder[C]) StopAndWait()

type WorkerManager

type WorkerManager[ID comparable, C any] struct {
	// contains filtered or unexported fields
}

func NewWorkerManager

func NewWorkerManager[ID comparable, C any](log *zap.Logger, fld func(ID) zap.Field, workerFactory WorkerFactory[ID, C], equal func(c1, c2 C) bool) *WorkerManager[ID, C]

func (*WorkerManager[ID, C]) ApplyConfiguration

func (m *WorkerManager[ID, C]) ApplyConfiguration(sources []WorkSource[ID, C]) error

func (*WorkerManager[ID, C]) StopAllWorkers

func (m *WorkerManager[ID, C]) StopAllWorkers()