kafka

Documentation

Index

• func NewClient(options ...func(*Client)) stream.Client
• func ParseKafkaURL(uri string) (string, string, string, error)
• type Client
  • func (c *Client) NewConsumer() stream.Consumer
  • func (c *Client) NewConsumerAndProducer() (stream.Consumer, stream.Producer)
  • func (c *Client) NewProducer() stream.Producer
• type Consumer
  • func (c *Consumer) Close() (err error)
  • func (c *Consumer) Messages() <-chan *stream.Message
• type Producer
  • func (p *Producer) Close() (err error)
  • func (p *Producer) Messages() chan<- *stream.Message
  • func (p *Producer) PartitionKey(f func(*stream.Message) []byte)

Functions

func NewClient

func NewClient(options ...func(*Client)) stream.Client

NewClient returns a new kafka client.

func ParseKafkaURL

func ParseKafkaURL(uri string) (string, string, string, error)

ParseKafkaURL parses a Kafka URL and returns the relevant components.

Types

type Client

type Client struct {
	ConsumerConfig  *cluster.Config
	ConsumerBrokers []string
	ConsumerTopics  []string
	ConsumerGroup   string

	ProducerConfig  *sarama.Config
	ProducerBrokers []string
	ProducerTopics  []string

	// Logger is the configurable logger instance to log messages from this
	// streamclient. If left undefined, a noop logger will be used.
	Logger *zap.Logger
}

Client provides access to the streaming capabilities.

func (*Client) NewConsumer

func (c *Client) NewConsumer() stream.Consumer

NewConsumer returns a consumer that can iterate over messages on a stream.

func (*Client) NewConsumerAndProducer

func (c *Client) NewConsumerAndProducer() (stream.Consumer, stream.Producer)

NewConsumerAndProducer is a convenience method that returns both a consumer and a producer, with a single function call.

func (*Client) NewProducer

func (c *Client) NewProducer() stream.Producer

NewProducer returns a producer that produces messages on a Kafka stream.

type Consumer

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

Consumer implements the stream.Consumer interface for standardstream.

func (*Consumer) Close

func (c *Consumer) Close() (err error)

Close closes the consumer connection.

func (*Consumer) Messages

func (c *Consumer) Messages() <-chan *stream.Message

Messages returns the read channel for the messages that are returned by the stream.

type Producer

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

Producer represents the object that will produce messages to a stream.

func (*Producer) Close

func (p *Producer) Close() (err error)

Close closes the producer connection.

func (*Producer) Messages

func (p *Producer) Messages() chan<- *stream.Message

Messages returns the read channel for the messages that are returned by the stream.

func (*Producer) PartitionKey

func (p *Producer) PartitionKey(f func(*stream.Message) []byte)

PartitionKey can be used to define the key to use for partitioning messages.

You pass in a function that accepts the currently processed message as its single value and should return byte slice, representing the partition key.

You can either set the key to a fixed byte slice, or determine the partition key based on the message's value or other properties.