reader

package
v3.65.0 Latest Latest
Warning

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

Go to latest
Published: Mar 7, 2022 License: MIT Imports: 70 Imported by: 1

Documentation

Overview

Package reader defines implementations of an interface for generic message reading from various third party sources.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func OptLinesSetDelimiter

func OptLinesSetDelimiter(delimiter string) func(r *Lines)

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

func OptLinesSetMaxBuffer

func OptLinesSetMaxBuffer(maxBuffer int) func(r *Lines)

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

func OptLinesSetMultipart

func OptLinesSetMultipart(multipart bool) func(r *Lines)

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

Types

type AMQP

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

AMQP is an input type that reads messages via the AMQP 0.91 protocol.

func (*AMQP) Acknowledge

func (a *AMQP) Acknowledge(err error) error

Acknowledge instructs whether unacknowledged messages have been successfully propagated.

func (*AMQP) CloseAsync

func (a *AMQP) CloseAsync()

CloseAsync shuts down the AMQP input and stops processing requests.

func (*AMQP) Connect

func (a *AMQP) Connect() (err error)

Connect establishes a connection to an AMQP server.

func (*AMQP) Read

func (a *AMQP) Read() (types.Message, error)

Read a new AMQP message.

func (*AMQP) WaitForClose

func (a *AMQP) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the AMQP input has closed down.

type AMQP09 added in v3.2.0

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

AMQP09 is an input type that reads messages via the AMQP09 0.9 protocol.

func NewAMQP09 added in v3.2.0

func NewAMQP09(conf AMQP09Config, log log.Modular, stats metrics.Type) (*AMQP09, error)

NewAMQP09 creates a new AMQP09 input type.

func (*AMQP09) CloseAsync added in v3.2.0

func (a *AMQP09) CloseAsync()

CloseAsync shuts down the AMQP09 input and stops processing requests.

func (*AMQP09) ConnectWithContext added in v3.2.0

func (a *AMQP09) ConnectWithContext(ctx context.Context) (err error)

ConnectWithContext establishes a connection to an AMQP09 server.

func (*AMQP09) ReadWithContext added in v3.2.0

func (a *AMQP09) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext a new AMQP09 message.

func (*AMQP09) WaitForClose added in v3.2.0

func (a *AMQP09) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the AMQP09 input has closed down.

type AMQP09BindingConfig added in v3.2.0

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 added in v3.2.0

type AMQP09Config struct {
	URL                string                   `json:"url" yaml:"url"`
	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"`

	// TODO: V4 remove this (maybe in V5 to allow a grace period)
	Batching batch.PolicyConfig `json:"batching" yaml:"batching"`
}

AMQP09Config contains configuration for the AMQP09 input type.

func NewAMQP09Config added in v3.2.0

func NewAMQP09Config() AMQP09Config

NewAMQP09Config creates a new AMQP09Config with default values.

type AMQP09QueueDeclareConfig added in v3.2.0

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 AMQP1 added in v3.19.0

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

AMQP1 is an input type that reads messages via the AMQP 1.0 protocol.

func NewAMQP1 added in v3.19.0

func NewAMQP1(conf AMQP1Config, log log.Modular, stats metrics.Type) (*AMQP1, error)

NewAMQP1 creates a new AMQP1 input type.

func (*AMQP1) CloseAsync added in v3.19.0

func (a *AMQP1) CloseAsync()

CloseAsync shuts down the AMQP1 input and stops processing requests.

func (*AMQP1) ConnectWithContext added in v3.19.0

func (a *AMQP1) ConnectWithContext(ctx context.Context) error

ConnectWithContext establishes a connection to an AMQP1 server.

func (*AMQP1) ReadWithContext added in v3.19.0

func (a *AMQP1) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext a new AMQP1 message.

func (*AMQP1) WaitForClose added in v3.19.0

func (a *AMQP1) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the AMQP1 input has closed down.

type AMQP1Config added in v3.19.0

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           sasl.Config `json:"sasl" yaml:"sasl"`
}

AMQP1Config contains configuration for the AMQP1 input type.

func NewAMQP1Config added in v3.19.0

func NewAMQP1Config() AMQP1Config

NewAMQP1Config creates a new AMQP1Config with default values.

type AMQPBindingConfig

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

AMQPBindingConfig contains fields describing a queue binding to be declared.

type AMQPConfig

type AMQPConfig struct {
	URL             string                 `json:"url" yaml:"url"`
	Queue           string                 `json:"queue" yaml:"queue"`
	QueueDeclare    AMQPQueueDeclareConfig `json:"queue_declare" yaml:"queue_declare"`
	BindingsDeclare []AMQPBindingConfig    `json:"bindings_declare" yaml:"bindings_declare"`
	ConsumerTag     string                 `json:"consumer_tag" yaml:"consumer_tag"`
	PrefetchCount   int                    `json:"prefetch_count" yaml:"prefetch_count"`
	PrefetchSize    int                    `json:"prefetch_size" yaml:"prefetch_size"`
	MaxBatchCount   int                    `json:"max_batch_count" yaml:"max_batch_count"`
	TLS             btls.Config            `json:"tls" yaml:"tls"`
}

AMQPConfig contains configuration for the AMQP input type.

func NewAMQPConfig

func NewAMQPConfig() AMQPConfig

NewAMQPConfig creates a new AMQPConfig with default values.

type AMQPQueueDeclareConfig

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

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

type AmazonS3

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

AmazonS3 is a benthos reader.Type implementation that reads messages from an Amazon S3 bucket.

func NewAmazonS3

func NewAmazonS3(
	conf AmazonS3Config,
	log log.Modular,
	stats metrics.Type,
) (*AmazonS3, error)

NewAmazonS3 creates a new Amazon S3 bucket reader.Type.

func (*AmazonS3) Acknowledge

func (a *AmazonS3) Acknowledge(err error) error

Acknowledge confirms whether or not our unacknowledged messages have been successfully propagated or not.

func (*AmazonS3) CloseAsync

func (a *AmazonS3) CloseAsync()

CloseAsync begins cleaning up resources used by this reader asynchronously.

func (*AmazonS3) Connect

func (a *AmazonS3) Connect() error

Connect attempts to establish a connection to the target S3 bucket and any relevant queues used to traverse the objects (SQS, etc).

func (*AmazonS3) ConnectWithContext added in v3.3.0

func (a *AmazonS3) ConnectWithContext(ctx context.Context) error

ConnectWithContext attempts to establish a connection to the target S3 bucket and any relevant queues used to traverse the objects (SQS, etc).

func (*AmazonS3) Read

func (a *AmazonS3) Read() (types.Message, error)

Read attempts to read a new message from the target S3 bucket.

func (*AmazonS3) ReadWithContext added in v3.3.0

func (a *AmazonS3) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the target S3 bucket.

func (*AmazonS3) WaitForClose

func (a *AmazonS3) WaitForClose(time.Duration) error

WaitForClose will block until either the reader is closed or a specified timeout occurs.

type AmazonS3Config

type AmazonS3Config struct {
	sess.Config        `json:",inline" yaml:",inline"`
	Bucket             string                  `json:"bucket" yaml:"bucket"`
	Prefix             string                  `json:"prefix" yaml:"prefix"`
	Retries            int                     `json:"retries" yaml:"retries"`
	ForcePathStyleURLs bool                    `json:"force_path_style_urls" yaml:"force_path_style_urls"`
	DownloadManager    S3DownloadManagerConfig `json:"download_manager" yaml:"download_manager"`
	DeleteObjects      bool                    `json:"delete_objects" yaml:"delete_objects"`
	SQSURL             string                  `json:"sqs_url" yaml:"sqs_url"`
	SQSEndpoint        string                  `json:"sqs_endpoint" yaml:"sqs_endpoint"`
	SQSBodyPath        string                  `json:"sqs_body_path" yaml:"sqs_body_path"`
	SQSBucketPath      string                  `json:"sqs_bucket_path" yaml:"sqs_bucket_path"`
	SQSEnvelopePath    string                  `json:"sqs_envelope_path" yaml:"sqs_envelope_path"`
	SQSMaxMessages     int64                   `json:"sqs_max_messages" yaml:"sqs_max_messages"`
	MaxBatchCount      int                     `json:"max_batch_count" yaml:"max_batch_count"`
	Timeout            string                  `json:"timeout" yaml:"timeout"`
}

