reader

package
v0.26.0 Latest Latest
Warning

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

 Go to latest Published: Aug 25, 2018 License: MIT Imports: 48 Imported by: 0

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

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

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

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

func OptLinesSetMultipart added in v0.7.7 

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

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

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

Acknowledge instructs whether unacknowledged messages have been successfully propagated.

func (*AMQP) CloseAsync added in v0.7.7 

func (a *AMQP) CloseAsync()

CloseAsync shuts down the AMQP input and stops processing requests.

func (*AMQP) Connect added in v0.7.7 

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

Connect establishes a connection to an AMQP server.

func (*AMQP) Read added in v0.7.7 

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

Read a new AMQP message.

func (*AMQP) WaitForClose added in v0.7.7 

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

WaitForClose blocks until the AMQP input has closed down.

type AMQPBindingConfig added in v0.24.0 

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

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

AMQPConfig contains configuration for the AMQP input type.

func NewAMQPConfig added in v0.7.7 

func NewAMQPConfig() AMQPConfig

NewAMQPConfig creates a new AMQPConfig with default values.

type AMQPQueueDeclareConfig added in v0.24.0 

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

type AmazonAWSCredentialsConfig struct {
	ID     string `json:"id" yaml:"id"`
	Secret string `json:"secret" yaml:"secret"`
	Token  string `json:"token" yaml:"token"`
	Role   string `json:"role" yaml:"role"`
}

AmazonAWSCredentialsConfig contains configuration params for AWS credentials.

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

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) Read 

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

Read 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 {
	Region          string                     `json:"region" yaml:"region"`
	Bucket          string                     `json:"bucket" yaml:"bucket"`
	Prefix          string                     `json:"prefix" yaml:"prefix"`
	Retries         int                        `json:"retries" yaml:"retries"`
	DeleteObjects   bool                       `json:"delete_objects" yaml:"delete_objects"`
	SQSURL          string                     `json:"sqs_url" yaml:"sqs_url"`
	SQSBodyPath     string                     `json:"sqs_body_path" yaml:"sqs_body_path"`
	SQSEnvelopePath string                     `json:"sqs_envelope_path" yaml:"sqs_envelope_path"`
	SQSMaxMessages  int64                      `json:"sqs_max_messages" yaml:"sqs_max_messages"`
	Credentials     AmazonAWSCredentialsConfig `json:"credentials" yaml:"credentials"`
	TimeoutS        int64                      `json:"timeout_s" yaml:"timeout_s"`
}

AmazonS3Config contains configuration values for the AmazonS3 input type.

func NewAmazonS3Config 

func NewAmazonS3Config() AmazonS3Config

NewAmazonS3Config creates a new AmazonS3Config with default values.

type AmazonSQS added in v0.7.5 

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

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

NewAmazonSQS creates a new Amazon SQS reader.Type.

func (*AmazonSQS) Acknowledge added in v0.7.5 

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

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

func (*AmazonSQS) CloseAsync added in v0.7.5 

func (a *AmazonSQS) CloseAsync()

CloseAsync begins cleaning up resources used by this reader asynchronously.

func (*AmazonSQS) Connect added in v0.7.5 

func (a *AmazonSQS) Connect() error

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

func (*AmazonSQS) Read added in v0.7.5 

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

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

func (*AmazonSQS) WaitForClose added in v0.7.5 

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

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

type AmazonSQSConfig added in v0.7.5 

type AmazonSQSConfig struct {
	Region      string                     `json:"region" yaml:"region"`
	URL         string                     `json:"url" yaml:"url"`
	Credentials AmazonAWSCredentialsConfig `json:"credentials" yaml:"credentials"`
	TimeoutS    int64                      `json:"timeout_s" yaml:"timeout_s"`
}

AmazonSQSConfig contains configuration values for the input type.

func NewAmazonSQSConfig added in v0.7.5 

func NewAmazonSQSConfig() AmazonSQSConfig

NewAmazonSQSConfig creates a new Config with default values.

type CutOff added in v0.9.0 

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

func NewCutOff(r Type) *CutOff

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

func (*CutOff) Acknowledge added in v0.9.0 

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

func (c *CutOff) CloseAsync()

CloseAsync triggers the asynchronous closing of the reader.

func (*CutOff) Connect added in v0.9.0 

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

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

Read attempts to read a new message from the source.

func (*CutOff) WaitForClose added in v0.9.0 

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

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

type Files added in v0.13.4 

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

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

func (*Files) Acknowledge added in v0.13.4 

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

Acknowledge instructs whether unacknowledged messages have been successfully propagated.

func (*Files) CloseAsync added in v0.13.4 

func (f *Files) CloseAsync()

