Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

reader

package

Go to main page

Versions in this module

v3

v3.15.4

May 30, 2020

v3.15.2

May 30, 2020

v3.15.1

May 30, 2020

Changes in this version

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

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

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

+ type AMQP struct

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

+ func (a *AMQP) CloseAsync()

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

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

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

+ type AMQP09 struct

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

+ func (a *AMQP09) CloseAsync()

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

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

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

+ type AMQP09BindingConfig struct

+ Exchange string

+ RoutingKey string

+ type AMQP09Config struct

+ Batching batch.PolicyConfig

+ BindingsDeclare []AMQP09BindingConfig

+ ConsumerTag string

+ PrefetchCount int

+ PrefetchSize int

+ Queue string

+ QueueDeclare AMQP09QueueDeclareConfig

+ TLS btls.Config

+ URL string

+ func NewAMQP09Config() AMQP09Config

+ type AMQP09QueueDeclareConfig struct

+ Durable bool

+ Enabled bool

+ type AMQPBindingConfig struct

+ Exchange string

+ RoutingKey string

+ type AMQPConfig struct

+ BindingsDeclare []AMQPBindingConfig

+ ConsumerTag string

+ MaxBatchCount int

+ PrefetchCount int

+ PrefetchSize int

+ Queue string

+ QueueDeclare AMQPQueueDeclareConfig

+ TLS btls.Config

+ URL string

+ func NewAMQPConfig() AMQPConfig

+ type AMQPQueueDeclareConfig struct

+ Durable bool

+ Enabled bool

+ type AmazonS3 struct

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

+ 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 struct

+ Bucket string

+ DeleteObjects bool

+ DownloadManager S3DownloadManagerConfig

+ ForcePathStyleURLs bool

+ MaxBatchCount int

+ Prefix string

+ Retries int

+ SQSBodyPath string

+ SQSBucketPath string

+ SQSEndpoint string

+ SQSEnvelopePath string

+ SQSMaxMessages int64

+ SQSURL string

+ Timeout string

+ func NewAmazonS3Config() AmazonS3Config

+ type AmazonSQS struct

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

+ 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 struct

+ DeleteMessage bool

+ MaxNumberOfMessages int64

+ Timeout string

+ URL string

+ func NewAmazonSQSConfig() AmazonSQSConfig

+ type Async interface

+ ConnectWithContext func(ctx context.Context) error

+ ReadWithContext func(ctx context.Context) (types.Message, AsyncAckFn, error)

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

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

+ type AsyncBatcher struct

+ func (p *AsyncBatcher) CloseAsync()

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

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

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

+ type AsyncBundleUnacks struct

+ func NewAsyncBundleUnacks(r Async) *AsyncBundleUnacks

+ func (p *AsyncBundleUnacks) CloseAsync()

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

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

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

+ type AsyncCutOff struct

+ func NewAsyncCutOff(r Async) *AsyncCutOff

+ func (c *AsyncCutOff) CloseAsync()

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

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

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

+ type AsyncPreserver struct

+ func NewAsyncPreserver(r Async) *AsyncPreserver

+ func (p *AsyncPreserver) CloseAsync()

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

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

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

+ type CutOff struct

+ func NewCutOff(r Type) *CutOff

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

+ func (c *CutOff) CloseAsync()

+ func (c *CutOff) Connect() error

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

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

+ type Files struct

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

+ 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 struct

+ DeleteFiles bool

+ Path string

+ func NewFilesConfig() FilesConfig

+ type GCPPubSub struct

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

+ 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 struct

+ Batching batch.PolicyConfig

+ MaxBatchCount int

+ MaxOutstandingBytes int

+ MaxOutstandingMessages int

+ ProjectID string

+ SubscriptionID string

+ func NewGCPPubSubConfig() GCPPubSubConfig

+ type HDFS struct

