input

package
v3.65.3 Latest Latest
Warning

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

Go to latest
Published: Jul 27, 2022 License: MIT Imports: 96 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 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"
	TypeAMQP09            = "amqp_0_9"
	TypeAMQP1             = "amqp_1"
	TypeAWSKinesis        = "aws_kinesis"
	TypeAWSS3             = "aws_s3"
	TypeAWSSQS            = "aws_sqs"
	TypeAzureBlobStorage  = "azure_blob_storage"
	TypeAzureQueueStorage = "azure_queue_storage"
	TypeBloblang          = "bloblang"
	TypeBroker            = "broker"
	TypeCSVFile           = "csv"
	TypeDynamic           = "dynamic"
	TypeFile              = "file"
	TypeFiles             = "files"
	TypeGCPCloudStorage   = "gcp_cloud_storage"
	TypeGCPPubSub         = "gcp_pubsub"
	TypeGenerate          = "generate"
	TypeHDFS              = "hdfs"
	TypeHTTPClient        = "http_client"
	TypeHTTPServer        = "http_server"
	TypeInproc            = "inproc"
	TypeKafka             = "kafka"
	TypeKafkaBalanced     = "kafka_balanced"
	TypeKinesis           = "kinesis"
	TypeKinesisBalanced   = "kinesis_balanced"
	TypeMQTT              = "mqtt"
	TypeNanomsg           = "nanomsg"
	TypeNATS              = "nats"
	TypeNATSJetStream     = "nats_jetstream"
	TypeNATSStream        = "nats_stream"
	TypeNSQ               = "nsq"
	TypePulsar            = "pulsar"
	TypeReadUntil         = "read_until"
	TypeRedisList         = "redis_list"
	TypeRedisPubSub       = "redis_pubsub"
	TypeRedisStreams      = "redis_streams"
	TypeResource          = "resource"
	TypeS3                = "s3"
	TypeSequence          = "sequence"
	TypeSFTP              = "sftp"
	TypeSocket            = "socket"
	TypeSocketServer      = "socket_server"
	TypeSQS               = "sqs"
	TypeSTDIN             = "stdin"
	TypeSubprocess        = "subprocess"
	TypeTCP               = "tcp"
	TypeTCPServer         = "tcp_server"
	TypeUDPServer         = "udp_server"
	TypeWebsocket         = "websocket"
	TypeZMQ4              = "zmq4"
)

String constants representing each input type. Deprecated: Do not add new components here. Instead, use the public plugin APIs. Examples can be found in: ./internal/impl

View Source
const (
	// ErrCodeKMSThrottlingException is defined in the API Reference
	// https://docs.aws.amazon.com/sdk-for-go/api/service/kinesis/#Kinesis.GetRecords
	ErrCodeKMSThrottlingException = "KMSThrottlingException"
)

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")
)
View Source
var (
	ErrLeaseNotAcquired = errors.New("the shard could not be leased due to a collision")
)

Common errors that might occur throughout checkpointing.

Functions

func AppendProcessorsFromConfig

func AppendProcessorsFromConfig(
	conf Config,
	mgr types.Manager,
	log log.Modular,
	stats metrics.Type,
	pipelines ...types.PipelineConstructorFunc,
) []types.PipelineConstructorFunc

AppendProcessorsFromConfig takes a variant arg of pipeline constructor functions and returns a new slice of them where the processors of the provided input configuration will also be initialized.

func Descriptions

func Descriptions() string

Descriptions returns a formatted string of descriptions for each type.

func DocumentPlugin

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 PluginCount

func PluginCount() int

PluginCount returns the number of registered plugins. This does NOT count the standard set of components.

func PluginDescriptions

func PluginDescriptions() string

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

func RegisterPlugin

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

RegisterPlugin registers a plugin by a unique name so that it can be constructed similar to regular inputs. If configuration is not needed for this plugin then configConstructor can be nil. A constructor for the plugin itself must be provided.

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.

func WalkConstructors

func WalkConstructors(fn func(ConstructorFunc, docs.ComponentSpec))

WalkConstructors iterates each component constructor.

Types

type AWSKinesisConfig

type AWSKinesisConfig struct {
	session.Config  `json:",inline" yaml:",inline"`
	Streams         []string                 `json:"streams" yaml:"streams"`
	DynamoDB        DynamoDBCheckpointConfig `json:"dynamodb" yaml:"dynamodb"`
	CheckpointLimit int                      `json:"checkpoint_limit" yaml:"checkpoint_limit"`
	CommitPeriod    string                   `json:"commit_period" yaml:"commit_period"`
	LeasePeriod     string                   `json:"lease_period" yaml:"lease_period"`
	RebalancePeriod string                   `json:"rebalance_period" yaml:"rebalance_period"`
	StartFromOldest bool                     `json:"start_from_oldest" yaml:"start_from_oldest"`
	Batching        batch.PolicyConfig       `json:"batching" yaml:"batching"`
}

