pure

Documentation

Overview

Package pure contains all component implementations that are pure, in that they do not interact with external systems. This includes all base component types such as brokers and is likely necessary as a base for all builds.

Index

• Variables
• func RetryOutputIndefinitely(mgr bundle.NewManagement, wrapped output.Streamed) (output.Streamed, error)
• func SwitchReorderFromGroup(group *message.SortGroup, parts []*message.Part)
• type Branch
  • func (b *Branch) CloseAsync()
  • func (b *Branch) ProcessMessage(msg *message.Batch) ([]*message.Batch, error)
  • func (b *Branch) WaitForClose(timeout time.Duration) error
• type CacheWriter
  • func NewCacheWriter(conf output.CacheConfig, mgr bundle.NewManagement, log log.Modular) (*CacheWriter, error)
  • func (c *CacheWriter) CloseAsync()
  • func (c *CacheWriter) ConnectWithContext(ctx context.Context) error
  • func (c *CacheWriter) WaitForClose(time.Duration) error
  • func (c *CacheWriter) WriteWithContext(ctx context.Context, msg *message.Batch) error
• type SyncResponseWriter
  • func (s SyncResponseWriter) CloseAsync()
  • func (s SyncResponseWriter) ConnectWithContext(ctx context.Context) error
  • func (s SyncResponseWriter) WaitForClose(time.Duration) error
  • func (s SyncResponseWriter) WriteWithContext(ctx context.Context, msg *message.Batch) error
• type Workflow
  • func NewWorkflow(conf processor.WorkflowConfig, mgr bundle.NewManagement) (*Workflow, error)
  • func (w *Workflow) CloseAsync()
  • func (w *Workflow) Flow() [][]string
  • func (w *Workflow) ProcessMessage(msg *message.Batch) ([]*message.Batch, error)
  • func (w *Workflow) WaitForClose(timeout time.Duration) error

Variables

var (
	// ErrSwitchNoConditionMet is returned when a message does not match any
	// output conditions.
	ErrSwitchNoConditionMet = errors.New("no switch output conditions were met by message")
	// ErrSwitchNoCasesMatched is returned when a message does not match any
	// output cases.
	ErrSwitchNoCasesMatched = errors.New("no switch cases were matched by message")
	// ErrSwitchNoOutputs is returned when creating a switchOutput type with less than
	// 2 outputs.
	ErrSwitchNoOutputs = errors.New("attempting to create switch with fewer than 2 cases")
)

var (
	// ErrBrokerNoInputs is returned when creating a broker with zero inputs.
	ErrBrokerNoInputs = errors.New("attempting to create broker input type with no inputs")
)

var (
	// ErrBrokerNoOutputs is returned when creating a Broker type with zero
	// outputs.
	ErrBrokerNoOutputs = errors.New("attempting to create broker output type with no outputs")
)

Functions

func RetryOutputIndefinitely

func RetryOutputIndefinitely(mgr bundle.NewManagement, wrapped output.Streamed) (output.Streamed, error)

RetryOutputIndefinitely returns a wrapped variant of the provided output where send errors downstream are automatically caught and retried rather than propagated upstream as nacks.

func SwitchReorderFromGroup

func SwitchReorderFromGroup(group *message.SortGroup, parts []*message.Part)

SwitchReorderFromGroup takes a message sort group and rearranges a slice of message parts so that they match up from their origins.

Types

type Branch

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

Branch contains conditions and maps for transforming a batch of messages into a subset of request messages, and mapping results from those requests back into the original message batch.

func (*Branch) CloseAsync

func (b *Branch) CloseAsync()

CloseAsync shuts down the processor and stops processing requests.

func (*Branch) ProcessMessage

func (b *Branch) ProcessMessage(msg *message.Batch) ([]*message.Batch, error)

ProcessMessage applies the processor to a message, either creating >0 resulting messages or a response to be sent back to the message source.

func (*Branch) WaitForClose

func (b *Branch) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the processor has closed down.

type CacheWriter

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

CacheWriter implements an output writer for caches.

func NewCacheWriter

func NewCacheWriter(conf output.CacheConfig, mgr bundle.NewManagement, log log.Modular) (*CacheWriter, error)

NewCacheWriter creates a writer for cache the output plugin.

func (*CacheWriter) CloseAsync

func (c *CacheWriter) CloseAsync()

CloseAsync does nothing.

func (*CacheWriter) ConnectWithContext

func (c *CacheWriter) ConnectWithContext(ctx context.Context) error

ConnectWithContext does nothing.

func (*CacheWriter) WaitForClose

func (c *CacheWriter) WaitForClose(time.Duration) error

WaitForClose does nothing.

func (*CacheWriter) WriteWithContext

func (c *CacheWriter) WriteWithContext(ctx context.Context, msg *message.Batch) error

WriteWithContext attempts to store a message within a cache.

type SyncResponseWriter

type SyncResponseWriter struct{}

SyncResponseWriter is a writer implementation that adds messages to a ResultStore located in the context of the first message part of each batch. This is essentially a mechanism that returns the result of a pipeline directly back to the origin of the message.

func (SyncResponseWriter) CloseAsync

func (s SyncResponseWriter) CloseAsync()

CloseAsync is a noop.

func (SyncResponseWriter) ConnectWithContext

func (s SyncResponseWriter) ConnectWithContext(ctx context.Context) error

ConnectWithContext is a noop.

func (SyncResponseWriter) WaitForClose

func (s SyncResponseWriter) WaitForClose(time.Duration) error

WaitForClose is a noop.

func (SyncResponseWriter) WriteWithContext

func (s SyncResponseWriter) WriteWithContext(ctx context.Context, msg *message.Batch) error

WriteWithContext writes a message batch to a ResultStore located in the first message of the batch.

type Workflow

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

Workflow is a processor that applies a list of child processors to a new payload mapped from the original, and after processing attempts to overlay the results back onto the original payloads according to more mappings.

func NewWorkflow

func NewWorkflow(conf processor.WorkflowConfig, mgr bundle.NewManagement) (*Workflow, error)

NewWorkflow instanciates a new workflow processor.

func (*Workflow) CloseAsync

func (w *Workflow) CloseAsync()

CloseAsync shuts down the processor and stops processing requests.

func (*Workflow) Flow

func (w *Workflow) Flow() [][]string

Flow returns the calculated workflow as a 2D slice.

func (*Workflow) ProcessMessage

func (w *Workflow) ProcessMessage(msg *message.Batch) ([]*message.Batch, error)

ProcessMessage applies workflow stages to each part of a message type.

func (*Workflow) WaitForClose

func (w *Workflow) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the processor has closed down.