networkstream

Documentation

Overview

Package networkstream presents a persistence.EventRepository as an eventstream.Stream via the dogma.messaging.v1 EventStream service.

Index

• func RegisterServer(s *grpc.Server, m marshaler.Marshaler, options ...ServerOption)
• type ServerOption
  • func WithApplication(ak string, s eventstream.Stream, types *sets.Set[message.Type]) ServerOption
• type Stream
  • func (s *Stream) Application() configkit.Identity
  • func (s *Stream) EventTypes(ctx context.Context) (*sets.Set[message.Type], error)
  • func (s *Stream) Open(ctx context.Context, o uint64, f *sets.Set[message.Type]) (eventstream.Cursor, error)

Functions

func RegisterServer

func RegisterServer(
	s *grpc.Server,
	m marshaler.Marshaler,
	options ...ServerOption,
)

RegisterServer registers an event stream server for the given streams.

Types

type ServerOption

type ServerOption struct {
	// contains filtered or unexported fields
}

ServerOption configures the behavior of a server.

func WithApplication

func WithApplication(
	ak string,
	s eventstream.Stream,
	types *sets.Set[message.Type],
) ServerOption

WithApplication returns a server option that configures the server to serve events for a specific application.

type Stream

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

	// Client is the gRPC client used to query the event stream server.
	Client eventstreamspec.StreamAPIClient

	// Marshaler is used to marshal and unmarshal messages and message types.
	Marshaler marshaler.Marshaler

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

Stream is an implementation of eventstream.Stream that consumes messages via the dogma.messaging.v1 EventStream gRPC API.

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(ctx 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().