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
  • func (h NamespaceHandler) DeleteNamespace(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.

func (NamespaceHandler) DeleteNamespace

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

DeleteNamespace implements the namespace handler interface.