README
¶
#edat-kafka-go - Kafka for edat
Installation
go get -u github.com/stackus/edat-kafka-go
Usage Example
import (
"github.com/stackus/edat-kafka-go"
"github.com/stackus/edat/msg"
)
// Create a consumer and use it in a message subscriber
consumer := edatkafkago.NewConsumer(brokers, groupID)
subscriber := msg.NewSubscriber(consumer)
// Create a producer and use it in a message publisher
producer := edatkafkago.NewProducer(brokers)
publisher := msg.NewPublisher(producer)
Prerequisites
Go 1.15
Features
- Message Consumer
NewConsumer(brokers, groupID, ...options) - Message Producer
NewProducer(brokers, ...options)
TODOs
- Documentation
- Tests, tests, and more tests
Contributing
Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.
Please make sure to update tests as appropriate.
License
MIT
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
View Source
var DefaultAckWait = time.Second * 30
DefaultAckWait is a time.Duration representing the maximum amount of time for a consumer to finish
View Source
var DefaultSerializer = KafkaGoSerializer{}
Functions ¶
This section is empty.
Types ¶
type Consumer ¶
type Consumer struct {
// contains filtered or unexported fields
}
Consumer implements msg.Consumer
func NewConsumer ¶
func NewConsumer(brokers []string, groupID string, options ...ConsumerOption) *Consumer
NewConsumer constructs a new instance of Consumer
type ConsumerOption ¶
type ConsumerOption func(*Consumer)
func WithConsumerAckWait ¶
func WithConsumerAckWait(ackWait time.Duration) ConsumerOption
func WithConsumerDialer ¶
func WithConsumerDialer(dialer *kafka.Dialer) ConsumerOption
func WithConsumerLogger ¶
func WithConsumerLogger(logger log.Logger) ConsumerOption
func WithConsumerSerializer ¶
func WithConsumerSerializer(serializer Serializer) ConsumerOption
type KafkaGoSerializer ¶
type KafkaGoSerializer struct{}
func (KafkaGoSerializer) Deserialize ¶
func (s KafkaGoSerializer) Deserialize(message kafka.Message) (msg.Message, error)
type Producer ¶
type Producer struct {
// contains filtered or unexported fields
}
Producer implements msg.Producer
func NewProducer ¶
func NewProducer(brokers []string, options ...ProducerOption) *Producer
NewProducer constructs a new instance of Producer
type ProducerOption ¶
type ProducerOption func(*Producer)
func WithProducerLogger ¶
func WithProducerLogger(logger log.Logger) ProducerOption
func WithProducerSerializer ¶
func WithProducerSerializer(serializer Serializer) ProducerOption
func WithProducerTransport ¶
func WithProducerTransport(transport *kafka.Transport) ProducerOption
Source Files
Click to show internal directories.
Click to hide internal directories.