writer

Feb 8, 2022

Changes in this version

type AmazonS3Config

+ ServerSideEncryption string

type HTTPClientConfig

+ Multipart []HTTPClientMultipartExpression

+ type HTTPClientMultipartExpression struct

+ Body string

+ ContentDisposition string

+ ContentType string

Jan 21, 2022

Changes in this version

type GCPPubSubConfig

+ OrderingKey string

Dec 1, 2021

Changes in this version

type SNS

+ func NewSNSV2(conf SNSConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*SNS, error)

type SNSConfig

+ MessageDeduplicationID string

+ MessageGroupID string

+ Metadata output.Metadata

Nov 22, 2021

Changes in this version

type MQTTConfig

+ RetainedInterpolated string

type NATSConfig

+ Headers map[string]string

Nov 2, 2021

Changes in this version

type AMQPConfig

+ URLs []string

type MQTTConfig

+ ConnectTimeout string

+ WriteTimeout string

type WebsocketConfig

+ TLS btls.Config

Oct 14, 2021

Changes in this version

type MQTTConfig

+ DynamicClientIDSuffix string

Sep 22, 2021

Changes in this version

type AMQP

+ func NewAMQPV2(mgr types.Manager, conf AMQPConfig, log log.Modular, stats metrics.Type) (*AMQP, error)

type AmazonS3

+ func NewAmazonS3V2(conf AmazonS3Config, mgr types.Manager, log log.Modular, stats metrics.Type) (*AmazonS3, error)

type AmazonS3Config

+ CacheControl string

+ ContentDisposition string

+ ContentLanguage string

+ WebsiteRedirectLocation string

type AmazonSQS

+ func NewAmazonSQSV2(conf AmazonSQSConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*AmazonSQS, error)

type AzureBlobStorage — darwin/amd64, linux/amd64, windows/amd64

+ func NewAzureBlobStorageV2(mgr types.Manager, conf AzureBlobStorageConfig, log log.Modular, ...) (*AzureBlobStorage, error)

type AzureQueueStorage — darwin/amd64, linux/amd64, windows/amd64

+ func NewAzureQueueStorageV2(conf AzureQueueStorageConfig, mgr types.Manager, log log.Modular, ...) (*AzureQueueStorage, error)

type AzureTableStorage — darwin/amd64, linux/amd64, windows/amd64

+ func NewAzureTableStorageV2(conf AzureTableStorageConfig, mgr types.Manager, log log.Modular, ...) (*AzureTableStorage, error)

type DynamoDB

+ func NewDynamoDBV2(conf DynamoDBConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*DynamoDB, error)

type Elasticsearch

+ func NewElasticsearchV2(conf ElasticsearchConfig, mgr types.Manager, log log.Modular, ...) (*Elasticsearch, error)

type Files

+ func NewFilesV2(conf FilesConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*Files, error)

type GCPPubSub

+ func NewGCPPubSubV2(conf GCPPubSubConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*GCPPubSub, error)

type HDFS

+ func NewHDFSV2(conf HDFSConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*HDFS, error)

type KafkaConfig

+ RackID string

type Kinesis

+ func NewKinesisV2(conf KinesisConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*Kinesis, error)

type MQTT

+ func NewMQTTV2(conf MQTTConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*MQTT, error)

type NATS

+ func NewNATSV2(conf NATSConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*NATS, error)

type NSQ

+ func NewNSQV2(conf NSQConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*NSQ, error)

type RedisHash

+ func NewRedisHashV2(conf RedisHashConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*RedisHash, error)

type RedisList

+ func NewRedisListV2(conf RedisListConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*RedisList, error)

type RedisPubSub

+ func NewRedisPubSubV2(conf RedisPubSubConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*RedisPubSub, error)

Sep 8, 2021

Changes in this version

type NATSConfig

+ Auth auth.Config

type NATSStreamConfig

+ Auth auth.Config

Sep 1, 2021

Changes in this version

type ElasticsearchConfig

+ GzipCompression bool

type KafkaConfig

+ Partition string

type MQTTConfig

