kafka

Documentation

Overview

Package kafka provides some shared interfaces for the Kafka components.

Index

• func DefaultSaramaConfig(name string) (*sarama.Config, error)
• type Batch
  • func NewBatch(messages []Message) Batch
• type BatchProcessorFunc
• type FailStrategy
• type Message
  • func NewMessage(ctx context.Context, sp opentracing.Span, msg *sarama.ConsumerMessage) Message

Functions

func DefaultSaramaConfig

func DefaultSaramaConfig(name string) (*sarama.Config, error)

DefaultSaramaConfig function creates a sarama config object with the default configuration set up.

Types

type Batch

type Batch interface {
	// Messages of the batch.
	Messages() []Message
}

Batch interface for multiple AWS SQS messages.

func NewBatch

func NewBatch(messages []Message) Batch

NewBatch initializes a new batch of messages returning an instance of the implementation of the kafka Batch interface

type BatchProcessorFunc

type BatchProcessorFunc func(Batch) error

BatchProcessorFunc definition of a batch async processor function.

type FailStrategy

type FailStrategy int

FailStrategy type definition.

const (
	// ExitStrategy does not commit failed message offsets and exits the application.
	ExitStrategy FailStrategy = iota
	// SkipStrategy commits the offset of messages that failed processing, and continues processing.
	SkipStrategy
)

type Message

type Message interface {
	// Context will contain the context to be used for processing.
	// Each context will have a logger setup which can be used to create a logger from context.
	Context() context.Context
	// Message will contain the raw Kafka message.
	Message() *sarama.ConsumerMessage
	// Span contains the tracing span of this message.
	Span() opentracing.Span
}

Message interface for wrapping messages that are handled by the kafka component.

func NewMessage

func NewMessage(ctx context.Context, sp opentracing.Span, msg *sarama.ConsumerMessage) Message

NewMessage initializes a new message which is an implementation of the kafka Message interface