writer

Documentation

Overview

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

Index

• type AmazonAWSCredentialsConfig
• type AmazonS3
  • func NewAmazonS3(conf AmazonS3Config, log log.Modular, stats metrics.Type) *AmazonS3
  • 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
  • func NewAmazonS3Config() AmazonS3Config
• type Files
  • 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
  • func NewFilesConfig() FilesConfig
• 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 AmazonAWSCredentialsConfig

type AmazonAWSCredentialsConfig struct {
	ID     string `json:"id"`
	Secret string `json:"secret"`
	Token  string `json:"token"`
}

AmazonAWSCredentialsConfig contains configuration params for AWS credentials.

type AmazonS3

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

AmazonS3 is a benthos writer.Type implementation that writes messages to an Amazon S3 bucket.

func NewAmazonS3

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

NewAmazonS3 creates a new Amazon S3 bucket writer.Type.

func (*AmazonS3) CloseAsync

func (a *AmazonS3) CloseAsync()

CloseAsync begins cleaning up resources used by this reader asynchronously.

func (*AmazonS3) Connect

func (a *AmazonS3) Connect() error

Connect attempts to establish a connection to the target S3 bucket and any relevant queues used to traverse the objects (SQS, etc).

func (*AmazonS3) WaitForClose

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

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

func (*AmazonS3) Write

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

Write attempts to write message contents to a target S3 bucket as files.

type AmazonS3Config

type AmazonS3Config struct {
	Region      string                     `json:"region" yaml:"region"`
	Bucket      string                     `json:"bucket" yaml:"bucket"`
	Path        string                     `json:"path" yaml:"path"`
	Credentials AmazonAWSCredentialsConfig `json:"credentials" yaml:"credentials"`
	TimeoutS    int64                      `json:"timeout_s" yaml:"timeout_s"`
}

AmazonS3Config is configuration values for the input type.

func NewAmazonS3Config

func NewAmazonS3Config() AmazonS3Config

NewAmazonS3Config creates a new Config with default values.

type Files

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

Files is a benthos writer.Type implementation that writes messages parts each to their own file.

func NewFiles

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

NewFiles creates a new file based writer.Type.

func (*Files) CloseAsync

func (f *Files) CloseAsync()

CloseAsync begins cleaning up resources used by this reader asynchronously.

func (*Files) Connect

func (f *Files) Connect() error

Connect is a noop.

func (*Files) WaitForClose

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

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

func (*Files) Write

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

Write attempts to write message contents to a directory as files.

type FilesConfig

type FilesConfig struct {
	Path string `json:"path" yaml:"path"`
}

FilesConfig is configuration values for the input type.

func NewFilesConfig

func NewFilesConfig() FilesConfig

NewFilesConfig creates a new Config with default values.

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.