output

package
v0.8.8 Latest Latest
Warning

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

 Go to latest Published: Feb 27, 2018 License: MIT Imports: 43 Imported by: 0

Documentation

Overview

Package output defines dispatchers of data for a variety of targets. All output types must implement interface output.Type.

Index

Constants

This section is empty.

Variables

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 

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 AMQP 

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

AMQP is an output type that serves AMQP messages.

func (*AMQP) CloseAsync 

func (a *AMQP) CloseAsync()

CloseAsync shuts down the AMQP output and stops processing messages.

func (*AMQP) ResponseChan 

func (a *AMQP) ResponseChan() <-chan types.Response

ResponseChan returns the errors channel.

func (*AMQP) StartReceiving 

func (a *AMQP) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a messages channel for the output to read.

func (*AMQP) WaitForClose 

func (a *AMQP) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the AMQP output has closed down.

type AMQPConfig 

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 

func NewAMQPConfig() AMQPConfig

NewAMQPConfig creates a new AMQPConfig with default values.

type BrokerConfig 

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

BrokerConfig is configuration 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"`
	AmazonS3    writer.AmazonS3Config  `json:"amazon_s3" yaml:"amazon_s3"`
	AmazonSQS   writer.AmazonSQSConfig `json:"amazon_sqs" yaml:"amazon_sqs"`
	AMQP        AMQPConfig             `json:"amqp" yaml:"amqp"`
	Broker      BrokerConfig           `json:"broker" yaml:"broker"`
	Dynamic     DynamicConfig          `json:"dynamic" yaml:"dynamic"`
	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"`
	MQTT        writer.MQTTConfig      `json:"mqtt" yaml:"mqtt"`
	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"`
	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 

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 

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 

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 

func NewDynamicConfig() DynamicConfig

NewDynamicConfig creates a new DynamicConfig 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 

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) ResponseChan 

func (w *LineWriter) ResponseChan() <-chan types.Response

ResponseChan returns the errors channel.

func (*LineWriter) StartReceiving 

func (w *LineWriter) StartReceiving(msgs <-chan types.Message) error

StartReceiving 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 NATS 

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

NATS is an output type that serves NATS messages.

func (*NATS) CloseAsync 

func (n *NATS) CloseAsync()

CloseAsync shuts down the NATS output and stops processing messages.

func (*NATS) ResponseChan 

func (n *NATS) ResponseChan() <-chan types.Response

ResponseChan returns the errors channel.

func (*NATS) StartReceiving 

func (n *NATS) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a messages channel for the output to read.

func (*NATS) WaitForClose 

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

WaitForClose blocks until the NATS output has closed down.

type NATSConfig 

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 

func NewNATSConfig() NATSConfig

NewNATSConfig creates a new NATSConfig with default values.

type NATSStream 

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

NATSStream is an output type that serves NATS messages.

func (*NATSStream) CloseAsync 

func (n *NATSStream) CloseAsync()

CloseAsync shuts down the NATSStream output and stops processing messages.

func (*NATSStream) ResponseChan 

func (n *NATSStream) ResponseChan() <-chan types.Response

ResponseChan returns the errors channel.

func (*NATSStream) StartReceiving 

func (n *NATSStream) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a messages channel for the output to read.

func (*NATSStream) WaitForClose 

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

WaitForClose blocks until the NATSStream output has closed down.

type NATSStreamConfig 

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 

func NewNATSStreamConfig() NATSStreamConfig

NewNATSStreamConfig creates a new NATSStreamConfig with default values.

type NSQ 

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

NSQ is an output type that serves NSQ messages.

func (*NSQ) CloseAsync 

func (n *NSQ) CloseAsync()

CloseAsync shuts down the NSQ output and stops processing messages.

func (*NSQ) ResponseChan 

func (n *NSQ) ResponseChan() <-chan types.Response

ResponseChan returns the errors channel.

func (*NSQ) StartReceiving 

func (n *NSQ) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a messages channel for the output to read.

func (*NSQ) WaitForClose 

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

WaitForClose blocks until the NSQ output has closed down.

type NSQConfig 

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 

func NewNSQConfig() NSQConfig

NewNSQConfig creates a new NSQConfig with default values.

type RedisPubSub 

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

RedisPubSub is an output type that serves RedisPubSub messages.

func (*RedisPubSub) CloseAsync 

func (r *RedisPubSub) CloseAsync()

CloseAsync shuts down the RedisPubSub output and stops processing messages.

func (*RedisPubSub) ResponseChan 

func (r *RedisPubSub) ResponseChan() <-chan types.Response

ResponseChan returns the errors channel.

func (*RedisPubSub) StartReceiving 

func (r *RedisPubSub) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a messages channel for the output to read.

func (*RedisPubSub) WaitForClose 

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

WaitForClose blocks until the RedisPubSub output has closed down.

type RedisPubSubConfig 

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 

func NewRedisPubSubConfig() RedisPubSubConfig

NewRedisPubSubConfig creates a new RedisPubSubConfig with default values.

type STDOUTConfig 

type STDOUTConfig struct {
	CustomDelim string `json:"custom_delimiter" yaml:"custom_delimiter"`
}

STDOUTConfig is configuration values for the stdout based output type.

func NewSTDOUTConfig 

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 Type interface {
	types.Closable
	types.Consumer
}

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 ...pipeline.ConstructorFunc,
) (Type, error)

New creates an input type based on an input 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.

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 NewAmazonSQS 

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

NewAmazonSQS creates a new AmazonSQS output type.

func NewBroker 

func NewBroker(conf Config, mgr types.Manager, log log.Modular, stats metrics.Type) (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 

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

NewDynamic creates a new Dynamic 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 File 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 input 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 

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 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 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 NewSTDOUT 

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

NewSTDOUT creates a new STDOUT output type.

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 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 ...pipeline.ConstructorFunc) (Type, error)

WrapWithPipelines wraps an output with a variadic number of pipelines.

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(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 

func (i *WithPipeline) CloseAsync()

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

func (*WithPipeline) ResponseChan 

func (i *WithPipeline) ResponseChan() <-chan types.Response

ResponseChan returns the channel used for reading response messages from this output.

func (*WithPipeline) StartReceiving 

func (i *WithPipeline) StartReceiving(msgChan <-chan types.Message) error

StartReceiving 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) ResponseChan 

func (w *Writer) ResponseChan() <-chan types.Response

ResponseChan returns the errors channel.

func (*Writer) StartReceiving 

func (w *Writer) StartReceiving(msgs <-chan types.Message) error

StartReceiving 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.

type ZMQ4Config 

type ZMQ4Config struct{}

ZMQ4Config empty stub for when ZMQ4 is not compiled.

func NewZMQ4Config 

func NewZMQ4Config() *ZMQ4Config

NewZMQ4Config returns nil.

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.