input

Documentation

Index

• Constants
• type Inputer
  • func NewInputer(typ string) Inputer
• type KafkaGo
  • func NewKafkaGo() *KafkaGo
  • func (k *KafkaGo) CommitMessages(ctx context.Context, msg *model.InputMessage) (err error)
  • func (k *KafkaGo) Description() string
  • func (k *KafkaGo) Init(cfg *config.Config, taskName string, putFn func(msg model.InputMessage)) error
  • func (k *KafkaGo) Run(ctx context.Context)
  • func (k *KafkaGo) Stop() error
• type KafkaSarama
  • func NewKafkaSarama() *KafkaSarama
  • func (k *KafkaSarama) CommitMessages(ctx context.Context, msg *model.InputMessage) error
  • func (k *KafkaSarama) Description() string
  • func (k *KafkaSarama) Init(cfg *config.Config, taskName string, putFn func(msg model.InputMessage)) error
  • func (k *KafkaSarama) Run(ctx context.Context)
  • func (k *KafkaSarama) Stop() error
• type MyConsumerGroupHandler
  • 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

Constants

const (
	TypeKafkaGo     = "kafka-go"
	TypeKafkaSarama = "sarama"
	TypePulsar      = "pulsar"
)

Types

type Inputer

type Inputer interface {
	Init(cfg *config.Config, taskName string, putFn func(msg model.InputMessage)) error
	Run(ctx context.Context)
	Stop() error
	CommitMessages(ctx context.Context, message *model.InputMessage) error
}

func NewInputer

func NewInputer(typ string) Inputer

type KafkaGo

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

KafkaGo implements input.Inputer

func NewKafkaGo

func NewKafkaGo() *KafkaGo

NewKafkaGo get instance of kafka reader

func (*KafkaGo) CommitMessages

func (k *KafkaGo) CommitMessages(ctx context.Context, msg *model.InputMessage) (err error)

func (*KafkaGo) Description

func (k *KafkaGo) Description() string

Description of this kafka consumer, which topic it reads from

func (*KafkaGo) Init

func (k *KafkaGo) Init(cfg *config.Config, taskName string, putFn func(msg model.InputMessage)) error

Init Initialise the kafka instance with configuration

func (*KafkaGo) Run

func (k *KafkaGo) Run(ctx context.Context)

kafka main loop

func (*KafkaGo) Stop

func (k *KafkaGo) Stop() error

Stop kafka consumer and close all connections

type KafkaSarama

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

KafkaSarama implements input.Inputer

func NewKafkaSarama

func NewKafkaSarama() *KafkaSarama

NewKafkaSarama get instance of kafka reader

func (*KafkaSarama) CommitMessages

func (k *KafkaSarama) CommitMessages(ctx context.Context, msg *model.InputMessage) error

func (*KafkaSarama) Description

func (k *KafkaSarama) Description() string

Description of this kafka consumer, which topic it reads from

func (*KafkaSarama) Init

func (k *KafkaSarama) Init(cfg *config.Config, taskName string, putFn func(msg model.InputMessage)) error

Init Initialise the kafka instance with configuration

func (*KafkaSarama) Run

func (k *KafkaSarama) Run(ctx context.Context)

kafka main loop

func (*KafkaSarama) Stop

func (k *KafkaSarama) Stop() error

Stop kafka consumer and close all connections

type MyConsumerGroupHandler

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

func (MyConsumerGroupHandler) Cleanup

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

func (MyConsumerGroupHandler) ConsumeClaim

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

func (MyConsumerGroupHandler) Setup

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