kafka

README

go-utils Kafka

The go-utils Kafka package provides handy methods to produce and read messages in Kafka topics.

Installation

To use the log package you must import the package.

import "github.com/eliona-smart-building-assistant/go-utils/kafka"

Optionally you can define an environment variable named BROKERS which sets the Kafka bootstrap servers.

export BROKERS=10.10.100.1:29092,192.168.178.1:9092

Usage

After installation, you can produce messages in Kafka topics.

import "github.com/eliona-smart-building-assistant/go-utils/kafka"

For example, you can push a temperature object to the climate topic. You have to design a new Producer and produce the temperature to the topic.

type Temperature struct {
    Value int
    Unit  string
}

producer := kafka.NewProducer()
defer producer.Close()
temperature := Temperature{Value: 24, Unit: "Celsius"}
_ = kafka.Produce(producer, "climate", temperature)

To read messages from a topic have to define a new consumer and subscribe a topic. After this, you can read the temperatures through a channel.

consumer := kafka.NewConsumer()
defer consumer.Close()
kafka.Subscribe(consumer, "climate")
temperatures := make(chan Temperature)
go kafka.Read(consumer, temperatures)

for temperature := range temperatures {
    fmt.Printf("Temperature is: %d %s", temperature.Value, temperature.Unit)
}

Documentation

Index

• func Brokers() string
• func NewConsumer() *kafka.Consumer
• func NewProducer() *kafka.Producer
• func Produce(producer *kafka.Producer, topic string, value any) error
• func Read[T any](consumer *kafka.Consumer, values chan T)
• func Subscribe(consumer *kafka.Consumer, topic string)

Functions

func Brokers

func Brokers() string

Brokers reads the kafka brokers from environment variable BROKERS

func NewConsumer

func NewConsumer() *kafka.Consumer

func NewProducer

func NewProducer() *kafka.Producer

func Produce

func Produce(producer *kafka.Producer, topic string, value any) error

Produce sends a typed message to a specific kafka topic

func Read

func Read[T any](consumer *kafka.Consumer, values chan T)

func Subscribe

func Subscribe(consumer *kafka.Consumer, topic string)