output

package
Version: v3.15.4 Latest Latest
Warning

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

Go to latest
Published: May 30, 2020 License: MIT Imports: 47 Imported by: 0

Documentation

Overview

Package output defines all sinks for sending Benthos messages to a variety of third party destinations. All output types must implement interface output.Type.

If the sink of an output provides a form of acknowledgement of message receipt then the output is responsible for propagating that acknowledgement back to the source of the data by sending it over the transaction response channel. Otherwise a standard acknowledgement is sent.

Index

Constants

View Source
const (
	TypeAMQP            = "amqp"
	TypeAMQP09          = "amqp_0_9"
	TypeBroker          = "broker"
	TypeCache           = "cache"
	TypeDrop            = "drop"
	TypeDropOnError     = "drop_on_error"
	TypeDynamic         = "dynamic"
	TypeDynamoDB        = "dynamodb"
	TypeElasticsearch   = "elasticsearch"
	TypeFile            = "file"
	TypeFiles           = "files"
	TypeGCPPubSub       = "gcp_pubsub"
	TypeHDFS            = "hdfs"
	TypeHTTPClient      = "http_client"
	TypeHTTPServer      = "http_server"
	TypeInproc          = "inproc"
	TypeKafka           = "kafka"
	TypeKinesis         = "kinesis"
	TypeKinesisFirehose = "kinesis_firehose"
	TypeMQTT            = "mqtt"
	TypeNanomsg         = "nanomsg"
	TypeNATS            = "nats"
	TypeNATSStream      = "nats_stream"
	TypeNSQ             = "nsq"
	TypeRedisHash       = "redis_hash"
	TypeRedisList       = "redis_list"
	TypeRedisPubSub     = "redis_pubsub"
	TypeRedisStreams    = "redis_streams"
	TypeResource        = "resource"
	TypeRetry           = "retry"
	TypeS3              = "s3"
	TypeSNS             = "sns"
	TypeSQS             = "sqs"
	TypeSTDOUT          = "stdout"
	TypeSwitch          = "switch"
	TypeSyncResponse    = "sync_response"
	TypeTCP             = "tcp"
	TypeTry             = "try"
	TypeUDP             = "udp"
	TypeSocket          = "socket"
	TypeWebsocket       = "websocket"
	TypeZMQ4            = "zmq4"
)

String constants representing each output type.

Variables

View Source
var (
	// ErrSwitchNoConditionMet is returned when a message does not match any
	// output conditions.
	ErrSwitchNoConditionMet = errors.New("no switch output conditions were met by message")
	// ErrSwitchNoOutputs is returned when creating a Switch type with less than
	// 2 outputs.
	ErrSwitchNoOutputs = errors.New("attempting to create switch with less than 2 outputs")
)
View Source
var Constructors = map[string]TypeSpec{}

Constructors is a map of all output types with their specs.

View Source
var DocsAsync = `
This output benefits from sending multiple messages in flight in parallel for
improved performance. You can tune the max number of in flight messages with the
field ` + "`max_in_flight`" + `.`

DocsAsync returns a documentation paragraph regarding outputs that support asynchronous sends.

View Source
var DocsBatches = `` /* 205-byte string literal not displayed */

DocsBatches returns a documentation paragraph regarding outputs that support batching.

View Source
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 Descriptions

func Descriptions() string

Descriptions returns a formatted string of collated descriptions of each type.

func DocumentPlugin

func DocumentPlugin(
	typeString, description string,
	configSanitiser PluginConfigSanitiser,
)

DocumentPlugin adds a description and an optional configuration sanitiser function to the definition of a registered plugin. This improves the documentation generated by PluginDescriptions.

func PluginCount

func PluginCount() int

PluginCount returns the number of registered plugins. This does NOT count the standard set of components.

func PluginDescriptions

func PluginDescriptions() string

PluginDescriptions generates and returns a markdown formatted document listing each registered plugin and an example configuration for it.

func RegisterPlugin

func RegisterPlugin(
	typeString string,
	configConstructor PluginConfigConstructor,
	constructor PluginConstructor,
)

RegisterPlugin registers a plugin by a unique name so that it can be constructed similar to regular outputs. If configuration is not needed for this plugin then configConstructor can be nil. A constructor for the plugin itself must be provided.

func SanitiseConfig

