producer

Documentation

Index

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

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) ConfigChanged

func (p *Producer) ConfigChanged(configMap *v1.ConfigMap) *Producer

ConfigChanged is called by the configMapObserver handler function in main() so that settings specific to the producer may be extracted and the producer restarted if necessary. The new configmap could technically have changes to the eventing-kafka section as well as the sarama section, but none of those matter to a currently-running Producer, so those are ignored here (which avoids the necessity of calling env.GetEnvironment). If those settings are needed in the future, the environment will also need to be re-parsed here. If there aren't any producer-specific differences between the current config and the new one, then just log that and move on; do not restart the Producer unnecessarily.

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