stream

Documentation

Index

• func StaticProvider(key string) func(_ string) string
• type CryptoKeyProvider
• type Filter
  • func ReadAll() Filter
  • func ReadDataType(t string) Filter
  • func ReadEventType(t event.Type) Filter
• type FilteredStream
  • func Init[BT any, T bcts.ReadWriter[BT]](st Stream, ctx context.Context) (out FilteredStream[BT, T], err error)
• type Stream

Functions

func StaticProvider

func StaticProvider(key string) func(_ string) string

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 FilteredStream

type FilteredStream[BT any, T bcts.ReadWriter[BT]] interface {
	Write() chan<- event.WriteEventReadStatus[BT, T]
	Store(event event.Event[BT, T]) (position uint64, err error)
	Stream(
		eventTypes []event.Type,
		from store.StreamPosition,
		filter Filter,
		ctx context.Context,
	) (out <-chan event.ReadEvent[BT, T], err error)
	End() (pos uint64, err error)
	Name() string
	FilteredEnd(eventTypes []event.Type, filter Filter) (pos uint64, err error)
}

func Init

func Init[BT any, T bcts.ReadWriter[BT]](
	st Stream,
	ctx context.Context,
) (out FilteredStream[BT, T], err error)

type Stream

type Stream interface {
	Write() chan<- store.WriteEvent
	Stream(from store.StreamPosition, ctx context.Context) (out <-chan store.ReadEvent, err error)
	End() (pos uint64, err error)
	Name() string
}