func SanitiseConfig(conf Config) (interface{}, error)

SanitiseConfig returns a sanitised version of the Config, meaning sections that aren't relevant to behaviour are removed.

Types

type AsyncSink

type AsyncSink interface {
	// ConnectWithContext attempts to establish a connection to the sink, if
	// unsuccessful returns an error. If the attempt is successful (or not
	// necessary) returns nil.
	ConnectWithContext(ctx context.Context) error

	// WriteWithContext should block until either the message is sent (and
	// acknowledged) to a sink, or a transport specific error has occurred, or
	// the Type is closed.
	WriteWithContext(ctx context.Context, msg types.Message) error

	types.Closable
}

AsyncSink is a type that writes Benthos messages to a third party sink. If the protocol supports a form of acknowledgement then it will be returned by the call to Write.

type AsyncWriter

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

AsyncWriter is an output type that writes messages to a writer.Type.

func (*AsyncWriter) CloseAsync

func (w *AsyncWriter) CloseAsync()

CloseAsync shuts down the File output and stops processing messages.

func (*AsyncWriter) Connected

func (w *AsyncWriter) Connected() bool

Connected returns a boolean indicating whether this output is currently connected to its target.

func (*AsyncWriter) Consume

func (w *AsyncWriter) Consume(ts <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*AsyncWriter) WaitForClose

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

WaitForClose blocks until the File output has closed down.

type Batcher

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

Batcher wraps an output with a batching policy.

func (*Batcher) CloseAsync

func (m *Batcher) CloseAsync()

CloseAsync shuts down the Batcher and stops processing messages.

func (*Batcher) Connected

func (m *Batcher) Connected() bool

Connected returns a boolean indicating whether this output is currently connected to its target.

func (*Batcher) Consume

func (m *Batcher) Consume(msgs <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*Batcher) WaitForClose

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

WaitForClose blocks until the Batcher output has closed down.

type BrokerConfig

type BrokerConfig struct {
	Copies      int                `json:"copies" yaml:"copies"`
	Pattern     string             `json:"pattern" yaml:"pattern"`
	MaxInFlight int                `json:"max_in_flight" yaml:"max_in_flight"`
	Outputs     brokerOutputList   `json:"outputs" yaml:"outputs"`
	Batching    batch.PolicyConfig `json:"batching" yaml:"batching"`
}

BrokerConfig contains configuration fields for the Broker output type.

func NewBrokerConfig

func NewBrokerConfig() BrokerConfig

NewBrokerConfig creates a new BrokerConfig with default values.

type Config

