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.65.6

Aug 10, 2022

v3.65.5

Aug 10, 2022

v3.65.4

Aug 10, 2022

v3.65.3

Jul 27, 2022

v3.65.2

Jul 26, 2022

v3.65.1

Jul 24, 2022

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

+ AutoAck bool

+ Batching batch.PolicyConfig

+ BindingsDeclare []AMQP09BindingConfig

+ ConsumerTag string

+ NackRejectPatterns []string

+ PrefetchCount int

+ PrefetchSize int

+ Queue string

+ QueueDeclare AMQP09QueueDeclareConfig

+ TLS btls.Config

+ URL string

+ URLs []string

+ func NewAMQP09Config() AMQP09Config

+ type AMQP09QueueDeclareConfig struct

+ Durable bool

+ Enabled bool

+ type AMQP1 struct

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

+ func (a *AMQP1) CloseAsync()

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

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

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

+ type AMQP1Config struct

+ AzureRenewLock bool

+ SASL sasl.Config

+ SourceAddress string

+ TLS btls.Config

+ URL string

+ func NewAMQP1Config() AMQP1Config

+ 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

+ Sync bool

+ 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, opts ...HTTPClientOptFunc) (*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 HTTPClientOptFunc func(*HTTPClient)

+ func HTTPClientOptSetDropEmpty(dropEmpty bool) HTTPClientOptFunc

+ 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

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

+ CheckpointLimit int

+ ClientID string

+ CommitPeriod string

+ ConsumerGroup string

+ ExtractTracingMap string

+ FetchBufferCap int

+ Group KafkaBalancedGroupConfig

+ MaxBatchCount int

+ MaxProcessingPeriod string

+ Partition int32

+ RackID string

+ SASL sasl.Config

+ StartFromOldest bool

+ TLS btls.Config

+ TargetVersion string

+ Topic string

+ Topics []string

+ func NewKafkaConfig() KafkaConfig

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

+ func (k KafkaConfig) IsDeprecated() bool

+ 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

+ ConnectTimeout string

+ DynamicClientIDSuffix string

+ KeepAlive int64

+ Password string

+ QoS uint8

+ StaleConnectionTimeout string

+ TLS tls.Config

+ Topics []string

+ URLs []string

+ User string

+ Will mqttconf.Will

+ 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

+ Auth auth.Config

+ PrefetchCount int

+ QueueID string

+ Subject string

+ TLS btls.Config

+ 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

+ Auth auth.Config

+ Batching batch.PolicyConfig

+ ClientID string

+ ClusterID string

+ DurableName string

+ MaxInflight int

+ QueueID string

+ StartFromOldest bool

+ Subject string

+ TLS btls.Config

+ 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

+ 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

+ 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

+ CreateStreams bool

+ Limit int64

+ StartFromOldest bool

+ Streams []string

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

+ TLS btls.Config

+ URL string

+ func NewWebsocketConfig() WebsocketConfig

+ type ZMQ4Config struct

+ Bind bool

+ HighWaterMark int

+ PollTimeout string

+ SocketType string

+ SubFilters []string

+ URLs []string

+ func NewZMQ4Config() *ZMQ4Config

Other modules containing this package

github.com/dafanshu/benthos