output

package
v3.6.1-olly-2 Latest Latest
Warning

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

 Go to latest Published: Dec 9, 2019 License: MIT Imports: 39 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"
	TypeRetry           = "retry"
	TypeS3              = "s3"
	TypeSNS             = "sns"
	TypeSQS             = "sqs"
	TypeSTDOUT          = "stdout"
	TypeSwitch          = "switch"
	TypeSyncResponse    = "sync_response"
	TypeTCP             = "tcp"
	TypeTry             = "try"
	TypeUDP             = "udp"
	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 (
	// 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 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"`
	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"`
	Retry           RetryConfig                  `json:"retry" yaml:"retry"`
	S3              writer.AmazonS3Config        `json:"s3" yaml:"s3"`
	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"`
	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 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 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 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 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 {
	// 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.

Source Files

View all Source files

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
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.