inproc

package
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 1, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Overview

Package inproc provides the in-process workflow runtime: it instantiates a workflow's executors and drives execution, streaming or to completion, with optional checkpointing.

Index

Constants

View Source 
const (
	RunStatusNotStarted      = execution.RunStatusNotStarted
	RunStatusIdle            = execution.RunStatusIdle
	RunStatusPendingRequests = execution.RunStatusPendingRequests
	RunStatusEnded           = execution.RunStatusEnded
	RunStatusRunning         = execution.RunStatusRunning
)

Variables

View Source 
var (
	// Default is the default execution environment.
	Default = OffThread

	// OffThread is an environment which will run steps in a background goroutine, streaming events out as they are raised.
	OffThread = newExecutionEnvironment(execution.ModeOffThread, false, nil)

	// Concurrent is like [OffThread], but enables concurrent execution.
	Concurrent = newExecutionEnvironment(execution.ModeOffThread, true, nil)

	// Lockstep is an environment which will run steps in the event watching tread,
	// accumulating events during each step and streaming them out after each step is completed.
	Lockstep = newExecutionEnvironment(execution.ModeLockstep, false, nil)
)

Functions

func BindSubworkflowAsExecutor 

func BindSubworkflowAsExecutor(wf *workflow.Workflow, id string) workflow.ExecutorBinding

BindSubworkflowAsExecutor returns an workflow.ExecutorBinding that hosts wf as a subworkflow with the supplied workflow-unique executor ID.

Types

type ExecutionEnvironment 

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

ExecutionEnvironment provides an in-process workflow execution environment for running, streaming, and checkpointing workflows.

func (*ExecutionEnvironment) IsCheckpointingEnabled 

func (e *ExecutionEnvironment) IsCheckpointingEnabled() bool

IsCheckpointingEnabled reports whether checkpointing is configured for this environment.

func (*ExecutionEnvironment) OpenStreaming 

func (e *ExecutionEnvironment) OpenStreaming(ctx context.Context, wf *workflow.Workflow, opts ...ExecutionOption) (*StreamingRun, error)

OpenStreaming opens a streaming workflow run without sending an initial message. The caller can submit messages with StreamingRun.TrySendMessage.

func (*ExecutionEnvironment) Resume 

func (e *ExecutionEnvironment) Resume(ctx context.Context, wf *workflow.Workflow, fromCheckpoint workflow.CheckpointInfo, opts ...ExecutionOption) (*Run, error)

func (*ExecutionEnvironment) ResumeStreaming 

func (e *ExecutionEnvironment) ResumeStreaming(ctx context.Context, wf *workflow.Workflow, fromCheckpoint workflow.CheckpointInfo, opts ...ExecutionOption) (*StreamingRun, error)

func (*ExecutionEnvironment) Run 

func (e *ExecutionEnvironment) Run(ctx context.Context, wf *workflow.Workflow, msg any, opts ...ExecutionOption) (*Run, error)

func (*ExecutionEnvironment) RunStreaming 

func (e *ExecutionEnvironment) RunStreaming(ctx context.Context, wf *workflow.Workflow, msg any, opts ...ExecutionOption) (*StreamingRun, error)

func (*ExecutionEnvironment) WithCheckpointing 

func (e *ExecutionEnvironment) WithCheckpointing(mgr checkpoint.Manager) *ExecutionEnvironment

WithCheckpointing returns a new execution environment configured with the given [checkpoint.Manager].

type ExecutionOption 

type ExecutionOption func(*executionOptions)

ExecutionOption configures an individual run started via the execution environment's Run, RunStreaming, Resume, or ResumeStreaming methods.

func WithPendingRequestRepublish 

func WithPendingRequestRepublish(enabled bool) ExecutionOption

WithPendingRequestRepublish controls whether outstanding external requests are re-emitted as RequestInfoEvent values when resuming from a checkpoint. It defaults to true. It is accepted by Resume and ResumeStreaming.

func WithSessionID 

func WithSessionID(sessionID string) ExecutionOption

WithSessionID sets the workflow session ID for a new run. It is honored by Run and RunStreaming. Resume and ResumeStreaming use the session ID from the checkpoint being resumed.

type Run 

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

Run represents a non-streaming, in-process workflow run. It advances the workflow to the next halt and accumulates the events raised so far so they can be replayed via OutgoingEvents or consumed incrementally via NewEvents.

func (*Run) Checkpoints 

func (run *Run) Checkpoints() []workflow.CheckpointInfo

Checkpoints returns the list of created checkpoints.

func (*Run) Close 

func (run *Run) Close(ctx context.Context) error

Close ends the run and releases its resources.

func (*Run) GetStatus 

func (run *Run) GetStatus(ctx context.Context) (RunStatus, error)

GetStatus returns the current execution status of the run.

func (*Run) IsCheckpointingEnabled 

func (run *Run) IsCheckpointingEnabled() bool

