input

Oct 29, 2020

Changes in this version

+ const TypePulsar

Jul 26, 2022

Changes in this version

+ const CommitRetries

+ const Krb5KeytabAuth

+ const RetryBackoff

+ const TypeKafkaFranz

+ const TypeKafkaGo

+ const TypeKafkaSarama

+ var SHA256 scram.HashGeneratorFcn = func() hash.Hash

+ var SHA512 scram.HashGeneratorFcn = func() hash.Hash

+ func GetFranzConfig(kfkCfg *config.KafkaConfig) (opts []kgo.Opt, err error)

+ func GetSaramaConfig(kfkCfg *config.KafkaConfig) (sarCfg *sarama.Config, err error)

+ type Inputer interface

+ CommitMessages func(message *model.InputMessage) error

+ Init func(cfg *config.Config, taskCfg *config.TaskConfig, ...) error

+ Run func()

+ Stop func() error

+ func NewInputer(typ string) Inputer

+ type KafkaFranz struct

+ func NewKafkaFranz() *KafkaFranz

+ func (k *KafkaFranz) CommitMessages(msg *model.InputMessage) error

+ func (k *KafkaFranz) Description() string

+ func (k *KafkaFranz) Init(cfg *config.Config, taskCfg *config.TaskConfig, ...) (err error)

+ func (k *KafkaFranz) Run()

+ func (k *KafkaFranz) Stop() error

+ type KafkaGo struct

+ func NewKafkaGo() *KafkaGo

+ func (k *KafkaGo) CommitMessages(msg *model.InputMessage) (err error)

+ func (k *KafkaGo) Description() string

+ func (k *KafkaGo) Init(cfg *config.Config, taskCfg *config.TaskConfig, ...) (err error)

+ func (k *KafkaGo) Run()

+ func (k *KafkaGo) Stop() error

+ type KafkaGoLogger struct

+ func (kgl *KafkaGoLogger) Printf(template string, args ...interface{})

+ type KafkaSarama struct

+ func NewKafkaSarama() *KafkaSarama

+ func (k *KafkaSarama) CommitMessages(msg *model.InputMessage) error

+ func (k *KafkaSarama) Description() string

+ func (k *KafkaSarama) Init(cfg *config.Config, taskCfg *config.TaskConfig, ...) (err error)

+ func (k *KafkaSarama) Run()

+ func (k *KafkaSarama) Stop() error

+ type MyConsumerGroupHandler struct

+ func (h MyConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error

+ func (h MyConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ func (h MyConsumerGroupHandler) Setup(sess sarama.ConsumerGroupSession) error

+ type XDGSCRAMClient struct

+ func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

+ func (x *XDGSCRAMClient) Done() bool

+ func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

Mar 26, 2020

Changes in this version

+ func ConsumerHealthCheck(consumerName string) healthcheck.Check

+ type Consumer struct

+ Name string

+ func (consumer *Consumer) Cleanup(session sarama.ConsumerGroupSession) error

+ func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ func (consumer *Consumer) Setup(session sarama.ConsumerGroupSession) error

+ type ConsumerError struct

+ Error error

+ UnixTime int64

+ type Kafka struct

+ Brokers string

+ ConsumerGroup string

+ Earliest bool

+ Name string

+ Sasl struct{ ... }

+ Topic string

+ Version string

+ func NewKafka() *Kafka

+ func (k *Kafka) Description() string

+ func (k *Kafka) GetName() string

+ func (k *Kafka) Init() error

+ func (k *Kafka) Msgs() chan []byte

+ func (k *Kafka) Start() error

+ func (k *Kafka) Stop() error