routine

Documentation

Index

• func NewLogExitedCallback(le *logrus.Entry) func(err error)
• type Option
  • func WithBackoff(bo cbackoff.BackOff) Option
  • func WithExitCb(cb func(err error)) Option
  • func WithExitLogger(le *logrus.Entry) Option
  • func WithRetry(boConf *backoff.Backoff) Option
• type Routine
• type RoutineContainer
  • func NewRoutineContainer(opts ...Option) *RoutineContainer
  • func NewRoutineContainerWithLogger(le *logrus.Entry, opts ...Option) *RoutineContainer
  • func (k *RoutineContainer) ClearContext() bool
  • func (k *RoutineContainer) RestartRoutine() bool
  • func (k *RoutineContainer) SetContext(ctx context.Context, restart bool) bool
  • func (k *RoutineContainer) SetRoutine(routine Routine) (waitReturn <-chan struct{}, reset bool)
  • func (k *RoutineContainer) WaitExited(ctx context.Context, returnIfNotRunning bool, errCh <-chan error) error
• type StateRoutine
• type StateRoutineContainer
  • func NewStateRoutineContainer[T comparable](compare func(t1, t2 T) bool, opts ...Option) *StateRoutineContainer[T]
  • func NewStateRoutineContainerVT[T proto.EqualVT[T]](opts ...Option) *StateRoutineContainer[T]
  • func NewStateRoutineContainerWithLogger[T comparable](compare func(t1, t2 T) bool, le *logrus.Entry, opts ...Option) *StateRoutineContainer[T]
  • func NewStateRoutineContainerWithLoggerVT[T proto.EqualVT[T]](le *logrus.Entry, opts ...Option) *StateRoutineContainer[T]
  • func (s *StateRoutineContainer[T]) ClearContext() bool
  • func (s *StateRoutineContainer[T]) GetState() T
  • func (s *StateRoutineContainer[T]) RestartRoutine() bool
  • func (s *StateRoutineContainer[T]) SetContext(ctx context.Context, restart bool) bool
  • func (s *StateRoutineContainer[T]) SetState(state T) (waitReturn <-chan struct{}, changed, reset, running bool)
  • func (s *StateRoutineContainer[T]) SetStateRoutine(routine StateRoutine[T]) (waitReturn <-chan struct{}, reset, running bool)
  • func (s *StateRoutineContainer[T]) SwapValue(cb func(val T) T) (nextState T, waitReturn <-chan struct{}, changed, reset, running bool)
  • func (s *StateRoutineContainer[T]) WaitExited(ctx context.Context, returnIfNotRunning bool, errCh <-chan error) error

Functions

func NewLogExitedCallback

func NewLogExitedCallback(le *logrus.Entry) func(err error)

NewLogExitedCallback returns a ExitedCb which logs when a controller exited.

Types

type Option

type Option interface {
	// ApplyToRoutineContainer applies the option to the RoutineContainer.
	ApplyToRoutineContainer(k *RoutineContainer)
}

Option is an option for a RoutineContainer instance.

func WithBackoff

func WithBackoff(bo cbackoff.BackOff) Option

WithBackoff configures a backoff to use when the routine returns an error.

resets the backoff if the routine returned successfully. disables the backoff if bo = nil

func WithExitCb

func WithExitCb(cb func(err error)) Option

WithExitCb adds a callback after a routine exits.

func WithExitLogger

func WithExitLogger(le *logrus.Entry) Option

WithExitLogger adds a exited callback which logs information about the exit.

func WithRetry

func WithRetry(boConf *backoff.Backoff) Option

WithRetry configures a backoff configuration to use when the routine returns an error.

resets the backoff if the routine returned successfully. disables the backoff if config is nil

type Routine

type Routine func(ctx context.Context) error

Routine is a function called as a goroutine. If nil is returned, exits cleanly permanently. If an error is returned, can be restarted later.

type RoutineContainer

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

RoutineContainer contains a Routine.

func NewRoutineContainer

func NewRoutineContainer(opts ...Option) *RoutineContainer

NewRoutineContainer constructs a new RoutineContainer. Note: routines won't start until SetContext is called.

func NewRoutineContainerWithLogger

func NewRoutineContainerWithLogger(le *logrus.Entry, opts ...Option) *RoutineContainer

NewRoutineContainerWithLogger constructs a new RoutineContainer instance. Logs when a controller exits without being canceled.

Note: routines won't start until SetContext is called.

func (*RoutineContainer) ClearContext

func (k *RoutineContainer) ClearContext() bool

ClearContext clears the context and shuts down all routines.

Returns if the routine was stopped or restarted.

func (*RoutineContainer) RestartRoutine

func (k *RoutineContainer) RestartRoutine() bool

RestartRoutine restarts the existing routine (if set). Returns if the routine was restarted. Returns false if the context is currently nil or the routine is unset.

func (*RoutineContainer) SetContext

func (k *RoutineContainer) SetContext(ctx context.Context, restart bool) bool

SetContext updates the root context.

nil context is valid and will shutdown the routines. if restart is true, errored routines will also restart.

Returns if the routine was stopped or restarted.

func (*RoutineContainer) SetRoutine

func (k *RoutineContainer) SetRoutine(routine Routine) (waitReturn <-chan struct{}, reset bool)