AWSKinesisConfig is configuration values for the input type.

func NewAWSKinesisConfig

func NewAWSKinesisConfig() AWSKinesisConfig

NewAWSKinesisConfig creates a new Config with default values.

type AWSS3Config

type AWSS3Config struct {
	sess.Config        `json:",inline" yaml:",inline"`
	Bucket             string         `json:"bucket" yaml:"bucket"`
	Codec              string         `json:"codec" yaml:"codec"`
	Prefix             string         `json:"prefix" yaml:"prefix"`
	ForcePathStyleURLs bool           `json:"force_path_style_urls" yaml:"force_path_style_urls"`
	DeleteObjects      bool           `json:"delete_objects" yaml:"delete_objects"`
	SQS                AWSS3SQSConfig `json:"sqs" yaml:"sqs"`
}

AWSS3Config contains configuration values for the aws_s3 input type.

func NewAWSS3Config

func NewAWSS3Config() AWSS3Config

NewAWSS3Config creates a new AWSS3Config with default values.

type AWSS3SQSConfig

type AWSS3SQSConfig struct {
	URL          string `json:"url" yaml:"url"`
	Endpoint     string `json:"endpoint" yaml:"endpoint"`
	EnvelopePath string `json:"envelope_path" yaml:"envelope_path"`
	KeyPath      string `json:"key_path" yaml:"key_path"`
	BucketPath   string `json:"bucket_path" yaml:"bucket_path"`
	DelayPeriod  string `json:"delay_period" yaml:"delay_period"`
	MaxMessages  int64  `json:"max_messages" yaml:"max_messages"`
}

AWSS3SQSConfig contains configuration for hooking up the S3 input with an SQS queue.

func NewAWSS3SQSConfig

func NewAWSS3SQSConfig() AWSS3SQSConfig

NewAWSS3SQSConfig creates a new AWSS3SQSConfig with default values.

type AWSSQSConfig

type AWSSQSConfig struct {
	sess.Config     `json:",inline" yaml:",inline"`
	URL             string `json:"url" yaml:"url"`
	DeleteMessage   bool   `json:"delete_message" yaml:"delete_message"`
	ResetVisibility bool   `json:"reset_visibility" yaml:"reset_visibility"`
}

AWSSQSConfig contains configuration values for the input type.

func NewAWSSQSConfig

func NewAWSSQSConfig() AWSSQSConfig

NewAWSSQSConfig creates a new Config with default values.

type AsyncReader

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

AsyncReader is an input implementation that reads messages from a reader.Async component.

func (*AsyncReader) CloseAsync

func (r *AsyncReader) CloseAsync()

CloseAsync shuts down the AsyncReader input and stops processing requests.

func (*AsyncReader) Connected

func (r *AsyncReader) Connected() bool

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

func (*AsyncReader) TransactionChan

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

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

func (*AsyncReader) WaitForClose

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

WaitForClose blocks until the AsyncReader input has closed down.

type AzureBlobStorageConfig

type AzureBlobStorageConfig struct {
	StorageAccount          string `json:"storage_account" yaml:"storage_account"`
	StorageAccessKey        string `json:"storage_access_key" yaml:"storage_access_key"`
	StorageSASToken         string `json:"storage_sas_token" yaml:"storage_sas_token"`
	StorageConnectionString string `json:"storage_connection_string" yaml:"storage_connection_string"`
	Container               string `json:"container" yaml:"container"`
	Prefix                  string `json:"prefix" yaml:"prefix"`
	Codec                   string `json:"codec" yaml:"codec"`
	DeleteObjects           bool   `json:"delete_objects" yaml:"delete_objects"`
}

AzureBlobStorageConfig contains configuration fields for the AzureBlobStorage input type.

func NewAzureBlobStorageConfig

func NewAzureBlobStorageConfig() AzureBlobStorageConfig

NewAzureBlobStorageConfig creates a new AzureBlobStorageConfig with default values.

type AzureQueueStorageConfig

type AzureQueueStorageConfig struct {
	StorageAccount           string `json:"storage_account" yaml:"storage_account"`
	StorageAccessKey         string `json:"storage_access_key" yaml:"storage_access_key"`
	StorageSASToken          string `json:"storage_sas_token" yaml:"storage_sas_token"`
	StorageConnectionString  string `json:"storage_connection_string" yaml:"storage_connection_string"`
	QueueName                string `json:"queue_name" yaml:"queue_name"`
	DequeueVisibilityTimeout string `json:"dequeue_visibility_timeout" yaml:"dequeue_visibility_timeout"`
	MaxInFlight              int32  `json:"max_in_flight" yaml:"max_in_flight"`
	TrackProperties          bool   `json:"track_properties" yaml:"track_properties"`
}

AzureQueueStorageConfig contains configuration fields for the AzureQueueStorage input type.

func NewAzureQueueStorageConfig

func NewAzureQueueStorageConfig() AzureQueueStorageConfig

