kafka

Oct 13, 2022

Changes in this version

+ func NewKafka(tag string, tracing tracing.Tracing, lo logger.Logger, ...) database.Kafka

+ type Consumer struct

+ func NewConsumer(tracing tracing.Tracing, lg logger.Logger, config database.KafkaProviderConfig, ...) *Consumer

+ func (c *Consumer) Run()

+ type Kafka struct

+ func (c *Kafka) Consumer(options database.KafkaOptions, callback database.ConsumerCallback)

+ func (c *Kafka) Producer(isReady database.ProducerIsReady)

+ func (c *Kafka) Push(ctx context.Context, id string, options database.KafkaOptions, ...) error

+ type MsgSend struct

+ Data []byte

+ Headers map[string]interface{}

+ ID string

+ Name string

+ type Producer struct

+ func NewProducer(lg logger.Logger, config database.KafkaProviderConfig, store *Stores) *Producer

+ func (c *Producer) Run(isReady database.ProducerIsReady) (err error)

+ func (c *Producer) SendingData(id string, options database.KafkaOptions, body interface{}, ...)

+ type Stores struct

+ func NewStore() *Stores

+ func (c *Stores) Delete(id string)

+ func (c *Stores) Get(id string) (cb database.ConsumerCallback, ok bool)

+ func (c *Stores) LoadClient(id string) (client *Kafka, ok bool)

+ func (c *Stores) LoadClientByTag(ta string) (client *Kafka, ok bool)

+ func (c *Stores) Put(id string, cb database.ConsumerCallback)

+ func (c *Stores) StoreClient(client *Kafka)