IsCheckpointingEnabled reports whether a checkpoint manager is configured for this run.

func (*Run) LastCheckpoint 

func (run *Run) LastCheckpoint() (workflow.CheckpointInfo, bool)

LastCheckpoint returns the most recently created checkpoint, or false if no checkpoint has been created yet.

func (*Run) NewEventCount 

func (run *Run) NewEventCount() int

NewEventCount returns the number of accumulated events not yet consumed by NewEvents.

func (*Run) NewEvents 

func (run *Run) NewEvents() iter.Seq[workflow.Event]

NewEvents returns a snapshot iterator over the events accumulated since the previous call to NewEvents. The internal bookmark advances when NewEvents is called, so stopping iteration early does not leave unread events available to a subsequent call.

func (*Run) OutgoingEvents 

func (run *Run) OutgoingEvents() iter.Seq[workflow.Event]

OutgoingEvents returns an iterator over all events accumulated by the run so far.

func (*Run) RestoreCheckpoint 

func (run *Run) RestoreCheckpoint(ctx context.Context, checkpointInfo workflow.CheckpointInfo) error

RestoreCheckpoint restores the workflow state from the given checkpoint.

func (*Run) Resume 

func (run *Run) Resume(ctx context.Context, messages ...any) (bool, error)

Resume enqueues the given messages and advances the workflow to the next halt, returning whether any events were raised.

func (*Run) RunToNextHalt 

func (run *Run) RunToNextHalt(ctx context.Context) (bool, error)

RunToNextHalt advances the workflow until it halts, appending any raised events to the run and returning whether any events were raised.

func (*Run) SessionID 

func (run *Run) SessionID() string

SessionID returns the unique identifier for this run's session.

type RunStatus 

type RunStatus = execution.RunStatus

RunStatus describes the current execution state of a Run or StreamingRun.

type StreamingRun 

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

StreamingRun represents a streaming, in-process workflow run. Events are delivered to the caller as they are raised rather than accumulated, and new messages or responses can be sent to the workflow while it is running.

func (*StreamingRun) CancelRun 

func (stream *StreamingRun) CancelRun() error

CancelRun cancels the run, stopping event delivery.

func (*StreamingRun) Checkpoints 

func (stream *StreamingRun) Checkpoints() []workflow.CheckpointInfo

Checkpoints returns the list of created checkpoints.

func (*StreamingRun) Close 

func (stream *StreamingRun) Close(ctx context.Context) error

Close ends the run and releases its resources.

func (*StreamingRun) GetStatus 

func (stream *StreamingRun) GetStatus(ctx context.Context) (RunStatus, error)

GetStatus returns the current execution status of the run.

func (*StreamingRun) IsCheckpointingEnabled 

func (stream *StreamingRun) IsCheckpointingEnabled() bool

IsCheckpointingEnabled reports whether a checkpoint manager is configured for this run.

func (*StreamingRun) LastCheckpoint 

func (stream *StreamingRun) LastCheckpoint() (workflow.CheckpointInfo, bool)

LastCheckpoint returns the most recently created checkpoint, or false if no checkpoint has been created yet.

func (*StreamingRun) ResponsePortExecutorID 

func (stream *StreamingRun) ResponsePortExecutorID(portID string) (string, bool)

ResponsePortExecutorID returns the executor that handles responses on the given port, or ("", false) if no such port is registered.

func (*StreamingRun) RestoreCheckpoint 

func (stream *StreamingRun) RestoreCheckpoint(ctx context.Context, checkpointInfo workflow.CheckpointInfo) error

RestoreCheckpoint restores the workflow state from the given checkpoint.

func (*StreamingRun) SendResponse 

func (stream *StreamingRun) SendResponse(ctx context.Context, response *workflow.ExternalResponse) error

SendResponse delivers an external response to a pending request in the workflow.

func (*StreamingRun) SessionID 

func (stream *StreamingRun) SessionID() string

SessionID returns the unique identifier for this run's session.

func (*StreamingRun) TrySendMessage 

func (stream *StreamingRun) TrySendMessage(ctx context.Context, message any) (bool, error)

TrySendMessage attempts to enqueue a message as input to the workflow. It returns false without an error when the workflow does not accept its type. TODO: Once Go 1.27 is the minimum supported version, add generic message entry points that preserve caller-declared types during validation and routing.

func (*StreamingRun) WatchStream 

func (stream *StreamingRun) WatchStream(ctx context.Context) iter.Seq2[workflow.Event, error]

WatchStream returns an iterator over the workflow's events, blocking on pending requests so the stream stays open until they are serviced. Only one consumer may watch the stream at a time.

func (*StreamingRun) WatchUntilHalt 

func (stream *StreamingRun) WatchUntilHalt(ctx context.Context) iter.Seq2[workflow.Event, error]

WatchUntilHalt returns an iterator over the workflow's events that completes once the workflow halts, without blocking on pending requests. Only one consumer may watch the stream at a time.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.