writer

Documentation

Overview

Package writer defines implementations of an interface for generic message writing that outputs to various third party sinks.

Index

• type Kafka
  • 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
  • func NewKafkaConfig() KafkaConfig
• type RedisList
  • 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
  • func NewRedisListConfig() RedisListConfig
• type Type

Types

type Kafka

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

Kafka is a writer type that writes messages into kafka.

func NewKafka

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

NewKafka creates a new Kafka writer type.

func (*Kafka) CloseAsync

func (k *Kafka) CloseAsync()

CloseAsync shuts down the Kafka writer and stops processing messages.

func (*Kafka) Connect

func (k *Kafka) Connect() error

Connect attempts to establish a connection to a Kafka broker.

func (*Kafka) WaitForClose

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

WaitForClose blocks until the Kafka writer has closed down.

func (*Kafka) Write

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

Write will attempt to write a message to Kafka, wait for acknowledgement, and returns an error if applicable.

type KafkaConfig

type KafkaConfig struct {
	Addresses   []string `json:"addresses" yaml:"addresses"`
	ClientID    string   `json:"client_id" yaml:"client_id"`
	Topic       string   `json:"topic" yaml:"topic"`
	MaxMsgBytes int      `json:"max_msg_bytes" yaml:"max_msg_bytes"`
	TimeoutMS   int      `json:"timeout_ms" yaml:"timeout_ms"`
	AckReplicas bool     `json:"ack_replicas" yaml:"ack_replicas"`
}

KafkaConfig is configuration for the Kafka output type.

func NewKafkaConfig

func NewKafkaConfig() KafkaConfig

NewKafkaConfig creates a new KafkaConfig with default values.

type RedisList

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

RedisList is an output type that serves RedisList messages.

func NewRedisList

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

NewRedisList creates a new RedisList output type.

func (*RedisList) CloseAsync

func (r *RedisList) CloseAsync()

CloseAsync shuts down the RedisList output and stops processing messages.

func (*RedisList) Connect

func (r *RedisList) Connect() error

Connect establishes a connection to an RedisList server.

func (*RedisList) WaitForClose

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

WaitForClose blocks until the RedisList output has closed down.

func (*RedisList) Write

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

Write attempts to write a message by pushing it to the end of a Redis list.

type RedisListConfig

type RedisListConfig struct {
	URL string `json:"url" yaml:"url"`
	Key string `json:"key" yaml:"key"`
}

RedisListConfig is configuration for the RedisList output type.

func NewRedisListConfig

func NewRedisListConfig() RedisListConfig

NewRedisListConfig creates a new RedisListConfig with default values.

type Type

type Type interface {
	// Connect attempts to establish a connection to the sink, if unsuccessful
	// returns an error. If the attempt is successful (or not necessary) returns
	// nil.
	Connect() error

	// Write should block until either the message is sent (and acknowledged) to
	// a sink, or a transport specific error has occurred, or the Type is
	// closed.
	Write(msg types.Message) error

	types.Closable
}

Type is a type that writes Benthos messages to a third party sink.