AmazonS3Config contains configuration values for the AmazonS3 input type.

func NewAmazonS3Config

func NewAmazonS3Config() AmazonS3Config

NewAmazonS3Config creates a new AmazonS3Config with default values.

type AmazonSQS

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

AmazonSQS is a benthos reader.Type implementation that reads messages from an Amazon SQS queue.

func NewAmazonSQS

func NewAmazonSQS(
	conf AmazonSQSConfig,
	log log.Modular,
	stats metrics.Type,
) (*AmazonSQS, error)

NewAmazonSQS creates a new Amazon SQS reader.Type.

func (*AmazonSQS) Acknowledge

func (a *AmazonSQS) Acknowledge(err error) error

Acknowledge confirms whether or not our unacknowledged messages have been successfully propagated or not.

func (*AmazonSQS) CloseAsync

func (a *AmazonSQS) CloseAsync()

CloseAsync begins cleaning up resources used by this reader asynchronously.

func (*AmazonSQS) Connect

func (a *AmazonSQS) Connect() error

Connect attempts to establish a connection to the target SQS queue.

func (*AmazonSQS) ConnectWithContext added in v3.3.0

func (a *AmazonSQS) ConnectWithContext(ctx context.Context) error

ConnectWithContext attempts to establish a connection to the target SQS queue.

func (*AmazonSQS) Read

func (a *AmazonSQS) Read() (types.Message, error)

Read attempts to read a new message from the target SQS.

func (*AmazonSQS) ReadWithContext added in v3.3.0

func (a *AmazonSQS) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the target SQS.

func (*AmazonSQS) WaitForClose

func (a *AmazonSQS) WaitForClose(time.Duration) error

WaitForClose will block until either the reader is closed or a specified timeout occurs.

type AmazonSQSConfig

type AmazonSQSConfig struct {
	sess.Config         `json:",inline" yaml:",inline"`
	URL                 string `json:"url" yaml:"url"`
	Timeout             string `json:"timeout" yaml:"timeout"`
	MaxNumberOfMessages int64  `json:"max_number_of_messages" yaml:"max_number_of_messages"`
	DeleteMessage       bool   `json:"delete_message" yaml:"delete_message"`
}

AmazonSQSConfig contains configuration values for the input type.

func NewAmazonSQSConfig

func NewAmazonSQSConfig() AmazonSQSConfig

NewAmazonSQSConfig creates a new Config with default values.

type Async added in v3.1.0

type Async interface {
	// ConnectWithContext attempts to establish a connection to the source, if
	// unsuccessful returns an error. If the attempt is successful (or not
	// necessary) returns nil.
	ConnectWithContext(ctx context.Context) error

	// ReadWithContext attempts to read a new message from the source. If
	// successful a message is returned along with a function used to
	// acknowledge receipt of the returned message. It's safe to process the
	// returned message and read the next message asynchronously.
	ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

	types.Closable
}

Async is a type that reads Benthos messages from an external source and allows acknowledgements for a message batch to be propagated asynchronously. If the source supports acknowledgements then it is the responsibility of Type implementations to ensure acknowledgements are not sent for consumed messages until a subsequent Acknowledge call contains a nil error.

func NewAsyncBatcher added in v3.2.0

func NewAsyncBatcher(
	batchConfig batch.PolicyConfig,
	r Async,
	mgr types.Manager,
	log log.Modular,
	stats metrics.Type,
) (Async, error)

NewAsyncBatcher returns a new AsyncBatcher wrapper around a reader.Async.

type AsyncAckFn added in v3.1.0

type AsyncAckFn func(context.Context, types.Response) error

AsyncAckFn is a function used to acknowledge receipt of a message batch. The provided response indicates whether the message batch was successfully delivered. Returns an error if the acknowledge was not propagated.

type AsyncBatcher added in v3.2.0

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

AsyncBatcher is a wrapper for reader.Async implementations that applies a batching policy to incoming payloads. Once a batch is created and sent the provided ack function ensures all messages within the batch are acked.

func (*AsyncBatcher) CloseAsync added in v3.2.0

func (p *AsyncBatcher) CloseAsync()

CloseAsync triggers the asynchronous closing of the reader.

func (*AsyncBatcher) ConnectWithContext added in v3.2.0

func (p *AsyncBatcher) ConnectWithContext(ctx context.Context) error

ConnectWithContext attempts to establish a connection to the source, if unsuccessful returns an error. If the attempt is successful (or not necessary) returns nil.

func (*AsyncBatcher) ReadWithContext added in v3.2.0

func (p *AsyncBatcher) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the source.

func (*AsyncBatcher) WaitForClose added in v3.2.0

func (p *AsyncBatcher) WaitForClose(tout time.Duration) error

WaitForClose blocks until either the reader is finished closing or a timeout occurs.

type AsyncBundleUnacks added in v3.3.0

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

AsyncBundleUnacks is a wrapper for reader.Async implementations that, whenever an unack is given as a response to an async message, keeps the underlying ack function buffered. Once a non-unack response is received it is provided to all buffered ack functions.

Unacks are only returned by the batch processor, and once it is removed this component can also be removed.

TODO: V4 Remove this.

func NewAsyncBundleUnacks added in v3.3.0

func NewAsyncBundleUnacks(r Async) *AsyncBundleUnacks

NewAsyncBundleUnacks returns a new AsyncBundleUnacks wrapper around a reader.Async.

func (*AsyncBundleUnacks) CloseAsync added in v3.3.0

func (p *AsyncBundleUnacks) CloseAsync()

CloseAsync triggers the asynchronous closing of the reader.

func (*AsyncBundleUnacks) ConnectWithContext added in v3.3.0

func (p *AsyncBundleUnacks) ConnectWithContext(ctx context.Context) error

ConnectWithContext attempts to establish a connection to the source, if unsuccessful returns an error. If the attempt is successful (or not necessary) returns nil.

func (*AsyncBundleUnacks) ReadWithContext added in v3.3.0

func (p *AsyncBundleUnacks) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the source.

func (*AsyncBundleUnacks) WaitForClose added in v3.3.0

func (p *AsyncBundleUnacks) WaitForClose(tout time.Duration) error

WaitForClose blocks until either the reader is finished closing or a timeout occurs.

type AsyncCutOff added in v3.3.0

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

AsyncCutOff is a wrapper for reader.Async implementations that exits from WaitForClose immediately. This is only useful when the underlying readable resource cannot be closed reliably and can block forever.

func NewAsyncCutOff added in v3.3.0

func NewAsyncCutOff(r Async) *AsyncCutOff

NewAsyncCutOff returns a new AsyncCutOff wrapper around a reader.Async.

func (*AsyncCutOff) CloseAsync added in v3.3.0

func (c *AsyncCutOff) CloseAsync()

CloseAsync triggers the asynchronous closing of the reader.

func (*AsyncCutOff) ConnectWithContext added in v3.3.0

func (c *AsyncCutOff) ConnectWithContext(ctx context.Context) error

ConnectWithContext attempts to establish a connection to the source, if unsuccessful returns an error. If the attempt is successful (or not necessary) returns nil.

func (*AsyncCutOff) ReadWithContext added in v3.3.0

func (c *AsyncCutOff) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the source.

func (*AsyncCutOff) WaitForClose added in v3.3.0

func (c *AsyncCutOff) WaitForClose(tout time.Duration) error

WaitForClose blocks until either the reader is finished closing or a timeout occurs.

type AsyncPreserver added in v3.1.0

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

AsyncPreserver is a wrapper for reader.Async implementations that keeps a buffer of sent messages until they are acknowledged. If an error occurs during message propagation the contents of the buffer will be resent instead of reading new messages until it is depleted. AsyncPreserver implements reader.Async.

