bus

Documentation

Index

• func ExecCollectValues(ctx context.Context, bus Bus, dir directive.Directive, ...) ([]directive.Value, directive.Reference, error)
• func ExecOneOff(ctx context.Context, bus Bus, dir directive.Directive, returnIfIdle bool, ...) (directive.AttachedValue, directive.Reference, error)
• func ExecOneOffWatchCh(b Bus, dir directive.Directive) (<-chan directive.AttachedValue, directive.Reference, error)
• func ExecWaitValue[T any](ctx context.Context, b Bus, dir directive.Directive, ...) (T, directive.Reference, error)
• func NewCallbackHandler(valCb func(directive.AttachedValue), removedCb func(directive.AttachedValue), ...) directive.ReferenceHandler
• func NewPassThruHandler(passthru directive.ResolverHandler, disposeCb func()) directive.ReferenceHandler
• type Bus
• type BusController
  • func NewBusController[T config.Config](le *logrus.Entry, b Bus, conf T, controllerID string, ...) *BusController[T]
  • func (c *BusController[T]) Close() error
  • func (c *BusController[T]) Execute(ctx context.Context) error
  • func (c *BusController[T]) GetBus() Bus
  • func (c *BusController[T]) GetConfig() T
  • func (c *BusController[T]) GetControllerInfo() *controller.Info
  • func (c *BusController[T]) GetLogger() *logrus.Entry
  • func (c *BusController[T]) HandleDirective(ctx context.Context, di directive.Instance) (directive.Resolver, error)
• type BusFactory
  • func NewBusControllerFactory[T config.Config, C controller.Controller](b Bus, configID string, controllerID string, version semver.Version, ...) *BusFactory[T, C]
  • func NewFactory[T config.Config, C controller.Controller](b Bus, configID string, version semver.Version, newConfig func() T, ...) *BusFactory[T, C]
  • func (f *BusFactory[T, C]) Construct(cc config.Config, opts controller.ConstructOpts) (controller.Controller, error)
  • func (f *BusFactory[T, C]) ConstructConfig() config.Config
  • func (f *BusFactory[T, C]) GetConfigID() string
  • func (f *BusFactory[T, C]) GetVersion() semver.Version
• type CallbackHandler
  • func (h *CallbackHandler) HandleInstanceDisposed(directive.Instance)
  • func (h *CallbackHandler) HandleValueAdded(_ directive.Instance, v directive.AttachedValue)
  • func (h *CallbackHandler) HandleValueRemoved(_ directive.Instance, v directive.AttachedValue)
• type PassThruHandler
  • func (h *PassThruHandler) HandleInstanceDisposed(di directive.Instance)
  • func (h *PassThruHandler) HandleValueAdded(_ directive.Instance, v directive.AttachedValue)
  • func (h *PassThruHandler) HandleValueRemoved(_ directive.Instance, v directive.AttachedValue)

Functions

func ExecCollectValues

func ExecCollectValues(
	ctx context.Context,
	bus Bus,
	dir directive.Directive,
	valDisposeCallback func(),
) ([]directive.Value, directive.Reference, error)

ExecCollectValues collects one or more values while ctx is active and directive is not idle.

Returns err if the directive is idle and a resolver returned an error. Does not return an error if ctx was canceled. Does not release ref if err == nil.

func ExecOneOff

func ExecOneOff(
	ctx context.Context,
	bus Bus,
	dir directive.Directive,
	returnIfIdle bool,
	valDisposeCallback func(),
) (directive.AttachedValue, directive.Reference, error)

ExecOneOff executes a one-off directive. If returnIfIdle is set, returns nil, nil, nil if idle.

func ExecOneOffWatchCh

func ExecOneOffWatchCh(
	b Bus,
	dir directive.Directive,
) (<-chan directive.AttachedValue, directive.Reference, error)

ExecOneOffWatchCh executes a one-off directive and watches for changes. Returns a channel with size 1 which will hold the latest value. Continues to watch for changes until the directive is released.

func ExecWaitValue

func ExecWaitValue[T any](
	ctx context.Context,
	b Bus,
	dir directive.Directive,
	checkCb func(val T) (bool, error),
) (T, directive.Reference, error)

ExecWaitValue executes a directive and waits for a value matching the cb. If cb returns true, nil, returns the value. If checkCb is nil, returns first value.

func NewCallbackHandler

func NewCallbackHandler(
	valCb func(directive.AttachedValue),
	removedCb func(directive.AttachedValue),
	disposeCb func(),
) directive.ReferenceHandler

NewCallbackHandler wraps callback functions into a handler object.

func NewPassThruHandler

func NewPassThruHandler(
	passthru directive.ResolverHandler,
	disposeCb func(),
) directive.ReferenceHandler

NewPassThruHandler builds a new pass-through handler.

Types

type Bus

