input

package
v0.41.1 Latest Latest
Warning

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

 Go to latest Published: Dec 13, 2018 License: MIT Imports: 35 Imported by: 8

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 an input consumer supports acknowledgements then the implementation of the input will wait for each message to reach a permanent destination before acknowledging it.

Index

Constants

View Source 
const (
	TypeAMQP          = "amqp"
	TypeBroker        = "broker"
	TypeDynamic       = "dynamic"
	TypeFile          = "file"
	TypeFiles         = "files"
	TypeGCPPubSub     = "gcp_pubsub"
	TypeHDFS          = "hdfs"
	TypeHTTPClient    = "http_client"
	TypeHTTPServer    = "http_server"
	TypeInproc        = "inproc"
	TypeKafka         = "kafka"
	TypeKafkaBalanced = "kafka_balanced"
	TypeKinesis       = "kinesis"
	TypeMQTT          = "mqtt"
	TypeNanomsg       = "nanomsg"
	TypeNATS          = "nats"
	TypeNATSStream    = "nats_stream"
	TypeNSQ           = "nsq"
	TypeReadUntil     = "read_until"
	TypeRedisList     = "redis_list"
	TypeRedisPubSub   = "redis_pubsub"
	TypeRedisStreams  = "redis_streams"
	TypeS3            = "s3"
	TypeSQS           = "sqs"
	TypeSTDIN         = "stdin"
	TypeWebsocket     = "websocket"
	TypeZMQ4          = "zmq4"
)

String constants representing each input type.

Variables

View Source 
var Constructors = map[string]TypeSpec{}

Constructors is a map of all input types with their specs.

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

Functions

func Descriptions 

func Descriptions() string

Descriptions returns a formatted string of descriptions for each type.

func DocumentPlugin added in v0.28.0 

func DocumentPlugin(
	typeString, description string,
	configSanitiser PluginConfigSanitiser,
)

DocumentPlugin adds a description and an optional configuration sanitiser function to the definition of a registered plugin. This improves the documentation generated by PluginDescriptions.

func PluginDescriptions added in v0.28.0 

func PluginDescriptions() string

PluginDescriptions generates and returns a markdown formatted document listing each registered plugin and an example configuration for it.

func RegisterPlugin added in v0.28.0 

func RegisterPlugin(
	typeString string,
	configConstructor PluginConfigConstructor,
	constructor PluginConstructor,
)

RegisterPlugin registers a plugin by a unique name so that it can be constucted similar to regular inputs. A constructor for both the plugin itself as well as its configuration struct must be provided.

WARNING: This API is experimental and could (is likely) to change.

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"`
	Inputs brokerInputList `json:"inputs" yaml:"inputs"`
}

