kafka

Documentation

Index

• type Consumer
  • func New(name, ct, topic string, brokers []string, oo ...OptionFunc) (*Consumer, error)
  • func (c *Consumer) Close() error
  • func (c *Consumer) Consume(ctx context.Context) (<-chan async.Message, <-chan error, error)
• type Offset
• type OptionFunc
  • func Buffer(buf int) OptionFunc
  • func Start(start Offset) OptionFunc
  • func Timeout(timeout time.Duration) OptionFunc
  • func Version(version string) OptionFunc

Types

type Consumer

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

Consumer definition of a Kafka consumer.

func New

func New(name, ct, topic string, brokers []string, oo ...OptionFunc) (*Consumer, error)

New creates a ew Kafka consumer with defaults. To override those default you should provide a option.

func (*Consumer) Close

func (c *Consumer) Close() error

Close handles closing channel and connection of AMQP.

func (*Consumer) Consume

func (c *Consumer) Consume(ctx context.Context) (<-chan async.Message, <-chan error, error)

Consume starts consuming messages from a Kafka topic.

type Offset

type Offset int64

Offset defines the offset of messages inside a topic.

const (
	// OffsetNewest starts consuming from the newest available message in the topic.
	OffsetNewest Offset = -1
	// OffsetOldest starts consuming from the oldest available message in the topic.
	OffsetOldest Offset = -2
)

type OptionFunc

type OptionFunc func(*Consumer) error

OptionFunc definition for configuring the consumer in a functional way.

func Buffer

func Buffer(buf int) OptionFunc

Buffer option for adjusting the incoming messages buffer.

func Start

func Start(start Offset) OptionFunc

Start option for adjusting the start point in the topic.

func Timeout

func Timeout(timeout time.Duration) OptionFunc

Timeout option for adjusting the timeout of the connection.

func Version

func Version(version string) OptionFunc

Version option for setting the Kafka version.