type Config struct {
	Type            string                       `json:"type" yaml:"type"`
	AMQP            writer.AMQPConfig            `json:"amqp" yaml:"amqp"`
	AMQP09          writer.AMQPConfig            `json:"amqp_0_9" yaml:"amqp_0_9"`
	Broker          BrokerConfig                 `json:"broker" yaml:"broker"`
	Cache           writer.CacheConfig           `json:"cache" yaml:"cache"`
	Drop            writer.DropConfig            `json:"drop" yaml:"drop"`
	DropOnError     DropOnErrorConfig            `json:"drop_on_error" yaml:"drop_on_error"`
	Dynamic         DynamicConfig                `json:"dynamic" yaml:"dynamic"`
	DynamoDB        writer.DynamoDBConfig        `json:"dynamodb" yaml:"dynamodb"`
	Elasticsearch   writer.ElasticsearchConfig   `json:"elasticsearch" yaml:"elasticsearch"`
	File            FileConfig                   `json:"file" yaml:"file"`
	Files           writer.FilesConfig           `json:"files" yaml:"files"`
	GCPPubSub       writer.GCPPubSubConfig       `json:"gcp_pubsub" yaml:"gcp_pubsub"`
	HDFS            writer.HDFSConfig            `json:"hdfs" yaml:"hdfs"`
	HTTPClient      writer.HTTPClientConfig      `json:"http_client" yaml:"http_client"`
	HTTPServer      HTTPServerConfig             `json:"http_server" yaml:"http_server"`
	Inproc          InprocConfig                 `json:"inproc" yaml:"inproc"`
	Kafka           writer.KafkaConfig           `json:"kafka" yaml:"kafka"`
	Kinesis         writer.KinesisConfig         `json:"kinesis" yaml:"kinesis"`
	KinesisFirehose writer.KinesisFirehoseConfig `json:"kinesis_firehose" yaml:"kinesis_firehose"`
	MQTT            writer.MQTTConfig            `json:"mqtt" yaml:"mqtt"`
	Nanomsg         writer.NanomsgConfig         `json:"nanomsg" yaml:"nanomsg"`
	NATS            writer.NATSConfig            `json:"nats" yaml:"nats"`
	NATSStream      writer.NATSStreamConfig      `json:"nats_stream" yaml:"nats_stream"`
	NSQ             writer.NSQConfig             `json:"nsq" yaml:"nsq"`
	Plugin          interface{}                  `json:"plugin,omitempty" yaml:"plugin,omitempty"`
	RedisHash       writer.RedisHashConfig       `json:"redis_hash" yaml:"redis_hash"`
	RedisList       writer.RedisListConfig       `json:"redis_list" yaml:"redis_list"`
	RedisPubSub     writer.RedisPubSubConfig     `json:"redis_pubsub" yaml:"redis_pubsub"`
	RedisStreams    writer.RedisStreamsConfig    `json:"redis_streams" yaml:"redis_streams"`
	Resource        string                       `json:"resource" yaml:"resource"`
	Retry           RetryConfig                  `json:"retry" yaml:"retry"`
	S3              mywriter.AmazonS3Config      `json:"s3" yaml:"s3"` //use modified s3 writer
	SNS             writer.SNSConfig             `json:"sns" yaml:"sns"`
	SQS             writer.AmazonSQSConfig       `json:"sqs" yaml:"sqs"`
	STDOUT          STDOUTConfig                 `json:"stdout" yaml:"stdout"`
	Switch          SwitchConfig                 `json:"switch" yaml:"switch"`
	SyncResponse    struct{}                     `json:"sync_response" yaml:"sync_response"`
	TCP             writer.TCPConfig             `json:"tcp" yaml:"tcp"`
	Try             TryConfig                    `json:"try" yaml:"try"`
	UDP             writer.UDPConfig             `json:"udp" yaml:"udp"`
	Socket          writer.SocketConfig          `json:"socket" yaml:"socket"`
	Websocket       writer.WebsocketConfig       `json:"websocket" yaml:"websocket"`
	ZMQ4            *writer.ZMQ4Config           `json:"zmq4,omitempty" yaml:"zmq4,omitempty"`
	Processors      []processor.Config           `json:"processors" yaml:"processors"`
}

Config is the all encompassing configuration struct for all output types.

func NewConfig

func NewConfig() Config

NewConfig returns a configuration struct fully populated with default values.

func (*Config) UnmarshalYAML

