stream

Documentation

Index

• Constants
• type CryptoKeyProvider
• type Filter
  • func ReadAll[MT any]() Filter[MT]
  • func ReadDataType[MT any](t string) Filter[MT]
  • func ReadEventType[MT any](t event.Type) Filter[MT]
• type Persistence
• type Stream
  • func Init[DT, MT any](st Persistence, stream string, ctx context.Context) (out Stream[DT, MT], err error)

Constants

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

Types

type CryptoKeyProvider

type CryptoKeyProvider func(key string) string

type Filter

type Filter[MT any] func(md event.Metadata[MT]) bool

func ReadAll

func ReadAll[MT any]() Filter[MT]

func ReadDataType

func ReadDataType[MT any](t string) Filter[MT]

func ReadEventType

func ReadEventType[MT any](t event.Type) Filter[MT]

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 Stream

type Stream[DT, MT any] interface {
	Store(event event.Event[DT, MT], cryptoKey CryptoKeyProvider) (transactionId uint64, err error)
	Stream(eventTypes []event.Type, from store.StreamPosition, filter Filter[MT], cryptKey CryptoKeyProvider, ctx context.Context) (out <-chan event.Event[DT, MT], err error)
	Name() string
}

func Init

func Init[DT, MT any](st Persistence, stream string, ctx context.Context) (out Stream[DT, MT], err error)