producer

Documentation

Index

• type Batch
  • func (b *Batch) AddEvents(ctx *replication.ListenerContext, messages []gokafka.Message, ...)
  • func (b *Batch) Close()
  • func (b *Batch) FlushMessages()
  • func (b *Batch) StartBatchTicker()
• type Metric
  • func NewMetric(pqCDC cdc.Connector, slotName string) Metric
• type Producer
  • func NewProducer(kafkaClient kafka.Client, config *config.Connector, ...) (Producer, error)
  • func (p *Producer) Close() error
  • func (p *Producer) GetMetric() Metric
  • func (p *Producer) Produce(ctx *replication.ListenerContext, eventTime time.Time, ...)
  • func (p *Producer) StartBatch()

Types

type Batch

type Batch struct {
	Writer *gokafka.Writer
	// contains filtered or unexported fields
}

func (*Batch) AddEvents

func (b *Batch) AddEvents(ctx *replication.ListenerContext, messages []gokafka.Message, eventTime time.Time, isLastChunk bool)

func (*Batch) Close

func (b *Batch) Close()

func (*Batch) FlushMessages

func (b *Batch) FlushMessages()

func (*Batch) StartBatchTicker

func (b *Batch) StartBatchTicker()

type Metric

type Metric interface {
	SetProcessLatency(latency int64)
	SetBulkRequestProcessLatency(latency int64)
	PrometheusCollectors() []prometheus.Collector
	IncrementSuccessOp(topicName string)
	IncrementErrOp(topicName string)
}

func NewMetric

func NewMetric(pqCDC cdc.Connector, slotName string) Metric

type Producer

type Producer struct {
	ProducerBatch *Batch
}

func NewProducer

func NewProducer(
	kafkaClient kafka.Client,
	config *config.Connector,
	responseHandler kafka.ResponseHandler,
	pqCDC cdc.Connector,
) (Producer, error)

func (*Producer) Close

func (p *Producer) Close() error

func (*Producer) GetMetric

func (p *Producer) GetMetric() Metric

func (*Producer) Produce

func (p *Producer) Produce(
	ctx *replication.ListenerContext,
	eventTime time.Time,
	messages []gokafka.Message,
	isLastChunk bool,
)

func (*Producer) StartBatch

func (p *Producer) StartBatch()