func (c *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 DropOnError

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

DropOnError is an output type that continuously writes a message to a child output until the send is successful.

func (*DropOnError) CloseAsync

func (d *DropOnError) CloseAsync()

CloseAsync shuts down the DropOnError input and stops processing requests.

func (*DropOnError) Connected

func (d *DropOnError) Connected() bool

Connected returns a boolean indicating whether this output is currently connected to its target.

func (*DropOnError) Consume

func (d *DropOnError) Consume(ts <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*DropOnError) WaitForClose

func (d *DropOnError) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the DropOnError input has closed down.

type DropOnErrorConfig

type DropOnErrorConfig struct {
	*Config `yaml:",inline" json:",inline"`
}

DropOnErrorConfig contains configuration values for the DropOnError output type.

func NewDropOnErrorConfig

func NewDropOnErrorConfig() DropOnErrorConfig

NewDropOnErrorConfig creates a new DropOnErrorConfig with default values.

func (DropOnErrorConfig) MarshalJSON

func (d DropOnErrorConfig) MarshalJSON() ([]byte, error)

MarshalJSON prints an empty object instead of nil.

func (DropOnErrorConfig) MarshalYAML

func (d DropOnErrorConfig) MarshalYAML() (interface{}, error)

MarshalYAML prints an empty object instead of nil.

func (*DropOnErrorConfig) UnmarshalJSON

func (d *DropOnErrorConfig) UnmarshalJSON(bytes []byte) error

UnmarshalJSON ensures that when parsing child config it is initialised.

func (*DropOnErrorConfig) UnmarshalYAML

func (d *DropOnErrorConfig) UnmarshalYAML(unmarshal func(interface{}) error) error

UnmarshalYAML ensures that when parsing child config it is initialised.

type DynamicConfig

type DynamicConfig struct {
	Outputs map[string]Config `json:"outputs" yaml:"outputs"`
	Prefix  string            `json:"prefix" yaml:"prefix"`
	Timeout string            `json:"timeout" yaml:"timeout"`
}

DynamicConfig contains configuration fields for the Dynamic output type.

func NewDynamicConfig

func NewDynamicConfig() DynamicConfig

NewDynamicConfig creates a new DynamicConfig with default values.

type FileConfig

type FileConfig struct {
	Path  string `json:"path" yaml:"path"`
	Delim string `json:"delimiter" yaml:"delimiter"`
}

FileConfig contains configuration fields for the file based output type.

func NewFileConfig

func NewFileConfig() FileConfig

NewFileConfig creates a new FileConfig with default values.

type HTTPServer

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

HTTPServer is an output type that serves HTTPServer GET requests.

func (*HTTPServer) CloseAsync

func (h *HTTPServer) CloseAsync()

CloseAsync shuts down the HTTPServer output and stops processing requests.

func (*HTTPServer) Connected

func (h *HTTPServer) Connected() bool

Connected returns a boolean indicating whether this output is currently connected to its target.

func (*HTTPServer) Consume

func (h *HTTPServer) Consume(ts <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*HTTPServer) WaitForClose

func (h *HTTPServer) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the HTTPServer output has closed down.

type HTTPServerConfig

type HTTPServerConfig struct {
	Address    string `json:"address" yaml:"address"`
	Path       string `json:"path" yaml:"path"`
	StreamPath string `json:"stream_path" yaml:"stream_path"`
	WSPath     string `json:"ws_path" yaml:"ws_path"`
	Timeout    string `json:"timeout" yaml:"timeout"`
	CertFile   string `json:"cert_file" yaml:"cert_file"`
	KeyFile    string `json:"key_file" yaml:"key_file"`
}

HTTPServerConfig contains configuration fields for the HTTPServer output type.

func NewHTTPServerConfig

func NewHTTPServerConfig() HTTPServerConfig

NewHTTPServerConfig creates a new HTTPServerConfig with default values.

type Inproc

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

Inproc is an output type that serves Inproc messages.

func (*Inproc) CloseAsync

func (i *Inproc) CloseAsync()

CloseAsync shuts down the Inproc output and stops processing messages.

func (*Inproc) Connected

func (i *Inproc) Connected() bool

Connected returns a boolean indicating whether this output is currently connected to its target.

func (*Inproc) Consume

func (i *Inproc) Consume(ts <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*Inproc) WaitForClose

func (i *Inproc) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the Inproc output has closed down.

type InprocConfig

type InprocConfig string

InprocConfig contains configuration fields for the Inproc output type.

func NewInprocConfig

func NewInprocConfig() InprocConfig

NewInprocConfig creates a new InprocConfig with default values.

type LineWriter

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

LineWriter is an output type that writes messages to an io.WriterCloser type as lines.

func (*LineWriter) CloseAsync

func (w *LineWriter) CloseAsync()

CloseAsync shuts down the File output and stops processing messages.

func (*LineWriter) Connected

func (w *LineWriter) Connected() bool

Connected returns a boolean indicating whether this output is currently connected to its target.

func (*LineWriter) Consume

func (w *LineWriter) Consume(ts <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*LineWriter) WaitForClose

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

WaitForClose blocks until the File output has closed down.

type PluginConfigConstructor

type PluginConfigConstructor func() interface{}

PluginConfigConstructor is a func that returns a pointer to a new and fully populated configuration struct for a plugin type.

type PluginConfigSanitiser

type PluginConfigSanitiser func(conf interface{}) interface{}

PluginConfigSanitiser is a function that takes a configuration object for a plugin and returns a sanitised (minimal) version of it for printing in examples and plugin documentation.

This function is useful for when a plugins configuration struct is very large and complex, but can sometimes be expressed in a more concise way without losing the original intent.

type PluginConstructor

type PluginConstructor func(
	config interface{},
	manager types.Manager,
	logger log.Modular,
	metrics metrics.Type,
) (types.Output, error)

PluginConstructor is a func that constructs a Benthos output plugin. These are plugins that are specific to certain use cases, experimental, private or otherwise unfit for widespread general use. Any number of plugins can be specified when using Benthos as a framework.

The configuration object will be the result of the PluginConfigConstructor after overlaying the user configuration.

type Resource

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

Resource is a processor that returns the result of a output resource.

func (*Resource) CloseAsync

func (r *Resource) CloseAsync()

CloseAsync shuts down the output and stops processing requests.

func (*Resource) Connected

func (r *Resource) Connected() bool

Connected returns a boolean indicating whether this output is currently connected to its target.

func (*Resource) Consume

func (r *Resource) Consume(ts <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*Resource) WaitForClose

func (r *Resource) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the output has closed down.

type Retry

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

Retry is an output type that continuously writes a message to a child output until the send is successful.

func (*Retry) CloseAsync

func (r *Retry) CloseAsync()

CloseAsync shuts down the Retry input and stops processing requests.

func (*Retry) Connected

func (r *Retry) Connected() bool

Connected returns a boolean indicating whether this output is currently connected to its target.

func (*Retry) Consume

func (r *Retry) Consume(ts <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*Retry) WaitForClose

func (r *Retry) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the Retry input has closed down.

type RetryConfig

type RetryConfig struct {
	Output         *Config `json:"output" yaml:"output"`
	retries.Config `json:",inline" yaml:",inline"`
}

RetryConfig contains configuration values for the Retry output type.

func NewRetryConfig

func NewRetryConfig() RetryConfig

NewRetryConfig creates a new RetryConfig with default values.

func (RetryConfig) MarshalJSON

func (r RetryConfig) MarshalJSON() ([]byte, error)

MarshalJSON prints an empty object instead of nil.

func (RetryConfig) MarshalYAML

func (r RetryConfig) MarshalYAML() (interface{}, error)

MarshalYAML prints an empty object instead of nil.

type STDOUTConfig

type STDOUTConfig struct {
	Delim string `json:"delimiter" yaml:"delimiter"`
}

STDOUTConfig contains configuration fields for the stdout based output type.

func NewSTDOUTConfig

func NewSTDOUTConfig() STDOUTConfig

NewSTDOUTConfig creates a new STDOUTConfig with default values.

type Switch

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

Switch is a broker that implements types.Consumer and broadcasts each message out to an array of outputs.

func (*Switch) CloseAsync

func (o *Switch) CloseAsync()

CloseAsync shuts down the Switch broker and stops processing requests.

func (*Switch) Connected

func (o *Switch) Connected() bool

Connected returns a boolean indicating whether this output is currently connected to its target.

func (*Switch) Consume

func (o *Switch) Consume(transactions <-chan types.Transaction) error

Consume assigns a new transactions channel for the broker to read.

func (*Switch) WaitForClose

func (o *Switch) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the Switch broker has closed down.

type SwitchConfig

type SwitchConfig struct {
	RetryUntilSuccess bool                 `json:"retry_until_success" yaml:"retry_until_success"`
	Outputs           []SwitchConfigOutput `json:"outputs" yaml:"outputs"`
}

SwitchConfig contains configuration fields for the Switch output type.

func NewSwitchConfig

func NewSwitchConfig() SwitchConfig

NewSwitchConfig creates a new SwitchConfig with default values.

type SwitchConfigOutput

type SwitchConfigOutput struct {
	Condition   condition.Config `json:"condition" yaml:"condition"`
	Fallthrough bool             `json:"fallthrough" yaml:"fallthrough"`
	Output      Config           `json:"output" yaml:"output"`
}

SwitchConfigOutput contains configuration fields per output of a switch type.

func NewSwitchConfigOutput

func NewSwitchConfigOutput() SwitchConfigOutput

NewSwitchConfigOutput creates a new switch output config with default values.

func (*SwitchConfigOutput) UnmarshalJSON

func (s *SwitchConfigOutput) UnmarshalJSON(bytes []byte) error

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

func (*SwitchConfigOutput) UnmarshalYAML

func (s *SwitchConfigOutput) UnmarshalYAML(unmarshal func(interface{}) error) error

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

type TryConfig

type TryConfig brokerOutputList

TryConfig contains configuration fields for the Try output type.

func NewTryConfig

func NewTryConfig() TryConfig

NewTryConfig creates a new BrokerConfig with default values.

type Type

type Type interface {
	types.Closable
	types.Consumer

	// Connected returns a boolean indicating whether this output is currently
	// connected to its target.
	Connected() bool
}

Type is the standard interface of an output type.

func New

func New(
	conf Config,
	mgr types.Manager,
	log log.Modular,
	stats metrics.Type,
	pipelines ...types.PipelineConstructorFunc,
) (Type, error)

New creates an output type based on an output configuration.

func NewAMQP

func NewAMQP(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewAMQP creates a new AMQP output type. TODO: V4 Remove this.

func NewAMQP09

func NewAMQP09(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewAMQP09 creates a new AMQP output type.

func NewAmazonS3

func NewAmazonS3(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewAmazonS3 creates a new AmazonS3 output type.

func NewAmazonSNS

func NewAmazonSNS(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewAmazonSNS creates a new AmazonSNS output type.

func NewAmazonSQS

func NewAmazonSQS(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewAmazonSQS creates a new AmazonSQS output type.

func NewAsyncWriter

func NewAsyncWriter(
	typeStr string,
	maxInflight int,
	w AsyncSink,
	log log.Modular,
	stats metrics.Type,
) (Type, error)

NewAsyncWriter creates a new AsyncWriter output type.

func NewBatcher

func NewBatcher(
	batcher *batch.Policy,
	child Type,
	log log.Modular,
	stats metrics.Type,
) Type

NewBatcher creates a new Producer/Consumer around a buffer.

func NewBroker

func NewBroker(
	conf Config,
	mgr types.Manager,
	log log.Modular,
	stats metrics.Type,
	pipelines ...types.PipelineConstructorFunc,
) (Type, error)

NewBroker creates a new Broker output type. Messages will be sent out to the list of outputs according to the chosen broker pattern.

func NewCache

func NewCache(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewCache creates a new Cache output type.

func NewDrop

func NewDrop(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewDrop creates a new Drop output type.

func NewDropOnError

func NewDropOnError(
	conf Config,
	mgr types.Manager,
	log log.Modular,
	stats metrics.Type,
) (Type, error)

NewDropOnError creates a new DropOnError input type.

func NewDynamic

func NewDynamic(
	conf Config,
	mgr types.Manager,
	log log.Modular,
	stats metrics.Type,
) (Type, error)

NewDynamic creates a new Dynamic output type.

func NewDynamoDB

func NewDynamoDB(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewDynamoDB creates a new DynamoDB output type.

func NewElasticsearch

func NewElasticsearch(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewElasticsearch creates a new Elasticsearch output type.

func NewFile

func NewFile(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewFile creates a new File output type.

func NewFiles

func NewFiles(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewFiles creates a new Files output type.

func NewGCPPubSub

func NewGCPPubSub(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewGCPPubSub creates a new GCPPubSub output type.

func NewHDFS

func NewHDFS(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewHDFS creates a new HDFS output type.

func NewHTTPClient

func NewHTTPClient(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewHTTPClient creates a new HTTPClient output type.

func NewHTTPServer

func NewHTTPServer(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewHTTPServer creates a new HTTPServer output type.

func NewInproc

func NewInproc(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewInproc creates a new Inproc output type.

func NewKafka

func NewKafka(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewKafka creates a new Kafka output type.

func NewKinesis

func NewKinesis(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewKinesis creates a new Kinesis output type.

func NewKinesisFirehose

func NewKinesisFirehose(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewKinesisFirehose creates a new KinesisFirehose output type.

func NewLineWriter

func NewLineWriter(
	handle io.WriteCloser,
	closeOnExit bool,
	customDelimiter []byte,
	typeStr string,
	log log.Modular,
	stats metrics.Type,
) (Type, error)

NewLineWriter creates a new LineWriter output type.

func NewMQTT

func NewMQTT(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewMQTT creates a new MQTT output type.

func NewNATS

func NewNATS(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewNATS creates a new NATS output type.

func NewNATSStream

func NewNATSStream(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewNATSStream creates a new NATSStream output type.

func NewNSQ

func NewNSQ(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewNSQ creates a new NSQ output type.

func NewNanomsg

func NewNanomsg(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewNanomsg creates a new Nanomsg output type.

func NewRedisHash

func NewRedisHash(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewRedisHash creates a new RedisHash output type.

func NewRedisList

func NewRedisList(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewRedisList creates a new RedisList output type.

func NewRedisPubSub

func NewRedisPubSub(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewRedisPubSub creates a new RedisPubSub output type.

func NewRedisStreams

func NewRedisStreams(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewRedisStreams creates a new RedisStreams output type.

func NewResource

func NewResource(
	conf Config, mgr types.Manager, log log.Modular, stats metrics.Type,
) (Type, error)

NewResource returns a resource output.

func NewRetry

func NewRetry(
	conf Config,
	mgr types.Manager,
	log log.Modular,
	stats metrics.Type,
) (Type, error)

NewRetry creates a new Retry input type.

func NewSTDOUT

func NewSTDOUT(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewSTDOUT creates a new STDOUT output type.

func NewSocket

func NewSocket(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewSocket creates a new Socket output type.

func NewSwitch

func NewSwitch(
	conf Config,
	mgr types.Manager,
	logger log.Modular,
	stats metrics.Type,
) (Type, error)

NewSwitch creates a new Switch type by providing outputs. Messages will be sent to a subset of outputs according to condition and fallthrough settings.

func NewTCP

func NewTCP(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewTCP creates a new TCP output type.

func NewTry

func NewTry(
	conf Config,
	mgr types.Manager,
	log log.Modular,
	stats metrics.Type,
	pipelines ...types.PipelineConstructorFunc,
) (Type, error)

NewTry creates a new try broker output type.

func NewUDP

func NewUDP(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewUDP creates a new UDP output type.

func NewWebsocket

func NewWebsocket(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)

NewWebsocket creates a new Websocket output type.

func NewWriter

func NewWriter(
	typeStr string,
	w writer.Type,
	log log.Modular,
	stats metrics.Type,
) (Type, error)

NewWriter creates a new Writer output type.

func WrapWithPipelines

func WrapWithPipelines(out Type, pipeConstructors ...types.PipelineConstructorFunc) (Type, error)

WrapWithPipelines wraps an output with a variadic number of pipelines.

type TypeSpec

type TypeSpec struct {
	Summary     string
	Description string
	Footnotes   string

	// Async indicates whether this output benefits from sending multiple
	// messages asynchronously over the protocol.
	Async bool

	// Batches indicates whether this output benefits from batching of messages.
	Batches bool

	// Deprecated indicates whether this component is deprecated.
	Deprecated bool

	FieldSpecs docs.FieldSpecs
	// contains filtered or unexported fields
}

TypeSpec is a constructor and a usage description for each output type.

type WithPipeline

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

WithPipeline is a type that wraps both an output type and a pipeline type by routing the pipeline through the output, and implements the output.Type interface in order to act like an ordinary output.

func WrapWithPipeline

func WrapWithPipeline(i *int, out Type, pipeConstructor types.PipelineConstructorFunc) (*WithPipeline, error)

WrapWithPipeline routes a processing pipeline directly into an output and returns a type that manages both and acts like an ordinary output.

func (*WithPipeline) CloseAsync

func (i *WithPipeline) CloseAsync()

CloseAsync triggers a closure of this object but does not block.

func (*WithPipeline) Connected

func (i *WithPipeline) Connected() bool

Connected returns a boolean indicating whether this output is currently connected to its target.

func (*WithPipeline) Consume

func (i *WithPipeline) Consume(tsChan <-chan types.Transaction) error

Consume starts the type listening to a message channel from a producer.

func (*WithPipeline) WaitForClose

func (i *WithPipeline) WaitForClose(timeout time.Duration) error

WaitForClose is a blocking call to wait until the object has finished closing down and cleaning up resources.

type Writer

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

Writer is an output type that writes messages to a writer.Type.

func (*Writer) CloseAsync

func (w *Writer) CloseAsync()

CloseAsync shuts down the File output and stops processing messages.

func (*Writer) Connected

func (w *Writer) Connected() bool

Connected returns a boolean indicating whether this output is currently connected to its target.

func (*Writer) Consume

func (w *Writer) Consume(ts <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*Writer) WaitForClose

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

WaitForClose blocks until the File output has closed down.

Directories

Path Synopsis
Package writer defines implementations of an interface for generic message writing that outputs to various third party sinks.
Package writer defines implementations of an interface for generic message writing that outputs to various third party sinks.

Jump to

Keyboard shortcuts

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