NewAzureQueueStorageConfig creates a new AzureQueueStorageConfig with default values.

type Batcher

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

Batcher wraps an input with a batch policy.

func (*Batcher) CloseAsync

func (m *Batcher) CloseAsync()

CloseAsync shuts down the Batcher and stops processing messages.

func (*Batcher) Connected

func (m *Batcher) Connected() bool

Connected returns true if the underlying input is connected.

func (*Batcher) TransactionChan

func (m *Batcher) TransactionChan() <-chan types.Transaction

TransactionChan returns the channel used for consuming messages from this buffer.

func (*Batcher) WaitForClose

func (m *Batcher) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the Batcher output has closed down.

type Bloblang

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

Bloblang executes a bloblang mapping with an empty context each time this input is read from. An interval period must be specified that determines how often a message is generated.

func (*Bloblang) CloseAsync

func (b *Bloblang) CloseAsync()

CloseAsync shuts down the bloblang reader.

func (*Bloblang) ConnectWithContext

func (b *Bloblang) ConnectWithContext(ctx context.Context) error

ConnectWithContext establishes a Bloblang reader.

func (*Bloblang) ReadWithContext

func (b *Bloblang) ReadWithContext(ctx context.Context) (types.Message, reader.AsyncAckFn, error)

ReadWithContext a new bloblang generated message.

func (*Bloblang) WaitForClose

func (b *Bloblang) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the bloblang input has closed down.

type BloblangConfig

type BloblangConfig struct {
	Mapping string `json:"mapping" yaml:"mapping"`
	// internal can be both duration string or cron expression
	Interval string `json:"interval" yaml:"interval"`
	Count    int    `json:"count" yaml:"count"`
}

BloblangConfig contains configuration for the Bloblang input type.

func NewBloblangConfig

func NewBloblangConfig() BloblangConfig

NewBloblangConfig creates a new BloblangConfig with default values.

type BrokerConfig

type BrokerConfig struct {
	Copies   int                `json:"copies" yaml:"copies"`
	Inputs   brokerInputList    `json:"inputs" yaml:"inputs"`
	Batching batch.PolicyConfig `json:"batching" yaml:"batching"`
}

BrokerConfig contains configuration fields for the Broker input type.

func NewBrokerConfig

func NewBrokerConfig() BrokerConfig

NewBrokerConfig creates a new BrokerConfig with default values.

type CSVFileConfig

type CSVFileConfig struct {
	Paths          []string `json:"paths" yaml:"paths"`
	ParseHeaderRow bool     `json:"parse_header_row" yaml:"parse_header_row"`
	Delim          string   `json:"delimiter" yaml:"delimiter"`
	BatchCount     int      `json:"batch_count" yaml:"batch_count"`
}

CSVFileConfig contains configuration values for the CSVFile input type.

func NewCSVFileConfig

func NewCSVFileConfig() CSVFileConfig

NewCSVFileConfig creates a new CSVFileConfig with default values.

type Category

type Category string

Category describes the general category of an input.

var (
	CategoryLocal    Category = "Local"
	CategoryAWS      Category = "AWS"
	CategoryGCP      Category = "GCP"
	CategoryAzure    Category = "Azure"
	CategoryServices Category = "Services"
	CategoryNetwork  Category = "Network"
	CategoryUtility  Category = "Utility"
)

Input categories

type Config

