kafka

Documentation

Index

• Variables
• func Consume(ctx context.Context, topicNames []Topic, batchHandler ConsumeBatchHandler) error
• func EnvAddrsVersion() (addrs []string, version sarama.KafkaVersion)
• func Produce(topic Topic, key, value sarama.Encoder) error
• func Unuse() (err error)
• func UseConsumer(addrs []string, version sarama.KafkaVersion, groupId string, ...) (err error)
• func UseProducer(addrs []string, version sarama.KafkaVersion) (err error)
• func Version(v string) (sarama.KafkaVersion, error)
• type ConsumeBatchHandler
• type Topic
  • func TopicName(topic string) (Topic, error)
  • func (t Topic) Equal(s string) bool
  • func (t Topic) String() string

Variables

var (
	TopicMembership = Topic{/* contains filtered or unexported fields */}
	TopicLiquidity  = Topic{/* contains filtered or unexported fields */}
	TopicOperation  = Topic{/* contains filtered or unexported fields */}
)

Functions

func Consume

func Consume(ctx context.Context, topicNames []Topic, batchHandler ConsumeBatchHandler) error

func EnvAddrsVersion

func EnvAddrsVersion() (addrs []string, version sarama.KafkaVersion)

func Produce

func Produce(topic Topic, key, value sarama.Encoder) error

func Unuse

func Unuse() (err error)

func UseConsumer

func UseConsumer(addrs []string, version sarama.KafkaVersion, groupId string, maxInterval time.Duration, batchSize int) (err error)

func UseProducer

func UseProducer(addrs []string, version sarama.KafkaVersion) (err error)

func Version

func Version(v string) (sarama.KafkaVersion, error)

Types

type ConsumeBatchHandler

type ConsumeBatchHandler func(msgs []*sarama.ConsumerMessage) (markFlags []bool, errs []error)

type Topic

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

func TopicName

func TopicName(topic string) (Topic, error)

func (Topic) Equal

func (t Topic) Equal(s string) bool

func (Topic) String

func (t Topic) String() string