kafka

Documentation

Overview

Package kafka implements the Apache Kafka connector.

Index

• type GroupHandler
  • func (handler *GroupHandler) Cleanup(sarama.ConsumerGroupSession) error
  • func (handler *GroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (handler *GroupHandler) Setup(sarama.ConsumerGroupSession) error
• type KafkaSink
  • func NewKafkaSink(addrs []string, config *sarama.Config, topic string) *KafkaSink
  • func (ks *KafkaSink) In() chan<- interface{}
• type KafkaSource
  • func NewKafkaSource(ctx context.Context, addrs []string, groupID string, config *sarama.Config, ...) *KafkaSource
  • func (ks *KafkaSource) Out() <-chan interface{}
  • func (ks *KafkaSource) Via(_flow streams.Flow) streams.Flow

Types

type GroupHandler

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

GroupHandler represents a Sarama consumer group handler

func (*GroupHandler) Cleanup

func (handler *GroupHandler) Cleanup(sarama.ConsumerGroupSession) error

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited

func (*GroupHandler) ConsumeClaim

func (handler *GroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().

func (*GroupHandler) Setup

func (handler *GroupHandler) Setup(sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim

type KafkaSink

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

KafkaSink connector

func NewKafkaSink

func NewKafkaSink(addrs []string, config *sarama.Config, topic string) *KafkaSink

NewKafkaSink returns a new KafkaSink instance

func (*KafkaSink) In

func (ks *KafkaSink) In() chan<- interface{}

In returns an input channel for receiving data

type KafkaSource

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

KafkaSource connector

func NewKafkaSource

func NewKafkaSource(ctx context.Context, addrs []string, groupID string,
	config *sarama.Config, topics ...string) *KafkaSource

NewKafkaSource returns a new KafkaSource instance

func (*KafkaSource) Out

func (ks *KafkaSource) Out() <-chan interface{}

Out returns an output channel for sending data

func (*KafkaSource) Via

func (ks *KafkaSource) Via(_flow streams.Flow) streams.Flow

Via streams data through the given flow