samsa

README

Samsa

This is a Golang based library that enables you to write to Kafka service from your application effortlessly. It offers a simplified interface to create logs and write them into a Kafka topic.

Installation

go get github.com/faridgh1991/samsa

Usage

Using Samsa with a logger

package main

import (
	"context"
	"log/slog"
	"os"
	"time"

	kw "github.com/faridgh1991/samsa"
)

// main is the entry point of the program.
func main() {
	kafkaWriter, err := kw.NewWithContext(
		context.Background(),
		kw.Config{
			Endpoints:  []string{"localhost:9093"},
			Topic:      "testing_topic",
			BufferSize: 100,
		},
	)
	if err != nil {
		panic(err)
	}

	logger := slog.New(
		slog.NewJSONHandler(
			kafkaWriter,
			&slog.HandlerOptions{
				Level: slog.LevelInfo,
			},
		),
	)

	logger.LogAttrs(
		context.Background(),
		slog.LevelInfo,
		"hello, world!",
		slog.String("user", os.Getenv("USER")),
	)

	logger.Log(
		context.Background(),
		slog.LevelWarn,
		"finished",
		"now", time.Now(),
	)

	time.Sleep(time.Second)
}

Call Write directly

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/faridgh1991/samsa"
)

// main is the entry point of the program.
func main() {
	kafkaWriter, err := samsa.NewWithContext(
		context.Background(),
		samsa.Config{
			Endpoints:  []string{"localhost:9093"},
			Topic:      "testing_topic",
			BufferSize: 100,
		},
	)

	if err != nil {
		panic(err)
	}

	n, err := kafkaWriter.Write([]byte("Hello world!"))
	if err != nil {
		fmt.Println(n, err)
		return
	}
	fmt.Println(n)

	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*300)
	defer cancel()
	n, err = kafkaWriter.WriteWithContext(ctx, []byte("finish"))
	if err != nil {
		fmt.Println(n, err)
		return
	}
	fmt.Println(n)

	time.Sleep(time.Second)
}

Documentation

Index

• type Config
• type KafkaWriter
  • func New(conf Config) (*KafkaWriter, error)
  • func NewWithContext(ctx context.Context, conf Config) (writer *KafkaWriter, err error)
  • func (h *KafkaWriter) Write(buf []byte) (n int, err error)
  • func (h *KafkaWriter) WriteWithContext(ctx context.Context, buf []byte) (n int, err error)

Types

type Config

type Config struct {
	BufferSize int

	Endpoints []string
	Topic     string
	Producer  sarama.AsyncProducer
}

Config represents a configuration for a KafkaWriter

type KafkaWriter

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

KafkaWriter represents a writer that sends messages to Kafka using an async producer

Fields: - produce: a channel used to send messages to the Kafka producer - conf: the configuration for the Kafka writer - debug: a flag indicating whether debug mode is enabled - wg: a WaitGroup used for synchronization - once: a sync.Once used for one-time initialization

func New

func New(conf Config) (*KafkaWriter, error)

New creates a new KafkaWriter with the provided configuration. It initializes a new Kafka configuration and sets the producer properties. If no producer is defined in the configuration, it creates a new asynchronous producer using the given endpoints and Kafka configuration. If an error occurs during producer creation, it returns an error. The KafkaWriter struct is then initialized with the configuration and a buffered channel for producing messages. It launches a goroutine that listens for producer errors and incoming messages to be produced. If an error occurs during message production, it prints the error message and the failed message to stderr. If the channel for producing messages is closed, the goroutine will close the producer and finish execution. Finally, it returns a pointer to the KafkaWriter and a nil error.

func NewWithContext

func NewWithContext(ctx context.Context, conf Config) (writer *KafkaWriter, err error)

NewWithContext creates a new KafkaWriter with a context and configuration. It launches a goroutine to asynchronously create a KafkaWriter using the provided configuration. The function waits for the creation to complete or for the context to be cancelled. If the context is cancelled before the creation is completed, it returns an error indicating a connect timeout. If the creation is completed successfully, it returns the created KafkaWriter.

func (*KafkaWriter) Write

func (h *KafkaWriter) Write(buf []byte) (n int, err error)

Write writes a byte buffer to the KafkaWriter's producer input channel. It appends a copy of the buffer to the producer input channel. If the input channel buffer is full, it returns an error indicating a buffer overflow. The error message will include the contents of the buffer that was dropped. This function returns the number of bytes written and a nil error on success.

func (*KafkaWriter) WriteWithContext

func (h *KafkaWriter) WriteWithContext(ctx context.Context, buf []byte) (n int, err error)

WriteWithContext writes a byte buffer to the KafkaWriter's producer input channel, but unlike the Write method it respects the context timeout It appends a copy of the buffer to the producer input channel. If the context expires before the method can write to the channel, it returns an error indicating a deadline exceeded. If the input channel buffer is full, it returns an error indicating a buffer overflow. The error message will include the contents of the buffer that was dropped. This function returns the number of bytes written and a nil error on success.