+ KeepAlive int64

+ Retained bool

+ Will mqttconf.Will

Aug 19, 2021

Changes in this version

type AMQPConfig

+ Priority string

Aug 2, 2021

Changes in this version

type ElasticsearchConfig

+ Action string

+ Routing string

Jul 26, 2021

Changes in this version

type AMQP1Config

+ Metadata output.Metadata

type RedisListConfig

+ Batching batch.PolicyConfig

type RedisPubSubConfig

+ Batching batch.PolicyConfig

type RedisStreamsConfig

+ Batching batch.PolicyConfig

Apr 23, 2021

Changes in this version

type KafkaConfig

+ InjectTracingMap string

type MQTTConfig

+ TLS tls.Config

Mar 31, 2021

Changes in this version

type AMQPConfig

+ Metadata output.Metadata

type AmazonS3Config

+ Metadata output.Metadata

type AmazonSQSConfig

+ Metadata output.Metadata

type GCPPubSubConfig

+ Metadata output.Metadata

type HTTPClientConfig

+ BatchAsMultipart bool

type KafkaConfig

+ Metadata output.Metadata

type NATSConfig

+ TLS btls.Config

type NATSStreamConfig

+ TLS btls.Config

type RedisStreamsConfig

+ Metadata output.Metadata

type Socket

+ func (s *Socket) ConnectWithContext(ctx context.Context) error

+ func (s *Socket) WriteWithContext(ctx context.Context, msg types.Message) error

type SocketConfig

+ Codec string

Feb 15, 2021

Changes in this version

type GCPPubSubConfig

+ PublishTimeout string

Feb 1, 2021

Changes in this version

type AmazonS3Config

+ Tags map[string]string

Jan 18, 2021

Changes in this version

type AzureBlobStorageConfig

+ StorageSASToken string

Jan 6, 2021

Changes in this version

type AMQPConfig

+ ContentEncoding string

+ ContentType string

type HDFSConfig

+ Batching batch.PolicyConfig

Dec 24, 2020

Changes in this version

+ type AzureQueueStorage struct

+ func NewAzureQueueStorage(conf AzureQueueStorageConfig, log log.Modular, stats metrics.Type) (*AzureQueueStorage, error)

+ func (a *AzureQueueStorage) CloseAsync()

+ func (a *AzureQueueStorage) Connect() error

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

+ func (a *AzureQueueStorage) WaitForClose(time.Duration) error

+ func (a *AzureQueueStorage) Write(msg types.Message) error

+ func (a *AzureQueueStorage) WriteWithContext(ctx context.Context, msg types.Message) error

+ type AzureQueueStorageConfig struct

+ Batching batch.PolicyConfig

+ MaxInFlight int

+ QueueName string

+ StorageAccessKey string

+ StorageAccount string

+ StorageConnectionString string

+ TTL string

+ func NewAzureQueueStorageConfig() AzureQueueStorageConfig

Dec 7, 2020

Changes in this version

type KafkaConfig

+ RetryAsBatch bool

Nov 16, 2020

Changes in this version

type CacheConfig

+ TTL string

Oct 6, 2020

Changes in this version

type AzureBlobStorageConfig

+ PublicAccessLevel string

+ StorageConnectionString string

type AzureTableStorageConfig

+ StorageConnectionString string

Aug 16, 2020

Changes in this version

type AmazonS3Config

+ Batching batch.PolicyConfig

Aug 2, 2020

Changes in this version

type AzureTableStorageConfig

+ Batching batch.PolicyConfig

Jul 26, 2020

Changes in this version

+ func NewAzureTableStorage(conf AzureTableStorageConfig, log log.Modular, stats metrics.Type) (dummy, error) — js/wasm

+ type AzureTableStorage struct — darwin/amd64, linux/amd64, windows/amd64

+ func NewAzureTableStorage(conf AzureTableStorageConfig, log log.Modular, stats metrics.Type) (*AzureTableStorage, error)

+ func (a *AzureTableStorage) CloseAsync()

+ func (a *AzureTableStorage) Connect() error

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

