Documentation
¶
Overview ¶
Package flowexec manages flow execution sessions. It orchestrates the flow lifecycle: variable resolution, node construction, runner creation, and result processing.
The ExecutionSession interface supports both local execution (ServerSession) and future distributed execution across regions.
Index ¶
- type AIProviderSnapshot
- type AISnapshot
- type ConditionSnapshot
- type ExecutionParams
- type ExecutionResult
- type ExecutionSession
- type ForEachSnapshot
- type ForSnapshot
- type GraphQLSnapshot
- type JSSnapshot
- type LocalSessionFactory
- type MemorySnapshot
- type NodeConfigResult
- type NodeConfigSnapshot
- type RequestSnapshot
- type RunSubFlowSnapshot
- type ServerSession
- type ServerSessionOpts
- type SessionFactory
- type SnapshotRegistry
- func (r *SnapshotRegistry) Get(kind mflow.NodeKind) (NodeConfigSnapshot, bool)
- func (r *SnapshotRegistry) ReadAll(ctx context.Context, nodes []mflow.Node, logger interface{ ... }) map[idwrap.IDWrap]any
- func (r *SnapshotRegistry) Register(s NodeConfigSnapshot)
- func (r *SnapshotRegistry) WriteAllTx(ctx context.Context, tx *sql.Tx, sourceNodes []mflow.Node, ...) ([]NodeConfigResult, error)
- type SubFlowReturnSnapshot
- type SubFlowTriggerSnapshot
- type WaitSnapshot
- type WsConnectionSnapshot
- type WsSendSnapshot
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AIProviderSnapshot ¶
type AIProviderSnapshot struct{ Service *sflow.NodeAiProviderService }
func (*AIProviderSnapshot) Kind ¶
func (s *AIProviderSnapshot) Kind() mflow.NodeKind
type AISnapshot ¶
type AISnapshot struct{ Service *sflow.NodeAIService }
func (*AISnapshot) Kind ¶
func (s *AISnapshot) Kind() mflow.NodeKind
type ConditionSnapshot ¶
type ConditionSnapshot struct{ Service *sflow.NodeIfService }
func (*ConditionSnapshot) Kind ¶
func (s *ConditionSnapshot) Kind() mflow.NodeKind
type ExecutionParams ¶
type ExecutionParams struct {
Flow mflow.Flow
Nodes []mflow.Node
Edges []mflow.Edge // Only valid edges (no orphaned source/target references)
FlowVars []mflow.FlowVariable
}
ExecutionParams contains the flow data needed for execution.
type ExecutionResult ¶
type ExecutionResult struct {
Duration int32
}
ExecutionResult contains the outcome of a flow execution.
type ExecutionSession ¶
type ExecutionSession interface {
// Prepare builds the execution graph from flow data.
// Must be called before Run.
Prepare(ctx context.Context, params ExecutionParams) error
// Run executes the prepared flow and returns the result.
// The processor lifecycle (Start/Wait) is managed internally.
Run(ctx context.Context) (ExecutionResult, error)
}
ExecutionSession manages a single flow execution lifecycle. Implementations may run the flow locally (ServerSession) or dispatch to remote workers for distributed execution.
type ForEachSnapshot ¶
type ForEachSnapshot struct{ Service *sflow.NodeForEachService }
func (*ForEachSnapshot) Kind ¶
func (s *ForEachSnapshot) Kind() mflow.NodeKind
type ForSnapshot ¶
type ForSnapshot struct{ Service *sflow.NodeForService }
func (*ForSnapshot) Kind ¶
func (s *ForSnapshot) Kind() mflow.NodeKind
type GraphQLSnapshot ¶
type GraphQLSnapshot struct{ Service *sflow.NodeGraphQLService }
func (*GraphQLSnapshot) Kind ¶
func (s *GraphQLSnapshot) Kind() mflow.NodeKind
type JSSnapshot ¶
type JSSnapshot struct{ Service *sflow.NodeJsService }
func (*JSSnapshot) Kind ¶
func (s *JSSnapshot) Kind() mflow.NodeKind
type LocalSessionFactory ¶
type LocalSessionFactory struct {
Builder *flowbuilder.Builder
JsClient node_js_executorv1connect.NodeJsExecutorServiceClient
}
LocalSessionFactory creates ServerSession instances for local execution.
func (*LocalSessionFactory) Create ¶
func (f *LocalSessionFactory) Create(processor flowresult.ResultProcessor) ExecutionSession
type MemorySnapshot ¶
type MemorySnapshot struct{ Service *sflow.NodeMemoryService }
func (*MemorySnapshot) Kind ¶
func (s *MemorySnapshot) Kind() mflow.NodeKind
type NodeConfigResult ¶
type NodeConfigResult struct {
NodeKind mflow.NodeKind
Config any // The created model (e.g., mflow.NodeFor, mflow.NodeJS)
}
NodeConfigResult holds the result of writing a single node's type-specific config.
type NodeConfigSnapshot ¶
type NodeConfigSnapshot interface {
Kind() mflow.NodeKind
// Read fetches the type-specific config for a node. Returns (nil, nil) if none exists.
Read(ctx context.Context, nodeID idwrap.IDWrap) (any, error)
// WriteTx creates a copy of the config with the new node ID inside the given transaction.
// Returns the created model (for event publishing) or (nil, nil) if skipped.
WriteTx(ctx context.Context, tx *sql.Tx, newNodeID idwrap.IDWrap, config any) (any, error)
}
NodeConfigSnapshot reads and writes a single node kind's type-specific configuration during flow version snapshots.
type RequestSnapshot ¶
type RequestSnapshot struct{ Service *sflow.NodeRequestService }
func (*RequestSnapshot) Kind ¶
func (s *RequestSnapshot) Kind() mflow.NodeKind
type RunSubFlowSnapshot ¶
type RunSubFlowSnapshot struct{ Service *sflow.NodeRunSubFlowService }
func (*RunSubFlowSnapshot) Kind ¶
func (s *RunSubFlowSnapshot) Kind() mflow.NodeKind
type ServerSession ¶
type ServerSession struct {
// contains filtered or unexported fields
}
ServerSession implements ExecutionSession for local server execution. It builds the execution graph, runs the flow via FlowLocalRunner, and delegates result processing to a ResultProcessor.
func NewServerSession ¶
func NewServerSession(opts ServerSessionOpts) *ServerSession
NewServerSession creates a new ServerSession for local flow execution.
func (*ServerSession) Prepare ¶
func (s *ServerSession) Prepare(ctx context.Context, params ExecutionParams) error
Prepare builds execution variables, constructs flow nodes, and creates the runner.
func (*ServerSession) Run ¶
func (s *ServerSession) Run(ctx context.Context) (ExecutionResult, error)
Run starts the result processor, executes the flow, waits for all result processing to complete, and returns the execution duration.
type ServerSessionOpts ¶
type ServerSessionOpts struct {
Builder *flowbuilder.Builder
JsClient node_js_executorv1connect.NodeJsExecutorServiceClient
Processor flowresult.ResultProcessor
}
ServerSessionOpts configures a ServerSession.
type SessionFactory ¶
type SessionFactory interface {
Create(processor flowresult.ResultProcessor) ExecutionSession
}
SessionFactory creates ExecutionSession instances. Implementations control where the flow runs: locally (LocalSessionFactory) or on remote workers for distributed execution.
type SnapshotRegistry ¶
type SnapshotRegistry struct {
// contains filtered or unexported fields
}
SnapshotRegistry maps node kinds to their snapshot handlers.
func NewSnapshotRegistry ¶
func NewSnapshotRegistry() *SnapshotRegistry
NewSnapshotRegistry creates an empty registry.
func (*SnapshotRegistry) Get ¶
func (r *SnapshotRegistry) Get(kind mflow.NodeKind) (NodeConfigSnapshot, bool)
Get returns the snapshot handler for a node kind, if registered. Returns (nil, false) if the registry is nil or the kind is not registered.
func (*SnapshotRegistry) ReadAll ¶
func (r *SnapshotRegistry) ReadAll(ctx context.Context, nodes []mflow.Node, logger interface{ Warn(msg string, args ...any) }) map[idwrap.IDWrap]any
ReadAll reads type-specific configurations for all nodes that have registered handlers. Returns a map from node ID to the config value. Errors are logged and the node is skipped (resulting in default config when written).
func (*SnapshotRegistry) Register ¶
func (r *SnapshotRegistry) Register(s NodeConfigSnapshot)
Register adds a snapshot handler for a node kind.
func (*SnapshotRegistry) WriteAllTx ¶
func (r *SnapshotRegistry) WriteAllTx( ctx context.Context, tx *sql.Tx, sourceNodes []mflow.Node, nodeIDMapping map[string]idwrap.IDWrap, configs map[idwrap.IDWrap]any, ) ([]NodeConfigResult, error)
WriteAllTx writes type-specific configurations for all nodes within a transaction. Uses configs from ReadAll. Returns the created configs for event publishing.
type SubFlowReturnSnapshot ¶
type SubFlowReturnSnapshot struct {
Service *sflow.NodeSubFlowReturnService
}
func (*SubFlowReturnSnapshot) Kind ¶
func (s *SubFlowReturnSnapshot) Kind() mflow.NodeKind
type SubFlowTriggerSnapshot ¶
type SubFlowTriggerSnapshot struct {
Service *sflow.NodeSubFlowTriggerService
}
func (*SubFlowTriggerSnapshot) Kind ¶
func (s *SubFlowTriggerSnapshot) Kind() mflow.NodeKind
type WaitSnapshot ¶
type WaitSnapshot struct{ Service *sflow.NodeWaitService }
func (*WaitSnapshot) Kind ¶
func (s *WaitSnapshot) Kind() mflow.NodeKind
type WsConnectionSnapshot ¶
type WsConnectionSnapshot struct {
Service *sflow.NodeWsConnectionService
}
func (*WsConnectionSnapshot) Kind ¶
func (s *WsConnectionSnapshot) Kind() mflow.NodeKind
type WsSendSnapshot ¶
type WsSendSnapshot struct{ Service *sflow.NodeWsSendService }
func (*WsSendSnapshot) Kind ¶
func (s *WsSendSnapshot) Kind() mflow.NodeKind
Source Files