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 (processor *KafkaProcessor) Consume()

Functions

func DeliveryReport

func DeliveryReport(deliveryChan chan ckafka.Event)

DeliveryReport monitories a channel kafka

func NewKafkaProducer

func NewKafkaProducer() *ckafka.Producer

NewKafkaProducer creates a new kafka producer

func Publish

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

Publish publishes a message to a topic

Types

type KafkaProcessor

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

KafkaProcessor represents a processor for kafka messages

func NewKafkaProcessor

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

NewKafkaProcessor creates a new processor for kafka messages

func (*KafkaProcessor) Consume

func (processor *KafkaProcessor) Consume()

Consume consumes messages