persistedstream

Documentation

Overview

Package persistedstream provides an implementation of eventstream.Stream that streams persisted events from an event repository.

Index

• type Stream
  • func (s *Stream) Application() configkit.Identity
  • func (s *Stream) EventTypes(context.Context) (*sets.Set[message.Type], error)
  • func (s *Stream) Open(ctx context.Context, o uint64, f *sets.Set[message.Type]) (eventstream.Cursor, error)

Types

type Stream

type Stream struct {
	// App is the identity of the application that owns the stream.
	App configkit.Identity

	// Types is the set of supported event types.
	Types *sets.Set[message.Type]

	// Repository is the event repository used to load events.
	Repository persistence.EventRepository

	// Marshaler is used to unmarshal messages.
	Marshaler marshaler.Marshaler

	// Cache is an in-memory stream that contains recently recorded events.
	//
	// The cache must be provided, otherwise the stream has no way to block
	// until a new event is recorded.
	Cache eventstream.Stream

	// PreFetch specifies how many messages to pre-load into memory.
	PreFetch int
}

Stream is an implementation of eventstream.Stream that reads events from a persistence.EventRepository.

func (*Stream) Application

func (s *Stream) Application() configkit.Identity

Application returns the identity of the application that owns the stream.

func (*Stream) EventTypes

func (s *Stream) EventTypes(context.Context) (*sets.Set[message.Type], error)

EventTypes returns the set of event types that may appear on the stream.

func (*Stream) Open

func (s *Stream) Open(
	ctx context.Context,
	o uint64,
	f *sets.Set[message.Type],
) (eventstream.Cursor, error)

Open returns a cursor that reads events from the stream.

o is the offset of the first event to read. The first event on a stream is always at offset 0.

f is the set of "filter" event types to be returned by Cursor.Next(). Any other event types are ignored.

It returns an error if any of the event types in f are not supported, as indicated by EventTypes().