stream

Documentation

Index

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

Functions

func StaticProvider

func StaticProvider(key log.RedactedString) func(_ string) log.RedactedString

Types

type CryptoKeyProvider

type CryptoKeyProvider func(key string) log.RedactedString

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 interface {
	Write() chan<- event.ByteWriteEvent
	Store(event event.ByteEvent) (position uint64, err error)
	Stream(eventTypes []event.Type, from store.StreamPosition, filter Filter, ctx context.Context) (out <-chan event.ByteReadEvent, err error)
	End() (pos uint64, err error)
	Name() string
}

func Init

func Init(st Stream, ctx context.Context) (out FilteredStream, err error)

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 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
}