input

package
v0.8.5 Latest Latest
Warning

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

 Go to latest Published: Feb 22, 2018 License: MIT Imports: 31 Imported by: 0

Documentation

Overview

Package input defines consumers for aggregating data from a variety of sources. All consumer types must implement interface input.Type.

If the source of a consumer is given assurances of message receipt then the consumer must ensure that all messages read from a source are propagated to a reader before shutting down gracefully. For example, a ZMQ subscriber model based consumer would not need to provide such ensurance, a HTTP POST based consumer, however, would be expected to propagate any requests where a 200 OK message has been returned.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrFanInNoInputs is returned when creating a FanIn type with zero inputs.
	ErrFanInNoInputs = errors.New("attempting to create fan_in input type with no inputs")
)

Functions

func Descriptions 

func Descriptions() string

Descriptions returns a formatted string of descriptions for each type.

func OptLineReaderSetDelimiter 

func OptLineReaderSetDelimiter(delimiter string) func(r *LineReader)

OptLineReaderSetDelimiter is a option func that sets the delimiter (default '\n') used to divide lines (message parts) in the stream of data.

func OptLineReaderSetMaxBuffer 

func OptLineReaderSetMaxBuffer(maxBuffer int) func(r *LineReader)

OptLineReaderSetMaxBuffer is a option func that sets the maximum size of the line parsing buffers.

func OptLineReaderSetMultipart 

func OptLineReaderSetMultipart(multipart bool) func(r *LineReader)

OptLineReaderSetMultipart is a option func that sets the boolean flag indicating whether lines should be parsed as multipart or not.

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 Config 

type Config struct {
	Type          string                     `json:"type" yaml:"type"`
	AmazonS3      reader.AmazonS3Config      `json:"amazon_s3" yaml:"amazon_s3"`
	AmazonSQS     reader.AmazonSQSConfig     `json:"amazon_sqs" yaml:"amazon_sqs"`
	AMQP          reader.AMQPConfig          `json:"amqp" yaml:"amqp"`
	Dynamic       DynamicConfig              `json:"dynamic" yaml:"dynamic"`
	FanIn         FanInConfig                `json:"fan_in" yaml:"fan_in"`
	File          FileConfig                 `json:"file" yaml:"file"`
	HTTPClient    HTTPClientConfig           `json:"http_client" yaml:"http_client"`
	HTTPServer    HTTPServerConfig           `json:"http_server" yaml:"http_server"`
	Kafka         reader.KafkaConfig         `json:"kafka" yaml:"kafka"`
	KafkaBalanced reader.KafkaBalancedConfig `json:"kafka_balanced" yaml:"kafka_balanced"`
	NATS          reader.NATSConfig          `json:"nats" yaml:"nats"`
	NATSStream    reader.NATSStreamConfig    `json:"nats_stream" yaml:"nats_stream"`
	NSQ           reader.NSQConfig           `json:"nsq" yaml:"nsq"`
	RedisList     reader.RedisListConfig     `json:"redis_list" yaml:"redis_list"`
	RedisPubSub   reader.RedisPubSubConfig   `json:"redis_pubsub" yaml:"redis_pubsub"`
	ScaleProto    reader.ScaleProtoConfig    `json:"scalability_protocols" yaml:"scalability_protocols"`
	STDIN         STDINConfig                `json:"stdin" yaml:"stdin"`
	ZMQ4          *ZMQ4Config                `json:"zmq4,omitempty" yaml:"zmq4,omitempty"`
	Processors    []processor.Config         `json:"processors" yaml:"processors"`
}

Config is the all encompassing configuration struct for all input 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 {
	Inputs    map[string]Config `json:"inputs" yaml:"inputs"`
	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 FanInConfig 

type FanInConfig struct {
	Inputs []interface{} `json:"inputs" yaml:"inputs"`
}

FanInConfig is configuration for the FanIn input type.

func NewFanInConfig 

func NewFanInConfig() FanInConfig

NewFanInConfig creates a new FanInConfig with default values.

type FileConfig 

type FileConfig struct {
	Path        string `json:"path" yaml:"path"`
	Multipart   bool   `json:"multipart" yaml:"multipart"`
	MaxBuffer   int    `json:"max_buffer" yaml:"max_buffer"`
	CustomDelim string `json:"custom_delimiter" yaml:"custom_delimiter"`
}

FileConfig is configuration values for the File input 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) MessageChan 

func (h *HTTPClient) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*HTTPClient) StartListening 

func (h *HTTPClient) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

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"`
	Payload           string             `json:"payload" yaml:"payload"`
	ContentType       string             `json:"content_type" yaml:"content_type"`
	Stream            bool               `json:"stream" yaml:"stream"`
	StreamMultipart   bool               `json:"stream_multipart" yaml:"stream_multipart"`
	StreamMaxBuffer   int                `json:"stream_max_buffer" yaml:"stream_max_buffer"`
	StreamCustomDelim string             `json:"stream_custom_delimiter" yaml:"stream_custom_delimiter"`
	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) MessageChan 

