kafka

Documentation

Index

• Constants
• func NewReader(opts *cli.Options) (*skafka.Reader, error)
• func NewWriter(opts *cli.Options) (*skafka.Writer, error)
• 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

Constants

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

Functions

func NewReader

func NewReader(opts *cli.Options) (*skafka.Reader, error)

func NewWriter

func NewWriter(opts *cli.Options) (*skafka.Writer, error)

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
	MessageDesc *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.