Documentation ¶
Overview ¶
Package output defines dispatchers of data for a variety of targets. All output types must implement interface output.Type.
Index ¶
- Variables
- func Descriptions() string
- func SanitiseConfig(conf Config) (interface{}, error)
- type AMQP
- type AMQPConfig
- type Config
- type DynamicConfig
- type FanOutConfig
- type FileConfig
- type HTTPClient
- type HTTPClientConfig
- type HTTPServer
- type HTTPServerConfig
- type LineWriter
- type NATS
- type NATSConfig
- type NATSStream
- type NATSStreamConfig
- type NSQ
- type NSQConfig
- type RedisPubSub
- type RedisPubSubConfig
- type RoundRobinConfig
- type STDOUTConfig
- type ScaleProto
- type ScaleProtoConfig
- type Type
- func New(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type, ...) (Type, error)
- func NewAMQP(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewAmazonS3(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewAmazonSQS(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewDynamic(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewFanOut(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewFile(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewFiles(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewHTTPClient(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewHTTPServer(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewKafka(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewLineWriter(handle io.WriteCloser, customDelimiter []byte, typeStr string, log log.Modular, ...) (Type, error)
- func NewNATS(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewNATSStream(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewNSQ(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewRedisList(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewRedisPubSub(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewRoundRobin(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewSTDOUT(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewScaleProto(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
- func NewWriter(typeStr string, w writer.Type, log log.Modular, stats metrics.Type) (Type, error)
- func WrapWithPipelines(out Type, pipeConstructors ...pipeline.ConstructorFunc) (Type, error)
- type WithPipeline
- type Writer
- type ZMQ4Config
Constants ¶
This section is empty.
Variables ¶
var ( // ErrFanOutNoOutputs is returned when creating a FanOut type with zero // outputs. ErrFanOutNoOutputs = errors.New("attempting to create fan_out output type with no outputs") )
var ( // ErrRoundRobinNoOutputs is returned when creating a RoundRobin type with // zero outputs. ErrRoundRobinNoOutputs = errors.New("attempting to create round_robin output 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
SanitiseConfig returns a sanitised version of the Config, meaning sections that aren't relevant to behaviour are removed.
Types ¶
type AMQP ¶ added in v0.0.2
type AMQP struct {
// contains filtered or unexported fields
}
AMQP is an output type that serves AMQP messages.
func (*AMQP) CloseAsync ¶ added in v0.0.2
func (a *AMQP) CloseAsync()
CloseAsync shuts down the AMQP output and stops processing messages.
func (*AMQP) ResponseChan ¶ added in v0.0.2
ResponseChan returns the errors channel.
func (*AMQP) StartReceiving ¶ added in v0.0.2
StartReceiving assigns a messages channel for the output to read.
type AMQPConfig ¶ added in v0.0.2
type AMQPConfig struct { URL string `json:"url" yaml:"url"` Exchange string `json:"exchange" yaml:"exchange"` ExchangeType string `json:"exchange_type" yaml:"exchange_type"` BindingKey string `json:"key" yaml:"key"` }
AMQPConfig is configuration for the AMQP output type.
func NewAMQPConfig ¶ added in v0.0.2
func NewAMQPConfig() AMQPConfig
NewAMQPConfig creates a new AMQPConfig with default values.
type Config ¶
type Config struct { Type string `json:"type" yaml:"type"` AmazonS3 writer.AmazonS3Config `json:"amazon_s3" yaml:"amazon_s3"` AmazonSQS writer.AmazonSQSConfig `json:"amazon_sqs" yaml:"amazon_sqs"` AMQP AMQPConfig `json:"amqp" yaml:"amqp"` Dynamic DynamicConfig `json:"dynamic" yaml:"dynamic"` FanOut FanOutConfig `json:"fan_out" yaml:"fan_out"` File FileConfig `json:"file" yaml:"file"` Files writer.FilesConfig `json:"files" yaml:"files"` HTTPClient HTTPClientConfig `json:"http_client" yaml:"http_client"` HTTPServer HTTPServerConfig `json:"http_server" yaml:"http_server"` Kafka writer.KafkaConfig `json:"kafka" yaml:"kafka"` 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"` RoundRobin RoundRobinConfig `json:"round_robin" yaml:"round_robin"` ScaleProto ScaleProtoConfig `json:"scalability_protocols" yaml:"scalability_protocols"` STDOUT STDOUTConfig `json:"stdout" yaml:"stdout"` ZMQ4 *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. Note that some configs are empty structs, as the type has no optional values but we want to list it as an option.
func NewConfig ¶
func NewConfig() Config
NewConfig returns a configuration struct fully populated with default values.
func (*Config) UnmarshalJSON ¶ added in v0.8.0
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
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 is configuration for the Dynamic input type.
func NewDynamicConfig ¶ added in v0.8.0
func NewDynamicConfig() DynamicConfig
NewDynamicConfig creates a new DynamicConfig with default values.
type FanOutConfig ¶
type FanOutConfig struct {
Outputs []interface{} `json:"outputs" yaml:"outputs"`
}
FanOutConfig is configuration for the FanOut output type.
func NewFanOutConfig ¶
func NewFanOutConfig() FanOutConfig
NewFanOutConfig creates a new FanOutConfig with default values.
type FileConfig ¶
type FileConfig struct { Path string `json:"path" yaml:"path"` CustomDelim string `json:"custom_delimiter" yaml:"custom_delimiter"` }
FileConfig is configuration values for the file based output type.
func NewFileConfig ¶
func NewFileConfig() FileConfig
NewFileConfig creates a new FileConfig with default values.
type HTTPClient ¶
type HTTPClient struct {
// contains filtered or unexported fields
}
HTTPClient is an output type that pushes messages to HTTPClient.
func (*HTTPClient) CloseAsync ¶
func (h *HTTPClient) CloseAsync()
CloseAsync shuts down the HTTPClient output and stops processing messages.
func (*HTTPClient) ResponseChan ¶
func (h *HTTPClient) ResponseChan() <-chan types.Response
ResponseChan returns the errors channel.
func (*HTTPClient) StartReceiving ¶
func (h *HTTPClient) StartReceiving(msgs <-chan types.Message) error
StartReceiving assigns a messages channel for the output to read.
func (*HTTPClient) WaitForClose ¶
func (h *HTTPClient) WaitForClose(timeout time.Duration) error
WaitForClose blocks until the HTTPClient output has closed down.
type HTTPClientConfig ¶
type HTTPClientConfig struct { URL string `json:"url" yaml:"url"` Verb string `json:"verb" yaml:"verb"` ContentType string `json:"content_type" yaml:"content_type"` OAuth oauth.ClientConfig `json:"oauth" yaml:"oauth"` TimeoutMS int64 `json:"timeout_ms" yaml:"timeout_ms"` RetryMS int64 `json:"retry_period_ms" yaml:"retry_period_ms"` NumRetries int `json:"retries" yaml:"retries"` SkipCertVerify bool `json:"skip_cert_verify" yaml:"skip_cert_verify"` }
HTTPClientConfig is configuration for the HTTPClient output type.
func NewHTTPClientConfig ¶
func NewHTTPClientConfig() HTTPClientConfig
NewHTTPClientConfig creates a new HTTPClientConfig with default values.
type HTTPServer ¶
type HTTPServer struct {
// contains filtered or unexported fields
}
HTTPServer is an input type that serves HTTPServer POST requests.
func (*HTTPServer) CloseAsync ¶
func (h *HTTPServer) CloseAsync()
CloseAsync shuts down the HTTPServer input and stops processing requests.
func (*HTTPServer) ResponseChan ¶
func (h *HTTPServer) ResponseChan() <-chan types.Response
ResponseChan returns the errors channel.
func (*HTTPServer) StartReceiving ¶
func (h *HTTPServer) StartReceiving(msgs <-chan types.Message) error
StartReceiving 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"` 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 is configuration for the HTTPServer input type.
func NewHTTPServerConfig ¶
func NewHTTPServerConfig() HTTPServerConfig
NewHTTPServerConfig creates a new HTTPServerConfig 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) ResponseChan ¶ added in v0.7.4
func (w *LineWriter) ResponseChan() <-chan types.Response
ResponseChan returns the errors channel.
func (*LineWriter) StartReceiving ¶ added in v0.7.4
func (w *LineWriter) StartReceiving(msgs <-chan types.Message) error
StartReceiving 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) ResponseChan ¶ added in v0.2.2
ResponseChan returns the errors channel.
func (*NATS) StartReceiving ¶ added in v0.2.2
StartReceiving assigns a messages channel for the output to read.
type NATSConfig ¶ added in v0.2.2
type NATSConfig struct { URLs []string `json:"urls" yaml:"urls"` Subject string `json:"subject" yaml:"subject"` }
NATSConfig is configuration 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) ResponseChan ¶ added in v0.4.9
func (n *NATSStream) ResponseChan() <-chan types.Response
ResponseChan returns the errors channel.
func (*NATSStream) StartReceiving ¶ added in v0.4.9
func (n *NATSStream) StartReceiving(msgs <-chan types.Message) error
StartReceiving 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 is configuration 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) ResponseChan ¶ added in v0.1.1
ResponseChan returns the errors channel.
func (*NSQ) StartReceiving ¶ added in v0.1.1
StartReceiving assigns a messages channel for the output to read.
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 is configuration for the NSQ output type.
func NewNSQConfig ¶ added in v0.1.1
func NewNSQConfig() NSQConfig
NewNSQConfig creates a new NSQConfig 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) ResponseChan ¶ added in v0.6.0
func (r *RedisPubSub) ResponseChan() <-chan types.Response
ResponseChan returns the errors channel.
func (*RedisPubSub) StartReceiving ¶ added in v0.6.0
func (r *RedisPubSub) StartReceiving(msgs <-chan types.Message) error
StartReceiving 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 is configuration for the RedisPubSub output type.
func NewRedisPubSubConfig ¶ added in v0.6.0
func NewRedisPubSubConfig() RedisPubSubConfig
NewRedisPubSubConfig creates a new RedisPubSubConfig with default values.
type RoundRobinConfig ¶
type RoundRobinConfig struct {
Outputs []interface{} `json:"outputs" yaml:"outputs"`
}
RoundRobinConfig is configuration for the RoundRobin output type.
func NewRoundRobinConfig ¶
func NewRoundRobinConfig() RoundRobinConfig
NewRoundRobinConfig creates a new RoundRobinConfig with default values.
type STDOUTConfig ¶ added in v0.4.4
type STDOUTConfig struct {
CustomDelim string `json:"custom_delimiter" yaml:"custom_delimiter"`
}
STDOUTConfig is configuration values 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 ScaleProto ¶
type ScaleProto struct {
// contains filtered or unexported fields
}
ScaleProto is an output type that serves ScaleProto messages.
func (*ScaleProto) CloseAsync ¶
func (s *ScaleProto) CloseAsync()
CloseAsync shuts down the ScaleProto output and stops processing messages.
func (*ScaleProto) ResponseChan ¶
func (s *ScaleProto) ResponseChan() <-chan types.Response
ResponseChan returns the errors channel.
func (*ScaleProto) StartReceiving ¶
func (s *ScaleProto) StartReceiving(msgs <-chan types.Message) error
StartReceiving assigns a messages channel for the output to read.
func (*ScaleProto) WaitForClose ¶
func (s *ScaleProto) WaitForClose(timeout time.Duration) error
WaitForClose blocks until the ScaleProto output has closed down.
type ScaleProtoConfig ¶
type ScaleProtoConfig 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"` }
ScaleProtoConfig is configuration for the ScaleProto output type.
func NewScaleProtoConfig ¶
func NewScaleProtoConfig() ScaleProtoConfig
NewScaleProtoConfig creates a new ScaleProtoConfig with default values.
type Type ¶
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 ...pipeline.ConstructorFunc, ) (Type, error)
New creates an input type based on an input configuration.
func NewAmazonS3 ¶ added in v0.7.5
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 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 NewFanOut ¶
NewFanOut creates a new FanOut output type. Messages will be sent out to ALL outputs, outputs which block will apply backpressure upstream, meaning other outputs will also stop receiving messages.
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 input type.
func NewLineWriter ¶ added in v0.7.4
func NewLineWriter( handle io.WriteCloser, customDelimiter []byte, typeStr string, log log.Modular, stats metrics.Type, ) (Type, error)
NewLineWriter creates a new LineWriter 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 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 NewRoundRobin ¶
func NewRoundRobin(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
NewRoundRobin creates a new RoundRobin output type. Messages will be sent out to an output chosen by following their original order. If an output blocks this will block all throughput.
func NewScaleProto ¶
func NewScaleProto(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (Type, error)
NewScaleProto creates a new ScaleProto output type.
func WrapWithPipelines ¶ added in v0.3.0
func WrapWithPipelines(out Type, pipeConstructors ...pipeline.ConstructorFunc) (Type, error)
WrapWithPipelines wraps an output with a variadic number of pipelines.
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 pipeline.ConstructorFunc) (*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) ResponseChan ¶ added in v0.3.0
func (i *WithPipeline) ResponseChan() <-chan types.Response
ResponseChan returns the channel used for reading response messages from this output.
func (*WithPipeline) StartReceiving ¶ added in v0.3.0
func (i *WithPipeline) StartReceiving(msgChan <-chan types.Message) error
StartReceiving 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) ResponseChan ¶ added in v0.7.4
ResponseChan returns the errors channel.
func (*Writer) StartReceiving ¶ added in v0.7.4
StartReceiving assigns a messages channel for the output to read.
Source Files ¶
- amazon_s3.go
- amazon_sqs.go
- amqp.go
- broker_out_common.go
- constructor.go
- dynamic.go
- fan_out.go
- file.go
- files.go
- http_client.go
- http_server.go
- interface.go
- kafka.go
- line_writer.go
- nats.go
- nats_stream.go
- nsq.go
- package.go
- redis_list.go
- redis_pubsub.go
- round_robin.go
- scale_proto.go
- stdout.go
- wrap_with_pipeline.go
- writer.go
- zmq4_stub.go