Wrapping an input with this type is useful when your source of messages doesn't have a concept of a NoAck (like Kafka), and instead of "rejecting" messages we always intend to simply retry them until success.

func NewAsyncPreserver added in v3.1.0

func NewAsyncPreserver(r Async) *AsyncPreserver

NewAsyncPreserver returns a new AsyncPreserver wrapper around a reader.Async.

func (*AsyncPreserver) CloseAsync added in v3.1.0

func (p *AsyncPreserver) CloseAsync()

CloseAsync triggers the asynchronous closing of the reader.

func (*AsyncPreserver) ConnectWithContext added in v3.2.0

func (p *AsyncPreserver) ConnectWithContext(ctx context.Context) error

ConnectWithContext attempts to establish a connection to the source, if unsuccessful returns an error. If the attempt is successful (or not necessary) returns nil.

func (*AsyncPreserver) ReadWithContext added in v3.2.0

func (p *AsyncPreserver) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the source.

func (*AsyncPreserver) WaitForClose added in v3.1.0

func (p *AsyncPreserver) WaitForClose(tout time.Duration) error

WaitForClose blocks until either the reader is finished closing or a timeout occurs.

type CutOff

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

CutOff is a wrapper for reader.Type implementations that exits from WaitForClose immediately. This is only useful when the underlying readable resource cannot be closed reliably and can block forever.

func NewCutOff

func NewCutOff(r Type) *CutOff

NewCutOff returns a new CutOff wrapper around a reader.Type.

func (*CutOff) Acknowledge

func (c *CutOff) Acknowledge(err error) error

Acknowledge instructs whether messages read since the last Acknowledge call were successfully propagated. If the error is nil this will be forwarded to the underlying wrapped reader. If a non-nil error is returned the buffer of messages will be resent.

func (*CutOff) CloseAsync

func (c *CutOff) CloseAsync()

CloseAsync triggers the asynchronous closing of the reader.

func (*CutOff) Connect

func (c *CutOff) Connect() error

Connect attempts to establish a connection to the source, if unsuccessful returns an error. If the attempt is successful (or not necessary) returns nil.

func (*CutOff) Read

func (c *CutOff) Read() (types.Message, error)

Read attempts to read a new message from the source.

func (*CutOff) WaitForClose

func (c *CutOff) WaitForClose(tout time.Duration) error

WaitForClose blocks until either the reader is finished closing or a timeout occurs.

type Files

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

Files is an input type that reads file contents at a path as messages.

func NewFiles

func NewFiles(conf FilesConfig) (*Files, error)

NewFiles creates a new Files input type.

func (*Files) Acknowledge

func (f *Files) Acknowledge(err error) error

Acknowledge instructs whether unacknowledged messages have been successfully propagated.

func (*Files) CloseAsync

func (f *Files) CloseAsync()

CloseAsync shuts down the Files input and stops processing requests.

func (*Files) Connect

func (f *Files) Connect() (err error)

Connect establishes a connection.

func (*Files) ConnectWithContext added in v3.3.0

func (f *Files) ConnectWithContext(ctx context.Context) (err error)

ConnectWithContext establishes a connection.

func (*Files) Read

func (f *Files) Read() (types.Message, error)

Read a new Files message.

func (*Files) ReadWithContext added in v3.3.0

func (f *Files) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext a new Files message.

func (*Files) WaitForClose

func (f *Files) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the Files input has closed down.

type FilesConfig

type FilesConfig struct {
	Path        string `json:"path" yaml:"path"`
	DeleteFiles bool   `json:"delete_files" yaml:"delete_files"`
}

FilesConfig contains configuration for the Files input type.

func NewFilesConfig

func NewFilesConfig() FilesConfig

NewFilesConfig creates a new FilesConfig with default values.

type GCPPubSub

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

GCPPubSub is a benthos reader.Type implementation that reads messages from a GCP Cloud Pub/Sub subscription.

func NewGCPPubSub

func NewGCPPubSub(
	conf GCPPubSubConfig,
	log log.Modular,
	stats metrics.Type,
) (*GCPPubSub, error)

NewGCPPubSub creates a new GCP pubsub reader.Type.

func (*GCPPubSub) Acknowledge

func (c *GCPPubSub) Acknowledge(err error) error

Acknowledge confirms whether or not our unacknowledged messages have been successfully propagated or not.

func (*GCPPubSub) CloseAsync

func (c *GCPPubSub) CloseAsync()

CloseAsync begins cleaning up resources used by this reader asynchronously.

func (*GCPPubSub) Connect

func (c *GCPPubSub) Connect() error

Connect attempts to establish a connection to the target subscription.

func (*GCPPubSub) ConnectWithContext added in v3.3.0

func (c *GCPPubSub) ConnectWithContext(ignored context.Context) error

ConnectWithContext attempts to establish a connection to the target subscription.

func (*GCPPubSub) Read

func (c *GCPPubSub) Read() (types.Message, error)

Read attempts to read a new message from the target subscription.

func (*GCPPubSub) ReadWithContext added in v3.3.0

func (c *GCPPubSub) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the target subscription.

func (*GCPPubSub) WaitForClose

func (c *GCPPubSub) WaitForClose(time.Duration) error

WaitForClose will block until either the reader is closed or a specified timeout occurs.

type GCPPubSubConfig

type GCPPubSubConfig struct {
	ProjectID              string `json:"project" yaml:"project"`
	SubscriptionID         string `json:"subscription" yaml:"subscription"`
	MaxOutstandingMessages int    `json:"max_outstanding_messages" yaml:"max_outstanding_messages"`
	MaxOutstandingBytes    int    `json:"max_outstanding_bytes" yaml:"max_outstanding_bytes"`
	Sync                   bool   `json:"sync" yaml:"sync"`
	// TODO: V4 Remove these.
	MaxBatchCount int                `json:"max_batch_count" yaml:"max_batch_count"`
	Batching      batch.PolicyConfig `json:"batching" yaml:"batching"`
}

GCPPubSubConfig contains configuration values for the input type.

func NewGCPPubSubConfig

func NewGCPPubSubConfig() GCPPubSubConfig

NewGCPPubSubConfig creates a new Config with default values.

type HDFS

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

HDFS is a benthos reader.Type implementation that reads messages from a HDFS directory.

func NewHDFS

func NewHDFS(
	conf HDFSConfig,
	log log.Modular,
	stats metrics.Type,
) *HDFS

NewHDFS creates a new HDFS writer.Type.

func (*HDFS) Acknowledge

func (h *HDFS) Acknowledge(err error) error

Acknowledge instructs whether unacknowledged messages have been successfully propagated.

func (*HDFS) CloseAsync

func (h *HDFS) CloseAsync()

CloseAsync shuts down the HDFS input and stops processing requests.

func (*HDFS) Connect

func (h *HDFS) Connect() error

Connect attempts to establish a connection to the target HDFS host.

func (*HDFS) ConnectWithContext added in v3.3.0

func (h *HDFS) ConnectWithContext(ctx context.Context) error

ConnectWithContext attempts to establish a connection to the target HDFS host.

func (*HDFS) Read

func (h *HDFS) Read() (types.Message, error)

Read a new HDFS message.

func (*HDFS) ReadWithContext added in v3.3.0

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

ReadWithContext reads a new HDFS message.

func (*HDFS) WaitForClose

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

WaitForClose blocks until the HDFS input has closed down.

type HDFSConfig

type HDFSConfig struct {
	Hosts     []string `json:"hosts" yaml:"hosts"`
	User      string   `json:"user" yaml:"user"`
	Directory string   `json:"directory" yaml:"directory"`
}

HDFSConfig contains configuration fields for the HDFS input type.

func NewHDFSConfig

func NewHDFSConfig() HDFSConfig

NewHDFSConfig creates a new Config with default values.

type HTTPClient deprecated added in v3.3.0

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

HTTPClient is a reader that continuously polls an HTTP endpoint, providing an optional payload each time.

Deprecated: This component is no longer used by Benthos, instead look at ./lib/input/http_client.go.