CloseAsync shuts down the Files input and stops processing requests.

func (*Files) Connect added in v0.13.4 

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

Connect establishes a connection.

func (*Files) Read added in v0.13.4 

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

Read a new Files message.

func (*Files) WaitForClose added in v0.13.4 

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

WaitForClose blocks until the Files input has closed down.

type FilesConfig added in v0.13.4 

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

FilesConfig contains configuration for the Files input type.

func NewFilesConfig added in v0.13.4 

func NewFilesConfig() FilesConfig

NewFilesConfig creates a new FilesConfig with default values.

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, 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) 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) Read 

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

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

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

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

NewKafkaBalanced creates a new KafkaBalanced input type.

func (*KafkaBalanced) Acknowledge added in v0.7.7 

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

Acknowledge instructs whether the current offset should be committed.

func (*KafkaBalanced) CloseAsync added in v0.7.7 

func (k *KafkaBalanced) CloseAsync()

CloseAsync shuts down the KafkaBalanced input and stops processing requests.

func (*KafkaBalanced) Connect added in v0.7.7 

func (k *KafkaBalanced) Connect() error

Connect establishes a KafkaBalanced connection.

func (*KafkaBalanced) Read added in v0.7.7 

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

Read attempts to read a message from a KafkaBalanced topic.

func (*KafkaBalanced) WaitForClose added in v0.7.7 

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

WaitForClose blocks until the KafkaBalanced input has closed down.

type KafkaBalancedConfig added in v0.7.7 

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"`
	CommitPeriodMS  int         `json:"commit_period_ms" yaml:"commit_period_ms"`
	Topics          []string    `json:"topics" yaml:"topics"`
	StartFromOldest bool        `json:"start_from_oldest" yaml:"start_from_oldest"`
	TLS             btls.Config `json:"tls" yaml:"tls"`
}

KafkaBalancedConfig contains configuration for the KafkaBalanced input type.

func NewKafkaBalancedConfig added in v0.7.7 

func NewKafkaBalancedConfig() KafkaBalancedConfig

NewKafkaBalancedConfig creates a new KafkaBalancedConfig with default values.

type KafkaConfig 

type KafkaConfig struct {
	Addresses       []string    `json:"addresses" yaml:"addresses"`
	ClientID        string      `json:"client_id" yaml:"client_id"`
	ConsumerGroup   string      `json:"consumer_group" yaml:"consumer_group"`
	CommitPeriodMS  int         `json:"commit_period_ms" yaml:"commit_period_ms"`
	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"`
	TLS             btls.Config `json:"tls" yaml:"tls"`
}

KafkaConfig contains configuration fields for the Kafka input type.

func NewKafkaConfig 

func NewKafkaConfig() KafkaConfig

NewKafkaConfig creates a new KafkaConfig with default values.

type Kinesis added in v0.26.0 

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

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

NewKinesis creates a new Amazon Kinesis stream reader.Type.

func (*Kinesis) Acknowledge added in v0.26.0 

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

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

func (*Kinesis) CloseAsync added in v0.26.0 

func (k *Kinesis) CloseAsync()

CloseAsync begins cleaning up resources used by this reader asynchronously.

func (*Kinesis) Connect added in v0.26.0 

func (k *Kinesis) Connect() error

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

func (*Kinesis) Read added in v0.26.0 

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

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

func (*Kinesis) WaitForClose added in v0.26.0 

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

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

type KinesisConfig added in v0.26.0 

type KinesisConfig struct {
	Region          string                     `json:"region" yaml:"region"`
	Credentials     AmazonAWSCredentialsConfig `json:"credentials" yaml:"credentials"`
	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"`
	CommitPeriodMS  int                        `json:"commit_period_ms" yaml:"commit_period_ms"`
	StartFromOldest bool                       `json:"start_from_oldest" yaml:"start_from_oldest"`
	TimeoutMS       int64                      `json:"timeout_ms" yaml:"timeout_ms"`
}

KinesisConfig is configuration values for the input type.

func NewKinesisConfig added in v0.26.0 

func NewKinesisConfig() KinesisConfig

NewKinesisConfig creates a new Config with default values.

type Lines added in v0.7.7 

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

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

NewLines creates a new reader input type.

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 (*Lines) Acknowledge added in v0.7.7 

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

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

func (*Lines) CloseAsync added in v0.7.7 

func (r *Lines) CloseAsync()

CloseAsync shuts down the reader input and stops processing requests.

func (*Lines) Connect added in v0.7.7 

func (r *Lines) Connect() error

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

func (*Lines) Read added in v0.7.7 

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

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

func (*Lines) WaitForClose added in v0.7.7 

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

