flowctx

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Aug 26, 2026 License: MIT Imports: 9 Imported by: 0

Documentation

Overview

Package flowctx 定义一次工作流执行的上下文、入口信封和结果协议。

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Envelope 

type Envelope struct {
	Env             string                 `json:"env"`
	Workspace       string                 `json:"workspace"`
	Service         string                 `json:"service"`
	TraceID         string                 `json:"traceId"`
	RunID           string                 `json:"runId"`
	ParentRunID     string                 `json:"parentRunId"`
	WorkflowID      string                 `json:"workflowId"`
	WorkflowVersion string                 `json:"workflowVersion"`
	Attempt         int                    `json:"attempt"`
	CreatedAt       time.Time              `json:"createdAt"`
	Payload         map[string]interface{} `json:"payload"`
}

Envelope 是一次 workflow 执行的入口上下文。

普通调用方只需要关心 Env、Workspace、WorkflowID、WorkflowVersion 和 Payload； TraceID、RunID、Attempt、CreatedAt 由框架在执行时补齐。

func (*Envelope) PayloadValue 

func (e *Envelope) PayloadValue(key string) (interface{}, bool)

PayloadValue 从入口参数中读取顶层 key。

func (*Envelope) SetPayloadValue 

func (e *Envelope) SetPayloadValue(key string, val interface{})

SetPayloadValue 写入入口参数顶层 key。

type FlowContext 

type FlowContext struct {
	context.Context
	// contains filtered or unexported fields
}

FlowContext 是一次 workflow 执行中的可变上下文。

它内嵌 context.Context 用于取消和超时，同时把 workflow 自身的数据分区保存： envelope 表示本次请求身份并携带入口参数，values 保存节点运行中写入的临时变量， results 按稳定节点 ID 归档节点输出，runtime 保存不应该出现在用户输出里的框架运行依赖， response 保存节点拼装出的业务响应。

func FromContext 

func FromContext(ctx context.Context) (*FlowContext, bool)

FromContext 读取 NewFlowContext 或 WithContext 写入 ctx 的 FlowContext。

func NewFlowContext 

func NewFlowContext(parent context.Context) *FlowContext

NewFlowContext 创建一次 workflow 执行使用的 FlowContext。

parent 为空时使用 context.Background。返回的 FlowContext 会创建空 Envelope， 并写入内嵌 context，方便只拿到 context.Context 的下游通过 FromContext 恢复当前执行状态。

func (*FlowContext) BindingData 

func (fc *FlowContext) BindingData() map[string]interface{}

BindingData 是对外暴露给模板和节点配置的固定入口，顶层只保留 ctx。

func (*FlowContext) ContextData 

func (fc *FlowContext) ContextData() map[string]interface{}

ContextData 返回节点配置、条件判断和扩展节点读取运行数据时使用的统一视图。

视图固定包含 params、vars、results、response、envelope 五个分区： params 来自入口 Envelope.Payload，vars 来自节点运行中写入的上下文变量， results 按节点 ID 保存节点输出，response 保存业务响应，envelope 保存本次执行身份。

func (*FlowContext) Debug 

func (fc *FlowContext) Debug(message string, fields ...interface{})

func (*FlowContext) Debugf 

func (fc *FlowContext) Debugf(format string, args ...interface{})

func (*FlowContext) Env 

func (fc *FlowContext) Env(env string) *FlowContext

Env 设置本次执行所在环境。

env 是 workflow 发布、拉取和执行范围的一部分。多链路服务可以在请求中显式写入； 单服务单链路服务也可以由启动期 runtime scope 在执行前补齐。

func (*FlowContext) Envelope 

func (fc *FlowContext) Envelope() *Envelope

Envelope 返回本次执行关联的入口 Envelope。

这里刻意返回指针，因为 env、workspace、workflowId 等身份字段是执行契约的一部分。 普通写入优先使用 FlowContext 的链式 setter，避免把 Envelope 当成通用数据袋使用。

func (*FlowContext) Error 

func (fc *FlowContext) Error(message string, fields ...interface{})

func (*FlowContext) ErrorErr 

func (fc *FlowContext) ErrorErr(err error, fields ...interface{})

ErrorErr 会忽略 nil，方便节点在返回错误前顺手记录而不写重复 nil 判断。

func (*FlowContext) Errorf 

func (fc *FlowContext) Errorf(format string, args ...interface{})

func (*FlowContext) Fork 

func (fc *FlowContext) Fork() *FlowContext

