mgr

Documentation

Overview

Package mgr provides simple managing of flow control and logging.

Index

• Variables
• func RunModules(ctx context.Context, modules ...Module) error
• type EventCallback
• type EventCallbackFunc
• type EventMgr
  • func NewEventMgr[T any](eventName string, mgr *Manager) *EventMgr[T]
  • func (em *EventMgr[T]) AddCallback(callbackName string, callback EventCallbackFunc[T])
  • func (em *EventMgr[T]) Submit(event T)
  • func (em *EventMgr[T]) Subscribe(subscriberName string, chanSize int) *EventSubscription[T]
• type EventSubscription
  • func (es *EventSubscription[T]) Cancel()
  • func (es *EventSubscription[T]) Done() bool
  • func (es *EventSubscription[T]) Events() <-chan T
• type Group
  • func NewGroup(modules ...Module) *Group
  • func (g *Group) Done() <-chan struct{}
  • func (g *Group) IsDone() bool
  • func (g *Group) Start() error
  • func (g *Group) Stop() (ok bool)
• type Manager
  • func New(name string) *Manager
  • func NewWithContext(ctx context.Context, name string) *Manager
  • func (m *Manager) Cancel()
  • func (m *Manager) Ctx() context.Context
  • func (m *Manager) Debug(msg string, args ...any)
  • func (m *Manager) Do(name string, fn func(w *WorkerCtx) error) error
  • func (m *Manager) Done() <-chan struct{}
  • func (m *Manager) Error(msg string, args ...any)
  • func (m *Manager) Go(name string, fn func(w *WorkerCtx) error)
  • func (m *Manager) Info(msg string, args ...any)
  • func (m *Manager) IsDone() bool
  • func (m *Manager) Log(level slog.Level, msg string, args ...any)
  • func (m *Manager) LogAttrs(level slog.Level, msg string, attrs ...slog.Attr)
  • func (m *Manager) LogEnabled(level slog.Level) bool
  • func (m *Manager) Name() string
  • func (m *Manager) WaitForWorkers(max time.Duration) (done bool)
  • func (m *Manager) Warn(msg string, args ...any)
• type Module
• type WorkerCtx
  • func WorkerFromCtx(ctx context.Context) *WorkerCtx
  • func (w *WorkerCtx) AddToCtx(ctx context.Context) context.Context
  • func (w *WorkerCtx) Cancel()
  • func (w *WorkerCtx) Ctx() context.Context
  • func (w *WorkerCtx) Debug(msg string, args ...any)
  • func (w *WorkerCtx) Done() <-chan struct{}
  • func (w *WorkerCtx) Error(msg string, args ...any)
  • func (w *WorkerCtx) Info(msg string, args ...any)
  • func (w *WorkerCtx) IsDone() bool
  • func (w *WorkerCtx) Log(level slog.Level, msg string, args ...any)
  • func (w *WorkerCtx) LogAttrs(level slog.Level, msg string, attrs ...slog.Attr)
  • func (w *WorkerCtx) LogEnabled(level slog.Level) bool
  • func (w *WorkerCtx) Logger() *slog.Logger
  • func (w *WorkerCtx) Warn(msg string, args ...any)

Variables

var WorkerCtxContextKey = workerContextKey{}

WorkerCtxContextKey is the key used to add the WorkerCtx to a context.

Functions

func RunModules

func RunModules(ctx context.Context, modules ...Module) error

RunModules is a simple wrapper function to start modules and stop them again when the given context is canceled.

Types

type EventCallback

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

EventCallback is a registered callback to an event.

type EventCallbackFunc

type EventCallbackFunc[T any] func(*WorkerCtx, T) (cancel bool, err error)

EventCallbackFunc defines the event callback function.

type EventMgr

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

EventMgr is a simple event manager.

func NewEventMgr

func NewEventMgr[T any](eventName string, mgr *Manager) *EventMgr[T]

NewEventMgr returns a new event manager. It is easiest used as a public field on a struct, so that others can simply Subscribe() oder AddCallback().

func (*EventMgr[T]) AddCallback

func (em *EventMgr[T]) AddCallback(callbackName string, callback EventCallbackFunc[T])

AddCallback adds a callback to executed on events. The received events are shared among all subscribers and callbacks. Be sure to apply proper concurrency safeguards, if applicable.

func (*EventMgr[T]) Submit

func (em *EventMgr[T]) Submit(event T)

Submit submits a new event.

func (*EventMgr[T]) Subscribe

func (em *EventMgr[T]) Subscribe(subscriberName string, chanSize int) *EventSubscription[T]

Subscribe subscribes to events. The received events are shared among all subscribers and callbacks. Be sure to apply proper concurrency safeguards, if applicable.

type EventSubscription

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

EventSubscription is a subscription to an event.

func (*EventSubscription[T]) Cancel

func (es *EventSubscription[T]) Cancel()

Cancel cancels the subscription. The events channel is not closed, but will not receive new events.

func (*EventSubscription[T]) Done

func (es *EventSubscription[T]) Done() bool

Done returns whether the event subscription has been canceled.

func (*EventSubscription[T]) Events

func (es *EventSubscription[T]) Events() <-chan T

Events returns a read channel for the events. The received events are shared among all subscribers and callbacks. Be sure to apply proper concurrency safeguards, if applicable.

type Group

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

Group describes a group of modules.

func NewGroup

func NewGroup(modules ...Module) *Group

NewGroup returns a new group of modules.

func (*Group) Done

func (g *Group) Done() <-chan struct{}

Done returns the context Done channel.

func (*Group) IsDone

func (g *Group) IsDone() bool

IsDone checks whether the manager context is done.