+ func (a *AzureTableStorage) WaitForClose(time.Duration) error

+ func (a *AzureTableStorage) Write(msg types.Message) error

+ func (a *AzureTableStorage) WriteWithContext(wctx context.Context, msg types.Message) error

+ type AzureTableStorageConfig struct

+ InsertType string

+ MaxInFlight int

+ PartitionKey string

+ Properties map[string]string

+ RowKey string

+ StorageAccessKey string

+ StorageAccount string

+ TableName string

+ Timeout string

+ func NewAzureTableStorageConfig() AzureTableStorageConfig

Jul 19, 2020

Changes in this version

+ func IterateBatchedSend(msg types.Message, fn func(int, types.Part) error) error

type ElasticsearchConfig

+ TLS btls.Config

type KafkaConfig

+ StaticHeaders map[string]string

Jul 12, 2020

Changes in this version

+ type AzureBlobStorage struct

+ func NewAzureBlobStorage(conf AzureBlobStorageConfig, log log.Modular, stats metrics.Type) (*AzureBlobStorage, error)

+ func (a *AzureBlobStorage) CloseAsync()

+ func (a *AzureBlobStorage) Connect() error

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

+ func (a *AzureBlobStorage) WaitForClose(time.Duration) error

+ func (a *AzureBlobStorage) Write(msg types.Message) error

+ func (a *AzureBlobStorage) WriteWithContext(wctx context.Context, msg types.Message) error

+ type AzureBlobStorageConfig struct

+ BlobType string

+ Container string

+ MaxInFlight int

+ Path string

+ StorageAccessKey string

+ StorageAccount string

+ Timeout string

+ func NewAzureBlobStorageConfig() AzureBlobStorageConfig

Jul 5, 2020

Changes in this version

type AMQP1Config

+ SASL sasl.Config

type ZMQ4Config

+ Bind bool

+ HighWaterMark int

+ PollTimeout string

+ SocketType string

+ URLs []string

Jun 28, 2020

Changes in this version

+ type AMQP1 struct

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

+ func (a *AMQP1) CloseAsync()

+ func (a *AMQP1) Connect() error

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

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

+ func (a *AMQP1) Write(msg types.Message) error

+ func (a *AMQP1) WriteWithContext(ctx context.Context, msg types.Message) error

+ type AMQP1Config struct

+ MaxInFlight int

+ TLS btls.Config

+ TargetAddress string

+ URL string

+ func NewAMQP1Config() AMQP1Config

Jun 7, 2020

Changes in this version

type AMQPConfig

+ Type string

Feb 5, 2020

Changes in this version

type NSQConfig

+ TLS btls.Config

Jan 27, 2020

Changes in this version

+ type Socket struct

+ func NewSocket(conf SocketConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*Socket, error)

+ func (s *Socket) CloseAsync()

+ func (s *Socket) Connect() error

+ func (s *Socket) WaitForClose(timeout time.Duration) error

+ func (s *Socket) Write(msg types.Message) error

+ type SocketConfig struct

+ Address string

+ Network string

+ func NewSocketConfig() SocketConfig

Jan 16, 2020

Changes in this version

type AMQP

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

+ func (a *AMQP) WriteWithContext(ctx context.Context, msg types.Message) error

type AMQPConfig

+ MaxInFlight int

type AmazonS3

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

+ func (a *AmazonS3) WriteWithContext(wctx context.Context, msg types.Message) error

type AmazonS3Config

+ MaxInFlight int

+ StorageClass string

type AmazonSQS

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

+ func (a *AmazonSQS) WriteWithContext(ctx context.Context, msg types.Message) error

type AmazonSQSConfig

+ Batching batch.PolicyConfig

+ MaxInFlight int

type Cache

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

+ func (c *Cache) WriteWithContext(ctx context.Context, msg types.Message) error

type CacheConfig

+ MaxInFlight int

type DynamoDB

+ func (d *DynamoDB) ConnectWithContext(ctx context.Context) error

+ func (d *DynamoDB) WriteWithContext(ctx context.Context, msg types.Message) error

type DynamoDBConfig

+ Batching batch.PolicyConfig

