kafka

Documentation

Index

• func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ...) adapter.Adapter
• func NewEnvConfig() adapter.EnvConfigAccessor
• type Adapter
  • func (a *Adapter) Handle(ctx context.Context, msg *sarama.ConsumerMessage) (bool, error)
  • func (a *Adapter) Start(stopCh <-chan struct{}) error
• type AdapterNet
• type AdapterSASL
• type AdapterTLS

Functions

func NewAdapter

func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ceClient client.Client, reporter source.StatsReporter) adapter.Adapter

func NewEnvConfig

func NewEnvConfig() adapter.EnvConfigAccessor

Types

type Adapter

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

func (*Adapter) Handle

func (a *Adapter) Handle(ctx context.Context, msg *sarama.ConsumerMessage) (bool, error)

func (*Adapter) Start

func (a *Adapter) Start(stopCh <-chan struct{}) error

type AdapterNet

type AdapterNet struct {
	SASL AdapterSASL
	TLS  AdapterTLS
}

type AdapterSASL

type AdapterSASL struct {
	Enable   bool   `envconfig:"KAFKA_NET_SASL_ENABLE" required:"false"`
	User     string `envconfig:"KAFKA_NET_SASL_USER" required:"false"`
	Password string `envconfig:"KAFKA_NET_SASL_PASSWORD" required:"false"`
}

type AdapterTLS

type AdapterTLS struct {
	Enable bool   `envconfig:"KAFKA_NET_TLS_ENABLE" required:"false"`
	Cert   string `envconfig:"KAFKA_NET_TLS_CERT" required:"false"`
	Key    string `envconfig:"KAFKA_NET_TLS_KEY" required:"false"`
	CACert string `envconfig:"KAFKA_NET_TLS_CA_CERT" required:"false"`
}