base

Documentation

Index

• func Batch[T any](n int, d time.Duration, in <-chan T) <-chan []T
• func EventKey(source string, subject string) string
• type Kafka
  • func NewKafka(config *eventbusv1alpha1.KafkaBus, logger *zap.SugaredLogger) *Kafka
  • func (k *Kafka) Brokers() []string
  • func (k *Kafka) Config() (*sarama.Config, error)
• type KafkaConnection
  • func NewKafkaConnection(logger *zap.SugaredLogger) *KafkaConnection

Functions

func Batch

func Batch[T any](n int, d time.Duration, in <-chan T) <-chan []T

Batch returns a read only channel that receives values from the input channel batched together into a slice. A value is sent to the output channel when the slice reaches n elements, or d time has elapsed, whichever happens first. Ordering is maintained.

func EventKey

func EventKey(source string, subject string) string

Types

type Kafka

type Kafka struct {
	Logger *zap.SugaredLogger
	// contains filtered or unexported fields
}

func NewKafka

func NewKafka(config *eventbusv1alpha1.KafkaBus, logger *zap.SugaredLogger) *Kafka

func (*Kafka) Brokers

func (k *Kafka) Brokers() []string

func (*Kafka) Config

func (k *Kafka) Config() (*sarama.Config, error)

type KafkaConnection

type KafkaConnection struct {
	Logger *zap.SugaredLogger
}

func NewKafkaConnection

func NewKafkaConnection(logger *zap.SugaredLogger) *KafkaConnection