SetRoutine sets the routine to execute, resetting the existing, if set. If the specified routine is nil, shuts down the current routine. Returns if the current routine was stopped or overwritten. Returns a channel which will be closed when the previous routine exits. The waitReturn channel will be nil if there was no previous routine (reset=false).

func (*RoutineContainer) WaitExited

func (k *RoutineContainer) WaitExited(ctx context.Context, returnIfNotRunning bool, errCh <-chan error) error

WaitExited waits for the routine to exit and returns the error if any. Note: Will NOT return after the routine is restarted normally. If returnIfNotRunning is set, returns nil if no routine is running. If returnIfNotRunning is not set, waits until a routine has started & exited. errCh is an optional error channel (can be nil)

type StateRoutine

type StateRoutine[T comparable] func(ctx context.Context, st T) error

StateRoutine is a function called as a goroutine with a state parameter. If the state changes, ctx will be canceled and the function restarted. If nil is returned, exits cleanly permanently. If an error is returned, can still be restarted later.

type StateRoutineContainer

type StateRoutineContainer[T comparable] struct {
	// contains filtered or unexported fields
}

StateRoutineContainer contains a Routine which is restarted when the input State changes.

func NewStateRoutineContainer

func NewStateRoutineContainer[T comparable](compare func(t1, t2 T) bool, opts ...Option) *StateRoutineContainer[T]

NewStateRoutineContainer constructs a new StateRoutineContainer.

Note: routines won't start until SetContext and SetState is called. If the state is equivalent to an empty T (nil if a pointer) the routine is stopped. compare must compare if the two states are equivalent. if compare is nil restarts the routine every time SetState is called.

func NewStateRoutineContainerVT

func NewStateRoutineContainerVT[T proto.EqualVT[T]](opts ...Option) *StateRoutineContainer[T]

NewStateRoutineContainer constructs a new StateRoutineContainer with protobuf comparison.

Note: routines won't start until SetContext and SetState is called. If the state is equivalent to an empty T (nil if a pointer) the routine is stopped. compare must compare if the two states are equivalent. if compare is nil restarts the routine every time SetState is called.

func NewStateRoutineContainerWithLogger

func NewStateRoutineContainerWithLogger[T comparable](compare func(t1, t2 T) bool, le *logrus.Entry, opts ...Option) *StateRoutineContainer[T]

NewRoutineContainerWithLogger constructs a new RoutineContainer instance. Logs when a controller exits without being canceled.

Note: routines won't start until SetContext is called.

func NewStateRoutineContainerWithLoggerVT

func NewStateRoutineContainerWithLoggerVT[T proto.EqualVT[T]](le *logrus.Entry, opts ...Option) *StateRoutineContainer[T]

NewRoutineContainerWithLogger constructs a new RoutineContainer instance. Logs when a controller exits without being canceled.

Note: routines won't start until SetContext is called.

func (*StateRoutineContainer[T]) ClearContext

func (s *StateRoutineContainer[T]) ClearContext() bool

ClearContext clears the context and shuts down all routines.

Returns if the routine was stopped or restarted.

func (*StateRoutineContainer[T]) GetState

func (s *StateRoutineContainer[T]) GetState() T

GetState returns the immediate state in the StateRoutineContainer.

func (*StateRoutineContainer[T]) RestartRoutine

func (s *StateRoutineContainer[T]) RestartRoutine() bool

RestartRoutine restarts the existing routine (if set). Returns if the routine was restarted. Returns false if the context is currently nil or the routine is unset.

func (*StateRoutineContainer[T]) SetContext

func (s *StateRoutineContainer[T]) SetContext(ctx context.Context, restart bool) bool

SetContext updates the root context.

nil context is valid and will shutdown the routines. if restart is true, errored routines will also restart.

Returns if the routine was stopped or restarted.

func (*StateRoutineContainer[T]) SetState

func (s *StateRoutineContainer[T]) SetState(state T) (waitReturn <-chan struct{}, changed, reset, running bool)

SetState sets the state in the StateRoutineContainer.

Returns if the state changed and if the routine is running. If reset=true the existing routine was canceled or restarted.

func (*StateRoutineContainer[T]) SetStateRoutine

func (s *StateRoutineContainer[T]) SetStateRoutine(routine StateRoutine[T]) (waitReturn <-chan struct{}, reset, running bool)

SetStateRoutine sets the routine to execute, resetting the existing, if set. If the specified routine is nil, shuts down the current routine. Returns if the current routine was stopped or overwritten. Returns a channel which will be closed when the previous routine exits. The waitReturn channel will be nil if there was no previous routine (reset=false). If SetContext has not been called or SetState is empty, returns false for running. Note: does not check if routine is equal to the current routine func (cannot compare generic funcs).

func (*StateRoutineContainer[T]) SwapValue

func (s *StateRoutineContainer[T]) SwapValue(cb func(val T) T) (nextState T, waitReturn <-chan struct{}, changed, reset, running bool)

SwapState locks the container, calls the callback, and stores the returned value.

Returns the updated value and if the state changed. If reset=true returns a channel which closes when the previous instance has exited.

func (*StateRoutineContainer[T]) WaitExited

func (s *StateRoutineContainer[T]) WaitExited(ctx context.Context, returnIfNotRunning bool, errCh <-chan error) error

WaitExited waits for the routine to exit and returns the error if any. Note: Will NOT return after the routine is restarted normally. If returnIfNotRunning is set, returns nil if no routine is running. errCh is an optional error channel (can be nil)