events

type Config ¶

type Config struct {
	Mode                  Mode
	Namespace             string
	MaxStreamLength       int64
	ApproxMaxStreamLength bool
}

Config defines the config of the events system.

func (c *Config) Validate() error

type Event ¶

type Event[T interface{}] struct {
	ID        string    `json:"id"`
	Timestamp time.Time `json:"timestamp"`
	Payload   T         `json:"payload"`
}

type EventType ¶

type EventType string

EventType describes the type of event.

type GenericReader ¶

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

GenericReader represents an event reader that supports registering type safe handlers for an arbitrary set of custom events within a given event category using the ReaderRegisterEvent method. NOTE: Optimally this should be an interface with RegisterEvent[T] method, but that's currently not possible in go. IMPORTANT: This reader should not be instantiated from external packages.

func (r *GenericReader) Configure(opts ...ReaderOption)

type GenericReporter ¶

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

GenericReporter represents an event reporter that supports sending typesafe messages for an arbitrary set of custom events within an event category using the ReporterSendEvent method. NOTE: Optimally this should be an interface with SendEvent[T] method, but that's not possible in go.

func NewReporter(system *System, category string) (*GenericReporter, error)

type HandlerFunc ¶

type HandlerFunc[T interface{}] func(context.Context, *Event[T]) error

type HandlerOption ¶

type HandlerOption stream.HandlerOption

HandlerOption can be used to configure event handlers.

func WithIdleTimeout(timeout time.Duration) HandlerOption

func WithMaxRetries(maxRetries int) HandlerOption

type Mode ¶

type Mode string

Mode defines the different modes of the event framework.

const (
	ModeRedis    Mode = "redis"
	ModeInMemory Mode = "inmemory"
)

type Reader ¶

type Reader interface {
	Configure(opts ...ReaderOption)
}

Reader specifies the minimum functionality a reader should expose. NOTE: we don't want to enforce any event registration methods here, allowing full control for customized readers.

type ReaderCanceler ¶

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

ReaderCanceler exposes the functionality to cancel a reader explicitly.

func (d *ReaderCanceler) Cancel() error

type ReaderFactory ¶

type ReaderFactory[R Reader] struct {
	// contains filtered or unexported fields
}

ReaderFactory allows to launch event readers of type [R] (can be GenericReader or customized readers).

func NewReaderFactory[R Reader](system *System, category string, fn ReaderFactoryFunc[R]) (*ReaderFactory[R], error)

func (f *ReaderFactory[R]) Launch(ctx context.Context,
	groupName string, readerName string, setup func(R) error) (*ReaderCanceler, error)

type ReaderFactoryFunc ¶

type ReaderFactoryFunc[R Reader] func(reader *GenericReader) (R, error)

ReaderFactoryFunc is an abstraction of a factory method that creates customized Reader implementations (type [R]). It is triggered by the ReaderFactory to create a new instance of the Reader to launch. The provided GenericReader object is available exclusively to the factory method (every call has a fresh instance) and should be used as base of any custom Reader implementation (use ReaderRegisterEvent to register custom handler).

type ReaderOption ¶

type ReaderOption stream.ConsumerOption

ReaderOption can be used to configure event readers.

func WithConcurrency(concurrency int) ReaderOption

func WithHandlerOptions(opts ...HandlerOption) ReaderOption

type StreamConsumer ¶

type StreamConsumer interface {
	Register(streamID string, handler stream.HandlerFunc, opts ...stream.HandlerOption) error
	Configure(opts ...stream.ConsumerOption)
	Start(ctx context.Context) error
	Errors() <-chan error
	Infos() <-chan string
}

StreamConsumer is an abstraction of a consumer from the streams package.

type StreamConsumerFactoryFunc ¶

type StreamConsumerFactoryFunc func(groupName string, consumerName string) (StreamConsumer, error)

StreamConsumerFactoryFunc is an abstraction of a factory method for stream consumers.

type StreamProducer ¶

type StreamProducer interface {
	Send(ctx context.Context, streamID string, payload map[string]interface{}) (string, error)
}

StreamProducer is an abstraction of a producer from the streams package.

type System ¶

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

System represents a single contained event system that is used to setup event Reporters and ReaderFactories.

func NewSystem(streamConsumerFactoryFunc StreamConsumerFactoryFunc, streamProducer StreamProducer) (*System, error)

func ProvideSystem(config Config, redisClient redis.UniversalClient) (*System, error)