producer

Documentation

Index

• type KafkaSyncProducer
  • func NewKafkaSyncProducer(ctx context.Context, brokers string) (*KafkaSyncProducer, error)
  • func (ksProducer *KafkaSyncProducer) Close() error
  • func (ksProducer *KafkaSyncProducer) GetBuffer() chan []string
  • func (ksProducer *KafkaSyncProducer) ProduceKafkaMessage(topic, key, val string) error
  • func (ksProducer *KafkaSyncProducer) ProduceKafkaMessages(topics, keys, vals []string) (int, error)
  • func (ksProducer *KafkaSyncProducer) Producer(ctx context.Context)

Types

type KafkaSyncProducer

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

func NewKafkaSyncProducer

func NewKafkaSyncProducer(ctx context.Context, brokers string) (*KafkaSyncProducer, error)

func (*KafkaSyncProducer) Close

func (ksProducer *KafkaSyncProducer) Close() error

func (*KafkaSyncProducer) GetBuffer

func (ksProducer *KafkaSyncProducer) GetBuffer() chan []string

This buffer can be used for consumer to put retry/dead letter/result message.

func (*KafkaSyncProducer) ProduceKafkaMessage

func (ksProducer *KafkaSyncProducer) ProduceKafkaMessage(topic, key, val string) error

message protocol: [topic type][retry count][reqID][req]

func (*KafkaSyncProducer) ProduceKafkaMessages

func (ksProducer *KafkaSyncProducer) ProduceKafkaMessages(topics, keys, vals []string) (int, error)

func (*KafkaSyncProducer) Producer

func (ksProducer *KafkaSyncProducer) Producer(ctx context.Context)