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 AddCompressFunc(name string, fn CompressFunc) struct{}
• func AddDecompressFunc(name string, fn DecompressFunc) struct{}
• func CommonRetryBackOffCtorFromParsed(pConf *service.ParsedConfig) (ctor func() backoff.BackOff, err error)
• func CommonRetryBackOffFields(defaultMaxRetries int, defaultInitInterval string, defaultMaxInterval string, ...) []*service.ConfigField
• func RetryOutputIndefinitely(mgr bundle.NewManagement, wrapped output.Streamed) (output.Streamed, error)
• func SwitchReorderFromGroup(group *message.SortGroup, parts []*message.Part)
• type Branch
  • func (b *Branch) Close(ctx context.Context) error
  • func (b *Branch) ProcessBatch(ctx context.Context, batch message.Batch) ([]message.Batch, error)
• type CacheWriter
  • func NewCacheWriter(conf output.CacheConfig, mgr bundle.NewManagement, log log.Modular) (*CacheWriter, error)
  • func (c *CacheWriter) Close(context.Context) error
  • func (c *CacheWriter) Connect(ctx context.Context) error
  • func (c *CacheWriter) WriteBatch(ctx context.Context, msg message.Batch) (err error)
• type CompressFunc
• type DecompressFunc
• type SyncResponseWriter
  • func (s SyncResponseWriter) Close(context.Context) error
  • func (s SyncResponseWriter) Connect(ctx context.Context) error
  • func (s SyncResponseWriter) WriteBatch(ctx context.Context, msg message.Batch) error
• type Workflow
  • func NewWorkflow(conf processor.WorkflowConfig, mgr bundle.NewManagement) (*Workflow, error)
  • func (w *Workflow) Close(ctx context.Context) error
  • func (w *Workflow) Flow() [][]string
  • func (w *Workflow) ProcessBatch(ctx context.Context, msg message.Batch) ([]message.Batch, 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 = errors.New("attempting to create broker input type with no inputs")

ErrBrokerNoInputs is returned when creating a broker with zero inputs.

var ErrBrokerNoOutputs = errors.New("attempting to create broker output type with no outputs")

ErrBrokerNoOutputs is returned when creating a Broker type with zero outputs.

Functions

func AddCompressFunc

func AddCompressFunc(name string, fn CompressFunc) struct{}

AddCompressFunc adds a compression algorithm to components. The return struct serves no purpose other than allowing you to call it within the global context as an assignment.

func AddDecompressFunc

func AddDecompressFunc(name string, fn DecompressFunc) struct{}

AddDecompressFunc adds a decompression algorithm to components. The return struct serves no purpose other than allowing you to call it within the global context as an assignment.

func CommonRetryBackOffCtorFromParsed

func CommonRetryBackOffCtorFromParsed(pConf *service.ParsedConfig) (ctor func() backoff.BackOff, err error)

func CommonRetryBackOffFields

func CommonRetryBackOffFields(
	defaultMaxRetries int,
	defaultInitInterval string,
	defaultMaxInterval string,
	defaultMaxElapsed string,
) []*service.ConfigField

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) Close

func (b *Branch) Close(ctx context.Context) error

Close blocks until the processor has closed down or the context is cancelled.

func (*Branch) ProcessBatch

func (b *Branch) ProcessBatch(ctx context.Context, batch message.Batch) ([]message.Batch, error)

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

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) Close

func (c *CacheWriter) Close(context.Context) error

Close does nothing.

func (*CacheWriter) Connect

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

Connect does nothing.

func (*CacheWriter) WriteBatch

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

WriteBatch attempts to store a message within a cache.

type CompressFunc

type CompressFunc func(level int, bytes []byte) ([]byte, error)

CompressFunc represents a compression algorithm and fully encapsulates it.

type DecompressFunc

type DecompressFunc func(bytes []byte) ([]byte, error)

DecompressFunc represents a decompression algorithm and fully encapsulates it.

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) Close

func (s SyncResponseWriter) Close(context.Context) error

Close is a noop.

func (SyncResponseWriter) Connect

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

Connect is a noop.

func (SyncResponseWriter) WriteBatch

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

WriteBatch 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) Close

func (w *Workflow) Close(ctx context.Context) error

Close 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) ProcessBatch

func (w *Workflow) ProcessBatch(ctx context.Context, msg message.Batch) ([]message.Batch, error)

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