kafka

Documentation

Index

• type Consumer
  • func New(ctx context.Context, opts ...ConsumerFunc) (*Consumer, error)
  • func (c *Consumer) Close()
  • func (c *Consumer) GetNextBatch(ctx context.Context, topic string, count int) ([][]byte, error)
  • func (c *Consumer) GetNextMessage(ctx context.Context, topic string) ([]byte, error)
• type ConsumerFunc
  • func WithBrokers(brokers []string) ConsumerFunc
  • func WithBufferSize(size int) ConsumerFunc
  • func WithGroupID(groupID string) ConsumerFunc
  • func WithTopics(topics []string) ConsumerFunc

Types

type Consumer

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

Consumer represents a buffered Kafka consumer.

func New

func New(ctx context.Context, opts ...ConsumerFunc) (*Consumer, error)

New creates a new Consumer instance.

func (*Consumer) Close

func (c *Consumer) Close()

Close closes the Consumer.

func (*Consumer) GetNextBatch

func (c *Consumer) GetNextBatch(ctx context.Context, topic string, count int) ([][]byte, error)

GetNextBatch gets the next count messages from the queue.

func (*Consumer) GetNextMessage

func (c *Consumer) GetNextMessage(ctx context.Context, topic string) ([]byte, error)

GetNextMessage gets the next message from the queue.

type ConsumerFunc

type ConsumerFunc func(*Consumer)

ConsumerFunc represents a function that configures the Consumer.

func WithBrokers

func WithBrokers(brokers []string) ConsumerFunc

WithBrokers sets the brokers on the Consumer.

func WithBufferSize

func WithBufferSize(size int) ConsumerFunc

WithBufferSize sets the buffer size on the Consumer.

func WithGroupID

func WithGroupID(groupID string) ConsumerFunc

WithGroupID sets the group id on the Consumer.

func WithTopics

func WithTopics(topics []string) ConsumerFunc

WithTopics sets the topics on the Consumer.