Fork 创建并行分支使用的子上下文。

子上下文会复制执行身份、values、results 和 response 状态，并共享 runtime 依赖和 trace 通道。 runtime 因此可以在分支成功后显式合并结果，避免并发分支直接修改父上下文。

func (*FlowContext) Get 

func (fc *FlowContext) Get(key string) (interface{}, bool)

Get 是 ValueOf 的简短别名。

func (*FlowContext) GetBool 

func (fc *FlowContext) GetBool(key string) (bool, bool)

GetBool 支持 bool 和 Go 能解析的布尔字符串。

func (*FlowContext) GetInt 

func (fc *FlowContext) GetInt(key string) (int, bool)

GetInt 支持常见整数、浮点数和十进制字符串；浮点数会按 Go 转 int 的规则截断。

func (*FlowContext) GetString 

func (fc *FlowContext) GetString(key string) (string, bool)

GetString 对非 string 值会走 fmt.Stringer/fmt.Sprint，适合宽松读取配置或上游结果。

func (*FlowContext) Info 

func (fc *FlowContext) Info(message string, fields ...interface{})

func (*FlowContext) Infof 

func (fc *FlowContext) Infof(format string, args ...interface{})

func (*FlowContext) Log 

func (fc *FlowContext) Log(level LogLevel, message string, fields ...interface{})

Log 是 best-effort 的 trace 钩子。trace 未打开或未挂接收器时直接 no-op， 这样业务节点不需要围绕诊断日志写开关判断。

func (*FlowContext) LookupBindingPath 

func (fc *FlowContext) LookupBindingPath(path string) (interface{}, bool)

LookupBindingPath 按点分路径读取 BindingData，例如 ctx.params.userId。

func (*FlowContext) MergeResponseFrom 

func (fc *FlowContext) MergeResponseFrom(src *FlowContext)

MergeResponseFrom 在节点成功后提交子上下文写入的 response。

func (*FlowContext) MergeResultsFrom 

func (fc *FlowContext) MergeResultsFrom(src *FlowContext)

MergeResultsFrom 把成功分支的节点结果归档合并回当前上下文。

func (*FlowContext) MergeValues 

func (fc *FlowContext) MergeValues(values map[string]interface{})

MergeValues 批量覆盖写入共享值。runtime 用它提交成功节点或成功并行分支的输出。

func (*FlowContext) MustGet 

func (fc *FlowContext) MustGet(key string) interface{}

MustGet 用于节点确实依赖某个上游值的场景；缺失时 panic，适合测试或明确的框架内部约束。

func (*FlowContext) OutputSnapshot 

func (fc *FlowContext) OutputSnapshot(returning *workflow.ReturningConfig) map[string]interface{}

OutputSnapshot 按 workflow.returning 配置导出本次执行后的 ctx 快照。

func (*FlowContext) Param 

func (fc *FlowContext) Param(path string) (interface{}, bool)

Param 从入口参数中按点分路径读取值。入口参数目前由 Envelope.Payload 承载。

func (*FlowContext) ParamString 

func (fc *FlowContext) ParamString(path string) (string, bool)

ParamString 读取入口参数并转换为字符串。

func (*FlowContext) ParamValue 

func (fc *FlowContext) ParamValue(key string, value interface{}) *FlowContext

ParamValue 写入单个入口参数。

它适合在构造 FlowContext 时逐步补充入口参数；空 key 会被忽略。

func (*FlowContext) Params 

func (fc *FlowContext) Params(params map[string]interface{}) *FlowContext

Params 覆盖本次执行入口参数。

参数会写入 Envelope.Payload，并作为 ctx.params 出现在模板、条件和节点配置绑定数据中。

func (*FlowContext) ParentRunID 

func (fc *FlowContext) ParentRunID() string

func (*FlowContext) Response 

func (fc *FlowContext) Response() (interface{}, bool)

Response 读取完整 response 区块。

func (*FlowContext) ResponseValue 

func (fc *FlowContext) ResponseValue(path string) (interface{}, bool)

ResponseValue 按点分路径读取 response 子字段。

func (*FlowContext) Result 

func (fc *FlowContext) Result(nodeID string) (interface{}, bool)

Result 按节点 ID 读取节点执行结果。

func (*FlowContext) ResultMap 

func (fc *FlowContext) ResultMap(nodeID string) (map[string]interface{}, bool)

ResultMap 按节点 ID 读取 map 形态的节点结果。