type Bus interface {
	// Controller is the directive controller.
	directive.Controller

	// GetControllers returns a list of all currently active controllers.
	GetControllers() []controller.Controller

	// ExecuteController adds a controller to the bus and calls Execute().
	// Any fatal error in the controller is returned.
	// The controller will receive directive callbacks.
	// If the controller returns nil, call RemoveController to remove the controller.
	ExecuteController(context.Context, controller.Controller) error
	// RemoveController removes the controller from the bus.
	// The controller will no longer receive callbacks.
	// Note: this might not cancel the Execute() context automatically.
	RemoveController(controller.Controller)
}

Bus manages running controllers. It has an attached directive controller, which is used to build declarative state requests between controllers.

type BusController

type BusController[T config.Config] struct {
	// contains filtered or unexported fields
}

BusController implements a base Controller with an attached Bus for directives. It has stubs for the functions implementing all required interfaces. Usually you should extend this type, overriding functions as needed.

func NewBusController

func NewBusController[T config.Config](
	le *logrus.Entry,
	b Bus,
	conf T,
	controllerID string,
	controllerVersion semver.Version,
	controllerDescrip string,
) *BusController[T]

NewBusController constructs a new Controller with details.

func (*BusController[T]) Close

func (c *BusController[T]) Close() error

Close releases any resources used by the controller.

func (*BusController[T]) Execute

func (c *BusController[T]) Execute(ctx context.Context) error

Execute executes the given controller.

func (*BusController[T]) GetBus

func (c *BusController[T]) GetBus() Bus

GetBus returns the bus.

func (*BusController[T]) GetConfig

func (c *BusController[T]) GetConfig() T

GetConfig returns the config. Note: don't modify this object!

func (*BusController[T]) GetControllerInfo

func (c *BusController[T]) GetControllerInfo() *controller.Info

GetControllerInfo returns information about the controller.

func (*BusController[T]) GetLogger

func (c *BusController[T]) GetLogger() *logrus.Entry

GetLogger returns the root logger.

func (*BusController[T]) HandleDirective

func (c *BusController[T]) HandleDirective(ctx context.Context, di directive.Instance) (directive.Resolver, error)

HandleDirective asks if the handler can resolve the directive.

type BusFactory

type BusFactory[T config.Config, C controller.Controller] struct {
	// contains filtered or unexported fields
}

BusFactory implements Factory with an attached Bus for directives.

func NewBusControllerFactory

func NewBusControllerFactory[T config.Config, C controller.Controller](
	b Bus,
	configID string,
	controllerID string,
	version semver.Version,
	controllerDescription string,
	newConfig func() T,
	newController func(base *BusController[T]) (C, error),
) *BusFactory[T, C]

NewBusControllerFactory builds a Factory for a type wrapping BusController.

func NewFactory

func NewFactory[T config.Config, C controller.Controller](
	b Bus,
	configID string,
	version semver.Version,
	newConfig func() T,
	newController func(le *logrus.Entry, b Bus, conf T) (C, error),
) *BusFactory[T, C]

NewFactory constructs a new Factory with details.

func (*BusFactory[T, C]) Construct

func (f *BusFactory[T, C]) Construct(cc config.Config, opts controller.ConstructOpts) (controller.Controller, error)

Construct constructs the associated controller given configuration.

func (*BusFactory[T, C]) ConstructConfig

func (f *BusFactory[T, C]) ConstructConfig() config.Config

ConstructConfig constructs an instance of the controller configuration.

func (*BusFactory[T, C]) GetConfigID

func (f *BusFactory[T, C]) GetConfigID() string

GetConfigID returns the unique config ID for the controller.

func (*BusFactory[T, C]) GetVersion

func (f *BusFactory[T, C]) GetVersion() semver.Version

GetVersion returns the version of this controller.

type CallbackHandler

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

CallbackHandler is a reference handler that uses function callbacks.

func (*CallbackHandler) HandleInstanceDisposed

func (h *CallbackHandler) HandleInstanceDisposed(directive.Instance)

HandleInstanceDisposed is called when a directive instance is disposed. This will occur if Close() is called on the directive instance.

func (*CallbackHandler) HandleValueAdded

func (h *CallbackHandler) HandleValueAdded(
	_ directive.Instance,
	v directive.AttachedValue,
)

HandleValueAdded is called when a value is added to the directive.

func (*CallbackHandler) HandleValueRemoved

func (h *CallbackHandler) HandleValueRemoved(
	_ directive.Instance,
	v directive.AttachedValue,
)

HandleValueRemoved is called when a value is removed from the directive.

type PassThruHandler

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

PassThruHandler is a reference handler that passes through to a resolver handler.

func (*PassThruHandler) HandleInstanceDisposed

func (h *PassThruHandler) HandleInstanceDisposed(di directive.Instance)

HandleInstanceDisposed is called when a directive instance is disposed. This will occur if Close() is called on the directive instance.

func (*PassThruHandler) HandleValueAdded

func (h *PassThruHandler) HandleValueAdded(
	_ directive.Instance,
	v directive.AttachedValue,
)

HandleValueAdded is called when a value is added to the directive.

func (*PassThruHandler) HandleValueRemoved

func (h *PassThruHandler) HandleValueRemoved(
	_ directive.Instance,
	v directive.AttachedValue,
)

HandleValueRemoved is called when a value is removed from the directive.