func NewHTTPClient deprecated added in v3.3.0

func NewHTTPClient(payload types.Message, httpClient *client.Type, opts ...HTTPClientOptFunc) (*HTTPClient, error)

NewHTTPClient creates a new HTTPClient reader type.

Deprecated: This component is no longer used by Benthos, instead look at ./lib/input/http_client.go.

func (*HTTPClient) CloseAsync added in v3.3.0

func (h *HTTPClient) CloseAsync()

CloseAsync shuts down the HTTPClient input and stops processing requests.

func (*HTTPClient) Connect added in v3.3.0

func (h *HTTPClient) Connect() (err error)

Connect establishes a connection.

func (*HTTPClient) ConnectWithContext added in v3.3.0

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

ConnectWithContext establishes a connection.

func (*HTTPClient) ReadWithContext added in v3.3.0

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

ReadWithContext a new HTTPClient message.

func (*HTTPClient) WaitForClose added in v3.3.0

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

WaitForClose blocks until the HTTPClient input has closed down.

type HTTPClientOptFunc added in v3.18.0

type HTTPClientOptFunc func(*HTTPClient)

HTTPClientOptFunc changes the behaviour of an HTTPClient reader.

func HTTPClientOptSetDropEmpty added in v3.18.0

func HTTPClientOptSetDropEmpty(dropEmpty bool) HTTPClientOptFunc

HTTPClientOptSetDropEmpty determines whether payloads received that are empty should be dropped rather than propagated as an empty benthos message.

type Kafka

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

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

func NewKafka

func NewKafka(
	conf KafkaConfig, mgr types.Manager, log log.Modular, stats metrics.Type,
) (*Kafka, error)

NewKafka creates a new Kafka input type.

func (*Kafka) Acknowledge

func (k *Kafka) Acknowledge(err error) error

Acknowledge instructs whether the current offset should be committed.

func (*Kafka) AcknowledgeWithContext added in v3.3.0

func (k *Kafka) AcknowledgeWithContext(ctx context.Context, err error) error

AcknowledgeWithContext instructs whether the current offset should be committed.

func (*Kafka) CloseAsync

func (k *Kafka) CloseAsync()

CloseAsync shuts down the Kafka input and stops processing requests.

func (*Kafka) Connect

func (k *Kafka) Connect() error

Connect establishes a Kafka connection.

func (*Kafka) ConnectWithContext added in v3.3.0

func (k *Kafka) ConnectWithContext(ctx context.Context) error

ConnectWithContext establishes a Kafka connection.

func (*Kafka) Read

func (k *Kafka) Read() (types.Message, error)

Read attempts to read a message from a Kafka topic.

func (*Kafka) ReadNextWithContext added in v3.3.0

func (k *Kafka) ReadNextWithContext(ctx context.Context) (types.Message, error)

ReadNextWithContext attempts to read a message from a Kafka topic.

func (*Kafka) WaitForClose

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

WaitForClose blocks until the Kafka input has closed down.

type KafkaBalanced

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

KafkaBalanced is an input type that reads from a Kafka cluster by balancing partitions across other consumers of the same consumer group.

func NewKafkaBalanced

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

NewKafkaBalanced creates a new KafkaBalanced input type.

func (*KafkaBalanced) Acknowledge

func (k *KafkaBalanced) Acknowledge(err error) error

Acknowledge instructs whether the current offset should be committed.

func (*KafkaBalanced) Cleanup

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited but before the offsets are committed for the very last time.

func (*KafkaBalanced) CloseAsync

func (k *KafkaBalanced) CloseAsync()

CloseAsync shuts down the KafkaBalanced input and stops processing requests.

func (*KafkaBalanced) Connect

func (k *KafkaBalanced) Connect() error

Connect establishes a KafkaBalanced connection.

func (*KafkaBalanced) ConsumeClaim

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages(). Once the Messages() channel is closed, the Handler must finish its processing loop and exit.

func (*KafkaBalanced) Read

func (k *KafkaBalanced) Read() (types.Message, error)

Read attempts to read a message from a KafkaBalanced topic.

func (*KafkaBalanced) Setup

Setup is run at the beginning of a new session, before ConsumeClaim.

func (*KafkaBalanced) WaitForClose

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

WaitForClose blocks until the KafkaBalanced input has closed down.

type KafkaBalancedConfig

type KafkaBalancedConfig struct {
	Addresses           []string                 `json:"addresses" yaml:"addresses"`
	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"`
	MaxProcessingPeriod string                   `json:"max_processing_period" yaml:"max_processing_period"`
	FetchBufferCap      int                      `json:"fetch_buffer_cap" yaml:"fetch_buffer_cap"`
	Topics              []string                 `json:"topics" yaml:"topics"`
	Batching            batch.PolicyConfig       `json:"batching" yaml:"batching"`
	StartFromOldest     bool                     `json:"start_from_oldest" yaml:"start_from_oldest"`
	TargetVersion       string                   `json:"target_version" yaml:"target_version"`
	// TODO: V4 Remove this.
	MaxBatchCount int         `json:"max_batch_count" yaml:"max_batch_count"`
	TLS           btls.Config `json:"tls" yaml:"tls"`
	SASL          sasl.Config `json:"sasl" yaml:"sasl"`
}

KafkaBalancedConfig contains configuration for the KafkaBalanced input type.

func NewKafkaBalancedConfig

func NewKafkaBalancedConfig() KafkaBalancedConfig

NewKafkaBalancedConfig creates a new KafkaBalancedConfig with default values. TODO: V4 Remove this unused implementation.

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 KafkaCG added in v3.1.0

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

KafkaCG is an input type that reads from a Kafka cluster by balancing partitions across other consumers of the same consumer group.

func NewKafkaCG added in v3.1.0

func NewKafkaCG(
	conf KafkaBalancedConfig, mgr types.Manager, log log.Modular, stats metrics.Type,
) (*KafkaCG, error)

NewKafkaCG creates a new KafkaCG input type.

func (*KafkaCG) Cleanup added in v3.1.0

func (k *KafkaCG) Cleanup(sesh sarama.ConsumerGroupSession) error

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited but before the offsets are committed for the very last time.

func (*KafkaCG) CloseAsync added in v3.1.0

func (k *KafkaCG) CloseAsync()

CloseAsync shuts down the KafkaCG input and stops processing requests.

func (*KafkaCG) ConnectWithContext added in v3.2.0

func (k *KafkaCG) ConnectWithContext(ctx context.Context) error

ConnectWithContext establishes a KafkaCG connection.

func (*KafkaCG) ConsumeClaim added in v3.1.0

func (k *KafkaCG) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages(). Once the Messages() channel is closed, the Handler must finish its processing loop and exit.

func (*KafkaCG) ReadWithContext added in v3.2.0

func (k *KafkaCG) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a message from a KafkaCG topic.

func (*KafkaCG) Setup added in v3.1.0

func (k *KafkaCG) Setup(sesh sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim.

func (*KafkaCG) WaitForClose added in v3.1.0

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

WaitForClose blocks until the KafkaCG input has closed down.

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            batch.PolicyConfig       `json:"batching" yaml:"batching"`

	// TODO: V4 Remove this.
	Topic         string `json:"topic" yaml:"topic"`
	Partition     int32  `json:"partition" yaml:"partition"`
	MaxBatchCount int    `json:"max_batch_count" yaml:"max_batch_count"`
	// contains filtered or unexported fields
}

KafkaConfig contains configuration fields for the Kafka input type.

func NewKafkaConfig

func NewKafkaConfig() KafkaConfig

NewKafkaConfig creates a new KafkaConfig with default values.

func (KafkaConfig) IsDeprecated added in v3.40.0

func (k KafkaConfig) IsDeprecated() bool

IsDeprecated returns a boolean indicating whether this configuration uses the old topic/partition fields.

func (*KafkaConfig) UnmarshalYAML added in v3.40.0

func (k *KafkaConfig) UnmarshalYAML(value *yaml.Node) error

