README
¶
Watermill Kafka Pub/Sub
This is Pub/Sub for the Watermill project.
All Pub/Sub implementations can be found at https://watermill.io/pubsubs/.
Watermill is a Go library for working efficiently with message streams. It is intended for building event driven applications, enabling event sourcing, RPC over messages, sagas and basically whatever else comes to your mind. You can use conventional pub/sub implementations like Kafka or RabbitMQ, but also HTTP or MySQL binlog if that fits your use case.
Documentation: https://watermill.io/
Getting started guide: https://watermill.io/docs/getting-started/
Issues: https://github.com/ThreeDotsLabs/watermill/issues
Contributing
All contributions are very much welcome. If you'd like to help with Watermill development, please see open issues and submit your pull request via GitHub.
Support
If you didn't find the answer to your question in the documentation, feel free to ask us directly!
Please join us on the #watermill channel on the Gophers slack: You can get an invite here.
License
Documentation
¶
Overview ¶
Package kafka fork from github.com/ThreeDotsLabs/watermill-kafka@v2.5.0
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 DefaultMarshaler
- type GeneratePartitionKey
- type Marshaler
- type MarshalerUnmarshaler
- type PartitionOffset
- type Publisher
- type PublisherConfig
- 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 DefaultMarshaler ¶
type DefaultMarshaler struct {
AppID string
}
func (DefaultMarshaler) Marshal ¶
func (d DefaultMarshaler) Marshal(topic string, msg *message.Message) (*sarama.ProducerMessage, error)
func (DefaultMarshaler) Unmarshal ¶
func (d 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 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.
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
}
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, ) (sub *Subscriber, err 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
}
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.
Source Files