backends

Documentation

Index

• type Backend
  • func New(connOpts *opts.ConnectionOptions) (Backend, error)

Types

type Backend

type Backend interface {
	// Read will read data from the bus and dump each message to the results
	// channel. This method should _not_ decode the message - that is left up
	// to the upstream user. The error channel _should_ be optional.
	//
	// Decoding should happen _outside_ the backend.
	Read(ctx context.Context, readOpts *opts.ReadOptions, resultsChan chan *records.ReadRecord, errorChan chan *records.ErrorRecord) error

	// Write will attempt to write the input messages as a batch (if the backend
	// supports batch writing). This call will block until success/error.
	//
	// Encoding should happen _outside_ the backend.
	//
	// NOTE: Key, headers and any other metadata is fetched from CLIOptions
	// (that are passed when instantiating the backend).
	Write(ctx context.Context, writeOpts *opts.WriteOptions, errorCh chan<- *records.ErrorRecord, messages ...*records.WriteRecord) error

	// Test performs a "test" to see if the connection to the backend is alive.
	// The test varies between backends (ie. in kafka, it might be just attempting
	// to connect to a broker, while with another backend, plumber might try to
	// put/get sample data).
	Test(ctx context.Context) error

	// Tunnel creates a tunnel to Batch and exposes the connected backend as a
	// destination. This is a blocking call.
	Tunnel(ctx context.Context, tunnelOpts *opts.TunnelOptions, tunnelSvc tunnel.ITunnel, errorCh chan<- *records.ErrorRecord) error

	// Relay will hook into a message bus as a consumer and relay all messages
	// to the relayCh; if an error channel is provided, any errors will be piped
	// to the channel as well. This method _usually_ blocks.
	Relay(ctx context.Context, relayOpts *opts.RelayOptions, relayCh chan interface{}, errorCh chan<- *records.ErrorRecord) error

	// DisplayMessage will parse a Read record and print (pretty) output to STDOUT
	DisplayMessage(cliOpts *opts.CLIOptions, msg *records.ReadRecord) error

	// DisplayError will parse an Error record and print (pretty) output to STDOUT
	DisplayError(msg *records.ErrorRecord) error

	// Close closes any connections the backend has open. Once this is ran, you
	// should create a new backend instance.
	Close(ctx context.Context) error

	// Name returns the name of the backend
	Name() string
}

Backend is the interface that all backends implement; the interface is used for facilitating all CLI and server functionality in plumber. NOTE: Most backends do not support _some_ part of the interface - in those cases the methods will either return types.NotImplementedErr or types.UnsupportedFeatureErr.

func New

func New(connOpts *opts.ConnectionOptions) (Backend, error)

New is a convenience function to instantiate the appropriate backend based on package name of the backend.