+ MaxInFlight int

type Elasticsearch

+ func (e *Elasticsearch) ConnectWithContext(ctx context.Context) error

+ func (e *Elasticsearch) WriteWithContext(ctx context.Context, msg types.Message) error

type ElasticsearchConfig

+ Batching batch.PolicyConfig

+ MaxInFlight int

type Files

+ func (f *Files) ConnectWithContext(ctx context.Context) error

+ func (f *Files) WriteWithContext(ctx context.Context, msg types.Message) error

type GCPPubSub

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

+ func (c *GCPPubSub) WriteWithContext(ctx context.Context, msg types.Message) error

type GCPPubSubConfig

+ MaxInFlight int

type HDFS

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

+ func (h *HDFS) WriteWithContext(ctx context.Context, msg types.Message) error

type HDFSConfig

+ MaxInFlight int

type HTTPClient

+ func (h *HTTPClient) ConnectWithContext(ctx context.Context) error

+ func (h *HTTPClient) WriteWithContext(ctx context.Context, msg types.Message) error

type HTTPClientConfig

+ Batching batch.PolicyConfig

+ MaxInFlight int

type Kafka

+ func (k *Kafka) ConnectWithContext(ctx context.Context) error

+ func (k *Kafka) WriteWithContext(ctx context.Context, msg types.Message) error

type KafkaConfig

+ Batching batch.PolicyConfig

+ MaxInFlight int

type Kinesis

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

+ func (a *Kinesis) WriteWithContext(ctx context.Context, msg types.Message) error

type KinesisConfig

+ Batching batch.PolicyConfig

+ MaxInFlight int

type KinesisFirehose

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

+ func (a *KinesisFirehose) WriteWithContext(ctx context.Context, msg types.Message) error

type KinesisFirehoseConfig

+ Batching batch.PolicyConfig

+ MaxInFlight int

type MQTT

+ func (m *MQTT) ConnectWithContext(ctx context.Context) error

+ func (m *MQTT) WriteWithContext(ctx context.Context, msg types.Message) error

type MQTTConfig

+ MaxInFlight int

type NATS

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

+ func (n *NATS) WriteWithContext(ctx context.Context, msg types.Message) error

type NATSConfig

+ MaxInFlight int

type NATSStream

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

+ func (n *NATSStream) WriteWithContext(ctx context.Context, msg types.Message) error

type NATSStreamConfig

+ MaxInFlight int

type NSQ

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

+ func (n *NSQ) WriteWithContext(ctx context.Context, msg types.Message) error

type NSQConfig

+ MaxInFlight int

type Nanomsg

+ func (s *Nanomsg) ConnectWithContext(ctx context.Context) error

+ func (s *Nanomsg) WriteWithContext(ctx context.Context, msg types.Message) error

type NanomsgConfig

+ MaxInFlight int

type RedisHash

+ func (r *RedisHash) ConnectWithContext(ctx context.Context) error

+ func (r *RedisHash) WriteWithContext(ctx context.Context, msg types.Message) error

type RedisHashConfig

+ MaxInFlight int

type RedisList

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

+ func (r *RedisList) WriteWithContext(ctx context.Context, msg types.Message) error

type RedisListConfig

+ MaxInFlight int

type RedisPubSub

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

+ func (r *RedisPubSub) WriteWithContext(ctx context.Context, msg types.Message) error

type RedisPubSubConfig

+ MaxInFlight int

type RedisStreams

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

+ func (r *RedisStreams) WriteWithContext(ctx context.Context, msg types.Message) error

type RedisStreamsConfig

+ MaxInFlight int

type SNS

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

+ func (a *SNS) WriteWithContext(wctx context.Context, msg types.Message) error

type SNSConfig

+ MaxInFlight int

Dec 21, 2019

Changes in this version

type KafkaConfig

+ Partitioner string

Oct 20, 2019

Changes in this version

type AmazonS3Config

+ KMSKeyID string

type ElasticsearchConfig

+ Healthcheck bool

Sep 27, 2019

Changes in this version

type MQTTConfig

+ Password string

+ User string

