engine

Documentation

Overview

Package engine contains an in-memory Dogma engine.

Index

• Constants
• func Run(ctx context.Context, e *Engine, d time.Duration, opts ...OperationOption) error
• func RunTimeScaled(ctx context.Context, e *Engine, d time.Duration, f float64, t time.Time, ...) error
• type CommandExecutor
  • func (e CommandExecutor) ExecuteCommand(ctx context.Context, m dogma.Message, _ ...dogma.ExecuteCommandOption) error
• type Engine
  • func MustNew(app configkit.RichApplication, options ...Option) *Engine
  • func New(app configkit.RichApplication, options ...Option) (_ *Engine, err error)
  • func (e *Engine) Dispatch(ctx context.Context, m dogma.Message, options ...OperationOption) error
  • func (e *Engine) Reset()
  • func (e *Engine) Tick(ctx context.Context, options ...OperationOption) error
• type OperationOption
  • func EnableAggregates(enabled bool) OperationOption
  • func EnableHandler(name string, enabled bool) OperationOption
  • func EnableIntegrations(enabled bool) OperationOption
  • func EnableProcesses(enabled bool) OperationOption
  • func EnableProjections(enabled bool) OperationOption
  • func WithCurrentTime(t time.Time) OperationOption
  • func WithObserver(o fact.Observer) OperationOption
• type Option
  • func EnableProjectionCompactionDuringHandling(enabled bool) Option
  • func WithResetter(fn func()) Option

Constants

const DefaultTickInterval = 250 * time.Millisecond

DefaultTickInterval is the default interval at which Run() and RunTimeScaled() will perform an engine tick.

Functions

func Run

func Run(
	ctx context.Context,
	e *Engine,
	d time.Duration,
	opts ...OperationOption,
) error

Run repeatedly calls e.Tick() until ctx is canceled or an error occurs.

d is the duration between ticks. If it is 0, DefaultTickInterval is used.

func RunTimeScaled

func RunTimeScaled(
	ctx context.Context,
	e *Engine,
	d time.Duration,
	f float64,
	t time.Time,
	opts ...OperationOption,
) error

RunTimeScaled repeatedly calls e.Tick() until ctx is canceled or an error occurs.

d is the duration between ticks. If it is 0, DefaultTickInterval is used.

Each tick is performed using a WithCurrentTime() option that scales time by a factor of f. For example, if f is 2.0, the engine will see time progress by 2 seconds for every 1 second of real time.

t is the "epoch time", used as current time for the first tick. If t.IsZero() is true, the current time is used.

Types

type CommandExecutor

type CommandExecutor struct {
	// Engine is the engine that handles the recorded events.
	Engine *Engine

	// Options is a set of options used when dispatching the message to the
	// engine.
	Options []OperationOption
}

CommandExecutor adapts an Engine to the dogma.CommandExecutor interface.

func (CommandExecutor) ExecuteCommand

func (e CommandExecutor) ExecuteCommand(ctx context.Context, m dogma.Message, _ ...dogma.ExecuteCommandOption) error

ExecuteCommand enqueues a command for execution.

It panics if the command is not routed to any handlers.

type Engine

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

Engine is an in-memory Dogma engine that is used to execute tests.

func MustNew

func MustNew(app configkit.RichApplication, options ...Option) *Engine

MustNew returns a new engine that uses the given app configuration, or panics if unable to do so.

func New

func New(app configkit.RichApplication, options ...Option) (_ *Engine, err error)

New returns a new engine that uses the given app configuration.

func (*Engine) Dispatch

func (e *Engine) Dispatch(
	ctx context.Context,
	m dogma.Message,
	options ...OperationOption,
) error

Dispatch processes a message.

It is not an error to process a message that is not routed to any handlers.

It panics if the message is invalid.

func (*Engine) Reset

func (e *Engine) Reset()

Reset clears the engine's state, such as aggregate and process roots.

func (*Engine) Tick

func (e *Engine) Tick(
	ctx context.Context,
	options ...OperationOption,
) error

Tick performs one "tick" of the engine.

This allows external control of time-based features of the engine. now is the time that the engine should treat as the current time.

type OperationOption

type OperationOption interface {
	// contains filtered or unexported methods
}

OperationOption applies optional settings while dispatching a message or performing a tick.

func EnableAggregates

func EnableAggregates(enabled bool) OperationOption

EnableAggregates returns an operation option that enables or disables aggregate message handlers.

All handler types are enabled by default.

func EnableHandler

func EnableHandler(name string, enabled bool) OperationOption

EnableHandler returns an operation option that enables or disables a specific handler.

This option takes precedence over any EnableAggregates(), EnableProcesses(), EnableIntegrations() or EnableProjections() options.

func EnableIntegrations

func EnableIntegrations(enabled bool) OperationOption

EnableIntegrations returns an operation option that enables or disables integration message handlers.

All handler types are enabled by default.

func EnableProcesses

func EnableProcesses(enabled bool) OperationOption

EnableProcesses returns an operation option that enables or disables process message handlers.

All handler types are enabled by default.

func EnableProjections

func EnableProjections(enabled bool) OperationOption

EnableProjections returns an operation option that enables or disables projection message handlers.

All handler types are enabled by default.

func WithCurrentTime

func WithCurrentTime(t time.Time) OperationOption

WithCurrentTime returns an operation option that sets the engine's current time.

func WithObserver

func WithObserver(o fact.Observer) OperationOption

WithObserver returns an option that registers the given observer for the duration of the operation.

Multiple observers can be registered during a single operation.

type Option

type Option interface {
	// contains filtered or unexported methods
}

Option applies optional engine-wide settings.

func EnableProjectionCompactionDuringHandling

func EnableProjectionCompactionDuringHandling(enabled bool) Option

EnableProjectionCompactionDuringHandling returns an engine option that causes projection to be compacted in parallel with each event handled.

This option is intended to faciliate testing of compaction logic alongside projection building. It is likely not much use when using the engine outside of the test runner.

func WithResetter

func WithResetter(fn func()) Option

WithResetter returns an engine option that registers a reset hook with the engine.

fn is a function to be called whenever the engine is reset.