kafka

Documentation

Index

• Constants
• func Read(opts *cli.Options) error
• func Relay(opts *cli.Options) error
• func Write(opts *cli.Options) error
• type Kafka
  • func (k *Kafka) Read() error
  • func (k *Kafka) Write(key, value []byte) error
• type KafkaReader
  • func NewReader(opts *cli.Options) (*KafkaReader, error)
• type KafkaWriter
  • func NewWriter(opts *cli.Options) (*KafkaWriter, error)

Constants

const (
	DefaultMaxBytes  = 1048576 // 1MB
	DefaultMaxWait   = 50 * time.Millisecond
	DefaultBatchSize = 1
)

Functions

func Read

func Read(opts *cli.Options) error

Read is the entry point function for performing read operations in Kafka.

This is where we verify that the provided arguments and flag combination makes sense/are valid; this is also where we will perform our initial conn.

func Relay

func Relay(opts *cli.Options) error

func Write

func Write(opts *cli.Options) error

Write is the entry point function for performing write operations in Kafka.

This is where we verify that the passed args and flags combo makes sense, attempt to establish a connection, parse protobuf before finally attempting to perform the write.

Types

type Kafka

type Kafka struct {
	Id      string
	Reader  *skafka.Reader
	Writer  *skafka.Writer
	Options *cli.Options
	MsgDesc *desc.MessageDescriptor
	// contains filtered or unexported fields
}

Kafka holds all attributes required for performing a write to Kafka. This struct should be instantiated via the kafka.Read(..) func.

func (*Kafka) Read

func (k *Kafka) Read() error

Read will attempt to consume one or more messages from a given topic, optionally decode it and/or convert the returned output.

This method SHOULD be able to recover from network hiccups.

func (*Kafka) Write

func (k *Kafka) Write(key, value []byte) error

Write writes a message to a kafka topic. It is a wrapper for WriteMessages.

type KafkaReader

type KafkaReader struct {
	Reader *skafka.Reader
	Conn   *skafka.Conn
}

func NewReader

func NewReader(opts *cli.Options) (*KafkaReader, error)

type KafkaWriter

type KafkaWriter struct {
	Writer *skafka.Writer
	Conn   *skafka.Conn
}

func NewWriter

func NewWriter(opts *cli.Options) (*KafkaWriter, error)