output

package
v0.6.4 Latest Latest
Warning

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

Go to latest
Published: Jan 3, 2018 License: MIT Imports: 39 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 (
	// ErrBadReply is returned when a client gives a reply that is not the
	// expected success string.
	ErrBadReply = errors.New("bad reply from client")
)
View Source
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")
)
View Source
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.

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 Config

type Config struct {
	Type        string             `json:"type" yaml:"type"`
	HTTPClient  HTTPClientConfig   `json:"http_client" yaml:"http_client"`
	HTTPServer  HTTPServerConfig   `json:"http_server" yaml:"http_server"`
	ScaleProto  ScaleProtoConfig   `json:"scalability_protocols" yaml:"scalability_protocols"`
	Kafka       KafkaConfig        `json:"kafka" yaml:"kafka"`
	AMQP        AMQPConfig         `json:"amqp" yaml:"amqp"`
	NSQ         NSQConfig          `json:"nsq" yaml:"nsq"`
	NATS        NATSConfig         `json:"nats" yaml:"nats"`
	NATSStream  NATSStreamConfig   `json:"nats_stream" yaml:"nats_stream"`
	RedisPubSub RedisPubSubConfig  `json:"redis_pubsub" yaml:"redis_pubsub"`
	ZMQ4        *ZMQ4Config        `json:"zmq4,omitempty" yaml:"zmq4,omitempty"`
	File        FileConfig         `json:"file" yaml:"file"`
	STDOUT      STDOUTConfig       `json:"stdout" yaml:"stdout"`
	FanOut      FanOutConfig       `json:"fan_out" yaml:"fan_out"`
	RoundRobin  RoundRobinConfig   `json:"round_robin" yaml:"round_robin"`
	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.

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"`
	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"`
	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 Kafka

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

Kafka is an output type that writes messages into kafka.

func (*Kafka) CloseAsync

func (k *Kafka) CloseAsync()

CloseAsync shuts down the Kafka output and stops processing messages.

func (*Kafka) ResponseChan

func (k *Kafka) ResponseChan() <-chan types.Response

ResponseChan returns the errors channel.

func (*Kafka) StartReceiving

func (k *Kafka) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a messages channel for the output to read.

func (*Kafka) WaitForClose

func (k *Kafka) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the Kafka output has closed down.

type KafkaConfig

type KafkaConfig struct {
	Addresses   []string `json:"addresses" yaml:"addresses"`
	ClientID    string   `json:"client_id" yaml:"client_id"`
	Topic       string   `json:"topic" yaml:"topic"`
	TimeoutMS   int      `json:"timeout_ms" yaml:"timeout_ms"`
	AckReplicas bool     `json:"ack_replicas" yaml:"ack_replicas"`
}

KafkaConfig is configuration for the Kafka output type.

func NewKafkaConfig

func NewKafkaConfig() KafkaConfig

NewKafkaConfig creates a new KafkaConfig with default values.

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

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"`
	SuccessStr    string   `json:"reply_success" yaml:"reply_success"`
	PollTimeoutMS int      `json:"poll_timeout_ms" yaml:"poll_timeout_ms"`
	RepTimeoutMS  int      `json:"reply_timeout_ms" yaml:"reply_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,
	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, log log.Modular, stats metrics.Type) (Type, error)

NewAMQP creates a new AMQP output type.

func NewFanOut

func NewFanOut(conf Config, log log.Modular, stats metrics.Type) (Type, error)

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 NewFile

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

NewFile creates a new File output type.

func NewHTTPClient

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

NewHTTPClient creates a new HTTPClient output type.

func NewHTTPServer

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

NewHTTPServer creates a new HTTPServer input type.

func NewKafka

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

NewKafka creates a new Kafka output type.

func NewNATS

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

NewNATS creates a new NATS output type.

func NewNATSStream

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

NewNATSStream creates a new NATSStream output type.

func NewNSQ

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

NewNSQ creates a new NSQ output type.

func NewRedisPubSub

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

NewRedisPubSub creates a new RedisPubSub output type.

func NewRoundRobin

func NewRoundRobin(conf Config, 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 NewSTDOUT

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

NewSTDOUT creates a new STDOUT output type.

func NewScaleProto

func NewScaleProto(conf Config, log log.Modular, stats metrics.Type) (Type, error)

NewScaleProto creates a new ScaleProto 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 ZMQ4Config

type ZMQ4Config struct{}

ZMQ4Config empty stub for when ZMQ4 is not compiled.

func NewZMQ4Config

func NewZMQ4Config() *ZMQ4Config

NewZMQ4Config returns nil.

Jump to

Keyboard shortcuts

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