func (h *HTTPServer) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*HTTPServer) StartListening 

func (h *HTTPServer) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

func (*HTTPServer) WaitForClose 

func (h *HTTPServer) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the HTTPServer input 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 LineReader 

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

LineReader is an input implementation that continuously reads line delimited messages from an io.Reader type.

func (*LineReader) CloseAsync 

func (r *LineReader) CloseAsync()

CloseAsync shuts down the reader input and stops processing requests.

func (*LineReader) MessageChan 

func (r *LineReader) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*LineReader) StartListening 

func (r *LineReader) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

func (*LineReader) WaitForClose 

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

WaitForClose blocks until the reader input has closed down.

type Reader 

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

Reader is an input type that reads from a Reader instance.

func (*Reader) CloseAsync 

func (r *Reader) CloseAsync()

CloseAsync shuts down the Reader input and stops processing requests.

func (*Reader) MessageChan 

func (r *Reader) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*Reader) StartListening 

func (r *Reader) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

func (*Reader) WaitForClose 

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

WaitForClose blocks until the Reader input has closed down.

type STDINConfig 

type STDINConfig struct {
	Multipart   bool   `json:"multipart" yaml:"multipart"`
	MaxBuffer   int    `json:"max_buffer" yaml:"max_buffer"`
	CustomDelim string `json:"custom_delimiter" yaml:"custom_delimiter"`
}

STDINConfig contains config fields for the STDIN input type.

func NewSTDINConfig 

func NewSTDINConfig() STDINConfig

NewSTDINConfig creates a STDINConfig populated with default values.

type Type 

type Type interface {
	types.Closable
	types.Producer
}

Type is the standard interface of an input 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 input type.

func NewAmazonS3 

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

NewAmazonS3 creates a new amazon S3 input type.

func NewAmazonSQS 

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

NewAmazonSQS creates a new Amazon SQS input type.

func NewDynamic 

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

NewDynamic creates a new Dynamic input type.

func NewFanIn 

func NewFanIn(
	conf Config,
	mgr types.Manager,
	log log.Modular,
	stats metrics.Type,
	pipelines ...pipeline.ConstructorFunc,
) (Type, error)

NewFanIn creates a new FanIn input type.

func NewFile 

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

NewFile creates a new File input 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 input type.

func NewKafkaBalanced 

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

NewKafkaBalanced creates a new KafkaBalanced input type.

func NewLineReader 

func NewLineReader(
	typeStr string,
	handleCtor func() (io.Reader, error),
	onClose func(),
	log log.Modular,
	stats metrics.Type,
	options ...func(r *LineReader),
) (Type, error)

NewLineReader creates a new reader input type.

Callers must provide a constructor function for the target io.Reader, which is called on start up and again each time a reader is exhausted. If the constructor is called but there is no more content to create a Reader for then the error `io.EOF` should be returned and the LineReader will close.

Callers must also provide an onClose function, which will be called if the LineReader has been instructed to shut down. This function should unblock any blocked Read calls.

func NewNATS 

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

NewNATS creates a new NATS input type.

func NewNATSStream 

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

NewNATSStream creates a new NATSStream input type.

func NewNSQ 

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

NewNSQ create a new NSQ input type.

func NewReader 

func NewReader(
	typeStr string,
	r reader.Type,
	log log.Modular,
	stats metrics.Type,
) (Type, error)

NewReader creates a new Reader input type.

func NewRedisList 

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

NewRedisList creates a new Redis List input type.

func NewRedisPubSub 

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

NewRedisPubSub creates a new RedisPubSub input type.

func NewSTDIN 

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

NewSTDIN creates a new STDIN input type.

func NewScaleProto 

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

NewScaleProto creates a new ScaleProto input type.

func WrapWithPipelines 

func WrapWithPipelines(in Type, pipeConstructors ...pipeline.ConstructorFunc) (Type, error)

WrapWithPipelines wraps an input with a variadic number of pipelines.

type WithPipeline 

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

WithPipeline is a type that wraps both an input type and a pipeline type by routing the input through the pipeline, and implements the input.Type interface in order to act like an ordinary input.

func WrapWithPipeline 

func WrapWithPipeline(in Type, pipeConstructor pipeline.ConstructorFunc) (*WithPipeline, error)

WrapWithPipeline routes an input directly into a processing pipeline and returns a type that manages both and acts like an ordinary input.

func (*WithPipeline) CloseAsync 

func (i *WithPipeline) CloseAsync()

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

func (*WithPipeline) MessageChan 

func (i *WithPipeline) MessageChan() <-chan types.Message

MessageChan returns the channel used for consuming messages from this input.

func (*WithPipeline) StartListening 

func (i *WithPipeline) StartListening(resChan <-chan types.Response) error

StartListening starts the type listening to a response channel from a consumer.

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 is an 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 reader defines implementations of an interface for generic message reading from various third party sources.
 Package reader defines implementations of an interface for generic message reading from various third party sources.

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.