type Config struct {
	Label             string                       `json:"label" yaml:"label"`
	Type              string                       `json:"type" yaml:"type"`
	AMQP              reader.AMQPConfig            `json:"amqp" yaml:"amqp"`
	AMQP09            reader.AMQP09Config          `json:"amqp_0_9" yaml:"amqp_0_9"`
	AMQP1             reader.AMQP1Config           `json:"amqp_1" yaml:"amqp_1"`
	AWSKinesis        AWSKinesisConfig             `json:"aws_kinesis" yaml:"aws_kinesis"`
	AWSS3             AWSS3Config                  `json:"aws_s3" yaml:"aws_s3"`
	AWSSQS            AWSSQSConfig                 `json:"aws_sqs" yaml:"aws_sqs"`
	AzureBlobStorage  AzureBlobStorageConfig       `json:"azure_blob_storage" yaml:"azure_blob_storage"`
	AzureQueueStorage AzureQueueStorageConfig      `json:"azure_queue_storage" yaml:"azure_queue_storage"`
	Bloblang          BloblangConfig               `json:"bloblang" yaml:"bloblang"`
	Broker            BrokerConfig                 `json:"broker" yaml:"broker"`
	CSVFile           CSVFileConfig                `json:"csv" yaml:"csv"`
	Dynamic           DynamicConfig                `json:"dynamic" yaml:"dynamic"`
	File              FileConfig                   `json:"file" yaml:"file"`
	Files             reader.FilesConfig           `json:"files" yaml:"files"`
	GCPCloudStorage   GCPCloudStorageConfig        `json:"gcp_cloud_storage" yaml:"gcp_cloud_storage"`
	GCPPubSub         reader.GCPPubSubConfig       `json:"gcp_pubsub" yaml:"gcp_pubsub"`
	Generate          BloblangConfig               `json:"generate" yaml:"generate"`
	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"`
	KinesisBalanced   reader.KinesisBalancedConfig `json:"kinesis_balanced" yaml:"kinesis_balanced"`
	MQTT              reader.MQTTConfig            `json:"mqtt" yaml:"mqtt"`
	Nanomsg           reader.ScaleProtoConfig      `json:"nanomsg" yaml:"nanomsg"`
	NATS              reader.NATSConfig            `json:"nats" yaml:"nats"`
	NATSJetStream     NATSJetStreamConfig          `json:"nats_jetstream" yaml:"nats_jetstream"`
	NATSStream        reader.NATSStreamConfig      `json:"nats_stream" yaml:"nats_stream"`
	NSQ               reader.NSQConfig             `json:"nsq" yaml:"nsq"`
	Plugin            interface{}                  `json:"plugin,omitempty" yaml:"plugin,omitempty"`
	Pulsar            PulsarConfig                 `json:"pulsar" yaml:"pulsar"`
	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"`
	Resource          string                       `json:"resource" yaml:"resource"`
	S3                reader.AmazonS3Config        `json:"s3" yaml:"s3"`
	Sequence          SequenceConfig               `json:"sequence" yaml:"sequence"`
	SFTP              SFTPConfig                   `json:"sftp" yaml:"sftp"`
	Socket            SocketConfig                 `json:"socket" yaml:"socket"`
	SocketServer      SocketServerConfig           `json:"socket_server" yaml:"socket_server"`
	SQS               reader.AmazonSQSConfig       `json:"sqs" yaml:"sqs"`
	STDIN             STDINConfig                  `json:"stdin" yaml:"stdin"`
	Subprocess        SubprocessConfig             `json:"subprocess" yaml:"subprocess"`
	TCP               TCPConfig                    `json:"tcp" yaml:"tcp"`
	TCPServer         TCPServerConfig              `json:"tcp_server" yaml:"tcp_server"`
	UDPServer         UDPServerConfig              `json:"udp_server" yaml:"udp_server"`
	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. Deprecated: Do not add new components here. Instead, use the public plugin APIs. Examples can be found in: ./internal/impl

func NewConfig

func NewConfig() Config

NewConfig returns a configuration struct fully populated with default values. Deprecated: Do not add new components here. Instead, use the public plugin APIs. Examples can be found in: ./internal/impl

func (Config) Sanitised

func (conf Config) Sanitised(removeDeprecated bool) (interface{}, error)

Sanitised returns a sanitised version of the config, meaning sections that aren't relevant to behaviour are removed. Also optionally removes deprecated fields.

func (*Config) UnmarshalYAML

func (conf *Config) UnmarshalYAML(value *yaml.Node) error

UnmarshalYAML ensures that when parsing configs that are in a map or slice the default values are still applied.

type ConstructorFunc

ConstructorFunc is a func signature able to construct an input.

func GetDeprecatedPlugin

func GetDeprecatedPlugin(name string) (ConstructorFunc, bool)

GetDeprecatedPlugin returns a constructor for an old plugin if it exists.

type DynamicConfig

type DynamicConfig struct {
	Inputs  map[string]Config `json:"inputs" yaml:"inputs"`
	Prefix  string            `json:"prefix" yaml:"prefix"`
	Timeout string            `json:"timeout" yaml:"timeout"`
}

DynamicConfig contains configuration for the Dynamic input type.

func NewDynamicConfig

func NewDynamicConfig() DynamicConfig

NewDynamicConfig creates a new DynamicConfig with default values.

type DynamoDBCheckpointConfig

type DynamoDBCheckpointConfig struct {
	Table              string `json:"table" yaml:"table"`
	Create             bool   `json:"create" yaml:"create"`
	ReadCapacityUnits  int64  `json:"read_capacity_units" yaml:"read_capacity_units"`
	WriteCapacityUnits int64  `json:"write_capacity_units" yaml:"write_capacity_units"`
	BillingMode        string `json:"billing_mode" yaml:"billing_mode"`
}

DynamoDBCheckpointConfig contains configuration parameters for a DynamoDB based checkpoint store for Kinesis.

func NewDynamoDBCheckpointConfig

func NewDynamoDBCheckpointConfig() DynamoDBCheckpointConfig

NewDynamoDBCheckpointConfig returns a DynamoDBCheckpoint config struct with default values.

type FileConfig

type FileConfig struct {
	Path           string   `json:"path" yaml:"path"`
	Paths          []string `json:"paths" yaml:"paths"`
	Codec          string   `json:"codec" yaml:"codec"`
	Multipart      bool     `json:"multipart" yaml:"multipart"`
	MaxBuffer      int      `json:"max_buffer" yaml:"max_buffer"`
	Delim          string   `json:"delimiter" yaml:"delimiter"`
	DeleteOnFinish bool     `json:"delete_on_finish" yaml:"delete_on_finish"`
}

FileConfig contains configuration values for the File input type.

func NewFileConfig

func NewFileConfig() FileConfig

NewFileConfig creates a new FileConfig with default values.

type GCPCloudStorageConfig

type GCPCloudStorageConfig struct {
	Bucket        string `json:"bucket" yaml:"bucket"`
	Prefix        string `json:"prefix" yaml:"prefix"`
	Codec         string `json:"codec" yaml:"codec"`
	DeleteObjects bool   `json:"delete_objects" yaml:"delete_objects"`
}

GCPCloudStorageConfig contains configuration fields for the Google Cloud Storage input type.

func NewGCPCloudStorageConfig

func NewGCPCloudStorageConfig() GCPCloudStorageConfig

NewGCPCloudStorageConfig creates a new GCPCloudStorageConfig with default values.

type HTTPClient

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

func (h *HTTPClient) CloseAsync()

CloseAsync shuts down the HTTPClient input and stops processing requests.

func (*HTTPClient) ConnectWithContext

func (h *HTTPClient) ConnectWithContext(ctx context.Context) (err error)

ConnectWithContext establishes a connection.

func (*HTTPClient) ReadWithContext

func (h *HTTPClient) ReadWithContext(ctx context.Context) (types.Message, reader.AsyncAckFn, error)

ReadWithContext a new HTTPClient message.

func (*HTTPClient) WaitForClose

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

WaitForClose blocks until the HTTPClient input has closed down.

type HTTPClientConfig

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

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

func (h *HTTPServer) Connected() bool

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

func (*HTTPServer) TransactionChan

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"`
	WSWelcomeMessage   string                   `json:"ws_welcome_message" yaml:"ws_welcome_message"`
	WSRateLimitMessage string                   `json:"ws_rate_limit_message" yaml:"ws_rate_limit_message"`
	AllowedVerbs       []string                 `json:"allowed_verbs" yaml:"allowed_verbs"`
	Timeout            string                   `json:"timeout" yaml:"timeout"`
	RateLimit          string                   `json:"rate_limit" yaml:"rate_limit"`
	CertFile           string                   `json:"cert_file" yaml:"cert_file"`
	KeyFile            string                   `json:"key_file" yaml:"key_file"`
	CORS               httpdocs.ServerCORS      `json:"cors" yaml:"cors"`
	Response           HTTPServerResponseConfig `json:"sync_response" yaml:"sync_response"`
}

