kafka

Documentation

Index

• func AddService(config service.Config)
• func BatchMessages(ctx context.Context, values <-chan kafka.Message, maxItems int, ...) chan []kafka.Message
• func GetWriterTopic() string
• func Handle(done, sendCtx context.Context, cancelSend context.CancelFunc, ...) <-chan kafka.Message
• func MessageAsString(message kafka.Message) string
• func MustProduce() bool
• func NewConsumer(done, sendCtx context.Context, kafkaURL string, wg *sync.WaitGroup) <-chan kafka.Message
• func NewProducer(done context.Context, cancelSend context.CancelFunc, kafkaURL string, ...)
• func Start(done context.Context)
• type SimpleMessage
  • func MessageSimplified(message kafka.Message) SimpleMessage
  • func MessagesSimplified(messages []kafka.Message) []SimpleMessage

Functions

func AddService

func AddService(config service.Config)

func BatchMessages

func BatchMessages(ctx context.Context, values <-chan kafka.Message, maxItems int, maxTimeout time.Duration) chan []kafka.Message

func GetWriterTopic

func GetWriterTopic() string

func Handle

func Handle(done, sendCtx context.Context, cancelSend context.CancelFunc, messages <-chan kafka.Message, wg *sync.WaitGroup) <-chan kafka.Message

func MessageAsString

func MessageAsString(message kafka.Message) string

func MustProduce

func MustProduce() bool

func NewConsumer

func NewConsumer(done, sendCtx context.Context, kafkaURL string, wg *sync.WaitGroup) <-chan kafka.Message

func NewProducer

func NewProducer(done context.Context, cancelSend context.CancelFunc, kafkaURL string, in <-chan kafka.Message, wg *sync.WaitGroup)

func Start

func Start(done context.Context)

Types

type SimpleMessage

type SimpleMessage struct {
	Topic         string
	Partition     int
	Offset        int64
	HighWaterMark int64
	Headers       []kafka.Header
	Time          time.Time
	Key           string
	Value         string
}

func MessageSimplified

func MessageSimplified(message kafka.Message) SimpleMessage

func MessagesSimplified

func MessagesSimplified(messages []kafka.Message) []SimpleMessage