rpatterns

func Await ¶

func Await(
	in context.Context, stream reflex.StreamFunc, pollFn func() (bool, error),
	foreignID string, eventTypes ...reflex.EventType,
) error

Await returns nil when a new event with foreignID and one of the eventTypes is received from the event streamer. It also returns nil if the pollFn returns true when it is periodically called. Cancel the input context to return early.

This function can be used to wait for some state that is associated with an event. pollFn is used to periodically query the state, while this logic waits for a new event. It is done in parallel to mitigate race conditions.

func MemCursorStore ¶

func MemCursorStore(opts ...MemOpt) reflex.CursorStore

MemCursorStore returns an in-memory cursor store. Note that it obviously does not provide any persistence guarantees.

Use cases:

• Testing
• Programmatic seeding of a cursor: See ReadThroughCursorStore above.

func NewAckSpec ¶

func NewAckSpec(stream reflex.StreamFunc, ac *AckConsumer,
	opts ...reflex.StreamOption,
) reflex.Spec

NewAckSpec returns a reflex spec for the AckConsumer.

func NewBatchSpec ¶

func NewBatchSpec(
	stream reflex.StreamFunc,
	bc *BatchConsumer,
	opts ...reflex.StreamOption,
) reflex.Spec

NewBatchSpec returns a reflex spec for the AckConsumer.

func NewBestEffortConsumer ¶

func NewBestEffortConsumer(name string, retries int, fn reflex.ConsumerFunc,
	opts ...reflex.ConsumerOption,
) reflex.Consumer

NewBestEffortConsumer returns a reflex consumer that ignores errors after the provided number of retries and therefore eventually continues to the next event.

func NewBootstrapConsumable ¶

func NewBootstrapConsumable(sFn reflex.StreamFunc, cstore reflex.CursorStore,
	opts ...reflex.StreamOption,
) reflex.Consumable

NewBootstrapConsumable returns a reflex consumable that will start streaming from head if no cursor is found. This is useful if old events should be skipped for new consumers. Once running (bootstrapped), one can safely revert to reflex.NewConsumable. Deprecated: Use NewBootstrapSpec.

func NewBootstrapSpec ¶

func NewBootstrapSpec(stream reflex.StreamFunc, cstore reflex.CursorStore,
	consumer reflex.Consumer, opts ...reflex.StreamOption,
) reflex.Spec

NewBootstrapSpec returns a reflex spec that will start streaming from head if no cursor is found. This is useful if old events should be skipped for new consumers. Once running (bootstrapped), one can safely revert to reflex.NewSpec.

func NewConcurrentSpec ¶

func NewConcurrentSpec(
	stream reflex.StreamFunc,
	rac *ConcurrentConsumer,
	opts ...reflex.StreamOption,
) reflex.Spec

NewConcurrentSpec wraps the ConcurrentConsumer in a reflex.Spec

func NewDeadLetterConsumer ¶

func NewDeadLetterConsumer(name string, retries int, fn reflex.ConsumerFunc, eFn reflex.ErrorInsertFunc,
	opts ...reflex.ConsumerOption,
) reflex.Consumer

NewDeadLetterConsumer returns a reflex consumer that records but ignores errors after the provided number of retries and therefore eventually continues to the next event. However, if the consumer cannot record the error it will return the error in a blocking fashion like a standard consumer.

func Parallel ¶

func Parallel(getCtx getCtxFn, getConsumer getConsumerFn, n int, stream reflex.StreamFunc,
	cstore reflex.CursorStore, opts ...ParallelOption,
)

Parallel starts N consumers which consume the stream in parallel. Each event is consistently hashed to a consumer using the field specified in HashOption. Role scheduling combined with an appropriate getCtxFn can be used to implement distributed parallel consuming.

NOTE: N should preferably be a power of 2, and modifying N will reset the cursors.

func ParallelAck ¶

func ParallelAck(getCtx getCtxFn, getConsumer getAckConsumerFn, n int, stream reflex.StreamFunc, opts ...ParallelOption)

ParallelAck starts N consumers which consume the stream in parallel. Each event is consistently hashed to a consumer using the field specified in HashOption. Role scheduling combined with an appropriate getCtxFn can be used to implement distributed parallel consuming. Events must be acked manually.

NOTE: N should preferably be a power of 2, and modifying N will reset the cursors.

func ParallelAckConsumer ¶

func ParallelAckConsumer(
	shard ConsumerShard,
	store reflex.CursorStore,
	consume AckConsumerFunc,
	opts ...reflex.ConsumerOption,
) reflex.Consumer

ParallelAckConsumer constructs a AckConsumer from a ConsumerShard. This AckConsumer can be used in NewAckSpec to make it runnable.

func ParallelConsumer ¶

func ParallelConsumer(
	shard ConsumerShard,
	consume reflex.ConsumerFunc,
	opts ...reflex.ConsumerOption,
) reflex.Consumer

ParallelConsumer constructs a reflex.Consumer from a ConsumerShard. This pattern is used when you need to customise the consume function for each shard. This reflex.Consumer can be used in reflex.NewSpec to make it runnable.

func ParallelSpecs ¶

func ParallelSpecs(name string, n int,
	stream reflex.StreamFunc, store reflex.CursorStore,
	consume reflex.ConsumerFunc,
	opts ...ParallelOption,
) []reflex.Spec

ParallelSpecs will create n reflex.Spec structs, one for each shard. stream and store are re-used for each spec. This pattern is used when consume is the same function for each shard. See ParallelOption for more details on passing through reflex.ConsumerOption or reflex.StreamOption

func ReadThroughCursorStore ¶

func ReadThroughCursorStore(primary, fallback reflex.CursorStore) reflex.CursorStore

ReadThroughCursorStore provides a cursor store that queries the fallback cursor store if the cursor is not found in the primary. It writes the cursor value to the primary if not. Cursor updates always go directly to the primary.

Use cases:

• Migrating cursor stores: Use the new cursor store as the primary and the old cursor store as the fallback. Revert to just the new cursor store after the migration.
• Programmatic seeding of a cursor: Use a MemCursorStore with the cursor seeded by WithMemCursor as the fallback and the target cursor store as the primary. Revert to just the target cursor store afterwards.

func RunForever ¶

func RunForever(getCtx func() context.Context, req reflex.Spec)

RunForever continuously calls the run function, backing off and logging on unexpected errors.

func WithPollBackoff ¶

func WithPollBackoff(d time.Duration) func(*Poller)

WithPollBackoff returns a option to configure the polling backoff (period) if no new events are available.

func WithSleep ¶

func WithSleep(_ *testing.T, fn func(d time.Duration) <-chan time.Time) func(*Poller)

WithSleep returns a option to configure the sleep function for testing.