kafka

Documentation

Index

• func DeliveryReport(deliveryChan chan ckafka.Event)
• func NewKafkaProducer() *ckafka.Producer
• func Publish(msg string, topic string, producer *ckafka.Producer, ...) error
• type KafkaProcessor
  • func NewKafkaProcessor(database *gorm.DB, producer *ckafka.Producer, deliveryChan chan ckafka.Event) *KafkaProcessor
  • func (k *KafkaProcessor) Consume()
  • func (k *KafkaProcessor) ProcessMessage(msg *ckafka.Message)
  • func (k *KafkaProcessor) ProcessTransaction(msg *ckafka.Message) error
  • func (k *KafkaProcessor) ProcessTransactionConfirmation(msg *ckafka.Message) error

Functions

func DeliveryReport

func DeliveryReport(deliveryChan chan ckafka.Event)

func NewKafkaProducer

func NewKafkaProducer() *ckafka.Producer

func Publish

func Publish(msg string, topic string, producer *ckafka.Producer, deliveryChan chan ckafka.Event) error

Types

type KafkaProcessor

type KafkaProcessor struct {
	Database     *gorm.DB
	Producer     *ckafka.Producer
	DeliveryChan chan ckafka.Event
}

func NewKafkaProcessor

func NewKafkaProcessor(database *gorm.DB, producer *ckafka.Producer, deliveryChan chan ckafka.Event) *KafkaProcessor

func (*KafkaProcessor) Consume

func (k *KafkaProcessor) Consume()

func (*KafkaProcessor) ProcessMessage

func (k *KafkaProcessor) ProcessMessage(msg *ckafka.Message)

func (*KafkaProcessor) ProcessTransaction

func (k *KafkaProcessor) ProcessTransaction(msg *ckafka.Message) error

func (*KafkaProcessor) ProcessTransactionConfirmation

func (k *KafkaProcessor) ProcessTransactionConfirmation(msg *ckafka.Message) error