func (*FlowContext) ResultValue 

func (fc *FlowContext) ResultValue(nodeID string, path string) (interface{}, bool)

ResultValue 按节点 ID 和点分路径读取节点结果中的字段。

func (*FlowContext) Results 

func (fc *FlowContext) Results() map[string]interface{}

Results 返回当前节点结果归档的防御性副本。

func (*FlowContext) RunID 

func (fc *FlowContext) RunID() string

func (*FlowContext) RuntimeValue 

func (fc *FlowContext) RuntimeValue(key string) (interface{}, bool)

RuntimeValue 读取框架运行期内部值。

func (*FlowContext) Service added in v0.2.0 

func (fc *FlowContext) Service(service string) *FlowContext

Service 设置本次执行所属的业务服务。

空值和纯空白值按历史协议归一为 default。

func (*FlowContext) Set 

func (fc *FlowContext) Set(key string, val interface{}) *FlowContext

Set 写入节点间共享变量，并返回当前 FlowContext 方便链式调用。

func (*FlowContext) SetEnvelope 

func (fc *FlowContext) SetEnvelope(envelope *Envelope) *FlowContext

SetEnvelope 显式替换本次执行的入口 Envelope。

普通业务入口优先使用 Env、Workspace、Params 等链式 setter。 只有 runtime、测试或从外部协议恢复完整执行现场时才需要直接设置 Envelope。

func (*FlowContext) SetNodeResult 

func (fc *FlowContext) SetNodeResult(nodeID string, result interface{})

SetNodeResult 由 runtime 在节点成功后调用，把节点输出保存到 results[nodeID]。

func (*FlowContext) SetResponse 

func (fc *FlowContext) SetResponse(response interface{})

SetResponse 写入本次 workflow 的业务响应区块。

response 是 FlowContext 快照的一部分，不负责承载 trace、runId 等框架诊断信息。

func (*FlowContext) SetResponseValue 

func (fc *FlowContext) SetResponseValue(path string, value interface{})

SetResponseValue 按点分路径写入 response 子字段，适合多个节点协作拼装业务响应。

func (*FlowContext) SetRuntimeValue 

func (fc *FlowContext) SetRuntimeValue(key string, val interface{})

SetRuntimeValue 写入框架运行期内部值。

这块数据不会进入 vars/output，也不会参与模板绑定，适合 plan 在入口执行前 放入 engine、workflow source、scope 等框架依赖。

func (*FlowContext) SetTraceLogger 

func (fc *FlowContext) SetTraceLogger(logger TraceLogger)

SetTraceLogger 由 runtime 在进入 node 前注入；普通节点代码应使用 Log/Info/Error 写日志。

func (*FlowContext) SetValue 

func (fc *FlowContext) SetValue(key string, val interface{})

SetValue 写入节点间共享变量。key 为空时不会特殊处理，调用方应在节点配置或 node 内保证命名有效。

func (*FlowContext) SetVar 

func (fc *FlowContext) SetVar(key string, val interface{})

SetVar 是 SetValue 的语义化别名，用于节点写入运行中临时变量。

func (*FlowContext) SetWorkflowID 

func (fc *FlowContext) SetWorkflowID(workflowID string) *FlowContext

SetWorkflowID 设置本次执行要加载的 workflow ID。

workflowId 可以是请求选择链路的显式输入；单服务单链路服务也可以在启动期配置 默认 workflow，由 runtime 在执行前写回上下文。

func (*FlowContext) SetWorkflowVersion 

func (fc *FlowContext) SetWorkflowVersion(version string) *FlowContext

SetWorkflowVersion 设置本次执行要加载的 workflow 版本。

version 为空时，是否可以解析为最新发布版本由当前 runtime 的 workflow source 决定。 单服务单链路服务可以在启动期配置默认版本。

func (*FlowContext) Trace 

func (fc *FlowContext) Trace(message string, fields ...interface{})

func (*FlowContext) TraceContext 

func (fc *FlowContext) TraceContext() TraceContext

func (*FlowContext) TraceID 

func (fc *FlowContext) TraceID() string

func (*FlowContext) Tracef 

func (fc *FlowContext) Tracef(format string, args ...interface{})

func (*FlowContext) ValueOf 

func (fc *FlowContext) ValueOf(key string) (interface{}, bool)

ValueOf 读取节点间共享变量。

func (*FlowContext) Values 

func (fc *FlowContext) Values() map[string]interface{}