func (*Group) Start

func (g *Group) Start() error

Start starts all modules in the group in the defined order. If a module fails to start, itself and all previous modules will be stopped in the reverse order.

func (*Group) Stop

func (g *Group) Stop() (ok bool)

Stop stops all modules in the group in the reverse order.

type Manager

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

Manager manages workers.

func New

func New(name string) *Manager

New returns a new manager.

func NewWithContext

func NewWithContext(ctx context.Context, name string) *Manager

NewWithContext returns a new manager that uses the given context.

func (*Manager) Cancel

func (m *Manager) Cancel()

Cancel cancels the worker context.

func (*Manager) Ctx

func (m *Manager) Ctx() context.Context

Ctx returns the worker context.

func (*Manager) Debug

func (m *Manager) Debug(msg string, args ...any)

Debug logs at LevelDebug. The manager context is automatically supplied.

func (*Manager) Do

func (m *Manager) Do(name string, fn func(w *WorkerCtx) error) error

Do directly executes the given function (as a "worker"). The worker context has - A separate context which is canceled when the functions returns. - Access to named structure logging. - Given function is re-run after failure (with backoff). - Panic catching. - Flow control helpers.

func (*Manager) Done

func (m *Manager) Done() <-chan struct{}

Done returns the context Done channel.

func (*Manager) Error

func (m *Manager) Error(msg string, args ...any)

Error logs at LevelError. The manager context is automatically supplied.

func (*Manager) Go

func (m *Manager) Go(name string, fn func(w *WorkerCtx) error)

Go starts the given function in a goroutine (as a "worker"). The worker context has - A separate context which is canceled when the functions returns. - Access to named structure logging. - Given function is re-run after failure (with backoff). - Panic catching. - Flow control helpers.

func (*Manager) Info

func (m *Manager) Info(msg string, args ...any)

Info logs at LevelInfo. The manager context is automatically supplied.

func (*Manager) IsDone

func (m *Manager) IsDone() bool

IsDone checks whether the manager context is done.

func (*Manager) Log

func (m *Manager) Log(level slog.Level, msg string, args ...any)

Log emits a log record with the current time and the given level and message. The manager context is automatically supplied.

func (*Manager) LogAttrs

func (m *Manager) LogAttrs(level slog.Level, msg string, attrs ...slog.Attr)

LogAttrs is a more efficient version of Log() that accepts only Attrs. The manager context is automatically supplied.

func (*Manager) LogEnabled

func (m *Manager) LogEnabled(level slog.Level) bool

LogEnabled reports whether the logger emits log records at the given level. The manager context is automatically supplied.

func (*Manager) Name

func (m *Manager) Name() string

Name returns the manager name.

func (*Manager) WaitForWorkers

func (m *Manager) WaitForWorkers(max time.Duration) (done bool)

WaitForWorkers waits for all workers of this manager to be done. The default maximum waiting time is one minute.

func (*Manager) Warn

func (m *Manager) Warn(msg string, args ...any)

Warn logs at LevelWarn. The manager context is automatically supplied.

type Module

type Module interface {
	Start(mgr *Manager) error
	Stop(mgr *Manager) error
}

Module is an manage-able instance of some component.

type WorkerCtx

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

WorkerCtx provides workers with the necessary environment for flow control and logging.

func WorkerFromCtx

func WorkerFromCtx(ctx context.Context) *WorkerCtx

WorkerFromCtx returns the WorkerCtx from the given context.

func (*WorkerCtx) AddToCtx

func (w *WorkerCtx) AddToCtx(ctx context.Context) context.Context

AddToCtx adds the WorkerCtx to the given context.

func (*WorkerCtx) Cancel

func (w *WorkerCtx) Cancel()

Cancel cancels the worker context. Is automatically called after the worker stops/returns, regardless of error.

func (*WorkerCtx) Ctx

func (w *WorkerCtx) Ctx() context.Context

Ctx returns the worker context. Is automatically canceled after the worker stops/returns, regardless of error.

func (*WorkerCtx) Debug

func (w *WorkerCtx) Debug(msg string, args ...any)

Debug logs at LevelDebug. The worker context is automatically supplied.

func (*WorkerCtx) Done

func (w *WorkerCtx) Done() <-chan struct{}

Done returns the context Done channel.

func (*WorkerCtx) Error

func (w *WorkerCtx) Error(msg string, args ...any)

Error logs at LevelError. The worker context is automatically supplied.

func (*WorkerCtx) Info

func (w *WorkerCtx) Info(msg string, args ...any)

Info logs at LevelInfo. The worker context is automatically supplied.

func (*WorkerCtx) IsDone

func (w *WorkerCtx) IsDone() bool

IsDone checks whether the worker context is done.

func (*WorkerCtx) Log

func (w *WorkerCtx) Log(level slog.Level, msg string, args ...any)

Log emits a log record with the current time and the given level and message. The worker context is automatically supplied.

func (*WorkerCtx) LogAttrs

func (w *WorkerCtx) LogAttrs(level slog.Level, msg string, attrs ...slog.Attr)

LogAttrs is a more efficient version of Log() that accepts only Attrs. The worker context is automatically supplied.

func (*WorkerCtx) LogEnabled

func (w *WorkerCtx) LogEnabled(level slog.Level) bool

LogEnabled reports whether the logger emits log records at the given level. The worker context is automatically supplied.

func (*WorkerCtx) Logger

func (w *WorkerCtx) Logger() *slog.Logger

Logger returns the logger used by the worker context.

func (*WorkerCtx) Warn

func (w *WorkerCtx) Warn(msg string, args ...any)

Warn logs at LevelWarn. The worker context is automatically supplied.