kafka

Documentation

Index

• type KafkaConsumer
  • func NewKafkaConsumer() *KafkaConsumer
  • func (k *KafkaConsumer) Consume(ctx context.Context, msg *sarama.ConsumerMessage, consumerGroup string, ...)
• type KafkaProducer
  • func NewKafkaProducer(producer sarama.AsyncProducer) *KafkaProducer
  • func (k *KafkaProducer) Publish(ctx context.Context, msg *sarama.ProducerMessage, errType *string)

Types

type KafkaConsumer

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

func NewKafkaConsumer

func NewKafkaConsumer() *KafkaConsumer

func (*KafkaConsumer) Consume

func (k *KafkaConsumer) Consume(
	ctx context.Context,
	msg *sarama.ConsumerMessage,
	consumerGroup string,
	consumeFunc func(context.Context) error,
)

type KafkaProducer

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

func NewKafkaProducer

func NewKafkaProducer(
	producer sarama.AsyncProducer,
) *KafkaProducer

func (*KafkaProducer) Publish

func (k *KafkaProducer) Publish(
	ctx context.Context,
	msg *sarama.ProducerMessage,
	errType *string,
)