Sep 23, 2019

Changes in this version

+ type KinesisFirehose struct

+ func NewKinesisFirehose(conf KinesisFirehoseConfig, log log.Modular, stats metrics.Type) (*KinesisFirehose, error)

+ func (a *KinesisFirehose) CloseAsync()

+ func (a *KinesisFirehose) Connect() error

+ func (a *KinesisFirehose) WaitForClose(time.Duration) error

+ func (a *KinesisFirehose) Write(msg types.Message) error

+ type KinesisFirehoseConfig struct

+ Stream string

+ func NewKinesisFirehoseConfig() KinesisFirehoseConfig

Sep 17, 2019

Changes in this version

+ type AMQP struct

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

+ func (a *AMQP) CloseAsync()

+ func (a *AMQP) Connect() error

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

+ func (a *AMQP) Write(msg types.Message) error

+ type AMQPConfig struct

+ BindingKey string

+ Exchange string

+ ExchangeDeclare AMQPExchangeDeclareConfig

+ Immediate bool

+ Mandatory bool

+ Persistent bool

+ TLS btls.Config

+ URL string

+ func NewAMQPConfig() AMQPConfig

+ type AMQPExchangeDeclareConfig struct

+ Durable bool

+ Enabled bool

+ Type string

+ type AmazonS3 struct

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

+ func (a *AmazonS3) CloseAsync()

+ func (a *AmazonS3) Connect() error

+ func (a *AmazonS3) WaitForClose(time.Duration) error

+ func (a *AmazonS3) Write(msg types.Message) error

+ type AmazonS3Config struct

+ Bucket string

+ ContentEncoding string

+ ContentType string

+ ForcePathStyleURLs bool

+ Path string

+ Timeout string

+ func NewAmazonS3Config() AmazonS3Config

+ type AmazonSQS struct

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

+ func (a *AmazonSQS) CloseAsync()

+ func (a *AmazonSQS) Connect() error

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

+ func (a *AmazonSQS) Write(msg types.Message) error

+ type AmazonSQSConfig struct

+ MessageDeduplicationID string

+ MessageGroupID string

+ URL string

+ func NewAmazonSQSConfig() AmazonSQSConfig

+ type Cache struct

+ func NewCache(conf CacheConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*Cache, error)

+ func (c *Cache) CloseAsync()

+ func (c *Cache) Connect() error

+ func (c *Cache) WaitForClose(time.Duration) error

+ func (c *Cache) Write(msg types.Message) error

+ type CacheConfig struct

+ Key string

+ Target string

+ func NewCacheConfig() CacheConfig

+ type Drop struct

+ func NewDrop(conf DropConfig, log log.Modular, stats metrics.Type) *Drop

+ func (d *Drop) CloseAsync()

+ func (d *Drop) Connect() error

+ func (d *Drop) WaitForClose(time.Duration) error

+ func (d *Drop) Write(msg types.Message) error

+ type DropConfig struct

+ func NewDropConfig() DropConfig

+ type DynamoDB struct

+ func NewDynamoDB(conf DynamoDBConfig, log log.Modular, stats metrics.Type) (*DynamoDB, error)

+ func (d *DynamoDB) CloseAsync()

+ func (d *DynamoDB) Connect() error

+ func (d *DynamoDB) WaitForClose(time.Duration) error

+ func (d *DynamoDB) Write(msg types.Message) error

+ type DynamoDBConfig struct

+ JSONMapColumns map[string]string

+ StringColumns map[string]string

+ TTL string

+ TTLKey string

+ Table string

+ func NewDynamoDBConfig() DynamoDBConfig

+ type Elasticsearch struct

+ func NewElasticsearch(conf ElasticsearchConfig, log log.Modular, stats metrics.Type) (*Elasticsearch, error)

+ func (e *Elasticsearch) CloseAsync()

+ func (e *Elasticsearch) Connect() error

+ func (e *Elasticsearch) WaitForClose(timeout time.Duration) error

+ func (e *Elasticsearch) Write(msg types.Message) error

+ type ElasticsearchConfig struct