UnmarshalYAML checks while parsing a Kafka config whether any deprecated fields (topic, partition) have been specified.

type Kinesis

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

Kinesis is a benthos reader.Type implementation that reads messages from an Amazon Kinesis stream.

func NewKinesis

func NewKinesis(
	conf KinesisConfig,
	log log.Modular,
	stats metrics.Type,
) (*Kinesis, error)

NewKinesis creates a new Amazon Kinesis stream reader.Type.

func (*Kinesis) Acknowledge

func (k *Kinesis) Acknowledge(err error) error

Acknowledge confirms whether or not our unacknowledged messages have been successfully propagated or not.

func (*Kinesis) AcknowledgeWithContext added in v3.3.0

func (k *Kinesis) AcknowledgeWithContext(ctx context.Context, err error) error

AcknowledgeWithContext confirms whether or not our unacknowledged messages have been successfully propagated or not.

func (*Kinesis) CloseAsync

func (k *Kinesis) CloseAsync()

CloseAsync begins cleaning up resources used by this reader asynchronously.

func (*Kinesis) Connect

func (k *Kinesis) Connect() error

Connect attempts to establish a connection to the target SQS queue.

func (*Kinesis) ConnectWithContext added in v3.3.0

func (k *Kinesis) ConnectWithContext(ctx context.Context) error

ConnectWithContext attempts to establish a connection to the target Kinesis shard.

func (*Kinesis) Read

func (k *Kinesis) Read() (types.Message, error)

Read attempts to read a new message from the target SQS.

func (*Kinesis) ReadNextWithContext added in v3.3.0

func (k *Kinesis) ReadNextWithContext(ctx context.Context) (types.Message, error)

ReadNextWithContext attempts to read a new message from the target Kinesis shard.

func (*Kinesis) WaitForClose

func (k *Kinesis) WaitForClose(time.Duration) error

WaitForClose will block until either the reader is closed or a specified timeout occurs.

type KinesisBalanced

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

KinesisBalanced is a benthos reader.Type implementation that reads messages from an Amazon Kinesis stream.

func NewKinesisBalanced

func NewKinesisBalanced(
	conf KinesisBalancedConfig,
	log log.Modular,
	stats metrics.Type,
) (*KinesisBalanced, error)

NewKinesisBalanced creates a new Amazon Kinesis stream reader.Type.

func (*KinesisBalanced) Acknowledge

func (k *KinesisBalanced) Acknowledge(err error) error

Acknowledge confirms whether or not our unacknowledged messages have been successfully propagated or not.

func (*KinesisBalanced) CloseAsync

func (k *KinesisBalanced) CloseAsync()

CloseAsync begins cleaning up resources used by this reader asynchronously.

func (*KinesisBalanced) Connect

func (k *KinesisBalanced) Connect() error

Connect attempts to establish a connection to the target Kinesis stream.

func (*KinesisBalanced) ConnectWithContext added in v3.3.0

func (k *KinesisBalanced) ConnectWithContext(ctx context.Context) error

ConnectWithContext attempts to establish a connection to the target Kinesis stream.

func (*KinesisBalanced) Init

func (k *KinesisBalanced) Init(shardID string) error

Init is required by the KinesisConsumer interface

func (*KinesisBalanced) ProcessRecords

func (k *KinesisBalanced) ProcessRecords(records []*gokini.Records, _ *gokini.KinesisConsumer)

ProcessRecords implements the KinesisConsumer interface

func (*KinesisBalanced) Read

func (k *KinesisBalanced) Read() (types.Message, error)

Read attempts to read a new message from the target Kinesis stream.

func (*KinesisBalanced) ReadWithContext added in v3.3.0

func (k *KinesisBalanced) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the target Kinesis stream.

func (*KinesisBalanced) Shutdown

func (k *KinesisBalanced) Shutdown()

Shutdown implements the KinesisConsumer interface

func (*KinesisBalanced) WaitForClose

func (k *KinesisBalanced) WaitForClose(time.Duration) error

WaitForClose will block until either the reader is closed or a specified timeout occurs.

type KinesisBalancedConfig

type KinesisBalancedConfig struct {
	sess.Config           `json:",inline" yaml:",inline"`
	Stream                string `json:"stream" yaml:"stream"`
	DynamoDBTable         string `json:"dynamodb_table" yaml:"dynamodb_table"`
	DynamoDBBillingMode   string `json:"dynamodb_billing_mode" yaml:"dynamodb_billing_mode"`
	DynamoDBReadCapacity  int64  `json:"dynamodb_read_provision" yaml:"dynamodb_read_provision"`
	DynamoDBWriteCapacity int64  `json:"dynamodb_write_provision" yaml:"dynamodb_write_provision"`
	// TODO: V4 Remove this.
	MaxBatchCount   int                `json:"max_batch_count" yaml:"max_batch_count"`
	Batching        batch.PolicyConfig `json:"batching" yaml:"batching"`
	StartFromOldest bool               `json:"start_from_oldest" yaml:"start_from_oldest"`
}

KinesisBalancedConfig is configuration values for the input type.

func NewKinesisBalancedConfig

func NewKinesisBalancedConfig() KinesisBalancedConfig

NewKinesisBalancedConfig creates a new Config with default values.

type KinesisConfig

type KinesisConfig struct {
	sess.Config     `json:",inline" yaml:",inline"`
	Limit           int64              `json:"limit" yaml:"limit"`
	Stream          string             `json:"stream" yaml:"stream"`
	Shard           string             `json:"shard" yaml:"shard"`
	DynamoDBTable   string             `json:"dynamodb_table" yaml:"dynamodb_table"`
	ClientID        string             `json:"client_id" yaml:"client_id"`
	CommitPeriod    string             `json:"commit_period" yaml:"commit_period"`
	StartFromOldest bool               `json:"start_from_oldest" yaml:"start_from_oldest"`
	Timeout         string             `json:"timeout" yaml:"timeout"`
	Batching        batch.PolicyConfig `json:"batching" yaml:"batching"`
}

KinesisConfig is configuration values for the input type.

func NewKinesisConfig

func NewKinesisConfig() KinesisConfig

NewKinesisConfig creates a new Config with default values.

type Lines

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

Lines is a reader implementation that continuously reads line delimited messages from an io.Reader type.

func NewLines

func NewLines(
	handleCtor func() (io.Reader, error),
	onClose func(),
	options ...func(r *Lines),
) (*Lines, error)

NewLines creates a new reader input type able to create a feed of line delimited messages from an io.Reader.

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

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

func NewLinesWithContext added in v3.3.0

func NewLinesWithContext(
	handleCtor func(ctx context.Context) (io.Reader, error),
	onClose func(ctx context.Context),
	options ...func(r *Lines),
) (*Lines, error)

NewLinesWithContext expands NewLines by requiring context.Context arguments in the provided closures.

func (*Lines) Acknowledge

func (r *Lines) Acknowledge(err error) error

Acknowledge confirms whether or not our unacknowledged messages have been successfully propagated or not.

func (*Lines) CloseAsync

func (r *Lines) CloseAsync()

CloseAsync shuts down the reader input and stops processing requests.

func (*Lines) Connect

func (r *Lines) Connect() error

Connect attempts to establish a new scanner for an io.Reader.

func (*Lines) ConnectWithContext added in v3.3.0

func (r *Lines) ConnectWithContext(ctx context.Context) error

ConnectWithContext attempts to establish a new scanner for an io.Reader.

func (*Lines) Read

func (r *Lines) Read() (types.Message, error)

Read attempts to read a new line from the io.Reader.

func (*Lines) ReadWithContext added in v3.3.0

func (r *Lines) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new line from the io.Reader.

func (*Lines) WaitForClose

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

WaitForClose blocks until the reader input has closed down.

type MQTT

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

MQTT is an input type that reads MQTT Pub/Sub messages.

func NewMQTT

func NewMQTT(
	conf MQTTConfig, log log.Modular, stats metrics.Type,
) (*MQTT, error)

NewMQTT creates a new MQTT input type.

func (*MQTT) Acknowledge

