output

package
v0.25.1 Latest Latest
Warning

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

 Go to latest Published: Aug 24, 2018 License: MIT Imports: 44 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"
	TypeBroker        = "broker"
	TypeDynamic       = "dynamic"
	TypeElasticsearch = "elasticsearch"
	TypeFile          = "file"
	TypeFiles         = "files"
	TypeHTTPClient    = "http_client"
	TypeHTTPServer    = "http_server"
	TypeInproc        = "inproc"
	TypeKafka         = "kafka"
	TypeMQTT          = "mqtt"
	TypeNanomsg       = "nanomsg"
	TypeNATS          = "nats"
	TypeNATSStream    = "nats_stream"
	TypeNSQ           = "nsq"
	TypeRedisList     = "redis_list"
	TypeRedisPubSub   = "redis_pubsub"
	TypeS3            = "s3"
	TypeSQS           = "sqs"
	TypeSTDOUT        = "stdout"
	TypeWebsocket     = "websocket"
	TypeZMQ4          = "zmq4"
)

String constants representing each output type.

Variables

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 SanitiseConfig added in v0.8.4 

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 BrokerConfig added in v0.8.8 

type BrokerConfig struct {
	Copies  int              `json:"copies" yaml:"copies"`
	Pattern string           `json:"pattern" yaml:"pattern"`
	Outputs brokerOutputList `json:"outputs" yaml:"outputs"`
}

BrokerConfig contains configuration fields for the Broker output type.

func NewBrokerConfig added in v0.8.8 

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"`
	Broker        BrokerConfig               `json:"broker" yaml:"broker"`
	Dynamic       DynamicConfig              `json:"dynamic" yaml:"dynamic"`
	Elasticsearch writer.ElasticsearchConfig `json:"elasticsearch" yaml:"elasticsearch"`
	File          FileConfig                 `json:"file" yaml:"file"`
	Files         writer.FilesConfig         `json:"files" yaml:"files"`
	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"`
	MQTT          writer.MQTTConfig          `json:"mqtt" yaml:"mqtt"`
	Nanomsg       NanomsgConfig              `json:"nanomsg" yaml:"nanomsg"`
	NATS          NATSConfig                 `json:"nats" yaml:"nats"`
	NATSStream    NATSStreamConfig           `json:"nats_stream" yaml:"nats_stream"`
	NSQ           NSQConfig                  `json:"nsq" yaml:"nsq"`
	RedisList     writer.RedisListConfig     `json:"redis_list" yaml:"redis_list"`
	RedisPubSub   RedisPubSubConfig          `json:"redis_pubsub" yaml:"redis_pubsub"`
	S3            writer.AmazonS3Config      `json:"s3" yaml:"s3"`
	SQS           writer.AmazonSQSConfig     `json:"sqs" yaml:"sqs"`
	STDOUT        STDOUTConfig               `json:"stdout" yaml:"stdout"`
	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) UnmarshalJSON added in v0.8.0 

func (c *Config) UnmarshalJSON(bytes []byte) error

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

func (*Config) UnmarshalYAML added in v0.8.0 

func (c *Config) 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 DynamicConfig added in v0.8.0 

type DynamicConfig struct {
	Outputs   map[string]Config `json:"outputs" yaml:"outputs"`
	Prefix    string            `json:"prefix" yaml:"prefix"`
	TimeoutMS int               `json:"timeout_ms" yaml:"timeout_ms"`
}

DynamicConfig contains configuration fields for the Dynamic output type.

func NewDynamicConfig added in v0.8.0 

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) Consume added in v0.19.0 

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"`
	TimeoutMS  int64  `json:"timeout_ms" yaml:"timeout_ms"`
	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 added in v0.15.4 

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

Inproc is an output type that serves Inproc messages.

func (*Inproc) CloseAsync added in v0.15.4 

func (i *Inproc) CloseAsync()

CloseAsync shuts down the Inproc output and stops processing messages.

func (*Inproc) Consume added in v0.19.0 

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

Consume assigns a messages channel for the output to read.

func (*Inproc) WaitForClose added in v0.15.4 

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

WaitForClose blocks until the Inproc output has closed down.

type InprocConfig added in v0.15.4 

type InprocConfig string

InprocConfig contains configuration fields for the Inproc output type.

