client

Documentation

Index

• Constants
• func EndHelp() string
• func FormatterHelp() string
• func ParseEnd(s string) (int64, error)
• func ParseStart(s string) (int64, error)
• func StartHelp() string
• type Consumer
  • func NewConsumer(broker string) (Consumer, error)
  • func (c Consumer) Close() error
  • func (c Consumer) Consume(ctx context.Context, req ConsumerRequest, messages chan<- Message) error
• type ConsumerRequest
  • func NewConsumerRequest(topic string, start int64, end int64) (ConsumerRequest, error)
• type Formatter
  • func NewFormatter(format string) Formatter
  • func (f Formatter) Format(m Message) string
• type Message
• type Producer
  • func NewProducer(broker string) (Producer, error)
  • func (p Producer) Close() error
  • func (p Producer) Produce(m Message) error

Constants

const (
	TopicPlaceholder     string = "%t"
	PartitionPlaceholder string = "%p"
	OffsetPlaceholder    string = "%o"
	TimestampPlaceholder string = "%e"
	KeyPlaceholder       string = "%k"
	ValuePlaceholder     string = "%v"
)

Valid format string placeholders.

const (
	StartOldest int64 = -1
	StartNow    int64 = -2
	EndNow      int64 = -3
	EndNever    int64 = -4
)

Special time values

const MessageBufferSize = 256

MessageBufferSize is the default buffer size for consuming/producing.

Functions

func EndHelp

func EndHelp() string

EndHelp returns a message describing valid end values.

func FormatterHelp

func FormatterHelp() string

FormatterHelp returns a message describing format tokens.

func ParseEnd

func ParseEnd(s string) (int64, error)

ParseEnd returns an end time parsed from s.

func ParseStart

func ParseStart(s string) (int64, error)

ParseStart returns a start time parsed from s.

func StartHelp

func StartHelp() string

StartHelp returns a message describing valid start values.

Types

type Consumer

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

Consumer is a high level API for a Kafka consumer.

func NewConsumer

func NewConsumer(broker string) (Consumer, error)

NewConsumer returns a Consumer for the given broker.

func (Consumer) Close

func (c Consumer) Close() error

Close shuts the consumer down.

func (Consumer) Consume

func (c Consumer) Consume(ctx context.Context, req ConsumerRequest, messages chan<- Message) error

Consume reads messages matching the ConsumerRequest and sends them to the given chan until ctx is done.

type ConsumerRequest

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

A ConsumerRequest specifies what to consume and how.

func NewConsumerRequest

func NewConsumerRequest(topic string, start int64, end int64) (ConsumerRequest, error)

NewConsumerRequest creates a valid ConsumerRequest.

type Formatter

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

Formatter describes a way of printing Kafka messages.

func NewFormatter

func NewFormatter(format string) Formatter

NewFormatter returns a Formatter initialized with a format string.

func (Formatter) Format

func (f Formatter) Format(m Message) string

Format returns a copy of the Formatter format string with the placeholders replaced with values from the Message.

type Message

type Message struct {
	Topic     string
	Partition int32
	Offset    int64
	Timestamp time.Time
	Key       string
	Value     string
}

Message represents a Kafka message.

type Producer

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

Producer is a high level API for a Kafka producer.

func NewProducer

func NewProducer(broker string) (Producer, error)

NewProducer returns a Producer for the given broker.

func (Producer) Close

func (p Producer) Close() error

Close shuts the producer down.

func (Producer) Produce

func (p Producer) Produce(m Message) error

Produce sends a Message to Kafka.