events

package module

Go to main page

Versions in this module

v0

v0.9.0

Jan 27, 2026

v0.8.1

Jan 23, 2026

v0.8.0

Jan 21, 2026

Changes in this version

+ const ErrTooBig

type CanConsumeBroadcast

+ TracerProvider func(context.Context) eventmodels.Tracer

type Library

+ func (lib *Library[ID, TX, DB]) SetPrefix(prefix string)

+ func (lib *Library[ID, TX, DB]) SetSizeCapLowerLimit(sizeCapDefaultAssumed int64)

+ func (lib *Library[ID, TX, DB]) SetTracerConfig(config eventmodels.TracerConfig)

+ func (lib *Library[ID, TX, DB]) SkipNotifierSupport()

+ func (lib *Library[ID, TX, DB]) TracerConfig() eventmodels.TracerConfig

+ func (lib *Library[ID, TX, DB]) TracerProvider(ctx context.Context) eventmodels.Tracer

type LibraryNoDB

+ func (lib *LibraryNoDB) Shutdown(ctx context.Context)

v0.7.0

Nov 22, 2025

v0.6.0

Sep 16, 2025

v0.5.1

Jun 30, 2025

v0.5.0

Jun 27, 2025

Changes in this version

type Library

+ func (lib *Library[ID, TX, DB]) DoNotLockBroadcastConsumerNumbers()

+ func (lib *Library[ID, TX, DB]) HasDB() bool

type LibraryNoDB

+ func (lib *LibraryNoDB) RecordError(category string, err error) error

+ func (lib *LibraryNoDB) RecordErrorNoWait(category string, err error) error

v0.4.0

May 7, 2025

v0.3.0

May 7, 2025

v0.2.0

May 5, 2025

Changes in this version

+ const UnregisteredTopicError

+ var HandlerSuccessCounts = mustRegister(prometheus.NewCounterVec(prometheus.CounterOpts{ ... }, []string{ ... }))

type Library

+ func (lib *Library[ID, TX, DB]) ValidateTopics(ctx context.Context, topics []string) error

v0.1.0

Apr 17, 2025

v0.0.2

Apr 15, 2025

v0.0.1

Apr 14, 2025

Changes in this version

+ var AckLatency = mustRegister(prometheus.NewHistogramVec(prometheus.HistogramOpts{ ... }, []string{ ... }))

+ var ConsumeCounts = mustRegister(prometheus.NewCounterVec(prometheus.CounterOpts{ ... }, []string{ ... }))

+ var ConsumersWaitingForQueueConcurrencyDemand = mustRegister(prometheus.NewGaugeVec(prometheus.GaugeOpts{ ... }, []string{ ... }))

+ var ConsumersWaitingForQueueConcurrencyLimit = mustRegister(prometheus.NewGaugeVec(prometheus.GaugeOpts{ ... }, []string{ ... }))

+ var DeadLetterConsumeCounts = mustRegister(prometheus.NewCounterVec(prometheus.CounterOpts{ ... }, []string{ ... }))

+ var DeadLetterProduceCounts = mustRegister(prometheus.NewCounterVec(prometheus.CounterOpts{ ... }, []string{ ... }))

+ var ErrTopicCreationTimeout errors.String = "event library topic creation deadline exceeded"

+ var ErrorCounts = mustRegister(prometheus.NewCounterVec(prometheus.CounterOpts{ ... }, []string{ ... }))

+ var HandlerBatchConcurrency = mustRegister(prometheus.NewGaugeVec(prometheus.GaugeOpts{ ... }, []string{ ... }))

+ var HandlerBatchQueued = mustRegister(prometheus.NewGaugeVec(prometheus.GaugeOpts{ ... }, []string{ ... }))

+ var HandlerCounts = mustRegister(prometheus.NewCounterVec(prometheus.CounterOpts{ ... }, []string{ ... }))

+ var HandlerErrorCounts = mustRegister(prometheus.NewCounterVec(prometheus.CounterOpts{ ... }, []string{ ... }))

+ var HandlerLatency = mustRegister(prometheus.NewHistogramVec(prometheus.HistogramOpts{ ... }, []string{ ... }))

+ var HandlerPanicCounts = mustRegister(prometheus.NewCounterVec(prometheus.CounterOpts{ ... }, []string{ ... }))

+ var HandlerWaitingForActiveConcurrencyDemand = mustRegister(prometheus.NewGaugeVec(prometheus.GaugeOpts{ ... }, []string{ ... }))

+ var HandlerWaitingForActiveConcurrencyLimit = mustRegister(prometheus.NewGaugeVec(prometheus.GaugeOpts{ ... }, []string{ ... }))

+ var HandlerWaitingForQueueConcurrencyDemand = mustRegister(prometheus.NewGaugeVec(prometheus.GaugeOpts{ ... }, []string{ ... }))

+ var HandlerWaitingForQueueConcurrencyLimit = mustRegister(prometheus.NewGaugeVec(prometheus.GaugeOpts{ ... }, []string{ ... }))

+ var LegalTopicNames = regexp.MustCompile(fmt.Sprintf(`^[-._a-zA-Z0-9]{1,%d}$`, maxTopicNameLength))

+ var LongestHandlerLatency = mustRegister(prometheus.NewGaugeVec(prometheus.GaugeOpts{ ... }, []string{ ... }))

+ var ProduceFromTxSplit = mustRegister(prometheus.NewCounterVec(prometheus.CounterOpts{ ... }, []string{ ... }))

+ var ProduceTopicCounts = mustRegister(prometheus.NewCounterVec(prometheus.CounterOpts{ ... }, []string{ ... }))

