buffer

package
v4.4.4 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 19, 2022 License: MIT Imports: 15 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type AckFunc

type AckFunc func(context.Context, error) error

AckFunc is a function used to acknowledge receipt of a message batch from a buffer. The provided error indicates whether the message batch was successfully delivered. Returns an error if the acknowledge was not propagated.

type Config

type Config struct {
	Type   string      `json:"type" yaml:"type"`
	Plugin interface{} `json:"plugin,omitempty" yaml:"plugin,omitempty"`
}

Config is the all encompassing configuration struct for all buffer types. Deprecated: Do not add new components here. Instead, use the public plugin APIs. Examples can be found in: ./internal/impl

func NewConfig

func NewConfig() Config

NewConfig returns a configuration struct fully populated with default values.

func (*Config) UnmarshalYAML

func (conf *Config) UnmarshalYAML(value *yaml.Node) error

UnmarshalYAML ensures that when parsing configs that are in a map or slice the default values are still applied.

type ReaderWriter

type ReaderWriter interface {
	// Read the next oldest message batch. If the buffer has a persisted store
	// the message is preserved until the returned AckFunc is called. Some
	// temporal buffer implementations such as windowers will ignore the ack
	// func.
	Read(context.Context) (*message.Batch, AckFunc, error)

	// Write a new message batch to the stack.
	Write(context.Context, *message.Batch, AckFunc) error

	// EndOfInput indicates to the buffer that the input has ended and that once
	// the buffer is depleted it should return component.ErrTypeClosed from Read in
	// order to gracefully shut down the pipeline.
	//
	// EndOfInput should be idempotent as it may be called more than once.
	EndOfInput()

	// Close the buffer and all resources it has, messages should no longer be
	// written or read by the implementation and it should clean up all
	// resources.
	Close(context.Context) error
}

ReaderWriter is a read/write interface implemented by buffers.

type Stream

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

Stream wraps a read/write buffer implementation with a channel based streaming component that satisfies the internal Benthos Consumer and Producer interfaces.

func (*Stream) CloseAsync

func (m *Stream) CloseAsync()

CloseAsync shuts down the Stream and stops processing messages.

func (*Stream) Consume

func (m *Stream) Consume(msgs <-chan message.Transaction) error

Consume assigns a messages channel for the output to read.

func (*Stream) StopConsuming

func (m *Stream) StopConsuming()

StopConsuming instructs the buffer to stop consuming messages and close once the buffer is empty.

func (*Stream) TransactionChan

func (m *Stream) TransactionChan() <-chan message.Transaction

TransactionChan returns the channel used for consuming messages from this buffer.

func (*Stream) WaitForClose

func (m *Stream) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the Stream output has closed down.

type Streamed

type Streamed interface {
	// TransactionChan returns a channel used for consuming transactions from
	// this type. Every transaction received must be resolved before another
	// transaction will be sent.
	TransactionChan() <-chan message.Transaction

	// StopConsuming instructs the buffer to cut off the producer it is
	// consuming from. It will then enter a mode whereby messages can only be
	// read, and when the buffer is empty it will shut down.
	StopConsuming()

	// Consume starts the type receiving transactions from a Transactor.
	Consume(<-chan message.Transaction) error

	// CloseAsync triggers the shut down of this component but should not block
	// the calling goroutine.
	CloseAsync()

	// WaitForClose is a blocking call to wait until the component has finished
	// shutting down and cleaning up resources.
	WaitForClose(timeout time.Duration) error
}

Streamed is an interface implemented by all buffer types that provides stream based methods.

func NewStream

func NewStream(typeStr string, buffer ReaderWriter, mgr component.Observability) Streamed

NewStream creates a new Producer/Consumer around a buffer.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL