kafka

Documentation

Overview

Kafka package used by Pachamama Group, uses Segmentio Kafka.

Index

• type KafkaConsumer
  • func NewConsumer(broker, port, topic, groupID string, maxB int, logger *logger.AggregatedLogger, ...) *KafkaConsumer
  • func (k *KafkaConsumer) Close() error
  • func (k *KafkaConsumer) Read(p []byte) (n int, err error)
• type KafkaProducer
  • func NewProducer(broker, port, topic string, logger *logger.AggregatedLogger, ...) *KafkaProducer
  • func (k *KafkaProducer) Write(p []byte) (n int, err error)
• type Message

Types

type KafkaConsumer

type KafkaConsumer struct {
	Timeout time.Duration
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(broker, port, topic, groupID string, maxB int, logger *logger.AggregatedLogger, timeout time.Duration) *KafkaConsumer

func (*KafkaConsumer) Close

func (k *KafkaConsumer) Close() error

func (*KafkaConsumer) Read

func (k *KafkaConsumer) Read(p []byte) (n int, err error)

Read will output a JSON to P in the form of Message Struct

type KafkaProducer

type KafkaProducer struct {
	Timeout time.Duration
	// contains filtered or unexported fields
}

func NewProducer

func NewProducer(broker, port, topic string, logger *logger.AggregatedLogger, timeout time.Duration) *KafkaProducer

Context should have a timeout which will signal the time to read or write a message

func (*KafkaProducer) Write

func (k *KafkaProducer) Write(p []byte) (n int, err error)

Write expects p[]byte to be in the form of a JSON with a Key and Value from Message Struct.

type Message

type Message struct {
	Topic     string    `json:"topic"`
	Key       string    `json:"key"`
	Value     string    `json:"value"`
	Timestamp time.Time `json:"timestamp"`
}