kafka

Documentation

Index

• func NewKafkaReader(kafkaConfig helpers.KafkaConfig) kafka.Reader
• func NewKafkaWriter(kafkaConfig helpers.KafkaConfig) kafka.Writer
• func SendAcknowledgement(logID string) error
• type KafkaConsumer
  • func NewKafkaConsumer(kafkaConfig helpers.KafkaConfig, logPersister LogPersister) *KafkaConsumer
  • func (c *KafkaConsumer) ConsumeLogKafka(ctx context.Context)
• type KafkaProducer
  • func NewKafkaProducer(kafkaConfig helpers.KafkaConfig) *KafkaProducer
  • func (p *KafkaProducer) Acknowledge(logID string)
  • func (p *KafkaProducer) CreateKafkaTopic() error
  • func (p *KafkaProducer) ProduceLogKafka(c echo.Context, logData types.LogData) error
  • func (p *KafkaProducer) WaitForAck(logID string, timeout time.Duration) (bool, error)
• type LogPersister

Functions

func NewKafkaReader

func NewKafkaReader(kafkaConfig helpers.KafkaConfig) kafka.Reader

func NewKafkaWriter

func NewKafkaWriter(kafkaConfig helpers.KafkaConfig) kafka.Writer

func SendAcknowledgement

func SendAcknowledgement(logID string) error

Types

type KafkaConsumer

type KafkaConsumer struct {
	KafkaConfig helpers.KafkaConfig
	Reader      kafka.Reader
	Persister   LogPersister
}

func NewKafkaConsumer

func NewKafkaConsumer(kafkaConfig helpers.KafkaConfig, logPersister LogPersister) *KafkaConsumer

func (*KafkaConsumer) ConsumeLogKafka

func (c *KafkaConsumer) ConsumeLogKafka(ctx context.Context)

type KafkaProducer

type KafkaProducer struct {
	KafkaConfig helpers.KafkaConfig
	Writer      kafka.Writer
	AckChan     map[string]chan bool
	AckMutex    sync.Mutex
}

func NewKafkaProducer

func NewKafkaProducer(kafkaConfig helpers.KafkaConfig) *KafkaProducer

func (*KafkaProducer) Acknowledge

func (p *KafkaProducer) Acknowledge(logID string)

func (*KafkaProducer) CreateKafkaTopic

func (p *KafkaProducer) CreateKafkaTopic() error

func (*KafkaProducer) ProduceLogKafka

func (p *KafkaProducer) ProduceLogKafka(c echo.Context, logData types.LogData) error

func (*KafkaProducer) WaitForAck

func (p *KafkaProducer) WaitForAck(logID string, timeout time.Duration) (bool, error)

type LogPersister

type LogPersister interface {
	PersistLog(context.Context, types.LogData) error
}