Values 返回当前 values 的浅拷贝。map 本身不会被调用方改回去，但其中嵌套对象仍然共享引用。

func (*FlowContext) Var 

func (fc *FlowContext) Var(path string) (interface{}, bool)

Var 从运行中变量中按点分路径读取值。

func (*FlowContext) VarString 

func (fc *FlowContext) VarString(path string) (string, bool)

VarString 读取运行中变量并转换为字符串。

func (*FlowContext) Warn 

func (fc *FlowContext) Warn(message string, fields ...interface{})

func (*FlowContext) Warnf 

func (fc *FlowContext) Warnf(format string, args ...interface{})

func (*FlowContext) WithContext 

func (fc *FlowContext) WithContext(parent context.Context) *FlowContext

WithContext 替换内嵌 context，同时保留 workflow 执行状态。

runtime 会在节点每次尝试执行前用它注入取消和超时边界。 节点实现应直接使用 FlowContext 的 context 方法，而不是再额外携带第二个 context。

func (*FlowContext) WorkflowID 

func (fc *FlowContext) WorkflowID() string

func (*FlowContext) WorkflowVersion 

func (fc *FlowContext) WorkflowVersion() string

func (*FlowContext) Workspace 

func (fc *FlowContext) Workspace(workspace string) *FlowContext

Workspace 设置本次执行所属空间。

workspace 是多空间隔离边界，也是 worker 拉取发布 workflow 时参与匹配的字段。

type LogEntry 

type LogEntry struct {
	Level   LogLevel               `json:"level"`
	Message string                 `json:"message"`
	Fields  map[string]interface{} `json:"fields,omitempty"`
}

LogEntry 是节点通过 FlowContext 写出的运行日志。 trace 打开时它会进入本次执行的内存收集器；是否落库由后续 collector 节点决定。

type LogField 

type LogField struct {
	Key   string
	Value interface{}
}

LogField 让节点代码可以用轻量写法添加结构化字段，不必每次手写 map。

func Field 

func Field(key string, value interface{}) LogField

type LogLevel 

type LogLevel string

LogLevel 由框架定义，避免节点实现直接依赖某个具体日志库。 

const (
	LogLevelTrace LogLevel = "trace"
	LogLevelDebug LogLevel = "debug"
	LogLevelInfo  LogLevel = "info"
	LogLevelWarn  LogLevel = "warn"
	LogLevelError LogLevel = "error"
)

type NodeResult 

type NodeResult struct {
	Pass    bool                   `json:"pass"`
	Output  map[string]interface{} `json:"output"`
	Message string                 `json:"message"`
}

NodeResult 是节点返回值的可选框架包装。 普通节点可以直接返回任意值；只有需要表达业务拒绝时才返回 NodeResult。 Pass=false 会被 runtime 解释为节点主动拒绝，并进入 retry/onError 策略。

func Fail 

func Fail(message string, output ...map[string]interface{}) *NodeResult

func Success 

func Success(output ...map[string]interface{}) *NodeResult

Success 和 Fail 是节点作者的便利函数。

func SuccessValue 

func SuccessValue(key string, value interface{}) *NodeResult

type TraceContext 

type TraceContext struct {
	Env             string    `json:"env"`
	Workspace       string    `json:"workspace"`
	TraceID         string    `json:"traceId"`
	RunID           string    `json:"runId"`
	ParentRunID     string    `json:"parentRunId"`
	WorkflowID      string    `json:"workflowId"`
	WorkflowVersion string    `json:"workflowVersion"`
	Attempt         int       `json:"attempt"`
	CreatedAt       time.Time `json:"createdAt"`
}

TraceContext 是 runtime 事件、节点日志和收集节点共享的链路身份。 它只描述“这次执行是谁”，不绑定存储；最终可以写 MySQL、ClickHouse，也可以只在内存里随执行结束释放。

type TraceLogger 

type TraceLogger interface {
	Log(ctx context.Context, fc *FlowContext, entry LogEntry)
}

TraceLogger 接收 FlowContext 发出的节点日志。

type WorkflowResult 

type WorkflowResult struct {
	TraceID string                 `json:"traceId"`
	RunID   string                 `json:"runId"`
	Success bool                   `json:"success"`
	Output  map[string]interface{} `json:"output"`
	Error   error                  `json:"-"`
}

WorkflowResult 是 runtime 对调用方返回的稳定摘要。 节点级诊断留在 trace/report 中，避免普通执行响应过重。

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.