eventbus

package module

Go to main page

Versions in this module

v1

v1.4.0

Oct 25, 2025

v1.3.0

Oct 8, 2025

Changes in this version

+ var ErrNATSConnectionNotEstablished = errors.New("NATS connection is not established")

type EventBus

+ func NewNatsEventBus(config map[string]interface{}) (EventBus, error)

+ type NatsConfig struct

+ AllowReconnect bool

+ ConnectionName string

+ MaxPingsOut int

+ MaxReconnects int

+ Password string

+ PingInterval int

+ ReconnectWait int

+ SubscribeTimeout int

+ Token string

+ URL string

+ Username string

+ type NatsEventBus struct

+ func (n *NatsEventBus) Publish(ctx context.Context, event Event) error

+ func (n *NatsEventBus) Start(ctx context.Context) error

+ func (n *NatsEventBus) Stop(ctx context.Context) error

+ func (n *NatsEventBus) Subscribe(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (n *NatsEventBus) SubscribeAsync(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (n *NatsEventBus) SubscriberCount(topic string) int

+ func (n *NatsEventBus) Topics() []string

+ func (n *NatsEventBus) Unsubscribe(ctx context.Context, subscription Subscription) error

v1.2.0

Oct 2, 2025

v1.1.6

Oct 1, 2025

v1.1.5

Aug 30, 2025

v1.1.4

Aug 29, 2025

Changes in this version

+ type DatadogStatsdExporter struct

+ func NewDatadogStatsdExporter(eventBus *EventBusModule, prefix, addr string, interval time.Duration, ...) (*DatadogStatsdExporter, error)

+ func (e *DatadogStatsdExporter) Close() error

+ func (e *DatadogStatsdExporter) Run(ctx context.Context)

+ type DeliveryStats struct

+ Delivered uint64

+ Dropped uint64

type EngineRouter

+ func (r *EngineRouter) CollectPerEngineStats() map[string]DeliveryStats

+ func (r *EngineRouter) CollectStats() (delivered uint64, dropped uint64)

type EventBusConfig

+ DeliveryMode string

+ PublishBlockTimeout time.Duration

+ RotateSubscriberOrder bool

type EventBusModule

+ func (m *EventBusModule) PerEngineStats() map[string]DeliveryStats

+ func (m *EventBusModule) Stats() (delivered uint64, dropped uint64)

type MemoryEventBus

+ func (m *MemoryEventBus) Stats() (delivered uint64, dropped uint64)

+ type PrometheusCollector struct

+ func NewPrometheusCollector(eventBus *EventBusModule, namespace string) *PrometheusCollector

+ func (c *PrometheusCollector) Collect(ch chan<- prometheus.Metric)

+ func (c *PrometheusCollector) Describe(ch chan<- *prometheus.Desc)

v1.1.3

Aug 25, 2025

v1.1.2

Aug 22, 2025

v1.1.1

Aug 21, 2025

v1.1.0

Aug 20, 2025

Changes in this version

+ const EventTypeBusStarted

+ const EventTypeBusStopped

+ const EventTypeConfigLoaded

+ const EventTypeMessageFailed

+ const EventTypeMessagePublished

+ const EventTypeMessageReceived

+ const EventTypeSubscriptionCreated

+ const EventTypeSubscriptionRemoved

+ const EventTypeTopicCreated

+ const EventTypeTopicDeleted

+ var ErrNoSubjectForEventEmission = errors.New("no subject available for event emission")

type EngineRouter

+ func (r *EngineRouter) SetModuleReference(module *EventBusModule)

type EventBusModule

+ func (m *EventBusModule) EmitEvent(ctx context.Context, event cloudevents.Event) error

+ func (m *EventBusModule) GetRegisteredEventTypes() []string

+ func (m *EventBusModule) RegisterObservers(subject modular.Subject) error

type MemoryEventBus

+ func (m *MemoryEventBus) SetModule(module *EventBusModule)

v1.0.0

Aug 14, 2025

Changes in this version

+ var ErrDuplicateEngineName = errors.New("duplicate engine name")

+ var ErrEngineNotFound = errors.New("engine not found")

+ var ErrInvalidShardCount = errors.New("invalid shard count")

+ var ErrSubscriptionNotFound = errors.New("subscription not found in any engine")

+ var ErrUnknownEngineRef = errors.New("routing rule references unknown engine")

+ var ErrUnknownEngineType = errors.New("unknown engine type")

+ func GetRegisteredEngines() []string

+ func RegisterEngine(engineType string, factory EngineFactory)

+ type CustomMemoryConfig struct

+ DefaultEventBufferSize int

+ EnableMetrics bool

+ EventFilters []map[string]interface{}

+ MaxEventQueueSize int

+ MetricsInterval time.Duration

+ type CustomMemoryEventBus struct

+ func (c *CustomMemoryEventBus) GetMetrics() *EventMetrics

+ func (c *CustomMemoryEventBus) Publish(ctx context.Context, event Event) error

+ func (c *CustomMemoryEventBus) Start(ctx context.Context) error

+ func (c *CustomMemoryEventBus) Stop(ctx context.Context) error

+ func (c *CustomMemoryEventBus) Subscribe(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (c *CustomMemoryEventBus) SubscribeAsync(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (c *CustomMemoryEventBus) SubscriberCount(topic string) int

+ func (c *CustomMemoryEventBus) Topics() []string

+ func (c *CustomMemoryEventBus) Unsubscribe(ctx context.Context, subscription Subscription) error

+ type EngineConfig struct

+ Config map[string]interface{}

+ Name string

+ Type string

+ type EngineFactory func(config map[string]interface{}) (EventBus, error)

+ type EngineRouter struct

+ func NewEngineRouter(config *EventBusConfig) (*EngineRouter, error)

+ func (r *EngineRouter) GetEngineForTopic(topic string) string

+ func (r *EngineRouter) GetEngineNames() []string

+ func (r *EngineRouter) Publish(ctx context.Context, event Event) error

+ func (r *EngineRouter) Start(ctx context.Context) error

+ func (r *EngineRouter) Stop(ctx context.Context) error

+ func (r *EngineRouter) Subscribe(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (r *EngineRouter) SubscribeAsync(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (r *EngineRouter) SubscriberCount(topic string) int

+ func (r *EngineRouter) Topics() []string

+ func (r *EngineRouter) Unsubscribe(ctx context.Context, subscription Subscription) error

type EventBus

+ func NewCustomMemoryEventBus(config map[string]interface{}) (EventBus, error)

+ func NewKafkaEventBus(config map[string]interface{}) (EventBus, error)

+ func NewKinesisEventBus(config map[string]interface{}) (EventBus, error)

+ func NewRedisEventBus(config map[string]interface{}) (EventBus, error)

type EventBusConfig

+ Engines []EngineConfig

+ Routing []RoutingRule

+ func (c *EventBusConfig) GetDefaultEngine() string

+ func (c *EventBusConfig) IsMultiEngine() bool

+ func (c *EventBusConfig) ValidateConfig() error

type EventBusModule

+ func (m *EventBusModule) GetRouter() *EngineRouter

+ type EventFilter interface

+ Name func() string

+ ShouldProcess func(event Event) bool

+ type EventMetrics struct

+ AverageProcessingTime time.Duration

+ EventsPerTopic map[string]int64

+ LastResetTime time.Time

+ TotalEvents int64

+ type KafkaConfig struct

+ Brokers []string

+ ConsumerConfig map[string]string

+ GroupID string

+ ProducerConfig map[string]string

+ SecurityConfig map[string]string

+ type KafkaConsumerGroupHandler struct

+ func (h *KafkaConsumerGroupHandler) Cleanup(sarama.ConsumerGroupSession) error

+ func (h *KafkaConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ func (h *KafkaConsumerGroupHandler) Setup(sarama.ConsumerGroupSession) error

+ type KafkaEventBus struct

+ func (k *KafkaEventBus) Publish(ctx context.Context, event Event) error

+ func (k *KafkaEventBus) Start(ctx context.Context) error

+ func (k *KafkaEventBus) Stop(ctx context.Context) error

+ func (k *KafkaEventBus) Subscribe(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (k *KafkaEventBus) SubscribeAsync(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (k *KafkaEventBus) SubscriberCount(topic string) int

+ func (k *KafkaEventBus) Topics() []string

+ func (k *KafkaEventBus) Unsubscribe(ctx context.Context, subscription Subscription) error

+ type KinesisConfig struct

+ AccessKeyID string

+ Region string

+ SecretAccessKey string

+ SessionToken string

+ ShardCount int32

+ StreamName string

+ type KinesisEventBus struct

+ func (k *KinesisEventBus) Publish(ctx context.Context, event Event) error

+ func (k *KinesisEventBus) Start(ctx context.Context) error

+ func (k *KinesisEventBus) Stop(ctx context.Context) error

+ func (k *KinesisEventBus) Subscribe(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (k *KinesisEventBus) SubscribeAsync(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (k *KinesisEventBus) SubscriberCount(topic string) int

+ func (k *KinesisEventBus) Topics() []string

+ func (k *KinesisEventBus) Unsubscribe(ctx context.Context, subscription Subscription) error

+ type RedisConfig struct

+ DB int

+ Password string

+ PoolSize int

+ URL string

+ Username string

+ type RedisEventBus struct

+ func (r *RedisEventBus) Publish(ctx context.Context, event Event) error

+ func (r *RedisEventBus) Start(ctx context.Context) error

+ func (r *RedisEventBus) Stop(ctx context.Context) error

+ func (r *RedisEventBus) Subscribe(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (r *RedisEventBus) SubscribeAsync(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (r *RedisEventBus) SubscriberCount(topic string) int

+ func (r *RedisEventBus) Topics() []string

+ func (r *RedisEventBus) Unsubscribe(ctx context.Context, subscription Subscription) error

+ type RoutingRule struct

+ Engine string

+ Topics []string

+ type TopicPrefixFilter struct

+ AllowedPrefixes []string

+ func (f *TopicPrefixFilter) Name() string

+ func (f *TopicPrefixFilter) ShouldProcess(event Event) bool

v0

v0.1.1

Aug 11, 2025

Changes in this version

+ var ErrEventBusNotStarted = errors.New("event bus not started")

+ var ErrEventBusShutdownTimeout = errors.New("event bus shutdown timed out")

+ var ErrEventHandlerNil = errors.New("event handler cannot be nil")

+ var ErrInvalidSubscriptionType = errors.New("invalid subscription type")

v0.1.0

Jul 10, 2025

Changes in this version

+ const ModuleName

+ const ServiceName

+ func NewModule() modular.Module

+ type Event struct

+ CreatedAt time.Time

+ Metadata map[string]interface{}

+ Payload interface{}

+ ProcessingCompleted *time.Time

+ ProcessingStarted *time.Time

+ Topic string

+ type EventBus interface

+ Publish func(ctx context.Context, event Event) error

+ Start func(ctx context.Context) error

+ Stop func(ctx context.Context) error

+ Subscribe func(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ SubscribeAsync func(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ SubscriberCount func(topic string) int

+ Topics func() []string

+ Unsubscribe func(ctx context.Context, subscription Subscription) error

+ type EventBusConfig struct

+ DefaultEventBufferSize int

+ Engine string

+ EventTTL int

+ ExternalBrokerPassword string

+ ExternalBrokerURL string

+ ExternalBrokerUser string

+ MaxEventQueueSize int

+ RetentionDays int

+ WorkerCount int

+ type EventBusModule struct

+ func (m *EventBusModule) Constructor() modular.ModuleConstructor

+ func (m *EventBusModule) Dependencies() []string

+ func (m *EventBusModule) Init(app modular.Application) error

+ func (m *EventBusModule) Name() string

+ func (m *EventBusModule) ProvidesServices() []modular.ServiceProvider

+ func (m *EventBusModule) Publish(ctx context.Context, topic string, payload interface{}) error

+ func (m *EventBusModule) RegisterConfig(app modular.Application) error

+ func (m *EventBusModule) RequiresServices() []modular.ServiceDependency

+ func (m *EventBusModule) Start(ctx context.Context) error

+ func (m *EventBusModule) Stop(ctx context.Context) error

+ func (m *EventBusModule) Subscribe(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (m *EventBusModule) SubscribeAsync(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (m *EventBusModule) SubscriberCount(topic string) int

+ func (m *EventBusModule) Topics() []string

+ func (m *EventBusModule) Unsubscribe(ctx context.Context, subscription Subscription) error

+ type EventHandler func(ctx context.Context, event Event) error

+ type MemoryEventBus struct

+ func NewMemoryEventBus(config *EventBusConfig) *MemoryEventBus

+ func (m *MemoryEventBus) Publish(ctx context.Context, event Event) error

+ func (m *MemoryEventBus) Start(ctx context.Context) error

+ func (m *MemoryEventBus) Stop(ctx context.Context) error

+ func (m *MemoryEventBus) Subscribe(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (m *MemoryEventBus) SubscribeAsync(ctx context.Context, topic string, handler EventHandler) (Subscription, error)

+ func (m *MemoryEventBus) SubscriberCount(topic string) int

+ func (m *MemoryEventBus) Topics() []string

+ func (m *MemoryEventBus) Unsubscribe(ctx context.Context, subscription Subscription) error

+ type Subscription interface

+ Cancel func() error

+ ID func() string

+ IsAsync func() bool

+ Topic func() string

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