Documentation
¶
Overview ¶
Package eventstream provides a catalog of the event streams that the engine consumes, both this application's own and those of remote engines.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrStreamNotFound = errors.New("event stream not found")
ErrStreamNotFound is returned by Catalog.Consume when the requested stream is not in the catalog.
Functions ¶
This section is empty.
Types ¶
type Catalog ¶
type Catalog struct {
// CacheSize is the number of each stream's most recent events to cache,
// per source. It must be positive.
CacheSize int
// Logger is the target for log messages produced by the catalog and the
// sources added to it.
Logger *slog.Logger
// contains filtered or unexported fields
}
A Catalog is an index of all event streams known to the engine, regardless of whether they belong to this engine or a remote one.
func (*Catalog) Consume ¶
func (c *Catalog) Consume( ctx context.Context, streamID *uuidpb.UUID, types *uuidcollection.Set, begin uint64, fn func(Segment) error, ) error
Consume delivers events from a stream to fn, beginning at the given offset.
Only events of the given types are delivered, in contiguous segments, as soon as they are available. The offset is typically the consumer's checkpoint offset. It blocks until ctx is canceled, fn returns an error, a read from the stream's source fails, or the catalog no longer has the stream, in which case it returns ErrStreamNotFound. If the offset is beyond the end of the stream, it returns an error that wraps source.ErrOffsetOutOfRange.
A consumer that keeps up with the stream is served from a cache of its most recent events, and one that falls behind is served by reading from the source until it catches up again.
types must not be empty, and every type must be one that Catalog.Watch was given for the stream's source.
func (*Catalog) StreamIDs ¶
func (c *Catalog) StreamIDs() cosyne.ObservableR[*uuidcollection.Set]
StreamIDs returns an observable of the IDs of the streams in the catalog.
The sets it holds must not be modified.
func (*Catalog) Watch ¶
Watch watches a source until ctx is canceled.
If the source reports that the catalog can not keep up with it, by returning an error that wraps source.ErrConsumerTooSlow, it watches the source again at once.
It adds the streams that the source delivers to the catalog, and removes them again before returning. Once it returns, no call to the source that the catalog made is in progress, and the catalog makes no more.
types is the set of message types to deliver from the source, or nil to deliver every type.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
internal
|
|
|
cache
Package cache provides a cache of an event stream's most recent events.
|
Package cache provides a cache of an event stream's most recent events. |
|
Package source defines where an event stream's events come from, the contract by which a source delivers them, and the gRPC service that provides a source's event streams to remote engines.
|
Package source defines where an event stream's events come from, the contract by which a source delivers them, and the gRPC service that provides a source's event streams to remote engines. |