source

Documentation

Overview

Package source provides high-level constructs to easily build plugins with event sourcing capability.

Index

• func Register(p Plugin)
• func WithInstanceBatchSize(size uint32) func(*builtinInstance)
• func WithInstanceClose(close func()) func(*builtinInstance)
• func WithInstanceContext(ctx context.Context) func(*builtinInstance)
• func WithInstanceEventSize(size uint32) func(*builtinInstance)
• func WithInstanceProgress(progress func() (float64, string)) func(*builtinInstance)
• func WithInstanceTimeout(timeout time.Duration) func(*builtinInstance)
• type BaseInstance
• type Instance
  • func NewPullInstance(pull PullFunc, options ...func(*builtinInstance)) (Instance, error)
  • func NewPushInstance(evtC <-chan PushEvent, options ...func(*builtinInstance)) (Instance, error)
• type Plugin
• type PullFunc
• type PushEvent

Functions

func Register

func Register(p Plugin)

Register registers the event sourcing capability in the framework for the given Plugin.

This function should be called from the provided plugins.FactoryFunc implementation. See the parent package for more detail. This function is idempotent.

func WithInstanceBatchSize

func WithInstanceBatchSize(size uint32) func(*builtinInstance)

WithInstanceBatchSize sets a custom size for the pre-allocated event batch used by NextBatch()

func WithInstanceClose

func WithInstanceClose(close func()) func(*builtinInstance)

WithInstanceClose sets a custom closing callback in the opened event source. The passed-in function is invoked when the event source gets closed.

func WithInstanceContext

func WithInstanceContext(ctx context.Context) func(*builtinInstance)

WithInstanceContext sets a custom context in the opened event source. If the context is cancelled, the event source is closed and sdk.ErrEOF is returned by the current invocation of NextBatch() and by any subsequent invocation.

func WithInstanceEventSize

func WithInstanceEventSize(size uint32) func(*builtinInstance)

WithInstanceEventSize sets a custom maximum size for each event returned by NextBatch()

func WithInstanceProgress

func WithInstanceProgress(progress func() (float64, string)) func(*builtinInstance)

WithInstanceProgress sets a custom callback for the framework to request a the progress state of the opened event stream

func WithInstanceTimeout

func WithInstanceTimeout(timeout time.Duration) func(*builtinInstance)

WithInstanceTimeout sets a custom timeout in the opened event source. When the timeout is reached, the current invocation of NextBatch() returns sdk.ErrTimeout.

Types

type BaseInstance

type BaseInstance struct {
	plugins.BaseEvents
	plugins.BaseProgress
}

BaseInstance is a base implementation of the Instance interface. Developer-defined Instance implementations should be composed with BaseInstance to have out-of-the-box compliance with all the required interfaces.

type Instance

type Instance interface {
	// (optional) sdk.Closer
	// (optional) sdk.Progresser
	sdk.Events
	sdk.NextBatcher
	sdk.ProgressBuffer
}

Instance is an interface representing a source capture session instance returned by a call to Open of a plugin with event sourcing capability.

Implementations of this interface must implement sdk.NextBatcher, and can optionally implement sdk.Closer and sdk.Progresser. If sdk.Closer is implemented, the Close method will be called while closing the source capture session.

func NewPullInstance

func NewPullInstance(pull PullFunc, options ...func(*builtinInstance)) (Instance, error)

NewPullInstance opens a new event source and starts a capture session, filling the event batches with a pull model.

The PullFunc required argument is a function that creates a new event and returns a non-nil error in case of success. The returned source.Instance provides a pre-built implementation of NextBatch() that correctly handles termination and timeouts. This should be used by developers to open an event source without defining a new type and by using a functional design.

The pull function is invoked sequentially and is blocking for the event source, meaning that it must not be a suspensive function. This implies avoiding suspending an execution through a select or through synchronization primitives.

Users can pass option parameters to influence the behavior of the opened event source, such as passing a context or setting a custom timeout duration.

The context passed-in to the pull function is cancelled automatically when the framework invokes Close() on the event source, or when the user-configured context is cancelled.

func NewPushInstance

func NewPushInstance(evtC <-chan PushEvent, options ...func(*builtinInstance)) (Instance, error)

NewPushInstance opens a new event source and starts a capture session, filling the event batches with a push model.

In this model, events are produced through a channel in the form of source.PushEvent messages. This is suitable for cases in which event production is suspensive, meaning that the time elapsed waiting for a new event to be produced is not deterministic or has no guaranteed limit.

Users can pass option parameters to influence the behavior of the opened event source, such as passing a context or setting a custom timeout duration.

The opened event source can be manually closed by cancelling the optional passed-in context, by closing the event cannel, or by sending source.PushEvent containing a non-nil Err.

type Plugin

type Plugin interface {
	plugins.Plugin
	sdk.StringerBuffer
	sdk.OpenParamsBuffer

	//
	// Open opens the source and starts a capture (e.g. stream of events).
	//
	// The argument string represents the user-defined parameters and
	// can be used to customize how the source is opened.
	// The return value is an Instance representing the source capture session.
	// There can be multiple instances of the same source open.
	// A successfull call to Open returns a nil error.
	//
	// The sdk.EventWriters event buffer, that is reused during each cycle
	// of new event creation, is initialized in automatic after the execution
	// of Open with the SetEvents method of the Instance interface.
	// Developers may override the default sdk.EventWriters by setting it
	// on the returned Instance with SetEvents, before returning from Open.
	// This can help specifying the data event size, the size of each
	// event batch, or just to use an implementation of the
	// sdk.EventWriters interface different from the SDK default one.
	Open(params string) (Instance, error)
}

Plugin is an interface representing a plugin with event sourcing capability

type PullFunc

type PullFunc func(context.Context, sdk.EventWriter) error

PullFunc produces a new event and returns a non-nil error in case of failure.

The event data is produced through the sdk.EventWriter interface. The context argument can be used to check for termination signals, which happen when the framework closes the event source or when the optional context passed-in by the user gets cancelled.

type PushEvent

type PushEvent struct {
	Err       error
	Data      []byte
	Timestamp time.Time
}

PushEvent represents an event produced from an event source with the push model.

If the event source produced the event successfully, then Data must be non-nil and Err must be ni. If the event source encountered a failure, Data must be nil and Err should contain an error describing the failure.

Timestamp can be optionally set to indicate a specific timestamp for the produced event.