producer

Documentation

Index

• type Config
• type KafkaProducer
  • func (kp *KafkaProducer) OnMessage(m *utils.Message, done utils.Done)
  • func (kp *KafkaProducer) Spawn(id int) utils.Composer
  • func (kp *KafkaProducer) Workers() int
• type RdKafkaFactory
  • func NewRdKafkaFactory(producer RdKafkaProducer) *RdKafkaFactory
  • func (pf *RdKafkaFactory) CreateProducer() RdKafkaProducer
• type RdKafkaProducer

Types

type Config

type Config struct {
	Workers    int
	Topic      string
	Attributes map[string]string
	Factory    *RdKafkaFactory
}

Config contains the configuration for a Counter.

type KafkaProducer

type KafkaProducer struct {
	Config

	KafkaProducer RdKafkaProducer
}

KafkaProducer send the message to a Kafka queue.

func (*KafkaProducer) OnMessage

func (kp *KafkaProducer) OnMessage(m *utils.Message, done utils.Done)

OnMessage is called when a new message is receive and send the message to a Kafka topic.

func (*KafkaProducer) Spawn

func (kp *KafkaProducer) Spawn(id int) utils.Composer

Spawn creates a new instance of a Counter worker.

func (*KafkaProducer) Workers

func (kp *KafkaProducer) Workers() int

Workers returns the number of workers.

type RdKafkaFactory

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

RdKafkaFactory is used to create multiple producers with the same attributes.

func NewRdKafkaFactory

func NewRdKafkaFactory(producer RdKafkaProducer) *RdKafkaFactory

NewRdKafkaFactory returns a new instance of a Producer Factory that always returns the same RdKafka Producer.

func (*RdKafkaFactory) CreateProducer

func (pf *RdKafkaFactory) CreateProducer() RdKafkaProducer

CreateProducer returns the internal RdKafkaProducer.

type RdKafkaProducer

type RdKafkaProducer interface {
	ProduceChannel() chan *rdkafka.Message
	Close()
}

RdKafkaProducer is an interface for rdkafka producer. Used for mocking purposes.