producer

Documentation

Index

• type Producer
  • func NewProducer(logger *zap.Logger, config *sarama.Config, brokers []string, ...) (*Producer, error)
  • func (p *Producer) Close()
  • func (p *Producer) ObserveMetrics(interval time.Duration)
  • func (p *Producer) ProduceKafkaMessage(ctx context.Context, channelReference eventingChannel.ChannelReference, ...) error
  • func (p *Producer) SecretChanged(ctx context.Context, secret *corev1.Secret) *Producer

Types

type Producer

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

Producer Struct

func NewProducer

func NewProducer(logger *zap.Logger,
	config *sarama.Config,
	brokers []string,
	statsReporter metrics.StatsReporter,
	healthServer *health.Server) (*Producer, error)

NewProducer returns a new Producer instance with specified configuration.

func (*Producer) Close

func (p *Producer) Close()

Close gracefully stops all Producer processing.

func (*Producer) ObserveMetrics

func (p *Producer) ObserveMetrics(interval time.Duration)

ObserveMetrics is an async process for observing Kafka metrics.

func (*Producer) ProduceKafkaMessage

func (p *Producer) ProduceKafkaMessage(ctx context.Context, channelReference eventingChannel.ChannelReference, message binding.Message, httpHeader http.Header, transformers ...binding.Transformer) error

ProduceKafkaMessage creates and sends a Sarama ProducerMessage to the specified Topic and waits for the delivery confirmation.

func (*Producer) SecretChanged

func (p *Producer) SecretChanged(ctx context.Context, secret *corev1.Secret) *Producer

SecretChanged is called by the secretObserver handler function in main() so that settings specific to the producer may be extracted and the producer restarted if necessary.