projection

Documentation

Overview

Package projection provides an adaptor that exposes Dogma projection message handlers as Verity event stream handlers.

Index

• Constants
• type Compactor
  • func (c *Compactor) Run(ctx context.Context) error
• type StreamAdaptor
  • func (a *StreamAdaptor) HandleEvent(ctx context.Context, o uint64, ev eventstream.Event) (err error)
  • func (a *StreamAdaptor) NextOffset(ctx context.Context, id configkit.Identity) (uint64, error)

Constants

const DefaultTimeout = 3 * time.Second

DefaultTimeout is the default timeout to use when applying an event.

Types

type Compactor

type Compactor struct {
	// Handler is the projection message handler to be compacted.
	Handler dogma.ProjectionMessageHandler

	// Interval is the interval at which the projection is compacted.
	Interval time.Duration

	// Timeout is the default timeout to use when compacting the projection.
	Timeout time.Duration

	// Semaphore is used to limit the number of projections being compacted
	// concurrently.
	Semaphore *semaphore.Weighted

	// Logger is the target for log messages produced about compaction.
	// If it is nil, logging.DefaultLogger is used.
	Logger logging.Logger
}

Compactor periodically compacts a projection.

func (*Compactor) Run

func (c *Compactor) Run(ctx context.Context) error

Run periodically compacts the projection until ctx is canceled or an error occurs.

type StreamAdaptor

type StreamAdaptor struct {
	// Identity is the handler's identity.
	Identity *envelopespec.Identity

	// Handler is the projection message handler that handles the events.
	Handler dogma.ProjectionMessageHandler

	// DefaultTimeout is the maximum time to allow for handling a single event
	// if the handler does not provide a timeout hint. If it is nil,
	// DefaultTimeout is used.
	DefaultTimeout time.Duration

	// Logger is the target for log messages produced within the handler.
	// If it is nil, logging.DefaultLogger is used.
	Logger logging.Logger
}

StreamAdaptor presents a dogma.ProjectionMessageHandler as an eventstream.Handler.

func (*StreamAdaptor) HandleEvent

func (a *StreamAdaptor) HandleEvent(
	ctx context.Context,
	o uint64,
	ev eventstream.Event,
) (err error)

HandleEvent handles a message consumed from the event stream.

o must be the offset that would be returned by NextOffset(). On success, the next call to NextOffset() will return ev.Offset + 1.

func (*StreamAdaptor) NextOffset

func (a *StreamAdaptor) NextOffset(
	ctx context.Context,
	id configkit.Identity,
) (uint64, error)

NextOffset returns the next offset to be consumed from the event stream.

id is the identity of the source application.