kafkaingest

Documentation

Index

• func KafkaProducerGroup(ctx context.Context, producer *kafka.Producer, logger *slog.Logger) (execute func() error, interrupt func(error))
• type Collector
  • func (s Collector) Close()
  • func (s Collector) Ingest(_ context.Context, namespace string, ev event.Event) error
• type NamespaceHandler
  • func (h NamespaceHandler) CreateNamespace(ctx context.Context, namespace string) error

Functions

func KafkaProducerGroup

func KafkaProducerGroup(ctx context.Context, producer *kafka.Producer, logger *slog.Logger) (execute func() error, interrupt func(error))

Types

type Collector

type Collector struct {
	Producer   *kafka.Producer
	Serializer serializer.Serializer

	// NamespacedTopicTemplate needs to contain at least one string parameter passed to fmt.Sprintf.
	// For example: "om_%s_events"
	NamespacedTopicTemplate string
}

Collector is a receiver of events that handles sending those events to a downstream Kafka broker.

func (Collector) Close

func (s Collector) Close()

Close closes the underlying producer.

func (Collector) Ingest

func (s Collector) Ingest(_ context.Context, namespace string, ev event.Event) error

Ingest produces an event to a Kafka topic.

type NamespaceHandler

type NamespaceHandler struct {
	AdminClient *kafka.AdminClient

	// NamespacedTopicTemplate needs to contain at least one string parameter passed to fmt.Sprintf.
	// For example: "om_%s_events"
	NamespacedTopicTemplate string

	Partitions int

	Logger *slog.Logger
}

NamespaceHandler is a namespace handler for Kafka ingest topics.

func (NamespaceHandler) CreateNamespace

func (h NamespaceHandler) CreateNamespace(ctx context.Context, namespace string) error

CreateNamespace implements the namespace handler interface.