eventbus

Documentation

Index

• Constants
• Variables
• func Merge[T any](ctx context.Context, chs []<-chan T) (T, error)
• func NewRateLimiter() workqueue.RateLimiter
• func SubscriberMatch(subs Subscribers, topic string) ([]chan events.Event, bool)
• type Consumer
  • func MergeConsumers(consumers ...Consumer) Consumer
  • func NewConsumer(bus *EventBus, topic string, consumerCh chan events.Event) Consumer
• type EventBus
  • func NewEventBus(workers int) *EventBus
  • func (d *EventBus) InjectLogger(logr logr.Logger) error
  • func (d *EventBus) Publish(topic string, event events.Event) error
  • func (d *EventBus) Start(ctx context.Context) error
  • func (d *EventBus) Subscribe(topic string) (Consumer, error)
• type EventConsumer
  • func (c EventConsumer) Id() string
  • func (c EventConsumer) OnEvent() <-chan events.Event
  • func (c EventConsumer) OnEventSync(ctx context.Context) events.Event
  • func (c EventConsumer) UnSubscribe() error
• type EventConsumerSet
  • func (c EventConsumerSet) Id() string
  • func (cs EventConsumerSet) OnEvent() <-chan events.Event
  • func (cs EventConsumerSet) OnEventSync(ctx context.Context) events.Event
  • func (cs EventConsumerSet) UnSubscribe() error
• type Subscribers

Constants

const CONSUMER_BUFFER_SIZE = 10

Variables

var ErrChannelClosed = errors.New("channel closed")

Functions

func Merge

func Merge[T any](ctx context.Context, chs []<-chan T) (T, error)

func NewRateLimiter

func NewRateLimiter() workqueue.RateLimiter

func SubscriberMatch

func SubscriberMatch(subs Subscribers, topic string) ([]chan events.Event, bool)

SubscriberMatch match all the subscribers to the topic name

Types

type Consumer

type Consumer interface {
	Id() string
	OnEventSync(ctx context.Context) events.Event
	OnEvent() <-chan events.Event
	UnSubscribe() error
}

func MergeConsumers

func MergeConsumers(consumers ...Consumer) Consumer

func NewConsumer

func NewConsumer(bus *EventBus, topic string, consumerCh chan events.Event) Consumer

type EventBus

type EventBus struct {
	// contains filtered or unexported fields
}

func NewEventBus

func NewEventBus(workers int) *EventBus

NewEventBus creates a new EventBus that uses multiple workers to publish events to any subscribers

func (*EventBus) InjectLogger

func (d *EventBus) InjectLogger(logr logr.Logger) error

InjectLogger injectable logger https://github.com/kubernetes-sigs/controller-runtime/blob/master/pkg/runtime/inject/inject.go

func (*EventBus) Publish

func (d *EventBus) Publish(topic string, event events.Event) error

Publish queue an event to a topic

func (*EventBus) Start

func (d *EventBus) Start(ctx context.Context) error

Start implements the Runnable interface so that it can be started by the Operator SDK manager.

func (*EventBus) Subscribe

func (d *EventBus) Subscribe(topic string) (Consumer, error)

Subscribe registers a subscription on the topic

type EventConsumer

type EventConsumer struct {
	// contains filtered or unexported fields
}

func (EventConsumer) Id

func (c EventConsumer) Id() string

func (EventConsumer) OnEvent

func (c EventConsumer) OnEvent() <-chan events.Event

func (EventConsumer) OnEventSync

func (c EventConsumer) OnEventSync(ctx context.Context) events.Event

func (EventConsumer) UnSubscribe

func (c EventConsumer) UnSubscribe() error

type EventConsumerSet

type EventConsumerSet struct {
	// contains filtered or unexported fields
}

func (EventConsumerSet) Id

func (c EventConsumerSet) Id() string

func (EventConsumerSet) OnEvent

func (cs EventConsumerSet) OnEvent() <-chan events.Event

func (EventConsumerSet) OnEventSync

func (cs EventConsumerSet) OnEventSync(ctx context.Context) events.Event

func (EventConsumerSet) UnSubscribe

func (cs EventConsumerSet) UnSubscribe() error

type Subscribers

type Subscribers map[string][]chan events.Event