BrokerConfig contains configuration fields for the Broker input 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          reader.AMQPConfig          `json:"amqp" yaml:"amqp"`
	Broker        BrokerConfig               `json:"broker" yaml:"broker"`
	Dynamic       DynamicConfig              `json:"dynamic" yaml:"dynamic"`
	File          FileConfig                 `json:"file" yaml:"file"`
	Files         reader.FilesConfig         `json:"files" yaml:"files"`
	GCPPubSub     reader.GCPPubSubConfig     `json:"gcp_pubsub" yaml:"gcp_pubsub"`
	HDFS          reader.HDFSConfig          `json:"hdfs" yaml:"hdfs"`
	HTTPClient    HTTPClientConfig           `json:"http_client" yaml:"http_client"`
	HTTPServer    HTTPServerConfig           `json:"http_server" yaml:"http_server"`
	Inproc        InprocConfig               `json:"inproc" yaml:"inproc"`
	Kafka         reader.KafkaConfig         `json:"kafka" yaml:"kafka"`
	KafkaBalanced reader.KafkaBalancedConfig `json:"kafka_balanced" yaml:"kafka_balanced"`
	Kinesis       reader.KinesisConfig       `json:"kinesis" yaml:"kinesis"`
	MQTT          reader.MQTTConfig          `json:"mqtt" yaml:"mqtt"`
	Nanomsg       reader.ScaleProtoConfig    `json:"nanomsg" yaml:"nanomsg"`
	NATS          reader.NATSConfig          `json:"nats" yaml:"nats"`
	NATSStream    reader.NATSStreamConfig    `json:"nats_stream" yaml:"nats_stream"`
	NSQ           reader.NSQConfig           `json:"nsq" yaml:"nsq"`
	Plugin        interface{}                `json:"plugin,omitempty" yaml:"plugin,omitempty"`
	ReadUntil     ReadUntilConfig            `json:"read_until" yaml:"read_until"`
	RedisList     reader.RedisListConfig     `json:"redis_list" yaml:"redis_list"`
	RedisPubSub   reader.RedisPubSubConfig   `json:"redis_pubsub" yaml:"redis_pubsub"`
	RedisStreams  reader.RedisStreamsConfig  `json:"redis_streams" yaml:"redis_streams"`
	S3            reader.AmazonS3Config      `json:"s3" yaml:"s3"`
	SQS           reader.AmazonSQSConfig     `json:"sqs" yaml:"sqs"`
	STDIN         STDINConfig                `json:"stdin" yaml:"stdin"`
	Websocket     reader.WebsocketConfig     `json:"websocket" yaml:"websocket"`
	ZMQ4          *reader.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.

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 {
	Inputs    map[string]Config `json:"inputs" yaml:"inputs"`
	Prefix    string            `json:"prefix" yaml:"prefix"`
	TimeoutMS int               `json:"timeout_ms" yaml:"timeout_ms"`
}

DynamicConfig contains 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 FileConfig 

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

FileConfig contains configuration values for the File input type.

func NewFileConfig 

func NewFileConfig() FileConfig

NewFileConfig creates a new FileConfig with default values.

type HTTPClient added in v0.7.0 

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

HTTPClient is an input type that continuously makes HTTP requests and reads the response bodies as message payloads.

func (*HTTPClient) CloseAsync added in v0.7.0 

func (h *HTTPClient) CloseAsync()

CloseAsync shuts down the HTTPClient input.

func (*HTTPClient) Connected added in v0.39.0 

func (h *HTTPClient) Connected() bool

Connected returns a boolean indicating whether this input is currently connected to its target.

func (*HTTPClient) TransactionChan added in v0.9.0 

func (h *HTTPClient) TransactionChan() <-chan types.Transaction

TransactionChan returns a transactions channel for consuming messages from this input type.

func (*HTTPClient) WaitForClose added in v0.7.0 

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

WaitForClose blocks until the HTTPClient input has closed down.

type HTTPClientConfig added in v0.7.0 

type HTTPClientConfig struct {
	client.Config `json:",inline" yaml:",inline"`
	Payload       string       `json:"payload" yaml:"payload"`
	Stream        StreamConfig `json:"stream" yaml:"stream"`
}

HTTPClientConfig contains configuration for the HTTPClient output type.

func NewHTTPClientConfig added in v0.7.0 

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 registers a range of HTTP endpoints where requests can send messages through Benthos. The endpoints are registered on the general Benthos HTTP server by default. It is also possible to specify a custom address to bind a new server to which the endpoints will be registered on instead.

func (*HTTPServer) CloseAsync 

func (h *HTTPServer) CloseAsync()

CloseAsync shuts down the HTTPServer input and stops processing requests.

func (*HTTPServer) Connected added in v0.39.0 

func (h *HTTPServer) Connected() bool

Connected returns a boolean indicating whether this input is currently connected to its target.

func (*HTTPServer) TransactionChan added in v0.9.0 

func (h *HTTPServer) TransactionChan() <-chan types.Transaction

TransactionChan returns a transactions channel for consuming messages from this input.

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"`
	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 for the HTTPServer input 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 input type that reads from a named pipe, which could be the output of a separate Benthos stream of the same process.

