reader

package
v4.0.0 Latest Latest
Warning

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

 Go to latest Published: Apr 20, 2022 License: MIT Imports: 36 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Async 

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

	// CloseAsync triggers the shut down of this component but should not block
	// the calling goroutine.
	CloseAsync()

	// WaitForClose is a blocking call to wait until the component has finished
	// shutting down and cleaning up resources.
	WaitForClose(timeout time.Duration) error
}

Async is a type that reads Benthos messages from an external source and allows acknowledgements for a message batch to be propagated asynchronously.

type AsyncAckFn 

type AsyncAckFn func(context.Context, error) error

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

type AsyncCutOff 

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 

func NewAsyncCutOff(r Async) *AsyncCutOff

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

func (*AsyncCutOff) CloseAsync 

func (c *AsyncCutOff) CloseAsync()

CloseAsync triggers the asynchronous closing of the reader.

func (*AsyncCutOff) ConnectWithContext 

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 

func (c *AsyncCutOff) ReadWithContext(ctx context.Context) (*message.Batch, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the source.

func (*AsyncCutOff) WaitForClose 

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

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

type AsyncPreserver 

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 

func NewAsyncPreserver(r Async) *AsyncPreserver

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

func (*AsyncPreserver) CloseAsync 

func (p *AsyncPreserver) CloseAsync()

CloseAsync triggers the asynchronous closing of the reader.

func (*AsyncPreserver) ConnectWithContext 

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 

func (p *AsyncPreserver) ReadWithContext(ctx context.Context) (*message.Batch, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the source.

func (*AsyncPreserver) WaitForClose 

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

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

type GCPPubSub 

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

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

func NewGCPPubSub 

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

NewGCPPubSub creates a new GCP pubsub reader.Type.

func (*GCPPubSub) CloseAsync 

func (c *GCPPubSub) CloseAsync()

CloseAsync begins cleaning up resources used by this reader asynchronously.

func (*GCPPubSub) Connect 

func (c *GCPPubSub) Connect() error

Connect attempts to establish a connection to the target subscription.

func (*GCPPubSub) ConnectWithContext 

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

ConnectWithContext attempts to establish a connection to the target subscription.

func (*GCPPubSub) ReadWithContext 

func (c *GCPPubSub) ReadWithContext(ctx context.Context) (*message.Batch, AsyncAckFn, error)

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

func (*GCPPubSub) WaitForClose 

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

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

type GCPPubSubConfig 

type GCPPubSubConfig struct {
	ProjectID              string `json:"project" yaml:"project"`
	SubscriptionID         string `json:"subscription" yaml:"subscription"`
	MaxOutstandingMessages int    `json:"max_outstanding_messages" yaml:"max_outstanding_messages"`
	MaxOutstandingBytes    int    `json:"max_outstanding_bytes" yaml:"max_outstanding_bytes"`
	Sync                   bool   `json:"sync" yaml:"sync"`
}

GCPPubSubConfig contains configuration values for the input type.

func NewGCPPubSubConfig 

func NewGCPPubSubConfig() GCPPubSubConfig

NewGCPPubSubConfig creates a new Config with default values.

type HDFS 

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

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

func NewHDFS 

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

NewHDFS creates a new HDFS writer.Type.

func (*HDFS) CloseAsync 

func (h *HDFS) CloseAsync()

CloseAsync shuts down the HDFS input and stops processing requests.

func (*HDFS) ConnectWithContext 

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

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

func (*HDFS) ReadWithContext 

func (h *HDFS) ReadWithContext(ctx context.Context) (*message.Batch, AsyncAckFn, error)

ReadWithContext reads a new HDFS message.

func (*HDFS) WaitForClose 

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

WaitForClose blocks until the HDFS input has closed down.

type HDFSConfig 

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

HDFSConfig contains configuration fields for the HDFS input type.

func NewHDFSConfig 

func NewHDFSConfig() HDFSConfig

NewHDFSConfig creates a new Config with default values.

type NATS 

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

NATS is an input type that receives NATS messages.

func NewNATS 

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

NewNATS creates a new NATS input type.

func (*NATS) CloseAsync 

func (n *NATS) CloseAsync()

CloseAsync shuts down the NATS input and stops processing requests.

func (*NATS) ConnectWithContext 

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

ConnectWithContext establishes a connection to a NATS server.

func (*NATS) ReadWithContext 

func (n *NATS) ReadWithContext(ctx context.Context) (*message.Batch, AsyncAckFn, error)

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

func (*NATS) WaitForClose 

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

WaitForClose blocks until the NATS input has closed down.

type NATSConfig 

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

NATSConfig contains configuration fields for the NATS input type.

func NewNATSConfig 

func NewNATSConfig() NATSConfig

NewNATSConfig creates a new NATSConfig with default values.

type NATSStream 

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

NATSStream is an input type that receives NATSStream messages.

func NewNATSStream 

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

NewNATSStream creates a new NATSStream input type.

func (*NATSStream) CloseAsync 

func (n *NATSStream) CloseAsync()

CloseAsync shuts down the NATSStream input and stops processing requests.

func (*NATSStream) ConnectWithContext 

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

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

func (*NATSStream) ReadWithContext 

func (n *NATSStream) ReadWithContext(ctx context.Context) (*message.Batch, AsyncAckFn, error)

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

func (*NATSStream) WaitForClose 

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

WaitForClose blocks until the NATSStream input has closed down.

type NATSStreamConfig 

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

NATSStreamConfig contains configuration fields for the NATSStream input type.

func NewNATSStreamConfig 

func NewNATSStreamConfig() NATSStreamConfig

NewNATSStreamConfig creates a new NATSStreamConfig with default values.

type NSQ 

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

NSQ is an input type that receives NSQ messages.

func NewNSQ 

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

NewNSQ creates a new NSQ input type.

func (*NSQ) CloseAsync 

func (n *NSQ) CloseAsync()

CloseAsync shuts down the NSQ input and stops processing requests.

func (*NSQ) ConnectWithContext 

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

ConnectWithContext establishes a connection to an NSQ server.

func (*NSQ) HandleMessage 

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

HandleMessage handles an NSQ message.

func (*NSQ) ReadWithContext 

func (n *NSQ) ReadWithContext(ctx context.Context) (*message.Batch, AsyncAckFn, error)

ReadWithContext attempts to read a new message from NSQ.

func (*NSQ) WaitForClose 

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

WaitForClose blocks until the NSQ input has closed down.

type NSQConfig 

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

NSQConfig contains configuration fields for the NSQ input type.

func NewNSQConfig 

func NewNSQConfig() NSQConfig

NewNSQConfig creates a new NSQConfig with default values.

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

func (r *RedisList) CloseAsync()

CloseAsync shuts down the RedisList input and stops processing requests.

func (*RedisList) ConnectWithContext 

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

ConnectWithContext establishes a connection to a Redis server.

func (*RedisList) ReadWithContext 

func (r *RedisList) ReadWithContext(ctx context.Context) (*message.Batch, AsyncAckFn, error)

ReadWithContext attempts to pop a message from a Redis list.

func (*RedisList) WaitForClose 

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

WaitForClose blocks until the RedisList input has closed down.

type RedisListConfig 

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

RedisListConfig contains configuration fields for the RedisList input type.

func NewRedisListConfig 

func NewRedisListConfig() RedisListConfig

NewRedisListConfig creates a new RedisListConfig with default values.

type RedisPubSub 

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

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

func NewRedisPubSub 

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

NewRedisPubSub creates a new RedisPubSub input type.

func (*RedisPubSub) CloseAsync 

func (r *RedisPubSub) CloseAsync()

CloseAsync shuts down the RedisPubSub input and stops processing requests.

func (*RedisPubSub) ConnectWithContext 

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

ConnectWithContext establishes a connection to an RedisPubSub server.

func (*RedisPubSub) ReadWithContext 

func (r *RedisPubSub) ReadWithContext(ctx context.Context) (*message.Batch, 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) CloseAsync 

func (r *RedisStreams) CloseAsync()

CloseAsync shuts down the RedisStreams input and stops processing requests.

func (*RedisStreams) ConnectWithContext 

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

ConnectWithContext establishes a connection to a Redis server.

func (*RedisStreams) ReadWithContext 

func (r *RedisStreams) ReadWithContext(ctx context.Context) (*message.Batch, AsyncAckFn, error)

ReadWithContext attempts to pop a message from a Redis list.

func (*RedisStreams) WaitForClose 

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

WaitForClose blocks until the RedisStreams input has closed down.

type RedisStreamsConfig 

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

RedisStreamsConfig contains configuration fields for the RedisStreams input type.

func NewRedisStreamsConfig 

func NewRedisStreamsConfig() RedisStreamsConfig

NewRedisStreamsConfig creates a new RedisStreamsConfig with default values.

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

func (s *ScaleProto) CloseAsync()

CloseAsync shuts down the ScaleProto input and stops processing requests.

func (*ScaleProto) ConnectWithContext 

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

ConnectWithContext establishes a nanomsg socket.

func (*ScaleProto) ReadWithContext 

func (s *ScaleProto) ReadWithContext(ctx context.Context) (*message.Batch, 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"`
}

ScaleProtoConfig contains configuration fields for the ScaleProto input type.

func NewScaleProtoConfig 

func NewScaleProtoConfig() ScaleProtoConfig

NewScaleProtoConfig creates a new ScaleProtoConfig with default values.

type Websocket 

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

Websocket is an input type that reads Websocket messages.

func NewWebsocket 

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

NewWebsocket creates a new Websocket input type.

func (*Websocket) CloseAsync 

func (w *Websocket) CloseAsync()

CloseAsync shuts down the Websocket input and stops reading messages.

func (*Websocket) ConnectWithContext 

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

ConnectWithContext establishes a connection to a Websocket server.

func (*Websocket) ReadWithContext 

func (w *Websocket) ReadWithContext(ctx context.Context) (*message.Batch, AsyncAckFn, error)

ReadWithContext attempts to read a new message from the websocket.

func (*Websocket) WaitForClose 

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

WaitForClose blocks until the Websocket input has closed down.

type WebsocketConfig 

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

WebsocketConfig contains configuration fields for the Websocket input type.

func NewWebsocketConfig 

func NewWebsocketConfig() WebsocketConfig

NewWebsocketConfig creates a new WebsocketConfig with default values.

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.