input

package
v4.0.0 Latest Latest
Warning

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

 Go to latest Published: Apr 20, 2022 License: MIT Imports: 73 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 (
	TypeAMQP09            = "amqp_0_9"
	TypeAWSKinesis        = "aws_kinesis"
	TypeAWSS3             = "aws_s3"
	TypeAWSSQS            = "aws_sqs"
	TypeAzureBlobStorage  = "azure_blob_storage"
	TypeAzureQueueStorage = "azure_queue_storage"
	TypeBroker            = "broker"
	TypeCSVFile           = "csv"
	TypeDynamic           = "dynamic"
	TypeFile              = "file"
	TypeGCPCloudStorage   = "gcp_cloud_storage"
	TypeGCPPubSub         = "gcp_pubsub"
	TypeGenerate          = "generate"
	TypeHDFS              = "hdfs"
	TypeHTTPClient        = "http_client"
	TypeHTTPServer        = "http_server"
	TypeInproc            = "inproc"
	TypeKafka             = "kafka"
	TypeKinesis           = "kinesis"
	TypeMQTT              = "mqtt"
	TypeNanomsg           = "nanomsg"
	TypeNATS              = "nats"
	TypeNATSJetStream     = "nats_jetstream"
	TypeNATSStream        = "nats_stream"
	TypeNSQ               = "nsq"
	TypeReadUntil         = "read_until"
	TypeRedisList         = "redis_list"
	TypeRedisPubSub       = "redis_pubsub"
	TypeRedisStreams      = "redis_streams"
	TypeResource          = "resource"
	TypeSequence          = "sequence"
	TypeSFTP              = "sftp"
	TypeSocket            = "socket"
	TypeSocketServer      = "socket_server"
	TypeSTDIN             = "stdin"
	TypeSubprocess        = "subprocess"
	TypeWebsocket         = "websocket"
)

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

Variables

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

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

Functions

func AppendProcessorsFromConfig 

func AppendProcessorsFromConfig(conf Config, mgr interop.Manager, pipelines ...iprocessor.PipelineConstructorFunc) []iprocessor.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 New 

func New(
	conf Config,
	mgr interop.Manager,
	log log.Modular,
	stats metrics.Type,
	pipelines ...iprocessor.PipelineConstructorFunc,
) (input.Streamed, error)

New creates an input type based on an input configuration.

func NewAsyncReader 

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

NewAsyncReader creates a new AsyncReader input type.

func NewBatcher 

func NewBatcher(
	batcher *policy.Batcher,
	child input.Streamed, log log.Modular,
	stats metrics.Type,
) input.Streamed

NewBatcher creates a new Batcher around an input.

func NewCSVFile 

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

NewCSVFile creates a new CSV file input type.

func NewFile 

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

NewFile creates a new File input type.

func NewGCPPubSub 

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

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

func NewHDFS 

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

NewHDFS creates a new Files input type.

func NewHTTPClient 

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

NewHTTPClient creates a new HTTPClient input type.

func NewHTTPServer 