func (*Inproc) CloseAsync added in v0.15.4 

func (i *Inproc) CloseAsync()

CloseAsync shuts down the Inproc input and stops processing requests.

func (*Inproc) Connected added in v0.39.0 

func (i *Inproc) Connected() bool

Connected returns a boolean indicating whether this input is currently connected to its target.

func (*Inproc) TransactionChan added in v0.15.4 

func (i *Inproc) TransactionChan() <-chan types.Transaction

TransactionChan returns a transactions channel for consuming messages from this input type.

func (*Inproc) WaitForClose added in v0.15.4 

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

WaitForClose blocks until the Inproc input has closed down.

type InprocConfig added in v0.15.4 

type InprocConfig string

InprocConfig is a configuration type for the inproc input.

func NewInprocConfig added in v0.15.4 

func NewInprocConfig() InprocConfig

NewInprocConfig creates a new inproc input config.

type PluginConfigConstructor added in v0.28.0 

type PluginConfigConstructor func() interface{}

PluginConfigConstructor is a func that returns a pointer to a new and fully populated configuration struct for a plugin type. It is valid to return a pointer to an empty struct (&struct{}{}) if no configuration fields are needed.

type PluginConfigSanitiser added in v0.28.0 

type PluginConfigSanitiser func(conf interface{}) interface{}

PluginConfigSanitiser is a function that takes a configuration object for a plugin and returns a sanitised (minimal) version of it for printing in examples and plugin documentation.

This function is useful for when a plugins configuration struct is very large and complex, but can sometimes be expressed in a more concise way without losing the original intent.

type PluginConstructor added in v0.28.0 

type PluginConstructor func(
	config interface{},
	manager types.Manager,
	logger log.Modular,
	metrics metrics.Type,
) (types.Input, error)

PluginConstructor is a func that constructs a Benthos input plugin. These are plugins that are specific to certain use cases, experimental, private or otherwise unfit for widespread general use. Any number of plugins can be specified when using Benthos as a framework.

The configuration object will be the result of the PluginConfigConstructor after overlaying the user configuration.

type ReadUntil added in v0.11.0 

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

ReadUntil is an input type that continuously reads another input type until a condition returns true on a message consumed.

func (*ReadUntil) CloseAsync added in v0.11.0 

func (r *ReadUntil) CloseAsync()

CloseAsync shuts down the ReadUntil input and stops processing requests.

func (*ReadUntil) Connected added in v0.39.0 

func (r *ReadUntil) Connected() bool

Connected returns a boolean indicating whether this input is currently connected to its target.

func (*ReadUntil) TransactionChan added in v0.11.0 

func (r *ReadUntil) TransactionChan() <-chan types.Transaction

TransactionChan returns a transactions channel for consuming messages from this input type.

func (*ReadUntil) WaitForClose added in v0.11.0 

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

WaitForClose blocks until the ReadUntil input has closed down.

type ReadUntilConfig added in v0.11.0 

type ReadUntilConfig struct {
	Input     *Config          `json:"input" yaml:"input"`
	Restart   bool             `json:"restart_input" yaml:"restart_input"`
	Condition condition.Config `json:"condition" yaml:"condition"`
}

ReadUntilConfig contains configuration values for the ReadUntil input type.

func NewReadUntilConfig added in v0.11.0 

func NewReadUntilConfig() ReadUntilConfig

NewReadUntilConfig creates a new ReadUntilConfig with default values.

func (ReadUntilConfig) MarshalJSON added in v0.11.0 

func (r ReadUntilConfig) MarshalJSON() ([]byte, error)

MarshalJSON prints an empty object instead of nil.

func (ReadUntilConfig) MarshalYAML added in v0.11.0 

func (r ReadUntilConfig) MarshalYAML() (interface{}, error)

MarshalYAML prints an empty object instead of nil.

type Reader added in v0.7.4 

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

Reader is an input implementation that reads messages from a reader.Type.

