producer

Documentation

Index

• type Response
• type T
  • func Spawn(parentActDesc *actor.Descriptor, cfg *config.Proxy) (*T, error)
  • func (p *T) AsyncProduce(topic string, key, message sarama.Encoder, headers []sarama.RecordHeader) <-chan Response
  • func (p *T) Produce(topic string, key, message sarama.Encoder, headers []sarama.RecordHeader) (*sarama.ProducerMessage, error)
  • func (p *T) Stop()

Types

type Response

type Response struct {
	Msg *sarama.ProducerMessage
	Err error
}

type T

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

T builds on top of `sarama.AsyncProducer` to improve the shutdown handling. The problem it solves is that `sarama.AsyncProducer` drops all buffered messages as soon as it is ordered to shutdown. On the contrary, when `T` is ordered to stop it allows some time for the buffered messages to be committed to the Kafka cluster, and only when that time has elapsed it drops uncommitted messages.

TODO Consider implementing some sort of dead message processing.

func Spawn

func Spawn(parentActDesc *actor.Descriptor, cfg *config.Proxy) (*T, error)

Spawn creates a producer instance and starts its internal goroutines.

func (*T) AsyncProduce

func (p *T) AsyncProduce(topic string, key, message sarama.Encoder, headers []sarama.RecordHeader) <-chan Response

AsyncProduce is an asynchronously counterpart of the `Produce` function. Errors are silently ignored.

func (*T) Produce

func (p *T) Produce(topic string, key, message sarama.Encoder, headers []sarama.RecordHeader) (*sarama.ProducerMessage, error)

Produce submits a message to the specified `topic` of the Kafka cluster using `key` to identify a destination partition. The exact algorithm used to map keys to partitions is implementation specific but it is guaranteed that it returns consistent results. If `key` is `nil`, then the message is placed into a random partition.

Errors usually indicate a catastrophic failure of the Kafka cluster, or missing topic if there cluster is not configured to auto create topics.

func (*T) Stop

func (p *T) Stop()

Stop shuts down all producer goroutines and releases all resources.