func NewHTTPServer(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

NewHTTPServer creates a new HTTPServer input type.

func NewInproc 

func NewInproc(
	conf Config,
	mgr interop.Manager,
	log log.Modular,
	stats metrics.Type,
) (input.Streamed, error)

NewInproc creates a new Inproc input type.

func NewKafka 

func NewKafka(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

NewKafka creates a new Kafka input type.

func NewNATS 

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

NewNATS creates a new NATS input type.

func NewNATSStream 

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

NewNATSStream creates a new NATSStream input type.

func NewNSQ 

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

NewNSQ creates a new NSQ input type.

func NewNanomsg 

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

NewNanomsg creates a new Nanomsg input type.

func NewReadUntil 

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

NewReadUntil creates a new ReadUntil input type.

func NewRedisList 

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

NewRedisList creates a new Redis List input type.

func NewRedisPubSub 

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

NewRedisPubSub creates a new RedisPubSub input type.

func NewRedisStreams 

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

NewRedisStreams creates a new Redis List input type.

func NewResource 

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

NewResource returns a resource input.

func NewSTDIN 

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

NewSTDIN creates a new STDIN input type.

func NewSequence 

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

NewSequence creates a new Sequence input type.

func NewSocket 

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

NewSocket creates a new Socket input type.

func NewSocketServer 

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

NewSocketServer creates a new SocketServer input type.

func NewWebsocket 

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

NewWebsocket creates a new Websocket input type.

func WalkConstructors 

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

WalkConstructors iterates each component constructor.

func WrapWithPipelines 

func WrapWithPipelines(in input.Streamed, pipeConstructors ...iprocessor.PipelineConstructorFunc) (input.Streamed, error)

WrapWithPipelines wraps an input with a variadic number of pipelines.

Types

type AMQP09BindingConfig 

type AMQP09BindingConfig struct {
	Exchange   string `json:"exchange" yaml:"exchange"`
	RoutingKey string `json:"key" yaml:"key"`
}

AMQP09BindingConfig contains fields describing a queue binding to be declared.

type AMQP09Config 

type AMQP09Config struct {
	URLs               []string                 `json:"urls" yaml:"urls"`
	Queue              string                   `json:"queue" yaml:"queue"`
	QueueDeclare       AMQP09QueueDeclareConfig `json:"queue_declare" yaml:"queue_declare"`
	BindingsDeclare    []AMQP09BindingConfig    `json:"bindings_declare" yaml:"bindings_declare"`
	ConsumerTag        string                   `json:"consumer_tag" yaml:"consumer_tag"`
	AutoAck            bool                     `json:"auto_ack" yaml:"auto_ack"`
	NackRejectPatterns []string                 `json:"nack_reject_patterns" yaml:"nack_reject_patterns"`
	PrefetchCount      int                      `json:"prefetch_count" yaml:"prefetch_count"`
	PrefetchSize       int                      `json:"prefetch_size" yaml:"prefetch_size"`
	TLS                btls.Config              `json:"tls" yaml:"tls"`
}

AMQP09Config contains configuration for the AMQP09 input type.

func NewAMQP09Config 

func NewAMQP09Config() AMQP09Config

NewAMQP09Config creates a new AMQP09Config with default values.

type AMQP09QueueDeclareConfig 

type AMQP09QueueDeclareConfig struct {
	Enabled bool `json:"enabled" yaml:"enabled"`
	Durable bool `json:"durable" yaml:"durable"`
}

AMQP09QueueDeclareConfig contains fields indicating whether the target AMQP09 queue needs to be declared and bound to an exchange, as well as any fields specifying how to accomplish that.

type AMQP1Config 

type AMQP1Config struct {
	URL            string            `json:"url" yaml:"url"`
	SourceAddress  string            `json:"source_address" yaml:"source_address"`
	AzureRenewLock bool              `json:"azure_renew_lock" yaml:"azure_renew_lock"`
	TLS            btls.Config       `json:"tls" yaml:"tls"`
	SASL           shared.SASLConfig `json:"sasl" yaml:"sasl"`
}

AMQP1Config contains configuration for the AMQP1 input type.

func NewAMQP1Config 

func NewAMQP1Config() AMQP1Config

NewAMQP1Config creates a new AMQP1Config with default values.

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        policy.Config            `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"`
	MaxNumberOfMessages int    `json:"max_number_of_messages" yaml:"max_number_of_messages"`
}

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 message.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 message.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) (*message.Batch, 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   []Config      `json:"inputs" yaml:"inputs"`
	Batching policy.Config `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 Config 

type Config struct {
	Label             string                    `json:"label" yaml:"label"`
	Type              string                    `json:"type" yaml:"type"`
	AMQP09            AMQP09Config              `json:"amqp_0_9" yaml:"amqp_0_9"`
	AMQP1             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"`
	Broker            BrokerConfig              `json:"broker" yaml:"broker"`
	CSVFile           CSVFileConfig             `json:"csv" yaml:"csv"`
	Dynamic           DynamicConfig             `json:"dynamic" yaml:"dynamic"`
	File              FileConfig                `json:"file" yaml:"file"`
	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             KafkaConfig               `json:"kafka" yaml:"kafka"`
	MQTT              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"`
	Resource          string                    `json:"resource" yaml:"resource"`
	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"`
	STDIN             STDINConfig               `json:"stdin" yaml:"stdin"`
	Subprocess        SubprocessConfig          `json:"subprocess" yaml:"subprocess"`
	Websocket         reader.WebsocketConfig    `json:"websocket" yaml:"websocket"`
	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) 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 

type ConstructorFunc func(Config, interop.Manager, log.Modular, metrics.Type, ...iprocessor.PipelineConstructorFunc) (input.Streamed, error)

ConstructorFunc is a func signature able to construct an input.

