Documentation ¶
Overview ¶
Package reader defines implementations of an interface for generic message reading from various third party sources.
Index ¶
- func OptLinesSetDelimiter(delimiter string) func(r *Lines)
- func OptLinesSetMaxBuffer(maxBuffer int) func(r *Lines)
- func OptLinesSetMultipart(multipart bool) func(r *Lines)
- type AMQP
- type AMQP09
- type AMQP09BindingConfig
- type AMQP09Config
- type AMQP09QueueDeclareConfig
- type AMQP1
- type AMQP1Config
- type AMQPBindingConfig
- type AMQPConfig
- type AMQPQueueDeclareConfig
- type AmazonS3
- func (a *AmazonS3) Acknowledge(err error) error
- func (a *AmazonS3) CloseAsync()
- func (a *AmazonS3) Connect() error
- func (a *AmazonS3) ConnectWithContext(ctx context.Context) error
- func (a *AmazonS3) Read() (types.Message, error)
- func (a *AmazonS3) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (a *AmazonS3) WaitForClose(time.Duration) error
- type AmazonS3Config
- type AmazonSQS
- func (a *AmazonSQS) Acknowledge(err error) error
- func (a *AmazonSQS) CloseAsync()
- func (a *AmazonSQS) Connect() error
- func (a *AmazonSQS) ConnectWithContext(ctx context.Context) error
- func (a *AmazonSQS) Read() (types.Message, error)
- func (a *AmazonSQS) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (a *AmazonSQS) WaitForClose(time.Duration) error
- type AmazonSQSConfig
- type Async
- type AsyncAckFn
- type AsyncBatcher
- type AsyncBundleUnacks
- type AsyncCutOff
- type AsyncPreserver
- type CutOff
- type Files
- func (f *Files) Acknowledge(err error) error
- func (f *Files) CloseAsync()
- func (f *Files) Connect() (err error)
- func (f *Files) ConnectWithContext(ctx context.Context) (err error)
- func (f *Files) Read() (types.Message, error)
- func (f *Files) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (f *Files) WaitForClose(timeout time.Duration) error
- type FilesConfig
- type GCPPubSub
- func (c *GCPPubSub) Acknowledge(err error) error
- func (c *GCPPubSub) CloseAsync()
- func (c *GCPPubSub) Connect() error
- func (c *GCPPubSub) ConnectWithContext(ignored context.Context) error
- func (c *GCPPubSub) Read() (types.Message, error)
- func (c *GCPPubSub) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (c *GCPPubSub) WaitForClose(time.Duration) error
- type GCPPubSubConfig
- type HDFS
- func (h *HDFS) Acknowledge(err error) error
- func (h *HDFS) CloseAsync()
- func (h *HDFS) Connect() error
- func (h *HDFS) ConnectWithContext(ctx context.Context) error
- func (h *HDFS) Read() (types.Message, error)
- func (h *HDFS) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (h *HDFS) WaitForClose(timeout time.Duration) error
- type HDFSConfig
- type HTTPClient
- func (h *HTTPClient) CloseAsync()
- func (h *HTTPClient) Connect() (err error)
- func (h *HTTPClient) ConnectWithContext(ctx context.Context) (err error)
- func (h *HTTPClient) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (h *HTTPClient) WaitForClose(timeout time.Duration) error
- type HTTPClientOptFunc
- type Kafka
- func (k *Kafka) Acknowledge(err error) error
- func (k *Kafka) AcknowledgeWithContext(ctx context.Context, err error) error
- func (k *Kafka) CloseAsync()
- func (k *Kafka) Connect() error
- func (k *Kafka) ConnectWithContext(ctx context.Context) error
- func (k *Kafka) Read() (types.Message, error)
- func (k *Kafka) ReadNextWithContext(ctx context.Context) (types.Message, error)
- func (k *Kafka) WaitForClose(timeout time.Duration) error
- type KafkaBalanced
- func (k *KafkaBalanced) Acknowledge(err error) error
- func (k *KafkaBalanced) Cleanup(sesh sarama.ConsumerGroupSession) error
- func (k *KafkaBalanced) CloseAsync()
- func (k *KafkaBalanced) Connect() error
- func (k *KafkaBalanced) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
- func (k *KafkaBalanced) Read() (types.Message, error)
- func (k *KafkaBalanced) Setup(sesh sarama.ConsumerGroupSession) error
- func (k *KafkaBalanced) WaitForClose(timeout time.Duration) error
- type KafkaBalancedConfig
- type KafkaBalancedGroupConfig
- type KafkaCG
- func (k *KafkaCG) Cleanup(sesh sarama.ConsumerGroupSession) error
- func (k *KafkaCG) CloseAsync()
- func (k *KafkaCG) ConnectWithContext(ctx context.Context) error
- func (k *KafkaCG) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
- func (k *KafkaCG) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (k *KafkaCG) Setup(sesh sarama.ConsumerGroupSession) error
- func (k *KafkaCG) WaitForClose(timeout time.Duration) error
- type KafkaConfig
- type Kinesis
- func (k *Kinesis) Acknowledge(err error) error
- func (k *Kinesis) AcknowledgeWithContext(ctx context.Context, err error) error
- func (k *Kinesis) CloseAsync()
- func (k *Kinesis) Connect() error
- func (k *Kinesis) ConnectWithContext(ctx context.Context) error
- func (k *Kinesis) Read() (types.Message, error)
- func (k *Kinesis) ReadNextWithContext(ctx context.Context) (types.Message, error)
- func (k *Kinesis) WaitForClose(time.Duration) error
- type KinesisBalanced
- func (k *KinesisBalanced) Acknowledge(err error) error
- func (k *KinesisBalanced) CloseAsync()
- func (k *KinesisBalanced) Connect() error
- func (k *KinesisBalanced) ConnectWithContext(ctx context.Context) error
- func (k *KinesisBalanced) Init(shardID string) error
- func (k *KinesisBalanced) ProcessRecords(records []*gokini.Records, _ *gokini.KinesisConsumer)
- func (k *KinesisBalanced) Read() (types.Message, error)
- func (k *KinesisBalanced) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (k *KinesisBalanced) Shutdown()
- func (k *KinesisBalanced) WaitForClose(time.Duration) error
- type KinesisBalancedConfig
- type KinesisConfig
- type Lines
- func (r *Lines) Acknowledge(err error) error
- func (r *Lines) CloseAsync()
- func (r *Lines) Connect() error
- func (r *Lines) ConnectWithContext(ctx context.Context) error
- func (r *Lines) Read() (types.Message, error)
- func (r *Lines) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (r *Lines) WaitForClose(timeout time.Duration) error
- type MQTT
- func (m *MQTT) Acknowledge(err error) error
- func (m *MQTT) CloseAsync()
- func (m *MQTT) Connect() error
- func (m *MQTT) ConnectWithContext(ctx context.Context) error
- func (m *MQTT) Read() (types.Message, error)
- func (m *MQTT) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (m *MQTT) WaitForClose(timeout time.Duration) error
- type MQTTConfig
- type NATS
- func (n *NATS) Acknowledge(err error) error
- func (n *NATS) CloseAsync()
- func (n *NATS) Connect() error
- func (n *NATS) ConnectWithContext(ctx context.Context) error
- func (n *NATS) Read() (types.Message, error)
- func (n *NATS) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (n *NATS) WaitForClose(timeout time.Duration) error
- type NATSConfig
- type NATSStream
- func (n *NATSStream) Acknowledge(err error) error
- func (n *NATSStream) CloseAsync()
- func (n *NATSStream) Connect() error
- func (n *NATSStream) ConnectWithContext(ctx context.Context) error
- func (n *NATSStream) Read() (types.Message, error)
- func (n *NATSStream) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (n *NATSStream) WaitForClose(timeout time.Duration) error
- type NATSStreamConfig
- type NSQ
- func (n *NSQ) Acknowledge(err error) error
- func (n *NSQ) CloseAsync()
- func (n *NSQ) Connect() (err error)
- func (n *NSQ) ConnectWithContext(ctx context.Context) (err error)
- func (n *NSQ) HandleMessage(message *nsq.Message) error
- func (n *NSQ) Read() (types.Message, error)
- func (n *NSQ) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (n *NSQ) WaitForClose(timeout time.Duration) error
- type NSQConfig
- type Preserver
- type RedisList
- func (r *RedisList) Acknowledge(err error) error
- func (r *RedisList) CloseAsync()
- func (r *RedisList) Connect() error
- func (r *RedisList) ConnectWithContext(ctx context.Context) error
- func (r *RedisList) Read() (types.Message, error)
- func (r *RedisList) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (r *RedisList) WaitForClose(timeout time.Duration) error
- type RedisListConfig
- type RedisPubSub
- func (r *RedisPubSub) Acknowledge(err error) error
- func (r *RedisPubSub) CloseAsync()
- func (r *RedisPubSub) Connect() error
- func (r *RedisPubSub) ConnectWithContext(ctx context.Context) error
- func (r *RedisPubSub) Read() (types.Message, error)
- func (r *RedisPubSub) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (r *RedisPubSub) WaitForClose(timeout time.Duration) error
- type RedisPubSubConfig
- type RedisStreams
- func (r *RedisStreams) Acknowledge(err error) error
- func (r *RedisStreams) CloseAsync()
- func (r *RedisStreams) Connect() error
- func (r *RedisStreams) ConnectWithContext(ctx context.Context) error
- func (r *RedisStreams) Read() (types.Message, error)
- func (r *RedisStreams) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (r *RedisStreams) WaitForClose(timeout time.Duration) error
- type RedisStreamsConfig
- type S3DownloadManagerConfig
- type ScaleProto
- func (s *ScaleProto) Acknowledge(err error) error
- func (s *ScaleProto) CloseAsync()
- func (s *ScaleProto) Connect() error
- func (s *ScaleProto) ConnectWithContext(ctx context.Context) error
- func (s *ScaleProto) Read() (types.Message, error)
- func (s *ScaleProto) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (s *ScaleProto) WaitForClose(timeout time.Duration) error
- type ScaleProtoConfig
- type Sync
- type SyncBatcher
- func (p *SyncBatcher) Acknowledge(err error) error
- func (p *SyncBatcher) AcknowledgeWithContext(ctx context.Context, err error) error
- func (p *SyncBatcher) CloseAsync()
- func (p *SyncBatcher) Connect() error
- func (p *SyncBatcher) ConnectWithContext(ctx context.Context) error
- func (p *SyncBatcher) Read() (types.Message, error)
- func (p *SyncBatcher) ReadNextWithContext(ctx context.Context) (types.Message, error)
- func (p *SyncBatcher) WaitForClose(tout time.Duration) error
- type Type
- type Websocket
- func (w *Websocket) Acknowledge(err error) error
- func (w *Websocket) CloseAsync()
- func (w *Websocket) Connect() error
- func (w *Websocket) ConnectWithContext(ctx context.Context) error
- func (w *Websocket) Read() (types.Message, error)
- func (w *Websocket) ReadWithContext(ctx context.Context) (types.Message, AsyncAckFn, error)
- func (w *Websocket) WaitForClose(timeout time.Duration) error
- type WebsocketConfig
- type ZMQ4Config
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func OptLinesSetDelimiter ¶
OptLinesSetDelimiter is a option func that sets the delimiter (default '\n') used to divide lines (message parts) in the stream of data.
func OptLinesSetMaxBuffer ¶
OptLinesSetMaxBuffer is a option func that sets the maximum size of the line parsing buffers.
func OptLinesSetMultipart ¶
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 ¶
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.
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 (*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
ConnectWithContext establishes a connection to an AMQP09 server.
func (*AMQP09) ReadWithContext ¶ added in v3.2.0
ReadWithContext a new AMQP09 message.
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"` 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"` 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 (*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
ConnectWithContext establishes a connection to an AMQP1 server.
func (*AMQP1) ReadWithContext ¶ added in v3.19.0
ReadWithContext a new AMQP1 message.
type AMQP1Config ¶ added in v3.19.0
type AMQP1Config struct { URL string `json:"url" yaml:"url"` SourceAddress string `json:"source_address" yaml:"source_address"` 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 ¶
NewAmazonS3 creates a new Amazon S3 bucket reader.Type.
func (*AmazonS3) Acknowledge ¶
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 ¶
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
ConnectWithContext attempts to establish a connection to the target S3 bucket and any relevant queues used to traverse the objects (SQS, etc).
func (*AmazonS3) ReadWithContext ¶ added in v3.3.0
ReadWithContext attempts to read a new message from the target S3 bucket.
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 ¶
NewAmazonSQS creates a new Amazon SQS reader.Type.
func (*AmazonSQS) Acknowledge ¶
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) ConnectWithContext ¶ added in v3.3.0
ConnectWithContext attempts to establish a connection to the target SQS queue.
func (*AmazonSQS) ReadWithContext ¶ added in v3.3.0
ReadWithContext attempts to read a new message from the target SQS.
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.
type AsyncAckFn ¶ added in v3.1.0
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 (*CutOff) Acknowledge ¶
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 ¶
Connect attempts to establish a connection to the source, if unsuccessful returns an error. If the attempt is successful (or not necessary) returns nil.
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 ¶
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) ConnectWithContext ¶ added in v3.3.0
ConnectWithContext establishes a connection.
func (*Files) ReadWithContext ¶ added in v3.3.0
ReadWithContext a new Files message.
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 ¶
NewGCPPubSub creates a new GCP pubsub reader.Type.
func (*GCPPubSub) Acknowledge ¶
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) ConnectWithContext ¶ added in v3.3.0
ConnectWithContext attempts to establish a connection to the target subscription.
func (*GCPPubSub) ReadWithContext ¶ added in v3.3.0
ReadWithContext attempts to read a new message from the target subscription.
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"` // 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 (*HDFS) Acknowledge ¶
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) ConnectWithContext ¶ added in v3.3.0
ConnectWithContext attempts to establish a connection to the target HDFS host.
func (*HDFS) ReadWithContext ¶ added in v3.3.0
ReadWithContext reads a new HDFS message.
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 ¶ 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.
func NewHTTPClient ¶ added in v3.3.0
func NewHTTPClient(payload types.Message, httpClient *client.Type, opts ...HTTPClientOptFunc) (*HTTPClient, error)
NewHTTPClient creates a new HTTPClient reader type.
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 ¶
Acknowledge instructs whether the current offset should be committed.
func (*Kafka) AcknowledgeWithContext ¶ added in v3.3.0
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) ConnectWithContext ¶ added in v3.3.0
ConnectWithContext establishes a Kafka connection.
func (*Kafka) ReadNextWithContext ¶ added in v3.3.0
ReadNextWithContext attempts to read a message from a Kafka topic.
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 ¶
func (k *KafkaBalanced) 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 (*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 ¶
func (k *KafkaBalanced) 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 (*KafkaBalanced) Read ¶
func (k *KafkaBalanced) Read() (types.Message, error)
Read attempts to read a message from a KafkaBalanced topic.
func (*KafkaBalanced) Setup ¶
func (k *KafkaBalanced) Setup(sesh sarama.ConsumerGroupSession) error
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"` 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
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
ReadWithContext attempts to read a message from a KafkaCG topic.
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"` 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"` MaxProcessingPeriod string `json:"max_processing_period" yaml:"max_processing_period"` FetchBufferCap int `json:"fetch_buffer_cap" yaml:"fetch_buffer_cap"` Topic string `json:"topic" yaml:"topic"` Partition int32 `json:"partition" yaml:"partition"` 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"` Batching batch.PolicyConfig `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 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 ¶
NewKinesis creates a new Amazon Kinesis stream reader.Type.
func (*Kinesis) Acknowledge ¶
Acknowledge confirms whether or not our unacknowledged messages have been successfully propagated or not.
func (*Kinesis) AcknowledgeWithContext ¶ added in v3.3.0
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) ConnectWithContext ¶ added in v3.3.0
ConnectWithContext attempts to establish a connection to the target Kinesis shard.
func (*Kinesis) ReadNextWithContext ¶ added in v3.3.0
ReadNextWithContext attempts to read a new message from the target Kinesis shard.
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 ¶
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) ConnectWithContext ¶ added in v3.3.0
ConnectWithContext attempts to establish a new scanner for an io.Reader.
func (*Lines) ReadWithContext ¶ added in v3.3.0
ReadWithContext attempts to read a new line from the io.Reader.
type MQTT ¶
type MQTT struct {
// contains filtered or unexported fields
}
MQTT is an input type that reads MQTT Pub/Sub messages.
func (*MQTT) Acknowledge ¶
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) ConnectWithContext ¶ added in v3.3.0
ConnectWithContext establishes a connection to an MQTT server.
func (*MQTT) ReadWithContext ¶ added in v3.3.0
ReadWithContext attempts to read a new message from an MQTT broker.
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"` CleanSession bool `json:"clean_session" yaml:"clean_session"` User string `json:"user" yaml:"user"` Password string `json:"password" yaml:"password"` StaleConnectionTimeout string `json:"stale_connection_timeout" yaml:"stale_connection_timeout"` }
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 (*NATS) Acknowledge ¶
Acknowledge is a noop since NATS messages do not support acknowledgments.
func (*NATS) CloseAsync ¶
func (n *NATS) CloseAsync()
CloseAsync shuts down the NATS input and stops processing requests.
func (*NATS) ConnectWithContext ¶ added in v3.3.0
ConnectWithContext establishes a connection to a NATS server.
func (*NATS) ReadWithContext ¶ added in v3.3.0
ReadWithContext attempts to read a new message from the NATS subject.
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"` }
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"` // 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 (*NSQ) Acknowledge ¶
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) ConnectWithContext ¶ added in v3.3.0
ConnectWithContext establishes a connection to an NSQ server.
func (*NSQ) HandleMessage ¶
HandleMessage handles an NSQ message.
func (*NSQ) ReadWithContext ¶ added in v3.3.0
ReadWithContext attempts to read a new message from NSQ.
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 ¶
NewPreserver returns a new Preserver wrapper around a reader.Type.
func (*Preserver) Acknowledge ¶
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 ¶
Connect attempts to establish a connection to the source, if unsuccessful returns an error. If the attempt is successful (or not necessary) returns nil.
type RedisList ¶
type RedisList struct {
// contains filtered or unexported fields
}
RedisList is an input type that reads Redis List messages.
func NewRedisList ¶
NewRedisList creates a new RedisList input type.
func (*RedisList) Acknowledge ¶
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) ConnectWithContext ¶ added in v3.3.0
ConnectWithContext establishes a connection to a Redis server.
func (*RedisList) ReadWithContext ¶ added in v3.3.0
ReadWithContext attempts to pop a message from a Redis list.
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"` 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
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.
type Websocket ¶
type Websocket struct {
// contains filtered or unexported fields
}
Websocket is an input type that reads Websocket messages.
func NewWebsocket ¶
NewWebsocket creates a new Websocket input type.
func (*Websocket) Acknowledge ¶
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) ConnectWithContext ¶ added in v3.3.0
ConnectWithContext establishes a connection to a Websocket server.
func (*Websocket) ReadWithContext ¶ added in v3.3.0
ReadWithContext attempts to read a new message from the websocket.
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"` }
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.
Source Files ¶
- amazon_s3.go
- amazon_sqs.go
- amqp.go
- amqp_0_9.go
- amqp_1.go
- async_batcher.go
- async_bundle_unacks.go
- async_cut_off.go
- async_preserver.go
- cut_off.go
- files.go
- gcp_pubsub.go
- hdfs.go
- http_client.go
- interface.go
- kafka.go
- kafka_balanced.go
- kafka_cg.go
- kinesis.go
- kinesis_balanced.go
- lines.go
- mqtt.go
- nats.go
- nats_stream.go
- nsq.go
- package.go
- preserver.go
- redis_list.go
- redis_pubsub.go
- redis_streams.go
- scale_proto.go
- sync_batcher.go
- websocket.go
- zmq4_config.go