HTTPServerConfig contains configuration for the HTTPServer input type.

func NewHTTPServerConfig

func NewHTTPServerConfig() HTTPServerConfig

NewHTTPServerConfig creates a new HTTPServerConfig with default values.

type HTTPServerResponseConfig

type HTTPServerResponseConfig struct {
	Status          string                        `json:"status" yaml:"status"`
	Headers         map[string]string             `json:"headers" yaml:"headers"`
	ExtractMetadata imetadata.IncludeFilterConfig `json:"metadata_headers" yaml:"metadata_headers"`
}

HTTPServerResponseConfig provides config fields for customising the response given from successful requests.

func NewHTTPServerResponseConfig

func NewHTTPServerResponseConfig() HTTPServerResponseConfig

NewHTTPServerResponseConfig creates a new HTTPServerConfig with default values.

type Inproc

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

func (i *Inproc) CloseAsync()

CloseAsync shuts down the Inproc input and stops processing requests.

func (*Inproc) Connected

func (i *Inproc) Connected() bool

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

func (*Inproc) TransactionChan

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

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

func (*Inproc) WaitForClose

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

WaitForClose blocks until the Inproc input has closed down.

type InprocConfig

type InprocConfig string

InprocConfig is a configuration type for the inproc input.

func NewInprocConfig

func NewInprocConfig() InprocConfig

NewInprocConfig creates a new inproc input config.

type NATSJetStreamConfig

type NATSJetStreamConfig struct {
	URLs          []string    `json:"urls" yaml:"urls"`
	Subject       string      `json:"subject" yaml:"subject"`
	Queue         string      `json:"queue" yaml:"queue"`
	Durable       string      `json:"durable" yaml:"durable"`
	Deliver       string      `json:"deliver" yaml:"deliver"`
	AckWait       string      `json:"ack_wait" yaml:"ack_wait"`
	MaxAckPending int         `json:"max_ack_pending" yaml:"max_ack_pending"`
	TLS           tls.Config  `json:"tls" yaml:"tls"`
	Auth          auth.Config `json:"auth" yaml:"auth"`
}

NATSJetStreamConfig contains configuration fields for the NATS Jetstream input type.

func NewNATSJetStreamConfig

func NewNATSJetStreamConfig() NATSJetStreamConfig