+ var TransmissionLatency = mustRegister(prometheus.NewHistogramVec(prometheus.HistogramOpts{ ... }, []string{ ... }))

+ func Async(limit int) func(*notifierConfig)

+ func DeadLetterTopic(topic string, consumerGroup ConsumerGroupName) string

+ func ObserveProduceLatency(err error, startTime time.Time)

+ func SASLConfigFromString(s string) sasl.Mechanism

+ type CanConsumeBroadcast interface

+ ConsumeBroadcast func(handlerName string, handler eventmodels.HandlerInterface, opts ...HandlerOpt)

+ InstanceID func() int32

+ Tracer func() eventmodels.Tracer

+ type ConsumerGroupName interface

+ String func() string

+ func NewConsumerGroup(name string) ConsumerGroupName

+ type Filtered map[int32]*keyeddistributor.Distributor[eventmodels.Event[E], K]

+ func RegisterFiltered[E any, K comparable](handlerName string, topic eventmodels.BoundTopic[E], ...) Filtered[K, E]

+ func (f Filtered[K, E]) Subscribe(lib CanConsumeBroadcast, key K) *keyeddistributor.Reader[eventmodels.Event[E]]

+ type HandlerOpt func(*registeredHandler, *LibraryNoDB)

+ func IsDeadLetterHandler(isDeadLetter bool) HandlerOpt

+ func WithBaseTopic(topic string) HandlerOpt

+ func WithBatch(size int) HandlerOpt

+ func WithConcurrency(parallelism int) HandlerOpt

+ func WithQueueDepthLimit(n int) HandlerOpt

+ func WithRetrying(retry bool) HandlerOpt

+ func WithTimeout(d time.Duration) HandlerOpt

+ type HeartbeatEvent struct

+ type Library struct

+ func New[ID eventmodels.AbstractID[ID], TX eventmodels.AbstractTX, ...]() *Library[ID, TX, DB]

+ func (lib *Library[ID, TX, DB]) BroadcastConsumerLastLatency() (lastGap time.Duration, configuredHeartbeat time.Duration)

+ func (lib *Library[ID, TX, DB]) CatchUpProduce(ctx context.Context, sleepTime time.Duration, batchSize int) (chan struct{}, error)

+ func (lib *Library[ID, TX, DB]) Configure(conn DB, tracer eventmodels.Tracer, mustRegisterTopics bool, ...)

+ func (lib *Library[ID, TX, DB]) ConfigureBroadcastHeartbeat(dur time.Duration)

+ func (lib *Library[ID, TX, DB]) ConsumeBroadcast(handlerName string, handler eventmodels.HandlerInterface, opts ...HandlerOpt)

+ func (lib *Library[ID, TX, DB]) ConsumeExactlyOnce(consumerGroup ConsumerGroupName, onFailure eventmodels.OnFailure, ...)

+ func (lib *Library[ID, TX, DB]) ConsumeIdempotent(consumerGroup ConsumerGroupName, onFailure eventmodels.OnFailure, ...)

+ func (lib *Library[ID, TX, DB]) DB() eventmodels.AbstractDB[ID, TX]

+ func (lib *Library[ID, TX, DB]) GetBroadcastConsumerGroupName() string

+ func (lib *Library[ID, TX, DB]) InstanceID() int32

+ func (lib *Library[ID, TX, DB]) IsConfigured() bool

+ func (lib *Library[ID, TX, DB]) Produce(ctx context.Context, method eventmodels.ProduceMethod, ...) error

+ func (lib *Library[ID, TX, DB]) ProduceFromTable(ctx context.Context, eventIDs []ID) error

+ func (lib *Library[ID, TX, DB]) RecordError(category string, err error) error

+ func (lib *Library[ID, TX, DB]) RecordErrorNoWait(category string, err error) error

+ func (lib *Library[ID, TX, DB]) SetBroadcastConsumerBaseName(name string)

+ func (lib *Library[ID, TX, DB]) SetBroadcastConsumerMaxLocks(max uint32)

+ func (lib *Library[ID, TX, DB]) SetEnhanceDB(enhance bool)

+ func (lib *Library[ID, TX, DB]) SetLazyTxProduce(lazy bool)

+ func (lib *Library[ID, TX, DB]) StartConsuming(ctx context.Context) (started chan struct{}, stopped chan struct{}, err error)

+ func (lib *Library[ID, TX, DB]) StartConsumingOrPanic(ctx context.Context) (stopped chan struct{})

+ func (lib *Library[ID, TX, DB]) Tracer() eventmodels.Tracer

+ type LibraryNoDB struct

+ ProduceSyncCount atomic.Uint64

+ func (lib *LibraryNoDB) CreateTopics(ctx context.Context, why string, topics []string) error

+ func (lib *LibraryNoDB) SetTopicConfig(topicConfig kafka.TopicConfig)

+ type NotifierOpt func(*notifierConfig)

+ func WrapHandlerOpt(handlerOpts ...HandlerOpt) NotifierOpt

+ type SASLConfig struct

+ Mechanism sasl.Mechanism

+ func (sc *SASLConfig) UnmarshalText(text []byte) error

+ type SASLMethod string

+ const SASLNone

+ const SASLPlain

+ const SASLSHA256

+ const SASLSHA512

+ type TLSConfig struct

+ func (sc *TLSConfig) UnmarshalJSON(b []byte) error

+ type Unfiltered map[int32]*eventdistributor.Distributor[eventmodels.Event[E]]

+ func RegisterUnfiltered[E any](handlerName string, topic eventmodels.BoundTopic[E], opts ...NotifierOpt) Unfiltered[E]

+ func (u Unfiltered[E]) Subscribe(lib CanConsumeBroadcast) eventdistributor.Reader[eventmodels.Event[E]]

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL