kafka

Documentation

Overview

Package kafka allows to process messages from a kafka topic ensuring at-least-once semantics.

Index

• type AloProcessor
  • func NewAloProcessor(config map[string]any) (AloProcessor, error)
  • func (proc AloProcessor) Close() error
  • func (proc AloProcessor) Process(ctx context.Context, entity string, h stream.MsgHandler) error

Types

type AloProcessor

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

An AloProcessor allows to process messages from a kafka topic ensuring at-least-once semantics.

func NewAloProcessor

func NewAloProcessor(config map[string]any) (AloProcessor, error)

NewAloProcessor returns an AloProcessor with the provided kafka configuration properties.

func (AloProcessor) Close

func (proc AloProcessor) Close() error

Close closes the underlaying kafka consumer.

func (AloProcessor) Process

func (proc AloProcessor) Process(ctx context.Context, entity string, h stream.MsgHandler) error

Process processes the messages received in the topic called entity by calling h. This method blocks the calling goroutine until the specified context is cancelled or an error occurs. It replaces the current kafka subscription, so it should not be called concurrently.