func NewInprocConfig added in v0.15.4 

func NewInprocConfig() InprocConfig

NewInprocConfig creates a new InprocConfig with default values.

type LineWriter added in v0.7.4 

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 added in v0.7.4 

func (w *LineWriter) CloseAsync()

CloseAsync shuts down the File output and stops processing messages.

func (*LineWriter) Consume added in v0.19.0 

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

Consume assigns a messages channel for the output to read.

func (*LineWriter) WaitForClose added in v0.7.4 

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

WaitForClose blocks until the File output has closed down.

type NATS added in v0.2.2 

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

NATS is an output type that serves NATS messages.

func (*NATS) CloseAsync added in v0.2.2 

func (n *NATS) CloseAsync()

CloseAsync shuts down the NATS output and stops processing messages.

func (*NATS) Consume added in v0.19.0 

func (n *NATS) Consume(ts <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*NATS) WaitForClose added in v0.2.2 

func (n *NATS) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the NATS output has closed down.

type NATSConfig added in v0.2.2 

type NATSConfig struct {
	URLs    []string `json:"urls" yaml:"urls"`
	Subject string   `json:"subject" yaml:"subject"`
}

NATSConfig contains configuration fields for the NATS output type.

func NewNATSConfig added in v0.2.2 

func NewNATSConfig() NATSConfig

NewNATSConfig creates a new NATSConfig with default values.

type NATSStream added in v0.4.9 

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

NATSStream is an output type that serves NATS messages.

func (*NATSStream) CloseAsync added in v0.4.9 

func (n *NATSStream) CloseAsync()

CloseAsync shuts down the NATSStream output and stops processing messages.

func (*NATSStream) Consume added in v0.19.0 

func (n *NATSStream) Consume(ts <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*NATSStream) WaitForClose added in v0.4.9 

func (n *NATSStream) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the NATSStream output has closed down.

type NATSStreamConfig added in v0.4.9 

type NATSStreamConfig struct {
	URLs      []string `json:"urls" yaml:"urls"`
	ClusterID string   `json:"cluster_id" yaml:"cluster_id"`
	ClientID  string   `json:"client_id" yaml:"client_id"`
	Subject   string   `json:"subject" yaml:"subject"`
}

NATSStreamConfig contains configuration fields for the NATSStream output type.

func NewNATSStreamConfig added in v0.4.9 

func NewNATSStreamConfig() NATSStreamConfig

NewNATSStreamConfig creates a new NATSStreamConfig with default values.

type NSQ added in v0.1.1 

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

NSQ is an output type that serves NSQ messages.

func (*NSQ) CloseAsync added in v0.1.1 

func (n *NSQ) CloseAsync()

CloseAsync shuts down the NSQ output and stops processing messages.

func (*NSQ) Consume added in v0.19.0 

func (n *NSQ) Consume(ts <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*NSQ) WaitForClose added in v0.1.1 

func (n *NSQ) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the NSQ output has closed down.

type NSQConfig added in v0.1.1 

type NSQConfig struct {
	Address     string `json:"nsqd_tcp_address" yaml:"nsqd_tcp_address"`
	Topic       string `json:"topic" yaml:"topic"`
	UserAgent   string `json:"user_agent" yaml:"user_agent"`
	MaxInFlight int    `json:"max_in_flight" yaml:"max_in_flight"`
}

NSQConfig contains configuration fields for the NSQ output type.

func NewNSQConfig added in v0.1.1 

func NewNSQConfig() NSQConfig

NewNSQConfig creates a new NSQConfig with default values.

type Nanomsg added in v0.22.0 

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

Nanomsg is an output type that serves Nanomsg messages.

func (*Nanomsg) CloseAsync added in v0.22.0 

func (s *Nanomsg) CloseAsync()

CloseAsync shuts down the Nanomsg output and stops processing messages.

func (*Nanomsg) Consume added in v0.22.0 

func (s *Nanomsg) Consume(ts <-chan types.Transaction) error

Consume assigns a messages channel for the output to read.

func (*Nanomsg) WaitForClose added in v0.22.0 

func (s *Nanomsg) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the Nanomsg output has closed down.

type NanomsgConfig added in v0.22.0 

