kafka

Documentation

Index

• type Consumer
  • func NewConsumer(cc ConsumerConfig) *Consumer
  • func (c *Consumer) Close()
  • func (c *Consumer) Consume(ctx context.Context)
  • func (c *Consumer) MsgChan() <-chan *pb.LeadpipeEvent
• type ConsumerConfig
• type Message
• type Producer
  • func NewProducer(broker string, topic string) *Producer
  • func (p *Producer) Close()
  • func (p *Producer) Send(ctx context.Context, evt *pb.LeadpipeEvent)

Types

type Consumer

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

Consumer coordinates the consuming of the queue

func NewConsumer

func NewConsumer(cc ConsumerConfig) *Consumer

NewConsumer creates a new consumer

func (*Consumer) Close

func (c *Consumer) Close()

Close closes the reader

func (*Consumer) Consume

func (c *Consumer) Consume(ctx context.Context)

Consume Mass Quantities

func (*Consumer) MsgChan

func (c *Consumer) MsgChan() <-chan *pb.LeadpipeEvent

MsgChan returns a message channel for consuming

type ConsumerConfig

type ConsumerConfig struct {
	ConsumerGroupID string
	Brokers         []string
	Topic           string
	MessageChan     chan *pb.LeadpipeEvent
}

ConsumerConfig configures a Consumer

type Message

type Message struct {
	VisitorID string
	Payload   string
}

Message models a key/value message

type Producer

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

Producer is a thin wrapper around a Sarama producer. It hides the Sarama API

func NewProducer

func NewProducer(broker string, topic string) *Producer

NewProducer creates a new producer given a list of broker addresses

func (*Producer) Close

func (p *Producer) Close()

Close closes the writer

func (*Producer) Send

func (p *Producer) Send(ctx context.Context, evt *pb.LeadpipeEvent)

Send wraps the sarama producer SendMessage