func (m *MQTT) Acknowledge(err error) error

Acknowledge instructs whether messages have been successfully propagated.

func (*MQTT) CloseAsync

func (m *MQTT) CloseAsync()

CloseAsync shuts down the MQTT input and stops processing requests.

func (*MQTT) Connect

func (m *MQTT) Connect() error

Connect establishes a connection to an MQTT server.

func (*MQTT) ConnectWithContext added in v3.3.0

func (m *MQTT) ConnectWithContext(ctx context.Context) error

ConnectWithContext establishes a connection to an MQTT server.

func (*MQTT) Read

func (m *MQTT) Read() (types.Message, error)

Read attempts to read a new message from an MQTT broker.

func (*MQTT) ReadWithContext added in v3.3.0

func (m *MQTT) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from an MQTT broker.

func (*MQTT) WaitForClose

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

WaitForClose blocks until the MQTT input has closed down.

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"`
	StaleConnectionTimeout string        `json:"stale_connection_timeout" yaml:"stale_connection_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 NATS

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

NATS is an input type that receives NATS messages.

func NewNATS

func NewNATS(conf NATSConfig, log log.Modular, stats metrics.Type) (*NATS, error)

NewNATS creates a new NATS input type.

func (*NATS) Acknowledge deprecated

func (n *NATS) Acknowledge(err error) error

Acknowledge confirms whether or not our unacknowledged messages have been successfully propagated or not.

Deprecated: Use ReadWithContext instead.

func (*NATS) CloseAsync

func (n *NATS) CloseAsync()

CloseAsync shuts down the NATS input and stops processing requests.

func (*NATS) Connect

func (n *NATS) Connect() error

Connect establishes a connection to a NATS server.

func (*NATS) ConnectWithContext added in v3.3.0

func (n *NATS) ConnectWithContext(ctx context.Context) error

ConnectWithContext establishes a connection to a NATS server.

func (*NATS) Read deprecated

func (n *NATS) Read() (types.Message, error)

Read attempts to read a new message from the NATS subject.

Deprecated: Use ReadWithContext instead.

func (*NATS) ReadWithContext added in v3.3.0

func (n *NATS) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the NATS subject.

func (*NATS) WaitForClose

func (n *NATS) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the NATS input has closed down.

type NATSConfig

type NATSConfig struct {
	URLs          []string    `json:"urls" yaml:"urls"`
	Subject       string      `json:"subject" yaml:"subject"`
	QueueID       string      `json:"queue" yaml:"queue"`
	PrefetchCount int         `json:"prefetch_count" yaml:"prefetch_count"`
	TLS           btls.Config `json:"tls" yaml:"tls"`
	Auth          auth.Config `json:"auth" yaml:"auth"`
}

NATSConfig contains configuration fields for the NATS input type.

func NewNATSConfig

func NewNATSConfig() NATSConfig

NewNATSConfig creates a new NATSConfig with default values.

type NATSStream

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

NATSStream is an input type that receives NATSStream messages.

func NewNATSStream

func NewNATSStream(conf NATSStreamConfig, log log.Modular, stats metrics.Type) (*NATSStream, error)

NewNATSStream creates a new NATSStream input type.

func (*NATSStream) Acknowledge

func (n *NATSStream) Acknowledge(err error) error

Acknowledge instructs whether unacknowledged messages have been successfully propagated.

func (*NATSStream) CloseAsync

func (n *NATSStream) CloseAsync()

CloseAsync shuts down the NATSStream input and stops processing requests.

func (*NATSStream) Connect

func (n *NATSStream) Connect() error

Connect attempts to establish a connection to a NATS streaming server.

func (*NATSStream) ConnectWithContext added in v3.3.0

func (n *NATSStream) ConnectWithContext(ctx context.Context) error

ConnectWithContext attempts to establish a connection to a NATS streaming server.

func (*NATSStream) Read

func (n *NATSStream) Read() (types.Message, error)

Read attempts to read a new message from the NATS streaming server.

func (*NATSStream) ReadWithContext added in v3.3.0

func (n *NATSStream) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the NATS streaming server.

func (*NATSStream) WaitForClose

func (n *NATSStream) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the NATSStream input has closed down.

type NATSStreamConfig

type NATSStreamConfig struct {
	URLs            []string    `json:"urls" yaml:"urls"`
	ClusterID       string      `json:"cluster_id" yaml:"cluster_id"`
	ClientID        string      `json:"client_id" yaml:"client_id"`
	QueueID         string      `json:"queue" yaml:"queue"`
	DurableName     string      `json:"durable_name" yaml:"durable_name"`
	UnsubOnClose    bool        `json:"unsubscribe_on_close" yaml:"unsubscribe_on_close"`
	StartFromOldest bool        `json:"start_from_oldest" yaml:"start_from_oldest"`
	Subject         string      `json:"subject" yaml:"subject"`
	MaxInflight     int         `json:"max_inflight" yaml:"max_inflight"`
	AckWait         string      `json:"ack_wait" yaml:"ack_wait"`
	TLS             btls.Config `json:"tls" yaml:"tls"`
	Auth            auth.Config `json:"auth" yaml:"auth"`

	// TODO: V4 remove this.
	Batching batch.PolicyConfig `json:"batching" yaml:"batching"`
}

NATSStreamConfig contains configuration fields for the NATSStream input type.

func NewNATSStreamConfig

func NewNATSStreamConfig() NATSStreamConfig

NewNATSStreamConfig creates a new NATSStreamConfig with default values.

type NSQ

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

NSQ is an input type that receives NSQ messages.

func NewNSQ

func NewNSQ(conf NSQConfig, log log.Modular, stats metrics.Type) (*NSQ, error)

NewNSQ creates a new NSQ input type.

func (*NSQ) Acknowledge

func (n *NSQ) Acknowledge(err error) error

Acknowledge instructs whether unacknowledged messages have been successfully propagated.

func (*NSQ) CloseAsync

func (n *NSQ) CloseAsync()

CloseAsync shuts down the NSQ input and stops processing requests.

func (*NSQ) Connect

func (n *NSQ) Connect() (err error)

Connect establishes a connection to an NSQ server.

func (*NSQ) ConnectWithContext added in v3.3.0

func (n *NSQ) ConnectWithContext(ctx context.Context) (err error)

ConnectWithContext establishes a connection to an NSQ server.

func (*NSQ) HandleMessage

func (n *NSQ) HandleMessage(message *nsq.Message) error

HandleMessage handles an NSQ message.

func (*NSQ) Read

func (n *NSQ) Read() (types.Message, error)

Read attempts to read a new message from NSQ.

func (*NSQ) ReadWithContext added in v3.3.0

func (n *NSQ) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from NSQ.

func (*NSQ) WaitForClose

func (n *NSQ) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the NSQ input has closed down.

type NSQConfig

type NSQConfig struct {
	Addresses       []string           `json:"nsqd_tcp_addresses" yaml:"nsqd_tcp_addresses"`
	LookupAddresses []string           `json:"lookupd_http_addresses" yaml:"lookupd_http_addresses"`
	Topic           string             `json:"topic" yaml:"topic"`
	Channel         string             `json:"channel" yaml:"channel"`
	UserAgent       string             `json:"user_agent" yaml:"user_agent"`
	TLS             btls.Config        `json:"tls" yaml:"tls"`
	MaxInFlight     int                `json:"max_in_flight" yaml:"max_in_flight"`
	Batching        batch.PolicyConfig `json:"batching" yaml:"batching"`
}

NSQConfig contains configuration fields for the NSQ input type.

func NewNSQConfig

func NewNSQConfig() NSQConfig

NewNSQConfig creates a new NSQConfig with default values.

type Preserver

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

Preserver is a wrapper for reader.Type implementations that keeps a buffer of sent messages until they are acknowledged. If an error occurs during message propagation the contents of the buffer will be resent instead of reading new messages until it is depleted. Preserver implements reader.Type.

func NewPreserver

func NewPreserver(r Type) *Preserver

NewPreserver returns a new Preserver wrapper around a reader.Type.

func (*Preserver) Acknowledge

func (p *Preserver) Acknowledge(err error) error

Acknowledge instructs whether messages read since the last Acknowledge call were successfully propagated. If the error is nil this will be forwarded to the underlying wrapped reader. If a non-nil error is returned the buffer of messages will be resent.

func (*Preserver) CloseAsync

func (p *Preserver) CloseAsync()

CloseAsync triggers the asynchronous closing of the reader.

func (*Preserver) Connect

func (p *Preserver) Connect() error

Connect attempts to establish a connection to the source, if unsuccessful returns an error. If the attempt is successful (or not necessary) returns nil.

func (*Preserver) Read

func (p *Preserver) Read() (types.Message, error)

Read attempts to read a new message from the source.

func (*Preserver) WaitForClose

func (p *Preserver) WaitForClose(tout time.Duration) error

WaitForClose blocks until either the reader is finished closing or a timeout occurs.

type RedisList

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

RedisList is an input type that reads Redis List messages.

func NewRedisList

func NewRedisList(
	conf RedisListConfig, log log.Modular, stats metrics.Type,
) (*RedisList, error)

NewRedisList creates a new RedisList input type.

func (*RedisList) Acknowledge

func (r *RedisList) Acknowledge(err error) error

Acknowledge is a noop since Redis Lists do not support acknowledgements.

func (*RedisList) CloseAsync

func (r *RedisList) CloseAsync()

CloseAsync shuts down the RedisList input and stops processing requests.

func (*RedisList) Connect

func (r *RedisList) Connect() error

Connect establishes a connection to a Redis server.

func (*RedisList) ConnectWithContext added in v3.3.0

func (r *RedisList) ConnectWithContext(ctx context.Context) error

ConnectWithContext establishes a connection to a Redis server.

func (*RedisList) Read

func (r *RedisList) Read() (types.Message, error)

Read attempts to pop a message from a Redis list.

func (*RedisList) ReadWithContext added in v3.3.0

func (r *RedisList) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to pop a message from a Redis list.

func (*RedisList) WaitForClose

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

WaitForClose blocks until the RedisList input has closed down.

type RedisListConfig

type RedisListConfig struct {
	bredis.Config `json:",inline" yaml:",inline"`
	Key           string `json:"key" yaml:"key"`
	Timeout       string `json:"timeout" yaml:"timeout"`
}

RedisListConfig contains configuration fields for the RedisList input type.

func NewRedisListConfig

func NewRedisListConfig() RedisListConfig

NewRedisListConfig creates a new RedisListConfig with default values.

type RedisPubSub

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

RedisPubSub is an input type that reads Redis Pub/Sub messages.

func NewRedisPubSub

func NewRedisPubSub(
	conf RedisPubSubConfig, log log.Modular, stats metrics.Type,
) (*RedisPubSub, error)

NewRedisPubSub creates a new RedisPubSub input type.

func (*RedisPubSub) Acknowledge

func (r *RedisPubSub) Acknowledge(err error) error

Acknowledge is a noop since Redis pub/sub channels do not support acknowledgements.

func (*RedisPubSub) CloseAsync

func (r *RedisPubSub) CloseAsync()

CloseAsync shuts down the RedisPubSub input and stops processing requests.

func (*RedisPubSub) Connect

func (r *RedisPubSub) Connect() error

Connect establishes a connection to a RedisPubSub server.

func (*RedisPubSub) ConnectWithContext added in v3.3.0

func (r *RedisPubSub) ConnectWithContext(ctx context.Context) error

ConnectWithContext establishes a connection to an RedisPubSub server.

func (*RedisPubSub) Read

func (r *RedisPubSub) Read() (types.Message, error)

Read attempts to pop a message from a redis pubsub channel.

func (*RedisPubSub) ReadWithContext added in v3.3.0

func (r *RedisPubSub) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to pop a message from a redis pubsub channel.

func (*RedisPubSub) WaitForClose

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

WaitForClose blocks until the RedisPubSub input has closed down.

type RedisPubSubConfig

type RedisPubSubConfig struct {
	bredis.Config `json:",inline" yaml:",inline"`
	Channels      []string `json:"channels" yaml:"channels"`
	UsePatterns   bool     `json:"use_patterns" yaml:"use_patterns"`
}

RedisPubSubConfig contains configuration fields for the RedisPubSub input type.

func NewRedisPubSubConfig

func NewRedisPubSubConfig() RedisPubSubConfig

NewRedisPubSubConfig creates a new RedisPubSubConfig with default values.

type RedisStreams

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

RedisStreams is an input type that reads Redis Streams messages.

func NewRedisStreams

func NewRedisStreams(
	conf RedisStreamsConfig, log log.Modular, stats metrics.Type,
) (*RedisStreams, error)

NewRedisStreams creates a new RedisStreams input type.

func (*RedisStreams) Acknowledge

func (r *RedisStreams) Acknowledge(err error) error

Acknowledge is a noop since Redis Lists do not support acknowledgements.

func (*RedisStreams) CloseAsync

func (r *RedisStreams) CloseAsync()

CloseAsync shuts down the RedisStreams input and stops processing requests.

func (*RedisStreams) Connect

func (r *RedisStreams) Connect() error

Connect establishes a connection to a Redis server.

func (*RedisStreams) ConnectWithContext added in v3.3.0

func (r *RedisStreams) ConnectWithContext(ctx context.Context) error

ConnectWithContext establishes a connection to a Redis server.

func (*RedisStreams) Read

func (r *RedisStreams) Read() (types.Message, error)

Read attempts to pop a message from a Redis list.

func (*RedisStreams) ReadWithContext added in v3.3.0

func (r *RedisStreams) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to pop a message from a Redis list.

func (*RedisStreams) WaitForClose

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

WaitForClose blocks until the RedisStreams input has closed down.

type RedisStreamsConfig

type RedisStreamsConfig struct {
	bredis.Config   `json:",inline" yaml:",inline"`
	BodyKey         string   `json:"body_key" yaml:"body_key"`
	Streams         []string `json:"streams" yaml:"streams"`
	CreateStreams   bool     `json:"create_streams" yaml:"create_streams"`
	ConsumerGroup   string   `json:"consumer_group" yaml:"consumer_group"`
	ClientID        string   `json:"client_id" yaml:"client_id"`
	Limit           int64    `json:"limit" yaml:"limit"`
	StartFromOldest bool     `json:"start_from_oldest" yaml:"start_from_oldest"`
	CommitPeriod    string   `json:"commit_period" yaml:"commit_period"`
	Timeout         string   `json:"timeout" yaml:"timeout"`

	// TODO: V4 remove this.
	Batching batch.PolicyConfig `json:"batching" yaml:"batching"`
}

RedisStreamsConfig contains configuration fields for the RedisStreams input type.

func NewRedisStreamsConfig

func NewRedisStreamsConfig() RedisStreamsConfig

NewRedisStreamsConfig creates a new RedisStreamsConfig with default values.

type S3DownloadManagerConfig

type S3DownloadManagerConfig struct {
	Enabled bool `json:"enabled" yaml:"enabled"`
}

S3DownloadManagerConfig is a config struct containing fields for an S3 download manager.

type ScaleProto

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

ScaleProto is an input type that contains Scalability Protocols messages.

func NewScaleProto

func NewScaleProto(conf ScaleProtoConfig, log log.Modular, stats metrics.Type) (*ScaleProto, error)

NewScaleProto creates a new ScaleProto input type.

func (*ScaleProto) Acknowledge

func (s *ScaleProto) Acknowledge(err error) error

Acknowledge instructs whether the pending messages were propagated successfully.

func (*ScaleProto) CloseAsync

func (s *ScaleProto) CloseAsync()

CloseAsync shuts down the ScaleProto input and stops processing requests.

func (*ScaleProto) Connect

func (s *ScaleProto) Connect() error

Connect establishes a nanomsg socket.

func (*ScaleProto) ConnectWithContext added in v3.3.0

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

ConnectWithContext establishes a nanomsg socket.

func (*ScaleProto) Read

func (s *ScaleProto) Read() (types.Message, error)

Read attempts to read a new message from the nanomsg socket.

func (*ScaleProto) ReadWithContext added in v3.3.0

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

ReadWithContext attempts to read a new message from the nanomsg socket.

func (*ScaleProto) WaitForClose

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

WaitForClose blocks until the ScaleProto input has closed down.

type ScaleProtoConfig

type ScaleProtoConfig struct {
	URLs        []string `json:"urls" yaml:"urls"`
	Bind        bool     `json:"bind" yaml:"bind"`
	SocketType  string   `json:"socket_type" yaml:"socket_type"`
	SubFilters  []string `json:"sub_filters" yaml:"sub_filters"`
	PollTimeout string   `json:"poll_timeout" yaml:"poll_timeout"`
	RepTimeout  string   `json:"reply_timeout" yaml:"reply_timeout"`
}

ScaleProtoConfig contains configuration fields for the ScaleProto input type.

func NewScaleProtoConfig

func NewScaleProtoConfig() ScaleProtoConfig

NewScaleProtoConfig creates a new ScaleProtoConfig with default values.

type Sync added in v3.3.0

type Sync interface {
	// ConnectWithContext attempts to establish a connection to the source, if
	// unsuccessful returns an error. If the attempt is successful (or not
	// necessary) returns nil.
	ConnectWithContext(ctx context.Context) error

	// ReadNextWithContext attempts to read a new message from the source. If
	// successful a message is returned. Messages returned remain unacknowledged
	// until the next AcknowledgeWithContext call.
	ReadNextWithContext(ctx context.Context) (types.Message, error)

	// Acknowledge, if applicable to the source, should send acknowledgments for
	// (or commit) all unacknowledged (or uncommitted) messages that have thus
	// far been consumed. If the error is non-nil this means the message was
	// unsuccessfully propagated down the pipeline, in which case it is up to
	// the implementation to decide whether to simply retry uncommitted messages
	// that are buffered locally, or to send the error upstream.
	AcknowledgeWithContext(ctx context.Context, err error) error

	types.Closable
}

Sync is a type that reads Benthos messages from an external source. Messages can be read continuously, but acknowledgements must be made synchronously and apply for all read messages.

type SyncBatcher added in v3.3.0

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

SyncBatcher is a wrapper for reader.Sync implementations that applies a batching policy to incoming payloads. Once a batch is created and sent the next Acknowledge call applies to all messages of the prior batches.

func NewSyncBatcher added in v3.3.0

func NewSyncBatcher(
	batchConfig batch.PolicyConfig,
	r Sync,
	mgr types.Manager,
	log log.Modular,
	stats metrics.Type,
) (*SyncBatcher, error)

NewSyncBatcher returns a new SyncBatcher wrapper around a reader.Async.

func (*SyncBatcher) Acknowledge added in v3.3.0

func (p *SyncBatcher) Acknowledge(err error) error

Acknowledge confirms whether or not our unacknowledged messages have been successfully propagated or not.

func (*SyncBatcher) AcknowledgeWithContext added in v3.3.0

func (p *SyncBatcher) AcknowledgeWithContext(ctx context.Context, err error) error

AcknowledgeWithContext confirms whether or not our unacknowledged messages have been successfully propagated or not.

func (*SyncBatcher) CloseAsync added in v3.3.0

func (p *SyncBatcher) CloseAsync()

CloseAsync triggers the asynchronous closing of the reader.

func (*SyncBatcher) Connect added in v3.3.0

func (p *SyncBatcher) Connect() error

Connect attempts to establish a connection to the source, if unsuccessful returns an error. If the attempt is successful (or not necessary) returns nil.

func (*SyncBatcher) ConnectWithContext added in v3.3.0

func (p *SyncBatcher) ConnectWithContext(ctx context.Context) error

ConnectWithContext attempts to establish a connection to the source, if unsuccessful returns an error. If the attempt is successful (or not necessary) returns nil.

func (*SyncBatcher) Read added in v3.3.0

func (p *SyncBatcher) Read() (types.Message, error)

Read attempts to read a new message from the source.

func (*SyncBatcher) ReadNextWithContext added in v3.3.0

func (p *SyncBatcher) ReadNextWithContext(ctx context.Context) (types.Message, error)

ReadNextWithContext attempts to read a new message from the source.

func (*SyncBatcher) WaitForClose added in v3.3.0

func (p *SyncBatcher) WaitForClose(tout time.Duration) error

WaitForClose blocks until either the reader is finished closing or a timeout occurs.

type Type

type Type interface {
	// Connect attempts to establish a connection to the source, if unsuccessful
	// returns an error. If the attempt is successful (or not necessary) returns
	// nil.
	Connect() error

	// Acknowledge, if applicable to the source, should send acknowledgments for
	// (or commit) all unacknowledged (or uncommitted) messages that have thus
	// far been consumed. If the error is non-nil this means the message was
	// unsuccessfully propagated down the pipeline, in which case it is up to
	// the implementation to decide whether to simply retry uncommitted messages
	// that are buffered locally, or to send the error upstream.
	Acknowledge(err error) error

	// Read attempts to read a new message from the source.
	Read() (types.Message, error)

	types.Closable
}

Type is a type that reads Benthos messages from an external source. If the source supports acknowledgements then it is the responsibility of Type implementations to ensure acknowledgements are not sent for consumed messages until a subsequent Acknowledge call contains a nil error.

func NewAMQP

func NewAMQP(conf AMQPConfig, log log.Modular, stats metrics.Type) (Type, error)

NewAMQP creates a new AMQP input type.

type Websocket

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

Websocket is an input type that reads Websocket messages.

func NewWebsocket

func NewWebsocket(
	conf WebsocketConfig,
	log log.Modular,
	stats metrics.Type,
) (*Websocket, error)

NewWebsocket creates a new Websocket input type.

func (*Websocket) Acknowledge

func (w *Websocket) Acknowledge(err error) error

Acknowledge instructs whether the pending messages were propagated successfully.

func (*Websocket) CloseAsync

func (w *Websocket) CloseAsync()

CloseAsync shuts down the Websocket input and stops reading messages.

func (*Websocket) Connect

func (w *Websocket) Connect() error

Connect establishes a connection to a Websocket server.

func (*Websocket) ConnectWithContext added in v3.3.0

func (w *Websocket) ConnectWithContext(ctx context.Context) error

ConnectWithContext establishes a connection to a Websocket server.

func (*Websocket) Read

func (w *Websocket) Read() (types.Message, error)

Read attempts to read a new message from the websocket.

func (*Websocket) ReadWithContext added in v3.3.0

func (w *Websocket) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the websocket.

func (*Websocket) WaitForClose

func (w *Websocket) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the Websocket input has closed down.

type WebsocketConfig

type WebsocketConfig struct {
	URL         string `json:"url" yaml:"url"`
	OpenMsg     string `json:"open_message" yaml:"open_message"`
	auth.Config `json:",inline" yaml:",inline"`
	TLS         btls.Config `json:"tls" yaml:"tls"`
}

WebsocketConfig contains configuration fields for the Websocket input type.

func NewWebsocketConfig

func NewWebsocketConfig() WebsocketConfig

NewWebsocketConfig creates a new WebsocketConfig with default values.

type ZMQ4Config

type ZMQ4Config struct {
	URLs          []string `json:"urls" yaml:"urls"`
	Bind          bool     `json:"bind" yaml:"bind"`
	SocketType    string   `json:"socket_type" yaml:"socket_type"`
	SubFilters    []string `json:"sub_filters" yaml:"sub_filters"`
	HighWaterMark int      `json:"high_water_mark" yaml:"high_water_mark"`
	PollTimeout   string   `json:"poll_timeout" yaml:"poll_timeout"`
}

ZMQ4Config contains configuration fields for the ZMQ4 input type.

func NewZMQ4Config

func NewZMQ4Config() *ZMQ4Config

NewZMQ4Config creates a new ZMQ4Config with default values.

Jump to

Keyboard shortcuts

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