stream

Documentation

Index

• Constants
• func NewStream[DT any](es Stream, eventTypes []event.Type, from store.StreamPosition, filter Filter, ...) (out <-chan event.Event[DT], err error)
• type CryptoKeyProvider
• type Filter
  • func ReadAll() Filter
  • func ReadDataType(t string) Filter
  • func ReadEventType(t event.Type) Filter
• type Map
  • func New[T any]() Map[T]
• type Persistence
• type SetHelper
  • func InitSetHelper[DT any](store, delete func(event.Event[DT]), stream Stream, ...) (out SetHelper)
• type Stream
  • func Init(st Persistence, stream string, ctx context.Context) (out Stream, err error)

Constants

const BATCH_SIZE = 5000 //5000 is an arbitrary number, should probably be based on something else.

Functions

func NewStream

func NewStream[DT any](es Stream, eventTypes []event.Type, from store.StreamPosition, filter Filter,
	cryptKey CryptoKeyProvider, ctx context.Context) (out <-chan event.Event[DT], err error)

Types

type CryptoKeyProvider

type CryptoKeyProvider func(key string) string

type Filter

type Filter func(md event.Metadata) bool

func ReadAll

func ReadAll() Filter

func ReadDataType

func ReadDataType(t string) Filter

func ReadEventType

func ReadEventType(t event.Type) Filter

type Map

type Map[T any] interface {
	Store(id string, data T)
	Load(id string) (data T, ok bool)
	Delete(id string)
	Range(f func(k string, v T) bool)
}

func New

func New[T any]() Map[T]

type Persistence

type Persistence interface {
	Store(streamName string, ctx context.Context, events ...store.Event) (transactionId uint64, err error)
	Stream(streamName string, from store.StreamPosition, ctx context.Context) (out <-chan store.Event, err error)
}

type SetHelper

type SetHelper interface {
	SetAndWait(event.StoreEvent) error
}

func InitSetHelper

func InitSetHelper[DT any](store, delete func(event.Event[DT]), stream Stream, keyProvider CryptoKeyProvider, eventChannel <-chan event.Event[DT], ctx context.Context) (out SetHelper)

type Stream

type Stream interface {
	Store(event event.StoreEvent, cryptoKey CryptoKeyProvider) (transactionId uint64, err error)
	Stream(from store.StreamPosition, ctx context.Context) (out <-chan store.Event, err error)
	Name() string
}

func Init

func Init(st Persistence, stream string, ctx context.Context) (out Stream, err error)