WaitForClose blocks until the reader input has closed down.

type MQTT added in v0.8.7 

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

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

func NewMQTT added in v0.8.7 

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

NewMQTT creates a new MQTT input type.

func (*MQTT) Acknowledge added in v0.8.7 

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

Acknowledge instructs whether messages have been successfully propagated.

func (*MQTT) CloseAsync added in v0.8.7 

func (m *MQTT) CloseAsync()

CloseAsync shuts down the MQTT input and stops processing requests.

func (*MQTT) Connect added in v0.8.7 

func (m *MQTT) Connect() error

Connect establishes a connection to an MQTT server.

func (*MQTT) Read added in v0.8.7 

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

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

func (*MQTT) WaitForClose added in v0.8.7 

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

WaitForClose blocks until the MQTT input has closed down.

type MQTTConfig added in v0.8.7 

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

MQTTConfig contains configuration fields for the MQTT input type.

func NewMQTTConfig added in v0.8.7 

func NewMQTTConfig() MQTTConfig

NewMQTTConfig creates a new MQTTConfig with default values.

type NATS added in v0.7.7 

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

NATS is an input type that receives NATS messages.

func (*NATS) Acknowledge added in v0.7.7 

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

Acknowledge is a noop since NATS messages do not support acknowledgments.

func (*NATS) CloseAsync added in v0.7.7 

func (n *NATS) CloseAsync()

CloseAsync shuts down the NATS input and stops processing requests.

func (*NATS) Connect added in v0.7.7 

func (n *NATS) Connect() error

Connect establishes a connection to a NATS server.

func (*NATS) Read added in v0.7.7 

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

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

func (*NATS) WaitForClose added in v0.7.7 

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

WaitForClose blocks until the NATS input has closed down.

type NATSConfig added in v0.7.7 

type NATSConfig struct {
	URLs    []string `json:"urls" yaml:"urls"`
	Subject string   `json:"subject" yaml:"subject"`
}

NATSConfig contains configuration fields for the NATS input type.

func NewNATSConfig added in v0.7.7 

func NewNATSConfig() NATSConfig

NewNATSConfig creates a new NATSConfig with default values.

type NATSStream added in v0.7.7 

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

NATSStream is an input type that receives NATSStream messages.

func (*NATSStream) Acknowledge added in v0.7.7 

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

Acknowledge instructs whether unacknowledged messages have been successfully propagated.

func (*NATSStream) CloseAsync added in v0.7.7 

func (n *NATSStream) CloseAsync()

CloseAsync shuts down the NATSStream input and stops processing requests.

func (*NATSStream) Connect added in v0.7.7 

func (n *NATSStream) Connect() error

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

func (*NATSStream) Read added in v0.7.7 

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

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

func (*NATSStream) WaitForClose added in v0.7.7 

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

WaitForClose blocks until the NATSStream input has closed down.

type NATSStreamConfig added in v0.7.7 

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"`
	StartFromOldest bool     `json:"start_from_oldest" yaml:"start_from_oldest"`
	Subject         string   `json:"subject" yaml:"subject"`
}

NATSStreamConfig contains configuration fields for the NATSStream input type.

func NewNATSStreamConfig added in v0.7.7 

func NewNATSStreamConfig() NATSStreamConfig

NewNATSStreamConfig creates a new NATSStreamConfig with default values.

type NSQ added in v0.7.7 

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

NSQ is an input type that receives NSQ messages.

func (*NSQ) Acknowledge added in v0.7.7 

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

Acknowledge instructs whether unacknowledged messages have been successfully propagated.

func (*NSQ) CloseAsync added in v0.7.7 

func (n *NSQ) CloseAsync()

CloseAsync shuts down the NSQ input and stops processing requests.

func (*NSQ) Connect added in v0.7.7 

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

Connect establishes a connection to an NSQ server.

func (*NSQ) HandleMessage added in v0.7.7 

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

HandleMessage handles an NSQ message.

func (*NSQ) Read added in v0.7.7 

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

Read attempts to read a new message from NSQ.

func (*NSQ) WaitForClose added in v0.7.7 

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

WaitForClose blocks until the NSQ input has closed down.

type NSQConfig added in v0.7.7 

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"`
	MaxInFlight     int      `json:"max_in_flight" yaml:"max_in_flight"`
}

NSQConfig contains configuration fields for the NSQ input type.

func NewNSQConfig added in v0.7.7 

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) Read 

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

Read 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 {
	URL       string `json:"url" yaml:"url"`
	Key       string `json:"key" yaml:"key"`
	TimeoutMS int    `json:"timeout_ms" yaml:"timeout_ms"`
}

