Documentation ¶
Index ¶
- Constants
- func DefaultSaramaSubscriberConfig() *sarama.Config
- func DefaultSaramaSyncPublisherConfig() *sarama.Config
- func MessageKeyFromCtx(ctx context.Context) ([]byte, bool)
- func MessagePartitionFromCtx(ctx context.Context) (int32, bool)
- func MessagePartitionOffsetFromCtx(ctx context.Context) (int64, bool)
- func MessageTimestampFromCtx(ctx context.Context) (time.Time, bool)
- type BatchConsumerConfig
- type ConsumerModel
- type DefaultMarshaler
- type GeneratePartitionKey
- type Marshaler
- type MarshalerUnmarshaler
- type MessageHandler
- func NewBatchedMessageHandler(outputChannel chan<- *message.Message, unmarshaler Unmarshaler, ...) MessageHandler
- func NewMessageHandler(outputChannel chan<- *message.Message, unmarshaler Unmarshaler, ...) MessageHandler
- func NewPartitionConcurrentMessageHandler(outputChannel chan<- *message.Message, unmarshaler Unmarshaler, ...) MessageHandler
- type OTELSaramaTracer
- func (t OTELSaramaTracer) WrapConsumer(c sarama.Consumer, consumerInfo otelsaramax.ConsumerInfo) sarama.Consumer
- func (t OTELSaramaTracer) WrapConsumerGroupHandler(h sarama.ConsumerGroupHandler, consumerInfo otelsaramax.ConsumerInfo) sarama.ConsumerGroupHandler
- func (t OTELSaramaTracer) WrapSyncProducer(cfg *sarama.Config, p sarama.SyncProducer) sarama.SyncProducer
- type PartitionOffset
- type Publisher
- type PublisherConfig
- type SaramaTracer
- type Subscriber
- type SubscriberConfig
- type Unmarshaler
Constants ¶
const NoSleep time.Duration = -1
NoSleep can be set to SubscriberConfig.NackResendSleep and SubscriberConfig.ReconnectRetrySleep.
const UUIDHeaderKey = "_watermill_message_uuid"
Variables ¶
This section is empty.
Functions ¶
func DefaultSaramaSubscriberConfig ¶
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 MessageKeyFromCtx ¶
MessageKeyFromCtx returns Kafka internal key of the consumed message
func MessagePartitionFromCtx ¶
MessagePartitionFromCtx returns Kafka partition of the consumed message
func MessagePartitionOffsetFromCtx ¶
MessagePartitionOffsetFromCtx returns Kafka partition offset of the consumed message
Types ¶
type BatchConsumerConfig ¶
type BatchConsumerConfig struct { // MaxBatchSize max amount of elements the batch will contain. // Default value is 100 if nothing is specified. MaxBatchSize int16 // MaxWaitTime max time that it will be waited until MaxBatchSize elements are received. // Default value is 100ms if nothing is specified. MaxWaitTime time.Duration }
BatchConsumerConfig configuration to be applied when the selected type of consumption is batch. Batch consumption means that the MaxBatchSize will be read or maxWaitTime waited the messages will then be sent to the output channel. ACK / NACK are handled properly to ensure at-least-once consumption.
type ConsumerModel ¶
type ConsumerModel int
ConsumerModel indicates the type of consumer model that will be used.
const ( // Default is a model when only one message is sent to the customer and customer needs to ACK the message // to receive the next. Default ConsumerModel = iota // Batch works by sending multiple messages in a batch Batch // PartitionConcurrent has one message sent to the customer per partition and customer needs to ACK the message // to receive the next message for the partition. PartitionConcurrent )
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 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 MessageHandler ¶
type MessageHandler interface { // Setup is called at the beginning of a new session, before ConsumeClaim (and thus ProcessMessages) is called. // It may have a consumer group session in the case of a consumer group. Setup(*sarama.ConsumerGroupSession) error Cleanup(*sarama.ConsumerGroupSession) error ProcessMessages( ctx context.Context, kafkaMessages <-chan *sarama.ConsumerMessage, sess sarama.ConsumerGroupSession, messageLogFields watermill.LogFields, ) error }
MessageHandler an message processor that is able to receive a ConsumerMessage and perform some task with it. Once consumed, if there is a session, it will the offset will be marked as processed.
func NewBatchedMessageHandler ¶
func NewBatchedMessageHandler( outputChannel chan<- *message.Message, unmarshaler Unmarshaler, logger watermill.LoggerAdapter, closing chan struct{}, maxBatchSize int16, maxWaitTime time.Duration, nackResendSleep time.Duration, ) MessageHandler
func NewMessageHandler ¶
func NewMessageHandler( outputChannel chan<- *message.Message, unmarshaler Unmarshaler, logger watermill.LoggerAdapter, closing chan struct{}, nackResendSleep time.Duration, ) MessageHandler
func NewPartitionConcurrentMessageHandler ¶
func NewPartitionConcurrentMessageHandler( outputChannel chan<- *message.Message, unmarshaler Unmarshaler, logger watermill.LoggerAdapter, closing chan struct{}, nackResendSleep time.Duration, ) MessageHandler
type OTELSaramaTracer ¶
type OTELSaramaTracer struct {
// contains filtered or unexported fields
}
func (OTELSaramaTracer) WrapConsumer ¶
func (t OTELSaramaTracer) WrapConsumer(c sarama.Consumer, consumerInfo otelsaramax.ConsumerInfo) sarama.Consumer
func (OTELSaramaTracer) WrapConsumerGroupHandler ¶
func (t OTELSaramaTracer) WrapConsumerGroupHandler(h sarama.ConsumerGroupHandler, consumerInfo otelsaramax.ConsumerInfo) sarama.ConsumerGroupHandler
func (OTELSaramaTracer) WrapSyncProducer ¶
func (t OTELSaramaTracer) WrapSyncProducer(cfg *sarama.Config, p sarama.SyncProducer) sarama.SyncProducer
type PartitionOffset ¶
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) BulkPublish ¶ added in v0.0.44
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. OTELEnabled bool // Tracer is used to trace Kafka messages. // If nil, then no tracing will be used. Tracer SaramaTracer }
func (PublisherConfig) Validate ¶
func (c PublisherConfig) Validate() error
type SaramaTracer ¶
type SaramaTracer interface { WrapConsumer(sarama.Consumer, otelsaramax.ConsumerInfo) sarama.Consumer // WrapPartitionConsumer(sarama.PartitionConsumer) sarama.PartitionConsumer WrapConsumerGroupHandler(sarama.ConsumerGroupHandler, otelsaramax.ConsumerInfo) sarama.ConsumerGroupHandler WrapSyncProducer(*sarama.Config, sarama.SyncProducer) sarama.SyncProducer }
func NewOTELSaramaTracer ¶
func NewOTELSaramaTracer(option ...otelsaramax.Option) SaramaTracer
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 ¶
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. // // Deprecated: pass OTELSaramaTracer to Tracer field instead. OTELEnabled bool // Tracer is used to trace Kafka messages. // If nil, then no tracing will be used. Tracer SaramaTracer // ConsumerModel indicates which type of consumer should be used ConsumerModel ConsumerModel // When set to not nil, consumption will be performed in batches and this configuration will be used BatchConsumerConfig *BatchConsumerConfig }
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.