+ AWS OptionalAWSConfig

+ Auth auth.BasicAuthConfig

+ ID string

+ Index string

+ Pipeline string

+ Sniff bool

+ Timeout string

+ Type string

+ URLs []string

+ func NewElasticsearchConfig() ElasticsearchConfig

+ type Files struct

+ func NewFiles(conf FilesConfig, log log.Modular, stats metrics.Type) *Files

+ func (f *Files) CloseAsync()

+ func (f *Files) Connect() error

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

+ func (f *Files) Write(msg types.Message) error

+ type FilesConfig struct

+ Path string

+ func NewFilesConfig() FilesConfig

+ type GCPPubSub struct

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

+ func (c *GCPPubSub) CloseAsync()

+ func (c *GCPPubSub) Connect() error

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

+ func (c *GCPPubSub) Write(msg types.Message) error

+ type GCPPubSubConfig struct

+ ProjectID string

+ TopicID string

+ func NewGCPPubSubConfig() GCPPubSubConfig

+ type HDFS struct

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

+ func (h *HDFS) CloseAsync()

+ func (h *HDFS) Connect() error

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

+ func (h *HDFS) Write(msg types.Message) error

+ type HDFSConfig struct

+ Directory string

+ Hosts []string

+ Path string

+ User string

+ func NewHDFSConfig() HDFSConfig

+ type HTTPClient struct

+ func NewHTTPClient(conf HTTPClientConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*HTTPClient, error)

+ func (h *HTTPClient) CloseAsync()

+ func (h *HTTPClient) Connect() error

+ func (h *HTTPClient) WaitForClose(timeout time.Duration) error

+ func (h *HTTPClient) Write(msg types.Message) error

+ type HTTPClientConfig struct

+ PropagateResponse bool

+ func NewHTTPClientConfig() HTTPClientConfig

+ type Kafka struct

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

+ func (k *Kafka) CloseAsync()

+ func (k *Kafka) Connect() error

+ func (k *Kafka) WaitForClose(timeout time.Duration) error

+ func (k *Kafka) Write(msg types.Message) error

+ type KafkaConfig struct

+ AckReplicas bool

+ Addresses []string

+ ClientID string

+ Compression string

+ Key string

+ MaxMsgBytes int

+ RoundRobinPartitions bool

+ SASL SASLConfig

+ TLS btls.Config

+ TargetVersion string

+ Timeout string

+ Topic string

+ func NewKafkaConfig() KafkaConfig

+ type Kinesis struct

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

+ func (a *Kinesis) CloseAsync()

+ func (a *Kinesis) Connect() error

+ func (a *Kinesis) WaitForClose(time.Duration) error

+ func (a *Kinesis) Write(msg types.Message) error

+ type KinesisConfig struct

+ HashKey string

+ PartitionKey string

+ Stream string

+ func NewKinesisConfig() KinesisConfig

+ type MQTT struct

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

+ func (m *MQTT) CloseAsync()

+ func (m *MQTT) Connect() error

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

+ func (m *MQTT) Write(msg types.Message) error

+ type MQTTConfig struct

+ ClientID string

+ QoS uint8

+ Topic string

+ URLs []string

+ func NewMQTTConfig() MQTTConfig

+ type NATS struct

+ func (n *NATS) CloseAsync()

+ func (n *NATS) Connect() error

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

+ func (n *NATS) Write(msg types.Message) error

+ type NATSConfig struct

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

+ func (n *NATSStream) Connect() error

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

+ func (n *NATSStream) Write(msg types.Message) error

+ type NATSStreamConfig struct

+ ClientID string

+ ClusterID string

+ Subject string

+ URLs []string

+ func NewNATSStreamConfig() NATSStreamConfig

+ type NSQ struct

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

+ func (n *NSQ) CloseAsync()

+ func (n *NSQ) Connect() error

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

+ func (n *NSQ) Write(msg types.Message) error

+ type NSQConfig struct

+ Address string

+ Topic string

+ UserAgent string

+ func NewNSQConfig() NSQConfig

+ type Nanomsg struct