NewNATSJetStreamConfig creates a new NATSJetstreamConfig with default values.

type PluginConfigConstructor

type PluginConfigConstructor func() interface{}

PluginConfigConstructor is a func that returns a pointer to a new and fully populated configuration struct for a plugin type.

type PluginConfigSanitiser

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

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 PulsarConfig

type PulsarConfig struct {
	URL              string      `json:"url" yaml:"url"`
	Topics           []string    `json:"topics" yaml:"topics"`
	SubscriptionName string      `json:"subscription_name" yaml:"subscription_name"`
	SubscriptionType string      `json:"subscription_type" yaml:"subscription_type"`
	Auth             auth.Config `json:"auth" yaml:"auth"`
}

PulsarConfig contains configuration for the Pulsar input type.

func NewPulsarConfig

func NewPulsarConfig() PulsarConfig

NewPulsarConfig creates a new PulsarConfig with default values.

type ReadUntil

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

func (r *ReadUntil) CloseAsync()

CloseAsync shuts down the ReadUntil input and stops processing requests.

func (*ReadUntil) Connected

func (r *ReadUntil) Connected() bool

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

func (*ReadUntil) TransactionChan

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

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

func (*ReadUntil) WaitForClose

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

WaitForClose blocks until the ReadUntil input has closed down.

type ReadUntilConfig

type ReadUntilConfig struct {
	Input        *Config          `json:"input" yaml:"input"`
	Restart      bool             `json:"restart_input" yaml:"restart_input"`
	LoopForever  bool             `json:"loop_forever" yaml:"loop_forever"`
	LoopInterval string           `json:"loop_interval" yaml:"loop_interval"`
	Condition    condition.Config `json:"condition" yaml:"condition"`
	Check        string           `json:"check" yaml:"check"`
}

ReadUntilConfig contains configuration values for the ReadUntil input type.

func NewReadUntilConfig

func NewReadUntilConfig() ReadUntilConfig

NewReadUntilConfig creates a new ReadUntilConfig with default values.

func (ReadUntilConfig) MarshalJSON

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

MarshalJSON prints an empty object instead of nil.

func (ReadUntilConfig) MarshalYAML

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

MarshalYAML prints an empty object instead of nil.

type Reader

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

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

func (*Reader) CloseAsync

func (r *Reader) CloseAsync()

CloseAsync shuts down the Reader input and stops processing requests.

func (*Reader) Connected

func (r *Reader) Connected() bool

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

func (*Reader) TransactionChan

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

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

func (*Reader) WaitForClose

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

WaitForClose blocks until the Reader input has closed down.

type Resource

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

Resource is an input that wraps an input resource.

func (*Resource) CloseAsync

func (r *Resource) CloseAsync()

CloseAsync shuts down the processor and stops processing requests.

func (*Resource) Connected

func (r *Resource) Connected() (isConnected bool)

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

func (*Resource) TransactionChan

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

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

func (*Resource) WaitForClose

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

WaitForClose blocks until the processor has closed down.

type SFTPConfig

type SFTPConfig struct {
	Address        string                `json:"address" yaml:"address"`
	Credentials    sftpSetup.Credentials `json:"credentials" yaml:"credentials"`
	Paths          []string              `json:"paths" yaml:"paths"`
	Codec          string                `json:"codec" yaml:"codec"`
	DeleteOnFinish bool                  `json:"delete_on_finish" yaml:"delete_on_finish"`
	MaxBuffer      int                   `json:"max_buffer" yaml:"max_buffer"`
	Watcher        watcherConfig         `json:"watcher" yaml:"watcher"`
}

SFTPConfig contains configuration fields for the SFTP input type.

func NewSFTPConfig

func NewSFTPConfig() SFTPConfig

NewSFTPConfig creates a new SFTPConfig with default values.

type STDINConfig