type NanomsgConfig struct {
	URLs          []string `json:"urls" yaml:"urls"`
	Bind          bool     `json:"bind" yaml:"bind"`
	SocketType    string   `json:"socket_type" yaml:"socket_type"`
	PollTimeoutMS int      `json:"poll_timeout_ms" yaml:"poll_timeout_ms"`
}

NanomsgConfig contains configuration fields for the Nanomsg output type.

func NewNanomsgConfig added in v0.22.0 

func NewNanomsgConfig() NanomsgConfig

NewNanomsgConfig creates a new NanomsgConfig with default values.

type RedisPubSub added in v0.6.0 

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

RedisPubSub is an output type that serves RedisPubSub messages.

func (*RedisPubSub) CloseAsync added in v0.6.0 

func (r *RedisPubSub) CloseAsync()

CloseAsync shuts down the RedisPubSub output and stops processing messages.

func (*RedisPubSub) Consume added in v0.19.0 

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

Consume assigns a messages channel for the output to read.

func (*RedisPubSub) WaitForClose added in v0.6.0 

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

WaitForClose blocks until the RedisPubSub output has closed down.

type RedisPubSubConfig added in v0.6.0 

type RedisPubSubConfig struct {
	URL     string `json:"url" yaml:"url"`
	Channel string `json:"channel" yaml:"channel"`
}

RedisPubSubConfig contains configuration fields for the RedisPubSub output type.

func NewRedisPubSubConfig added in v0.6.0 

func NewRedisPubSubConfig() RedisPubSubConfig

NewRedisPubSubConfig creates a new RedisPubSubConfig with default values.

type STDOUTConfig added in v0.4.4 

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

STDOUTConfig contains configuration fields for the stdout based output type.

func NewSTDOUTConfig added in v0.4.4 

func NewSTDOUTConfig() STDOUTConfig

NewSTDOUTConfig creates a new STDOUTConfig with default values.

type Type 

type Type interface {
	types.Closable
	types.Consumer
}

Type is the standard interface of an output type.

func New added in v0.0.2 

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 added in v0.0.2 

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

NewAMQP creates a new AMQP output type.

func NewAmazonS3 added in v0.7.5 

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

NewAmazonS3 creates a new AmazonS3 output type.

func NewAmazonSQS added in v0.8.1 

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

NewAmazonSQS creates a new AmazonSQS output type.

func NewBroker added in v0.8.8 

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 NewDynamic added in v0.8.0 

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

NewDynamic creates a new Dynamic output type.

func NewElasticsearch added in v0.10.7 

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 added in v0.7.7 

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

NewFiles creates a new Files 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 added in v0.15.4 

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 NewLineWriter added in v0.7.4 

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 added in v0.8.7 

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

NewMQTT creates a new MQTT output type.

func NewNATS added in v0.2.2 

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

NewNATS creates a new NATS output type.

func NewNATSStream added in v0.4.9 

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

NewNATSStream creates a new NATSStream output type.

func NewNSQ added in v0.1.1 

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

NewNSQ creates a new NSQ output type.

func NewNanomsg added in v0.22.0 

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

NewNanomsg creates a new Nanomsg output type.

func NewRedisList added in v0.7.4 

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

NewRedisList creates a new RedisList output type.

func NewRedisPubSub added in v0.6.0 

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

NewRedisPubSub creates a new RedisPubSub output 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 NewWebsocket added in v0.13.2 

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

NewWebsocket creates a new Websocket output type.

func NewWriter added in v0.7.4 

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

NewWriter creates a new Writer output type.

func WrapWithPipelines added in v0.3.0 

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

WrapWithPipelines wraps an output with a variadic number of pipelines.

type TypeSpec added in v0.9.1 

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

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

type WithPipeline added in v0.3.0 

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 added in v0.3.0 

func WrapWithPipeline(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 added in v0.3.0 

func (i *WithPipeline) CloseAsync()

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

func (*WithPipeline) Consume added in v0.19.0 

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

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

func (*WithPipeline) WaitForClose added in v0.3.0 

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 added in v0.7.4 

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

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

func (*Writer) CloseAsync added in v0.7.4 

func (w *Writer) CloseAsync()

CloseAsync shuts down the File output and stops processing messages.

func (*Writer) Consume added in v0.19.0 

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

Consume assigns a messages channel for the output to read.

func (*Writer) WaitForClose added in v0.7.4 

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.