func (*Reader) CloseAsync added in v0.7.4 

func (r *Reader) CloseAsync()

CloseAsync shuts down the Reader input and stops processing requests.

func (*Reader) Connected added in v0.39.0 

func (r *Reader) Connected() bool

Connected returns a boolean indicating whether this input is currently connected to its target.

func (*Reader) TransactionChan added in v0.9.0 

func (r *Reader) TransactionChan() <-chan types.Transaction

TransactionChan returns a transactions channel for consuming messages from this input type.

func (*Reader) WaitForClose added in v0.7.4 

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

WaitForClose blocks until the Reader input has closed down.

type STDINConfig added in v0.0.2 

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

STDINConfig contains config fields for the STDIN input type.

func NewSTDINConfig added in v0.0.2 

func NewSTDINConfig() STDINConfig

NewSTDINConfig creates a STDINConfig populated with default values.

type StreamConfig added in v0.9.1 

type StreamConfig struct {
	Enabled   bool   `json:"enabled" yaml:"enabled"`
	Reconnect bool   `json:"reconnect" yaml:"reconnect"`
	Multipart bool   `json:"multipart" yaml:"multipart"`
	MaxBuffer int    `json:"max_buffer" yaml:"max_buffer"`
	Delim     string `json:"delimiter" yaml:"delimiter"`
}

StreamConfig contains fields for specifying consumption behaviour when the body of a request is a constant stream of bytes.

type Type 

type Type interface {
	types.Closable
	types.Producer

	// Connected returns a boolean indicating whether this input is currently
	// connected to its target.
	Connected() bool
}

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

func NewAmazonS3 added in v0.7.4 

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

NewAmazonS3 creates a new AWS S3 input type.

func NewAmazonSQS added in v0.7.5 

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

NewAmazonSQS creates a new AWS SQS input 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 input type.

func NewDynamic added in v0.8.0 

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

NewDynamic creates a new Dynamic 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 NewFiles added in v0.13.4 

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

NewFiles creates a new Files input type.

func NewGCPPubSub added in v0.33.0 

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

NewGCPPubSub creates a new GCP Cloud Pub/Sub input type.

func NewHDFS added in v0.30.0 

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

NewHDFS creates a new Files input type.

func NewHTTPClient added in v0.7.0 

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

NewHTTPClient creates a new HTTPClient input 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 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 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 added in v0.4.1 

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

NewKafkaBalanced creates a new KafkaBalanced input type.

func NewKinesis added in v0.26.0 

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

NewKinesis creates a new AWS Kinesis input 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 input 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 input type.

func NewNATSStream added in v0.4.10 

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

NewNATSStream creates a new NATSStream input 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 input 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 input type.

func NewReadUntil added in v0.11.0 

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

NewReadUntil creates a new ReadUntil input type.

func NewReader added in v0.7.4 

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

NewReader creates a new Reader input 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 Redis List input 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 input type.

func NewRedisStreams added in v0.26.1 

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

NewRedisStreams creates a new Redis List 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 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 input type.

func WrapWithPipelines added in v0.2.9 

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

WrapWithPipelines wraps an input with a variadic number of pipelines.

type TypeSpec added in v0.9.1 

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

TypeSpec is a struct containing constructors, markdown descriptions and an optional sanitisation function for each input type.

type WithPipeline added in v0.2.9 

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

func WrapWithPipeline(procs *int, in Type, pipeConstructor types.PipelineConstructorFunc) (*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 added in v0.2.9 

func (i *WithPipeline) CloseAsync()

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

func (*WithPipeline) Connected added in v0.39.0 

func (i *WithPipeline) Connected() bool

Connected returns a boolean indicating whether this input is currently connected to its target.

func (*WithPipeline) TransactionChan added in v0.9.0 

func (i *WithPipeline) TransactionChan() <-chan types.Transaction

TransactionChan returns the channel used for consuming transactions from this input.

func (*WithPipeline) WaitForClose added in v0.2.9 

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.

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.