+ func NewHDFS(conf HDFSConfig, log log.Modular, stats metrics.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 struct

+ Directory string

+ Hosts []string

+ User string

+ func NewHDFSConfig() HDFSConfig

+ type HTTPClient struct

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

+ 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 Kafka struct

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

+ 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 struct

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

+ 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 struct

+ Addresses []string

+ Batching batch.PolicyConfig

+ ClientID string

+ CommitPeriod string

+ ConsumerGroup string

+ FetchBufferCap int

+ Group KafkaBalancedGroupConfig

+ MaxBatchCount int

+ MaxProcessingPeriod string

+ SASL sasl.Config

+ StartFromOldest bool

+ TLS btls.Config

+ TargetVersion string

+ Topics []string

+ func NewKafkaBalancedConfig() KafkaBalancedConfig

+ type KafkaBalancedGroupConfig struct

+ HeartbeatInterval string

+ RebalanceTimeout string

+ SessionTimeout string

+ func NewKafkaBalancedGroupConfig() KafkaBalancedGroupConfig

+ type KafkaCG struct

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

+ 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 struct

+ Addresses []string

+ Batching batch.PolicyConfig

+ ClientID string

+ CommitPeriod string

+ ConsumerGroup string

+ FetchBufferCap int

+ MaxBatchCount int

+ MaxProcessingPeriod string

+ Partition int32

+ SASL sasl.Config

+ StartFromOldest bool

+ TLS btls.Config

+ TargetVersion string

+ Topic string

+ func NewKafkaConfig() KafkaConfig

+ type Kinesis struct

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

+ 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 struct

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

+ 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 struct

+ Batching batch.PolicyConfig

+ DynamoDBBillingMode string

+ DynamoDBReadCapacity int64

+ DynamoDBTable string

+ DynamoDBWriteCapacity int64

+ MaxBatchCount int

+ StartFromOldest bool

+ Stream string

+ func NewKinesisBalancedConfig() KinesisBalancedConfig

+ type KinesisConfig struct

+ Batching batch.PolicyConfig

+ ClientID string

+ CommitPeriod string

+ DynamoDBTable string

+ Limit int64

+ Shard string

+ StartFromOldest bool

+ Stream string

+ Timeout string

+ func NewKinesisConfig() KinesisConfig

+ type Lines struct

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

+ func NewLinesWithContext(handleCtor func(ctx context.Context) (io.Reader, error), ...) (*Lines, error)

+ 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 struct

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

+ 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 struct

+ CleanSession bool

+ ClientID string

+ Password string

+ QoS uint8

+ Topics []string

+ URLs []string

+ User string

+ func NewMQTTConfig() MQTTConfig

+ type NATS struct

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

+ 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 struct

+ PrefetchCount int

+ QueueID string

+ Subject string

+ URLs []string

+ func NewNATSConfig() NATSConfig

+ type NATSStream struct

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

+ 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 struct

+ AckWait string

+ Batching batch.PolicyConfig

+ ClientID string

+ ClusterID string

+ DurableName string

+ MaxInflight int

+ QueueID string

+ StartFromOldest bool

+ Subject string

+ URLs []string

+ UnsubOnClose bool

+ func NewNATSStreamConfig() NATSStreamConfig

+ type NSQ struct

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

+ 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 struct

+ Addresses []string

+ Batching batch.PolicyConfig

+ Channel string

+ LookupAddresses []string

+ MaxInFlight int

+ TLS btls.Config

+ Topic string

+ UserAgent string

+ func NewNSQConfig() NSQConfig

+ type Preserver struct

+ func NewPreserver(r Type) *Preserver

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

+ func (p *Preserver) CloseAsync()

+ func (p *Preserver) Connect() error

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

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

+ type RedisList struct

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

+ 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 struct

+ Key string

+ Timeout string

+ URL string

+ func NewRedisListConfig() RedisListConfig

+ type RedisPubSub struct

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

+ 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 struct

+ Channels []string

+ URL string

+ UsePatterns bool

+ func NewRedisPubSubConfig() RedisPubSubConfig

+ type RedisStreams struct

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

+ 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 struct

+ Batching batch.PolicyConfig

+ BodyKey string

+ ClientID string

+ CommitPeriod string

+ ConsumerGroup string

+ Limit int64

+ StartFromOldest bool

+ Streams []string

+ Timeout string

+ URL string

+ func NewRedisStreamsConfig() RedisStreamsConfig

+ type S3DownloadManagerConfig struct

+ Enabled bool

+ type ScaleProto struct

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

+ 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 struct

+ Bind bool

+ PollTimeout string

+ RepTimeout string

+ SocketType string

+ SubFilters []string

+ URLs []string

+ func NewScaleProtoConfig() ScaleProtoConfig

+ type Sync interface

+ AcknowledgeWithContext func(ctx context.Context, err error) error

+ ConnectWithContext func(ctx context.Context) error

+ ReadNextWithContext func(ctx context.Context) (types.Message, error)

+ type SyncBatcher struct

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

+ 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 interface

+ Acknowledge func(err error) error

+ Connect func() error

+ Read func() (types.Message, error)

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

+ type Websocket struct

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

+ 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 struct

+ OpenMsg string

+ URL string

+ func NewWebsocketConfig() WebsocketConfig

+ type ZMQ4Config struct

+ func NewZMQ4Config() *ZMQ4Config

Other modules containing this package

github.com/windhooked/benthos