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)

Initialize The Producer

func (*Producer) Close

func (p *Producer) Close()

Close The Producer (Stop Processing)

func (*Producer) ObserveMetrics

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

Async Process For Observing Kafka Metrics

func (*Producer) ProduceKafkaMessage

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

Produce A KafkaMessage From The Specified CloudEvent To The Specified Topic And Wait For The Delivery Report

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.