kafka

Documentation

Index

• Constants
• func DefaultSaramaSubscriberConfig() *sarama.Config
• func DefaultSaramaSyncPublisherConfig() *sarama.Config
• func MessagePartitionFromCtx(ctx context.Context) (int32, bool)
• func MessagePartitionOffsetFromCtx(ctx context.Context) (int64, bool)
• func MessageTimestampFromCtx(ctx context.Context) (time.Time, bool)
• type DefaultMarshaler
  • func (DefaultMarshaler) Marshal(topic string, msg *message.Message) (*sarama.ProducerMessage, error)
  • func (DefaultMarshaler) Unmarshal(kafkaMsg *sarama.ConsumerMessage) (*message.Message, error)
• type GeneratePartitionKey
• type Marshaler
• type MarshalerUnmarshaler
  • func NewWithPartitioningMarshaler(generatePartitionKey GeneratePartitionKey) MarshalerUnmarshaler
• type PartitionOffset
• type Publisher
  • func NewPublisher(config PublisherConfig, logger watermill.LoggerAdapter) (*Publisher, error)
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(topic string, msgs ...*message.Message) error
• type PublisherConfig
  • func (c PublisherConfig) Validate() error
• type Subscriber
  • func NewSubscriber(config SubscriberConfig, logger watermill.LoggerAdapter) (*Subscriber, error)
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) PartitionOffset(topic string) (PartitionOffset, error)
  • func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)
  • func (s *Subscriber) SubscribeInitialize(topic string) (err error)
• type SubscriberConfig
  • func (c SubscriberConfig) Validate() error
• type Unmarshaler

Constants

const NoSleep time.Duration = -1

NoSleep can be set to SubscriberConfig.NackResendSleep and SubscriberConfig.ReconnectRetrySleep.

const UUIDHeaderKey = "_watermill_message_uuid"

Functions

func DefaultSaramaSubscriberConfig

func DefaultSaramaSubscriberConfig() *sarama.Config

DefaultSaramaSubscriberConfig creates default Sarama config used by Watermill.

Custom config can be passed to NewSubscriber and NewPublisher.

saramaConfig := DefaultSaramaSubscriberConfig()
saramaConfig.Consumer.Offsets.Initial = sarama.OffsetOldest

subscriberConfig.OverwriteSaramaConfig = saramaConfig

subscriber, err := NewSubscriber(subscriberConfig, logger)
// ...

func DefaultSaramaSyncPublisherConfig

func DefaultSaramaSyncPublisherConfig() *sarama.Config

func MessagePartitionFromCtx

func MessagePartitionFromCtx(ctx context.Context) (int32, bool)

MessagePartitionFromCtx returns Kafka partition of the consumed message

func MessagePartitionOffsetFromCtx

func MessagePartitionOffsetFromCtx(ctx context.Context) (int64, bool)

MessagePartitionOffsetFromCtx returns Kafka partition offset of the consumed message

func MessageTimestampFromCtx

func MessageTimestampFromCtx(ctx context.Context) (time.Time, bool)

MessageTimestampFromCtx returns Kafka internal timestamp of the consumed message

Types

type DefaultMarshaler

type DefaultMarshaler struct{}

func (DefaultMarshaler) Marshal

func (DefaultMarshaler) Marshal(topic string, msg *message.Message) (*sarama.ProducerMessage, error)

func (DefaultMarshaler) Unmarshal

func (DefaultMarshaler) Unmarshal(kafkaMsg *sarama.ConsumerMessage) (*message.Message, error)

type GeneratePartitionKey

type GeneratePartitionKey func(topic string, msg *message.Message) (string, error)

type Marshaler

type Marshaler interface {
	Marshal(topic string, msg *message.Message) (*sarama.ProducerMessage, error)
}

Marshaler marshals Watermill's message to Kafka message.

type MarshalerUnmarshaler

type MarshalerUnmarshaler interface {
	Marshaler
	Unmarshaler
}

func NewWithPartitioningMarshaler

func NewWithPartitioningMarshaler(generatePartitionKey GeneratePartitionKey) MarshalerUnmarshaler

type PartitionOffset

type PartitionOffset map[int32]int64

type Publisher

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

func NewPublisher

func NewPublisher(
	config PublisherConfig,
	logger watermill.LoggerAdapter,
) (*Publisher, error)

NewPublisher creates a new Kafka Publisher.

func (*Publisher) Close

func (p *Publisher) Close() error

func (*Publisher) Publish

func (p *Publisher) Publish(topic string, msgs ...*message.Message) error

Publish publishes message to Kafka.

Publish is blocking and wait for ack from Kafka. When one of messages delivery fails - function is interrupted.

type PublisherConfig

type PublisherConfig struct {
	// Kafka brokers list.
	Brokers []string

	// Marshaler is used to marshal messages from Watermill format into Kafka format.
	Marshaler Marshaler

	// OverwriteSaramaConfig holds additional sarama settings.
	OverwriteSaramaConfig *sarama.Config

	// If true then each sent message will be wrapped with Opentelemetry tracing, provided by otelsarama.
	TracingEnabled bool
}

func (PublisherConfig) Validate

func (c PublisherConfig) Validate() error

type Subscriber

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

func NewSubscriber

func NewSubscriber(
	config SubscriberConfig,
	logger watermill.LoggerAdapter,
) (*Subscriber, error)

NewSubscriber creates a new Kafka Subscriber.

func (*Subscriber) Close

func (s *Subscriber) Close() error

func (*Subscriber) PartitionOffset

func (s *Subscriber) PartitionOffset(topic string) (PartitionOffset, error)

func (*Subscriber) Subscribe

func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)

Subscribe subscribers for messages in Kafka.

There are multiple subscribers spawned

func (*Subscriber) SubscribeInitialize

func (s *Subscriber) SubscribeInitialize(topic string) (err error)

type SubscriberConfig

type SubscriberConfig struct {
	// Kafka brokers list.
	Brokers []string

	// Unmarshaler is used to unmarshal messages from Kafka format into Watermill format.
	Unmarshaler Unmarshaler

	// OverwriteSaramaConfig holds additional sarama settings.
	OverwriteSaramaConfig *sarama.Config

	// Kafka consumer group.
	// When empty, all messages from all partitions will be returned.
	ConsumerGroup string

	// How long after Nack message should be redelivered.
	NackResendSleep time.Duration

	// How long about unsuccessful reconnecting next reconnect will occur.
	ReconnectRetrySleep time.Duration

	InitializeTopicDetails *sarama.TopicDetail

	// If true then each consumed message will be wrapped with Opentelemetry tracing, provided by otelsarama.
	TracingEnabled bool
}

func (SubscriberConfig) Validate

func (c SubscriberConfig) Validate() error

type Unmarshaler

type Unmarshaler interface {
	Unmarshal(*sarama.ConsumerMessage) (*message.Message, error)
}

Unmarshaler unmarshals Kafka's message to Watermill's message.