type STDINConfig struct {
	Codec     string `json:"codec" yaml:"codec"`
	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

func NewSTDINConfig() STDINConfig

NewSTDINConfig creates a STDINConfig populated with default values.

type Sequence

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

Sequence is an input type that reads from a sequence of inputs, starting with the first, and when it ends gracefully it moves onto the next, and so on.

func (*Sequence) CloseAsync

func (r *Sequence) CloseAsync()

CloseAsync shuts down the Sequence input and stops processing requests.

func (*Sequence) Connected

func (r *Sequence) Connected() bool

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

func (*Sequence) TransactionChan

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

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

func (*Sequence) WaitForClose

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

WaitForClose blocks until the Sequence input has closed down.

type SequenceConfig

type SequenceConfig struct {
	ShardedJoin SequenceShardedJoinConfig `json:"sharded_join" yaml:"sharded_join"`
	Inputs      []Config                  `json:"inputs" yaml:"inputs"`
}

SequenceConfig contains configuration values for the Sequence input type.

func NewSequenceConfig

func NewSequenceConfig() SequenceConfig

NewSequenceConfig creates a new SequenceConfig with default values.

type SequenceShardedJoinConfig

type SequenceShardedJoinConfig struct {
	Type          string `json:"type" yaml:"type"`
	IDPath        string `json:"id_path" yaml:"id_path"`
	Iterations    int    `json:"iterations" yaml:"iterations"`
	MergeStrategy string `json:"merge_strategy" yaml:"merge_strategy"`
}

SequenceShardedJoinConfig describes an optional mechanism for performing sharded joins of structured data resulting from the input sequence. This is a way to merge the structured fields of fragmented datasets within memory even when the overall size of the data surpasses the memory available on the machine.

When configured the sequence of inputs will be consumed multiple times according to the number of iterations, and each iteration will process an entirely different set of messages by sharding them by the ID field.

Each message must be structured (JSON or otherwise processed into a structured form) and the fields will be aggregated with those of other messages sharing the ID. At the end of each iteration the joined messages are flushed downstream before the next iteration begins.

func NewSequenceShardedJoinConfig

func NewSequenceShardedJoinConfig() SequenceShardedJoinConfig

NewSequenceShardedJoinConfig creates a new sequence sharding configuration with default values.

type SocketConfig

type SocketConfig struct {
	Network   string `json:"network" yaml:"network"`
	Address   string `json:"address" yaml:"address"`
	Codec     string `json:"codec" yaml:"codec"`
	MaxBuffer int    `json:"max_buffer" yaml:"max_buffer"`
	// TODO: V4 remove these fields.
	Multipart bool   `json:"multipart" yaml:"multipart"`
	Delim     string `json:"delimiter" yaml:"delimiter"`
}

SocketConfig contains configuration values for the Socket input type.

func NewSocketConfig

func NewSocketConfig() SocketConfig

NewSocketConfig creates a new SocketConfig with default values.

type SocketServer

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

SocketServer is an input type that binds to an address and consumes streams of messages over Socket.

func (*SocketServer) Addr

func (t *SocketServer) Addr() net.Addr

Addr returns the underlying Socket listeners address.

func (*SocketServer) CloseAsync

func (t *SocketServer) CloseAsync()

CloseAsync shuts down the SocketServer input and stops processing requests.

func (*SocketServer) Connected

func (t *SocketServer) Connected() bool

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

func (*SocketServer) TransactionChan

func (t *SocketServer) TransactionChan() <-chan types.Transaction

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

func (*SocketServer) WaitForClose

func (t *SocketServer) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the SocketServer input has closed down.

type SocketServerConfig

type SocketServerConfig struct {
	Network   string `json:"network" yaml:"network"`
	Address   string `json:"address" yaml:"address"`
	Codec     string `json:"codec" yaml:"codec"`
	MaxBuffer int    `json:"max_buffer" yaml:"max_buffer"`
	Multipart bool   `json:"multipart" yaml:"multipart"`
	Delim     string `json:"delimiter" yaml:"delimiter"`
}

SocketServerConfig contains configuration for the SocketServer input type.

func NewSocketServerConfig

func NewSocketServerConfig() SocketServerConfig

NewSocketServerConfig creates a new SocketServerConfig with default values.

type StreamConfig

type StreamConfig struct {
	Enabled   bool   `json:"enabled" yaml:"enabled"`
	Reconnect bool   `json:"reconnect" yaml:"reconnect"`
	Codec     string `json:"codec" yaml:"codec"`
	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 Subprocess

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

Subprocess executes a bloblang mapping with an empty context each time this input is read from. An interval period must be specified that determines how often a message is generated.

func (*Subprocess) CloseAsync

func (s *Subprocess) CloseAsync()

CloseAsync shuts down the bloblang reader.

func (*Subprocess) ConnectWithContext

func (s *Subprocess) ConnectWithContext(ctx context.Context) error

ConnectWithContext establishes a Subprocess reader.

func (*Subprocess) ReadWithContext

func (s *Subprocess) ReadWithContext(ctx context.Context) (types.Message, reader.AsyncAckFn, error)

ReadWithContext a new bloblang generated message.

func (*Subprocess) WaitForClose

func (s *Subprocess) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the bloblang input has closed down.

type SubprocessConfig

type SubprocessConfig struct {
	Name          string   `json:"name" yaml:"name"`
	Args          []string `json:"args" yaml:"args"`
	Codec         string   `json:"codec" yaml:"codec"`
	RestartOnExit bool     `json:"restart_on_exit" yaml:"restart_on_exit"`
	MaxBuffer     int      `json:"max_buffer" yaml:"max_buffer"`
}

SubprocessConfig contains configuration for the Subprocess input type.

func NewSubprocessConfig

func NewSubprocessConfig() SubprocessConfig

NewSubprocessConfig creates a new SubprocessConfig with default values.

type TCPConfig

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

TCPConfig contains configuration values for the TCP input type.

func NewTCPConfig

func NewTCPConfig() TCPConfig

NewTCPConfig creates a new TCPConfig with default values.

type TCPServer

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

TCPServer is an input type that binds to an address and consumes streams of messages over TCP.

func (*TCPServer) Addr

func (t *TCPServer) Addr() net.Addr

Addr returns the underlying TCP listeners address.

func (*TCPServer) CloseAsync

func (t *TCPServer) CloseAsync()

CloseAsync shuts down the TCPServer input and stops processing requests.

func (*TCPServer) Connected

func (t *TCPServer) Connected() bool

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

func (*TCPServer) TransactionChan

func (t *TCPServer) TransactionChan() <-chan types.Transaction

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

func (*TCPServer) WaitForClose

func (t *TCPServer) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the TCPServer input has closed down.

type TCPServerConfig

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

TCPServerConfig contains configuration for the TCPServer input type.

func NewTCPServerConfig

func NewTCPServerConfig() TCPServerConfig

NewTCPServerConfig creates a new TCPServerConfig with default values.

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

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

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

NewAMQP creates a new AMQP input type. TODO: V4 Remove this.

func NewAMQP09

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

NewAMQP09 creates a new AMQP09 input type.

func NewAMQP1

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

NewAMQP1 creates a new AMQP1 input type.

func NewAmazonS3

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

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

NewAmazonSQS creates a new AWS SQS input type.

func NewAsyncReader

func NewAsyncReader(
	typeStr string,
	allowSkipAcks bool,
	r reader.Async,
	log log.Modular,
	stats metrics.Type,
) (Type, error)

NewAsyncReader creates a new AsyncReader input type.

func NewBatcher

func NewBatcher(
	batcher *batch.Policy,
	child Type,
	log log.Modular,
	stats metrics.Type,
) Type

NewBatcher creates a new Batcher around an input.

func NewBroker

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 NewCSVFile

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

NewCSVFile creates a new CSV file input type.

func NewDynamic

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

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

NewFiles creates a new Files input type.

func NewGCPPubSub

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

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

NewHDFS creates a new Files input type.

func NewHTTPClient

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

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

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

NewKafkaBalanced creates a new KafkaBalanced input type.

func NewKinesis

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

NewKinesis creates a new AWS Kinesis input type.

func NewKinesisBalanced

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

NewKinesisBalanced creates a new AWS KinesisBalanced input type.

func NewMQTT

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

NewMQTT creates a new MQTT input type.

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

func NewNanomsg

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

NewNanomsg creates a new Nanomsg input type.

func NewReadUntil

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

NewReadUntil creates a new ReadUntil 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 NewRedisStreams

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

NewRedisStreams creates a new Redis List input type.

func NewResource

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

NewResource returns a resource input.

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 NewSequence

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

NewSequence creates a new Sequence input type.

func NewSocket

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

NewSocket creates a new Socket input type.

func NewSocketServer

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

NewSocketServer creates a new SocketServer input type.

func NewTCP

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

NewTCP creates a new TCP input type.

func NewTCPServer

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

NewTCPServer creates a new TCPServer input type.

func NewUDPServer

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

NewUDPServer creates a new UDPServer input type.

func NewWebsocket

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

NewWebsocket creates a new Websocket input type.

func WrapWithPipelines

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

WrapWithPipelines wraps an input with a variadic number of pipelines.

type TypeSpec

type TypeSpec struct {
	Status      docs.Status
	Version     string
	Summary     string
	Description string
	Categories  []Category
	Footnotes   string

	FieldSpecs docs.FieldSpecs
	Examples   []docs.AnnotatedExample
	// contains filtered or unexported fields
}

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

type UDPServer

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

UDPServer is an input type that binds to an address and consumes streams of messages over UDP.

func (*UDPServer) Addr

func (t *UDPServer) Addr() net.Addr

Addr returns the underlying UDP listeners address.

func (*UDPServer) CloseAsync

func (t *UDPServer) CloseAsync()

CloseAsync shuts down the UDPServer input and stops processing requests.

func (*UDPServer) Connected

func (t *UDPServer) Connected() bool

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

func (*UDPServer) TransactionChan

func (t *UDPServer) TransactionChan() <-chan types.Transaction

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

func (*UDPServer) WaitForClose

func (t *UDPServer) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the UDPServer input has closed down.

type UDPServerConfig

type UDPServerConfig struct {
	Address   string `json:"address" yaml:"address"`
	MaxBuffer int    `json:"max_buffer" yaml:"max_buffer"`
	Delim     string `json:"delimiter" yaml:"delimiter"`
}

UDPServerConfig contains configuration for the UDPServer input type.

func NewUDPServerConfig

func NewUDPServerConfig() UDPServerConfig

NewUDPServerConfig creates a new UDPServerConfig with default values.

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

func (i *WithPipeline) CloseAsync()

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

func (*WithPipeline) Connected

func (i *WithPipeline) Connected() bool

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

func (*WithPipeline) TransactionChan

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

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

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.

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