RedisListConfig contains configuration fields for the RedisList input type.

func NewRedisListConfig 

func NewRedisListConfig() RedisListConfig

NewRedisListConfig creates a new RedisListConfig with default values.

type RedisPubSub added in v0.7.7 

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

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

func NewRedisPubSub added in v0.7.7 

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

NewRedisPubSub creates a new RedisPubSub input type.

func (*RedisPubSub) Acknowledge added in v0.7.7 

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

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

func (*RedisPubSub) CloseAsync added in v0.7.7 

func (r *RedisPubSub) CloseAsync()

CloseAsync shuts down the RedisPubSub input and stops processing requests.

func (*RedisPubSub) Connect added in v0.7.7 

func (r *RedisPubSub) Connect() error

Connect establishes a connection to an RedisPubSub server.

func (*RedisPubSub) Read added in v0.7.7 

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

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

func (*RedisPubSub) WaitForClose added in v0.7.7 

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

WaitForClose blocks until the RedisPubSub input has closed down.

type RedisPubSubConfig added in v0.7.7 

type RedisPubSubConfig struct {
	URL      string   `json:"url" yaml:"url"`
	Channels []string `json:"channels" yaml:"channels"`
}

RedisPubSubConfig contains configuration fields for the RedisPubSub input type.

func NewRedisPubSubConfig added in v0.7.7 

func NewRedisPubSubConfig() RedisPubSubConfig

NewRedisPubSubConfig creates a new RedisPubSubConfig with default values.

type ScaleProto added in v0.7.7 

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

ScaleProto is an input type that contains Scalability Protocols messages.

func (*ScaleProto) Acknowledge added in v0.7.7 

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

Acknowledge instructs whether the pending messages were propagated successfully.

func (*ScaleProto) CloseAsync added in v0.7.7 

func (s *ScaleProto) CloseAsync()

CloseAsync shuts down the ScaleProto input and stops processing requests.

func (*ScaleProto) Connect added in v0.7.7 

func (s *ScaleProto) Connect() error

Connect establishes a nanomsg socket.

func (*ScaleProto) Read added in v0.7.7 

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

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

func (*ScaleProto) WaitForClose added in v0.7.7 

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

WaitForClose blocks until the ScaleProto input has closed down.

type ScaleProtoConfig added in v0.7.7 

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"`
	PollTimeoutMS int      `json:"poll_timeout_ms" yaml:"poll_timeout_ms"`
	RepTimeoutMS  int      `json:"reply_timeout_ms" yaml:"reply_timeout_ms"`
}

ScaleProtoConfig contains configuration fields for the ScaleProto input type.

func NewScaleProtoConfig added in v0.7.7 

func NewScaleProtoConfig() ScaleProtoConfig

NewScaleProtoConfig creates a new ScaleProtoConfig with default values.

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

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

NewAMQP creates a new AMQP input type.

func NewFiles added in v0.13.4 

func NewFiles(conf FilesConfig) (Type, error)

NewFiles creates a new Files input type.

func NewNATS added in v0.7.7 

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

NewNATS creates a new NATS input type.

func NewNATSStream added in v0.7.7 

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

NewNATSStream creates a new NATSStream input type.

func NewNSQ added in v0.7.7 

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

NewNSQ creates a new NSQ input type.

func NewScaleProto added in v0.7.7 

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

NewScaleProto creates a new ScaleProto input type.

type Websocket added in v0.13.2 

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

Websocket is an input type that reads Websocket messages.

func NewWebsocket added in v0.13.2 

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

NewWebsocket creates a new Websocket input type.

func (*Websocket) Acknowledge added in v0.13.2 

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

Acknowledge instructs whether the pending messages were propagated successfully.

func (*Websocket) CloseAsync added in v0.13.2 

func (w *Websocket) CloseAsync()

CloseAsync shuts down the Websocket input and stops reading messages.

func (*Websocket) Connect added in v0.13.2 

func (w *Websocket) Connect() error

Connect establishes a connection to a Websocket server.

func (*Websocket) Read added in v0.13.2 

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

Read attempts to read a new message from the websocket.

func (*Websocket) WaitForClose added in v0.13.2 

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

WaitForClose blocks until the Websocket input has closed down.

type WebsocketConfig added in v0.13.2 

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

func NewWebsocketConfig() WebsocketConfig

NewWebsocketConfig creates a new WebsocketConfig with default values.

type ZMQ4Config added in v0.9.0 

type ZMQ4Config struct{}

ZMQ4Config is an empty stub for when ZMQ4 is not compiled.

func NewZMQ4Config added in v0.9.0 

func NewZMQ4Config() *ZMQ4Config

NewZMQ4Config returns nil.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.