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 AddKnownCompressionAlgorithm(name string, a KnownCompressionAlgorithm) struct{}
• func CacheOutputSpec() *service.ConfigSpec
• func CompressionAlgsList() (v []string)
• func DecompressionAlgsList() (v []string)
• 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 *service.ParsedConfig, mgr bundle.NewManagement) (*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 CombinedReadCloser
  • func (c *CombinedReadCloser) Close() error
  • func (c *CombinedReadCloser) Read(b []byte) (int, error)
• type CombinedWriteCloser
  • func (c *CombinedWriteCloser) Close() error
  • func (c *CombinedWriteCloser) Write(b []byte) (int, error)
• type CompressFunc
• type CompressWriter
• type DecompressFunc
• type DecompressReader
• type KnownCompressionAlgorithm
• 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 *service.ParsedConfig, 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 AddKnownCompressionAlgorithm

func AddKnownCompressionAlgorithm(name string, a KnownCompressionAlgorithm) struct{}

func CacheOutputSpec

func CacheOutputSpec() *service.ConfigSpec

func CompressionAlgsList

func CompressionAlgsList() (v []string)

func DecompressionAlgsList

func DecompressionAlgsList() (v []string)

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
}

func NewCacheWriter

func NewCacheWriter(conf *service.ParsedConfig, mgr bundle.NewManagement) (*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 CombinedReadCloser

type CombinedReadCloser struct {
	Primary, Source io.Reader
}

The Primary is read from and closed second. The Source is closed first.

func (*CombinedReadCloser) Close

func (c *CombinedReadCloser) Close() error

func (*CombinedReadCloser) Read

func (c *CombinedReadCloser) Read(b []byte) (int, error)

type CombinedWriteCloser

type CombinedWriteCloser struct {
	Primary, Sink io.Writer
}

The Primary is written to and closed first. The Sink is closed second.

func (*CombinedWriteCloser) Close

func (c *CombinedWriteCloser) Close() error

func (*CombinedWriteCloser) Write

func (c *CombinedWriteCloser) Write(b []byte) (int, error)

type CompressFunc

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

type CompressWriter

type CompressWriter func(level int, w io.Writer) (io.Writer, error)

type DecompressFunc

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

type DecompressReader

type DecompressReader func(r io.Reader) (io.Reader, error)

type KnownCompressionAlgorithm

type KnownCompressionAlgorithm struct {
	CompressFunc     CompressFunc
	CompressWriter   CompressWriter
	DecompressFunc   DecompressFunc
	DecompressReader DecompressReader
}

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 *service.ParsedConfig, 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.