+ func NewNanomsg(conf NanomsgConfig, log log.Modular, stats metrics.Type) (*Nanomsg, error)

+ func (s *Nanomsg) CloseAsync()

+ func (s *Nanomsg) Connect() error

+ func (s *Nanomsg) WaitForClose(timeout time.Duration) error

+ func (s *Nanomsg) Write(msg types.Message) error

+ type NanomsgConfig struct

+ Bind bool

+ PollTimeout string

+ SocketType string

+ URLs []string

+ func NewNanomsgConfig() NanomsgConfig

+ type OptionalAWSConfig struct

+ Enabled bool

+ type RedisHash struct

+ func NewRedisHash(conf RedisHashConfig, log log.Modular, stats metrics.Type) (*RedisHash, error)

+ func (r *RedisHash) CloseAsync()

+ func (r *RedisHash) Connect() error

+ func (r *RedisHash) WaitForClose(timeout time.Duration) error

+ func (r *RedisHash) Write(msg types.Message) error

+ type RedisHashConfig struct

+ Fields map[string]string

+ Key string

+ URL string

+ WalkJSONObject bool

+ WalkMetadata bool

+ func NewRedisHashConfig() RedisHashConfig

+ type RedisList struct

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

+ func (r *RedisList) CloseAsync()

+ func (r *RedisList) Connect() error

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

+ func (r *RedisList) Write(msg types.Message) error

+ type RedisListConfig struct

+ Key string

+ URL string

+ func NewRedisListConfig() RedisListConfig

+ type RedisPubSub struct

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

+ func (r *RedisPubSub) CloseAsync()

+ func (r *RedisPubSub) Connect() error

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

+ func (r *RedisPubSub) Write(msg types.Message) error

+ type RedisPubSubConfig struct

+ Channel string

+ URL string

+ func NewRedisPubSubConfig() RedisPubSubConfig

+ type RedisStreams struct

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

+ func (r *RedisStreams) CloseAsync()

+ func (r *RedisStreams) Connect() error

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

+ func (r *RedisStreams) Write(msg types.Message) error

+ type RedisStreamsConfig struct

+ BodyKey string

+ MaxLenApprox int64

+ Stream string

+ URL string

+ func NewRedisStreamsConfig() RedisStreamsConfig

+ type SASLConfig struct

+ Enabled bool

+ Password string

+ User string

+ type SNS struct

+ func NewSNS(conf SNSConfig, log log.Modular, stats metrics.Type) (*SNS, error)

+ func (a *SNS) CloseAsync()

+ func (a *SNS) Connect() error

+ func (a *SNS) WaitForClose(time.Duration) error

+ func (a *SNS) Write(msg types.Message) error

+ type SNSConfig struct

+ Timeout string

+ TopicArn string

+ func NewSNSConfig() SNSConfig

+ type TCP struct

+ func NewTCP(conf TCPConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*TCP, error)

+ func (t *TCP) CloseAsync()

+ func (t *TCP) Connect() error

+ func (t *TCP) WaitForClose(timeout time.Duration) error

+ func (t *TCP) Write(msg types.Message) error

+ type TCPConfig struct

+ Address string

+ func NewTCPConfig() TCPConfig

+ type Type interface

+ Connect func() error

+ Write func(msg types.Message) error

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

+ type UDP struct

+ func NewUDP(conf UDPConfig, mgr types.Manager, log log.Modular, stats metrics.Type) (*UDP, error)

+ func (t *UDP) CloseAsync()

+ func (t *UDP) Connect() error

+ func (t *UDP) WaitForClose(timeout time.Duration) error

+ func (t *UDP) Write(msg types.Message) error

+ type UDPConfig struct

+ Address string

+ func NewUDPConfig() UDPConfig

+ type Websocket struct

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

+ func (w *Websocket) CloseAsync()

+ func (w *Websocket) Connect() error

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

+ func (w *Websocket) Write(msg types.Message) error

+ type WebsocketConfig struct

+ URL string

+ func NewWebsocketConfig() WebsocketConfig

+ type ZMQ4Config struct

+ func NewZMQ4Config() *ZMQ4Config