Versions in this module Expand all Collapse all v0 v0.1.1 Nov 4, 2026 v0.1.0 Nov 3, 2026 Changes in this version + const Version + func Effect[T, S, O any](ctx *Ctx[S, O], name string, fn func() (T, error)) (T, error) + func NonRetryable(err error) error + type CallSpec struct + AnswerChannel string + Channel string + Codec ChannelCodec + DeadlineMs int64 + Factory string + NoCreate bool + Target string + type Change interface + ApplyTo func(state *S) + type Changes map[string]Change[S] + type ChannelBuilder struct + type ChannelCodec string + const Bytes + const JSON + const Text + type Control struct + func (c *Control) BindPlugin(name, relayID string, hooks []any, channel string) (any, error) + func (c *Control) Call(verb string, args ...any) (any, error) + func (c *Control) CellState(relayID, cell string) (any, error) + func (c *Control) DisablePlugin(name string) (any, error) + func (c *Control) EnablePlugin(name, provider string, opts map[string]any, single bool, relayID string, ...) (any, error) + func (c *Control) ListPlugins() (any, error) + func (c *Control) ListRelays() (any, error) + func (c *Control) StartRelay(factoryID, id string, initValues, opts any) (StartResult, error) + func (c *Control) StopRelay(relayID string) (any, error) + func (c *Control) UnbindPlugin(name, relayID string) (any, error) + type Ctx struct + func (c *Ctx[S, O]) CancelTimer(kind TimerKind, at uint64) error + func (c *Ctx[S, O]) CellKey() (string, bool) + func (c *Ctx[S, O]) Change(change Change[S]) error + func (c *Ctx[S, O]) Checkpoint() + func (c *Ctx[S, O]) Complete(result any) error + func (c *Ctx[S, O]) Emit(value O, opts ...EmitOption) error + func (c *Ctx[S, O]) Now() uint64 + func (c *Ctx[S, O]) Part(value O) error + func (c *Ctx[S, O]) Replay() bool + func (c *Ctx[S, O]) SpawnSteps(factoryID string, payload any, opts ...SpawnOption) (*SpawnHandle, error) + func (c *Ctx[S, O]) StableID() string + func (c *Ctx[S, O]) State() S + func (c *Ctx[S, O]) TimerAfter(ms uint64) (uint64, error) + func (c *Ctx[S, O]) TimerAtEventTime(at uint64) (uint64, error) + type EmitOption func(*emitOptions) + func WithChannel(name string) EmitOption + func WithKey(key string) EmitOption + func WithMetadata(metadata map[string]string) EmitOption + type Error = native.Error + type InputOptions struct + Codec ChannelCodec + Description string + Schema any + type LocalOutput struct + Channel string + Metadata map[string]string + Payload []byte + type LocalRelay struct + func StartLocal(relay Relay, opts ...RunnerOption) (*LocalRelay, error) + func (l *LocalRelay) Close() error + func (l *LocalRelay) Feed(channel string, payload []byte) error + func (l *LocalRelay) OnOutput(channel string, callback func(LocalOutput)) error + func (l *LocalRelay) RunUntilIdle(budget time.Duration) error + type Merge struct + func From[T, M any](m *Merge[M], s Stream[T], lift func(T) M) *Merge[M] + func NewMerge[M any](c *ChannelBuilder) *Merge[M] + func (m *Merge[M]) KeyBy(key func(M) string) *Merge[M] + type NoOutput struct + type NoState struct + type Overflow string + const DropNewest + const DropNothing + const DropOldest + type Relay interface + Channels func(c *ChannelBuilder) + Spec func() RelaySpec + func ShortRelay[I, S, O any](factoryID string, fn func(in I, ctx *Ctx[S, O]) (O, bool), ...) Relay + func ShortTasklet[I, S, O any](factoryID string, t Tasklet[I, S, O], opts ...ShortOptions) Relay + type RelaySpec struct + Description string + Durable any + FactoryID string + MaxDurationMs int64 + Run any + type Response struct + Created bool + Err *Error + Request Req + Target string + Value Ans + func (r Response[Req, Ans]) Ok() bool + type Routed struct + Document D + Key string + func (r *Routed[D]) UnmarshalJSON(b []byte) error + func (r Routed[D]) MarshalJSON() ([]byte, error) + type Runner struct + func CreateRunner(namespace string, opts ...RunnerOption) (*Runner, error) + func (r *Runner) AwaitTermination() error + func (r *Runner) Call(relayID, channel string, value any) (any, error) + func (r *Runner) Close() + func (r *Runner) Control() *Control + func (r *Runner) Namespace() string + func (r *Runner) RegisterFactory(relay Relay) error + func (r *Runner) Relays() ([]string, error) + func (r *Runner) Run() error + func (r *Runner) Send(relayID, channel string, message []byte, opts ...SendOption) (*SendVerdict, error) + func (r *Runner) Start() error + func (r *Runner) UnregisterFactory(relay Relay) error + type RunnerOption func(*runnerOptions) + func WithLogger(logger *slog.Logger) RunnerOption + type SendOption func(*sendOptions) + func Local() SendOption + func MsgID(id string) SendOption + func SendKey(key string) SendOption + type SendVerdict struct + To string + Verdict string + type ShortOptions struct + Codec ChannelCodec + Description string + Durable any + Input string + MaxDurationMs int64 + Output string + Run any + type SpawnHandle struct + func (h *SpawnHandle) TryResult() (*SpawnResult, error) + func (h *SpawnHandle) Wait() (SpawnResult, error) + type SpawnOption func(*spawnOptions) + func IfRunning(policy string) SpawnOption + func SpawnID(id string) SpawnOption + type SpawnOutcome string + const Delivered + const SkippedOnReplay + const Spawned + type SpawnResult struct + ID string + Outcome SpawnOutcome + type Spec struct + CellKeys []string + CompactAfterBytes int64 + IdleTTLSeconds int64 + LogicalClock bool + MailboxCapacity *int + MaxCells int + MaxTimersPerCell int + MinReadableSchemaVersion uint32 + Name string + SchemaVersion uint32 + type StartResult struct + ID string + Outcome string + type Stream struct + func Call[Req, Ans any](s Stream[Req], spec CallSpec) Stream[Response[Req, Ans]] + func FlatMap[T, U any](s Stream[T], f func(T) []U) Stream[U] + func InputDecoded[T any](c *ChannelBuilder, name string, decode func([]byte) (T, error), ...) Stream[T] + func Input[T any](c *ChannelBuilder, name string, opts ...InputOptions) Stream[T] + func IntoNamed[I, S, O any](s Stream[I], t Tasklet[I, S, O], nodeName string) Stream[O] + func Into[I, S, O any](s Stream[I], t Tasklet[I, S, O]) Stream[O] + func Map[T, U any](s Stream[T], f func(T) U) Stream[U] + func MergeIntoNamed[S, O, M any](m *Merge[M], t Tasklet[M, S, O], nodeName string) Stream[O] + func MergeInto[S, O, M any](m *Merge[M], t Tasklet[M, S, O]) Stream[O] + func RouteLatestNamed[D, T any](c *ChannelBuilder, documents Stream[D], traffic Stream[T], key func(T) string, ...) Stream[Routed[D]] + func RouteLatest[D, T any](c *ChannelBuilder, documents Stream[D], traffic Stream[T], key func(T) string) Stream[Routed[D]] + func (s Stream[T]) Branch(arms ...func(Stream[T])) Stream[T] + func (s Stream[T]) Filter(pred func(T) bool) Stream[T] + func (s Stream[T]) KeyBy(key func(T) string) Stream[T] + func (s Stream[T]) Keyed() Stream[T] + func (s Stream[T]) Named(name string) Stream[T] + func (s Stream[T]) OnFull(policy Overflow) Stream[T] + func (s Stream[T]) Output(name string, codec ...ChannelCodec) Stream[T] + func (s Stream[T]) OutputEncoded(name string, encode func(T) ([]byte, error)) Stream[T] + func (s Stream[T]) RetryBackoffMs(ms int) Stream[T] + func (s Stream[T]) RetryMax(attempts int) Stream[T] + type Tasklet interface + Run func(in I, ctx *Ctx[S, O]) error + Spec func() Spec + type TimerFiring struct + Kind TimerKind + ScheduledAt uint64 + type TimerKind string + const EventTime + const ProcessingTime
Go to main page