kafka

Documentation

Index

• type KafkaInput
  • func NewKafkaInput() *KafkaInput
  • func (in *KafkaInput) Ack(msg *core.Message, err error)
  • func (in *KafkaInput) Configure(config core.StringMap) (err error)
  • func (in *KafkaInput) GetState() ([]byte, bool)
  • func (in *KafkaInput) SetState(state []byte) error
  • func (in *KafkaInput) Start() (err error)
  • func (in *KafkaInput) Stop()
• type KafkaInputConfig
• type KafkaPosition
• type MessageConsumer
  • func (mc *MessageConsumer) Cleanup(sarama.ConsumerGroupSession) error
  • func (mc *MessageConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) (err error)
  • func (mc *MessageConsumer) Setup(sarama.ConsumerGroupSession) error

Types

type KafkaInput

type KafkaInput struct {
	*core.BaseInput
	// contains filtered or unexported fields
}

func NewKafkaInput

func NewKafkaInput() *KafkaInput

func (*KafkaInput) Ack

func (in *KafkaInput) Ack(msg *core.Message, err error)

func (*KafkaInput) Configure

func (in *KafkaInput) Configure(config core.StringMap) (err error)

func (*KafkaInput) GetState

func (in *KafkaInput) GetState() ([]byte, bool)

func (*KafkaInput) SetState

func (in *KafkaInput) SetState(state []byte) error

func (*KafkaInput) Start

func (in *KafkaInput) Start() (err error)

func (*KafkaInput) Stop

func (in *KafkaInput) Stop()

type KafkaInputConfig

type KafkaInputConfig struct {
	ServerAddresses []string
	Topics          []string
	ConsumerGroup   string
	FromBeginning   bool
}

type KafkaPosition

type KafkaPosition struct {
	Topic     string
	Partition int32
	Offset    int64
	Timestamp int64
}

type MessageConsumer

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

func (*MessageConsumer) Cleanup

func (mc *MessageConsumer) Cleanup(sarama.ConsumerGroupSession) error

func (*MessageConsumer) ConsumeClaim

func (mc *MessageConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) (err error)

func (*MessageConsumer) Setup

func (mc *MessageConsumer) Setup(sarama.ConsumerGroupSession) error