type DynamicConfig 

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

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 {
	Paths          []string `json:"paths" yaml:"paths"`
	Codec          string   `json:"codec" yaml:"codec"`
	MaxBuffer      int      `json:"max_buffer" yaml:"max_buffer"`
	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) (*message.Batch, 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 {
	ihttpdocs.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 message.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 message.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 KafkaBalancedGroupConfig 

type KafkaBalancedGroupConfig struct {
	SessionTimeout    string `json:"session_timeout" yaml:"session_timeout"`
	HeartbeatInterval string `json:"heartbeat_interval" yaml:"heartbeat_interval"`
	RebalanceTimeout  string `json:"rebalance_timeout" yaml:"rebalance_timeout"`
}

KafkaBalancedGroupConfig contains config fields for Kafka consumer groups.

func NewKafkaBalancedGroupConfig 

func NewKafkaBalancedGroupConfig() KafkaBalancedGroupConfig

NewKafkaBalancedGroupConfig returns a KafkaBalancedGroupConfig with default values.

type KafkaConfig 

type KafkaConfig struct {
	Addresses           []string                 `json:"addresses" yaml:"addresses"`
	Topics              []string                 `json:"topics" yaml:"topics"`
	ClientID            string                   `json:"client_id" yaml:"client_id"`
	RackID              string                   `json:"rack_id" yaml:"rack_id"`
	ConsumerGroup       string                   `json:"consumer_group" yaml:"consumer_group"`
	Group               KafkaBalancedGroupConfig `json:"group" yaml:"group"`
	CommitPeriod        string                   `json:"commit_period" yaml:"commit_period"`
	CheckpointLimit     int                      `json:"checkpoint_limit" yaml:"checkpoint_limit"`
	ExtractTracingMap   string                   `json:"extract_tracing_map" yaml:"extract_tracing_map"`
	MaxProcessingPeriod string                   `json:"max_processing_period" yaml:"max_processing_period"`
	FetchBufferCap      int                      `json:"fetch_buffer_cap" yaml:"fetch_buffer_cap"`
	StartFromOldest     bool                     `json:"start_from_oldest" yaml:"start_from_oldest"`
	TargetVersion       string                   `json:"target_version" yaml:"target_version"`
	TLS                 btls.Config              `json:"tls" yaml:"tls"`
	SASL                sasl.Config              `json:"sasl" yaml:"sasl"`
	Batching            policy.Config            `json:"batching" yaml:"batching"`
}

KafkaConfig contains configuration fields for the Kafka input type.

func NewKafkaConfig 

func NewKafkaConfig() KafkaConfig

NewKafkaConfig creates a new KafkaConfig with default values.

type MQTTConfig 

type MQTTConfig struct {
	URLs                  []string      `json:"urls" yaml:"urls"`
	QoS                   uint8         `json:"qos" yaml:"qos"`
	Topics                []string      `json:"topics" yaml:"topics"`
	ClientID              string        `json:"client_id" yaml:"client_id"`
	DynamicClientIDSuffix string        `json:"dynamic_client_id_suffix" yaml:"dynamic_client_id_suffix"`
	Will                  mqttconf.Will `json:"will" yaml:"will"`
	CleanSession          bool          `json:"clean_session" yaml:"clean_session"`
	User                  string        `json:"user" yaml:"user"`
	Password              string        `json:"password" yaml:"password"`
	ConnectTimeout        string        `json:"connect_timeout" yaml:"connect_timeout"`
	KeepAlive             int64         `json:"keepalive" yaml:"keepalive"`
	TLS                   tls.Config    `json:"tls" yaml:"tls"`
}

MQTTConfig contains configuration fields for the MQTT input type.

func NewMQTTConfig 

func NewMQTTConfig() MQTTConfig

NewMQTTConfig creates a new MQTTConfig 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 message.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"`
	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 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 message.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"`
	MaxBuffer int    `json:"max_buffer" yaml:"max_buffer"`
}

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 message.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"`
}

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 message.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"`
}

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"`
	MaxBuffer int    `json:"max_buffer" yaml:"max_buffer"`
}

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) (*message.Batch, 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 TypeSpec 

type TypeSpec struct {
	Status      docs.Status
	Version     string
	Summary     string
	Description string
	Categories  []string
	Footnotes   string
	Config      docs.FieldSpec
	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 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 input.Streamed, pipeConstructor iprocessor.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 message.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.

Source Files

View all Source files

Directories

Path Synopsis

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.