plan

package module
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: 26 Imported by: 0

Documentation

Overview

Package plan 是 FlowLink 的运行编排入口。

宿主服务应在启动时调用 ConfigureFromConfig 或 ConfigureFromFile 安装运行环境。 需要完全手工组装依赖时，也可以使用底层的 Configure(RuntimeOptions)。 单服务单链路场景下，每次请求只创建 core.FlowContext 并写入请求参数， 最后调用 Execute；多 workflow 服务才需要在请求上下文中补充 workflow 引用， 或者直接调用 ExecuteRef。

更底层的包仍然保持单一职责：node 负责节点注册，validation/schema 负责校验， flowlink-plan/source 负责加载已发布 Workflow，flowlink-plan/planner 负责把 Workflow 转成 Plan，runtime 负责执行 Plan。plan 根包把这些能力串成业务 worker 与控制面都可以复用的稳定入口。

Index

Constants

View Source 
const (
	// SourceTypeAdmin 表示直接通过 flowlink-admin worker pull 接口按需拉取 workflow。
	SourceTypeAdmin = source.SourceTypeAdmin
	// SourceTypePollingAdmin 表示在 admin 拉取之上启用本地 bootstrap 轮询缓存。
	SourceTypePollingAdmin = source.SourceTypePollingAdmin
	// SourceTypeMemory 表示从内存中的已发布 workflow 集合读取。
	SourceTypeMemory = source.SourceTypeMemory
)
View Source 
const (
	// EnvRuntimeConfigPath 是 ConfigureFromFile 默认读取的配置路径环境变量。
	EnvRuntimeConfigPath = configfile.EnvPath
)

Variables

View Source 
var (
	// ErrFlowContextRequired 表示静态入口缺少本次执行的 FlowContext。
	ErrFlowContextRequired = errors.New("flow context is required")
	// ErrFlowContextAssemblyRequired 表示 FlowContext 中没有设置 Assembly。
	ErrFlowContextAssemblyRequired = errors.New("flow context assembly is required")
)
View Source 
var (
	// ErrWorkflowSourceRequired 表示发布 workflow 执行缺少加载来源。
	ErrWorkflowSourceRequired = source.ErrWorkflowSourceRequired
	// ErrWorkerRequired 表示 plan worker 调用缺少可用 Worker。
	ErrWorkerRequired = source.ErrWorkerRequired
	// ErrRuntimeRequired 表示 plan 静态执行入口缺少启动期 Runtime。
	ErrRuntimeRequired = errors.New("flowlink runtime is required")
	// ErrAdminSourceRequired 表示 AdminSourceConfig 缺少控制面地址。
	ErrAdminSourceRequired = source.ErrAdminSourceRequired
	// ErrWorkerScopeRequired 表示 worker scope 缺少 env 或 workspace 信息。
	ErrWorkerScopeRequired = source.ErrWorkerScopeRequired
	// ErrWorkerScopeMismatch 表示启动期 scope 与请求 FlowContext 中的 scope 不一致。
	ErrWorkerScopeMismatch = source.ErrWorkerScopeMismatch
	// ErrPublishedWorkflowRefRequired 表示缺少发布 workflow 引用。
	ErrPublishedWorkflowRefRequired = source.ErrPublishedWorkflowRefRequired
	// ErrPublishedWorkflowRefMismatch 表示启动期默认 workflow 与请求中的 workflow 引用不一致。
	ErrPublishedWorkflowRefMismatch = source.ErrPublishedWorkflowRefMismatch
	// ErrPublishedWorkflowResolverNeeded 表示处理消息时缺少 workflow 解析器。
	ErrPublishedWorkflowResolverNeeded = source.ErrPublishedWorkflowResolverNeeded
	// ErrWorkflowIDRequired 表示发布 workflow 引用缺少 workflowId。
	ErrWorkflowIDRequired = source.ErrWorkflowIDRequired
	// ErrWorkerHTTPClientRequired 表示轮询加载器缺少 HTTP client。
	ErrWorkerHTTPClientRequired = source.ErrWorkerHTTPClientRequired
	// ErrWorkflowSourceTypeUnsupported 表示配置中的 workflow 来源类型没有注册工厂。
	ErrWorkflowSourceTypeUnsupported = source.ErrWorkflowSourceTypeUnsupported
)
View Source 
var (
	// ErrSwitchResolverRequired 表示 workflow 使用了 Switch 绑定，但执行环境没有解析器。
	ErrSwitchResolverRequired = errors.New("switch resolver is required")
	// ErrSwitchNameRequired 表示 tracing Switch 绑定没有配置开关名。
	ErrSwitchNameRequired = errors.New("switch name is required")
)
View Source 
var ErrAssemblyRequired = errors.New("flowlink assembly is required")

ErrAssemblyRequired 表示 plan 调用缺少可用 Assembly。

View Source 
var ErrDebugBaseURLRequired = errors.New("debug base url is required")

ErrDebugBaseURLRequired 表示调试 HTTP 客户端缺少业务服务地址。

View Source 
var ErrDebugWorkflowRequired = errors.New("debug workflow is required")

ErrDebugWorkflowRequired 表示业务调试请求缺少 workflow 定义。

Functions

func Configure 

func Configure(options RuntimeOptions) error

Configure 配置 plan.Execute 使用的默认 Runtime。

宿主服务应在启动时调用一次，把配置系统解析出的 Source 或 Assembly 传进来。 请求处理时只需要构造 core.FlowContext 并调用 Execute。

func ConfigureFromConfig 

func ConfigureFromConfig(ctx context.Context, config Config) error

ConfigureFromConfig 使用 FlowLink 配置创建并安装默认 Runtime。

宿主服务已经有自己的配置加载体系时，推荐把配置映射成 plan.Config 后调用本函数。

func ConfigureFromFile 

func ConfigureFromFile(ctx context.Context, path string, options ...ConfigLoadOption) error

ConfigureFromFile 从配置文件读取 FlowLink 配置，并安装默认 Runtime。

path 可以是具体文件，也可以是配置根目录。目录模式会沿用 Switch 的环境约定： APP_ENV 选择子目录，APP_ENV 为空时默认 dev。path 为空时优先读取 FLOWLINK_CONFIG， 仍为空则使用当前工作目录下的 configs。

func Debug 

func Debug(ctx context.Context, req *workerdebug.DebugRequest) (*workerdebug.DebugResponse, error)

Debug 使用启动期默认 Runtime 或默认 Assembly 执行业务调试请求。

调试请求中的 workflow 可以是编辑器当前定义或已保存版本；真正执行发生在当前业务进程， 因此会使用业务服务自己注册的节点实现。

func Execute 

func Execute(fc *core.FlowContext) (*core.WorkflowResult, error)

Execute 加载并执行 FlowContext 中引用或启动期默认的已发布 workflow。

单服务单链路场景推荐在启动配置里写入 scope 和 workflow，调用方每次请求只写 params。 多链路场景可以在同一个 FlowContext 中写入 workflowId、可选 workflowVersion， 或直接调用 ExecuteRef。Assembly、source、admin 地址等运行依赖通常在服务启动时配好。 从加载、校验、planning 到 runtime 执行，之后都属于框架职责。

func ExecutePlan 

func ExecutePlan(fc *core.FlowContext, plan *core.Plan) (*core.WorkflowResult, error)

ExecutePlan 使用 FlowContext 上的 Assembly 执行预先构建好的 plan。

普通调用方通常使用 Execute 或 ExecuteWorkflow。 当 worker 按已发布版本缓存编译后的 plan 时，ExecutePlan 更合适。

func ExecuteRef 

func ExecuteRef(fc *core.FlowContext, ref *PublishedWorkflowRef) (*core.WorkflowResult, error)

ExecuteRef 加载并执行指定的已发布 workflow 引用。

与 Execute 不同，workflow 引用由参数传入；runtime 依赖和请求数据仍然来自 FlowContext。

func ExecuteWorkflow 

func ExecuteWorkflow(fc *core.FlowContext, workflow *core.Workflow) (*core.WorkflowResult, error)

ExecuteWorkflow 执行调用方直接提供的 workflow 定义。

这条路径不需要已发布 workflow loader 或 admin 地址，但仍然需要通过 UseAssembly 在 FlowContext 上设置 Assembly。它适合内嵌 workflow、测试或不接 flowlink-admin 的服务。

func NewDebugHandler 

func NewDebugHandler(options DebugHandlerOptions) http.Handler

NewDebugHandler 创建业务服务侧 workflow 调试和节点清单 HTTP Handler。

Handler 只依赖标准库 net/http，业务服务可以挂到 Gin、Echo、原生 ServeMux 或网关之后。

func RegisterNode 

func RegisterNode(name string, ex node.Node) error

RegisterNode 把节点实现注册到默认 Assembly。

name 必须和 workflow JSON 中保存的 node type 保持一致。

func RegisterNodeDefinition 

func RegisterNodeDefinition(def node.Definition) error

RegisterNodeDefinition 把完整节点定义注册到默认 Assembly。

func RegisterNodeFunc 

func RegisterNodeFunc(name string, fn node.Func) error

RegisterNodeFunc 将函数适配成节点并注册到默认 Assembly。

func RegisterNodeFuncWithMetadata 

func RegisterNodeFuncWithMetadata(metadata node.Metadata, fn node.Func) error

RegisterNodeFuncWithMetadata 将函数及其编排元信息注册到默认 Assembly。

func RegisterNodeWithMetadata 

func RegisterNodeWithMetadata(metadata node.Metadata, ex node.Node) error

RegisterNodeWithMetadata 把节点实现及编排元信息注册到默认 Assembly。

func RegisterWorkflowSourceFactory 

func RegisterWorkflowSourceFactory(sourceType string, factory WorkflowSourceFactory) error

RegisterWorkflowSourceFactory 注册自定义 workflow 来源类型。

func SetWorkflowRef 

func SetWorkflowRef(fc *core.FlowContext, workflowID string, version string) error

SetWorkflowRef 写入本次请求要执行的已发布 workflow 引用。

它是 FlowContext 链式 setter 的函数式便利封装。 当配置的 workflow 来源支持解析最新发布版本时，version 可以为空。

func UseAssembly 

func UseAssembly(fc *core.FlowContext, assembly *Assembly) error

UseAssembly 把本次 FlowContext 执行使用的 Assembly 写入 runtime 区。

ExecuteWorkflow、ExecutePlan 和已发布 workflow 执行都需要 Assembly。 Assembly 刻意存放在 FlowContext runtime 数据中，使宿主服务可以在同一进程中按请求使用不同 Assembly。

func UseRuntime 

func UseRuntime(fc *core.FlowContext, options RuntimeOptions) error

UseRuntime 把本次执行需要的 FlowLink 运行依赖写入指定 FlowContext。

它不会修改包级全局状态。Execute 和 ExecuteRef 后续只会从同一个 FlowContext 读取这些依赖。如果只需要 Assembly，例如 ExecuteWorkflow 或 ExecutePlan 场景， 调用方可以直接使用 UseAssembly。

func UseSwitchResolver 

func UseSwitchResolver(fc *core.FlowContext, resolver SwitchResolver) error

UseSwitchResolver 把本次请求使用的外部开关解析器写入 FlowContext。

func UseWorkflowSource 

func UseWorkflowSource(fc *core.FlowContext, source WorkflowSource, scope ServiceScope) error

UseWorkflowSource 把 workflow 来源及其允许访问的范围写入 FlowContext。

这是高级覆盖入口，普通业务服务优先使用启动期 ConfigureFromConfig 或 ConfigureFromFile。 scope 会补齐空的 envelope env/workspace；如果 envelope 已经带有不同值， 会立即返回 ErrWorkerScopeMismatch，避免用一个 scope 拉 workflow、再用另一个 scope 执行。

Types

type AdminSourceConfig 

type AdminSourceConfig = source.AdminSourceConfig

type Assembly 

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

Assembly 是 plan 层复用的运行组件装配体。

它负责把节点注册、workflow 校验、planning 和 runtime 执行连接起来。 底层能力仍然保留在独立包中，调用方需要更窄集成点时可以直接使用对应包。

func DefaultAssembly 

func DefaultAssembly() (*Assembly, error)

DefaultAssembly 返回包级节点注册 helper 使用的进程内默认 Assembly。

它主要服务于简单服务和示例。更复杂的宿主可以直接调用 NewAssembly， 再通过 Configure 把 Assembly 放进启动期 Runtime。

func NewAssembly 

func NewAssembly(opts *AssemblyOptions) (*Assembly, error)

NewAssembly 创建 Assembly，并为缺失的 AssemblyOptions 填充默认实现。

func (*Assembly) Build 

func (e *Assembly) Build(ctx context.Context, workflow *core.Workflow) (*core.Plan, error)

Build 校验 workflow，并将其转换成 runtime Plan。

Build 是执行前的标准边界：调用方会在任何节点执行前拿到校验失败， runtime 则接收 planner 生成的结构坐标，而不是重新解释可编辑 workflow 树。

func (*Assembly) CanonicalizeWorkflow 

func (e *Assembly) CanonicalizeWorkflow(workflow *core.Workflow) *core.Workflow

CanonicalizeWorkflow 返回 workflow 的防御性副本，并归一旧版本协议字段。

校验、planning 和执行不应该修改调用方或缓存持有的对象， 尤其是 worker 跨请求复用已发布 workflow 定义时。

func (*Assembly) Debug 

func (e *Assembly) Debug(ctx context.Context, req *workerdebug.DebugRequest) (*workerdebug.DebugResponse, error)

Debug 执行请求体中的 workflow 定义，并返回 result、plan 摘要和 trace 事件。

func (*Assembly) NodeRegistry 

func (e *Assembly) NodeRegistry() *node.Registry

NodeRegistry 返回当前 Assembly 使用的节点注册中心。

func (*Assembly) Planner 

func (e *Assembly) Planner() *flowplanner.Planner

Planner 返回当前 Assembly 用于构建 runtime plan 的 planner。

func (*Assembly) RegisterNode 

func (e *Assembly) RegisterNode(name string, ex node.Node) error

RegisterNode 在当前 Assembly 上安装节点实现。

底层注册中心会拒绝重复注册，保证一个 node type 只有一个明确实现。

func (*Assembly) RegisterNodeDefinition 

func (e *Assembly) RegisterNodeDefinition(def node.Definition) error

RegisterNodeDefinition 在当前 Assembly 上安装完整节点定义。

func (*Assembly) RegisterNodeFunc 

func (e *Assembly) RegisterNodeFunc(name string, fn node.Func) error

RegisterNodeFunc 将函数适配成节点并注册到当前 Assembly。

func (*Assembly) RegisterNodeFuncWithMetadata 

func (e *Assembly) RegisterNodeFuncWithMetadata(metadata node.Metadata, fn node.Func) error

RegisterNodeFuncWithMetadata 将函数及其编排元信息注册到当前 Assembly。

func (*Assembly) RegisterNodeWithMetadata 

func (e *Assembly) RegisterNodeWithMetadata(metadata node.Metadata, ex node.Node) error

RegisterNodeWithMetadata 在当前 Assembly 上安装节点实现及编排元信息。

func (*Assembly) Run 

func (e *Assembly) Run(ctx context.Context, workflow *core.Workflow, envelope *core.Envelope) (*core.WorkflowResult, error)

Run 校验、planning 并执行一个 workflow 定义。

当调用方不需要单独缓存 plan 时，这是最简单的执行路径。

func (*Assembly) RunPlan 

func (e *Assembly) RunPlan(ctx context.Context, plan *core.Plan, envelope *core.Envelope) (*core.WorkflowResult, error)

RunPlan 执行已经构建好的 Plan。

当 worker 按已发布 workflow 版本缓存 plan，并希望避免每次请求重复构建树时使用它。

func (*Assembly) RunReport 

func (e *Assembly) RunReport(ctx context.Context, workflow *core.Workflow, envelope *core.Envelope) (*runtime.ExecutionReport, error)

RunReport 执行 workflow，并返回 plan、result 和 trace report。

它主要面向控制面调试入口和本地工具，不适合高吞吐业务消息处理路径。

func (*Assembly) Runtime 

func (e *Assembly) Runtime() *runtime.Runtime

Runtime 返回当前 Assembly 使用的 runtime 执行器。

func (*Assembly) SchemaValidator 

func (e *Assembly) SchemaValidator() schema.Validator

SchemaValidator 返回当前 Assembly 使用的节点配置 schema 校验器。

func (*Assembly) SetSwitchResolver 

func (e *Assembly) SetSwitchResolver(resolver SwitchResolver)

SetSwitchResolver 为 Assembly 安装外部开关解析器。

它主要服务控制面调试或低阶 Assembly.Run 调用；普通业务服务更推荐在启动期 RuntimeOptions.SwitchResolver 中传入。

func (*Assembly) SwitchResolver 

func (e *Assembly) SwitchResolver() SwitchResolver

SwitchResolver 返回 Assembly 执行路径使用的外部开关解析器。

func (*Assembly) UseSecretCodec 

func (e *Assembly) UseSecretCodec(codec *workflowsecret.Codec) *Assembly

UseSecretCodec configures startup-time resolution of persisted workflow secretRefs.

func (*Assembly) Validate 

func (e *Assembly) Validate(ctx context.Context, workflow *core.Workflow) []*schema.ValidationError

Validate 在 planning 或执行前校验 workflow。

校验范围包括通用 workflow 协议、节点是否可用、节点配置 schema、数据流引用， 以及框架拥有的节点策略。它返回结构化错误列表，而不是只返回第一个错误。

func (*Assembly) Validator 

func (e *Assembly) Validator() *validation.WorkflowValidator

Validator 返回 New 组装出的 workflow 校验器。

type AssemblyOptions 

type AssemblyOptions struct {
	// NodeRegistry 将 workflow JSON 中的 node type 映射到可执行节点实现。
	NodeRegistry *node.Registry
	// SchemaValidator 在 workflow 进入 planner 前校验节点配置 JSON。
	SchemaValidator schema.Validator
	// Planner 将可编辑 workflow 定义转换为 runtime 执行计划。
	Planner *flowplanner.Planner
	// SwitchResolver 在执行前解析 workflow.tracing.switchBinding；使用 Switch 模式时必填。
	SwitchResolver SwitchResolver
	// SecretCodec 在 planning 前解析 workspace 绑定的 node config secretRefs。
	SecretCodec *workflowsecret.Codec
	// SkipNodeAvailabilityValidation 仅供不执行节点的控制面使用。
	// worker 和普通业务 Assembly 应保持 false，让执行前校验真实注册表。
	SkipNodeAvailabilityValidation bool
}

AssemblyOptions 描述构建 Assembly 时可以替换的组件。

大多数业务服务传 nil 即可使用默认实现。只有在宿主服务需要共享节点注册中心、 使用更严格的 schema 校验器，或定制 planner 但仍保留 plan 公共入口时，才需要设置这些选项。

type Config 

type Config struct {
	// Scope 限定当前业务服务可拉取和执行的已发布 workflow 范围。
	Scope ServiceScope `json:"scope" yaml:"scope"`
	// Workflow 是单服务单链路时的默认 workflow 引用。
	// 配好后，请求 FlowContext 可以只携带 params。
	Workflow PublishedWorkflowRef `json:"workflow" yaml:"workflow"`
	// Source 描述已发布 workflow 的来源。
	Source SourceConfig `json:"source" yaml:"source"`
	// Worker 描述当前业务进程对控制面上报节点能力时使用的身份和心跳周期。
	Worker WorkerConfig `json:"worker" yaml:"worker"`
	// Assembly 允许高级调用方复用自己构造的 Assembly；配置文件不会填充它。
	Assembly *Assembly `json:"-" yaml:"-"`
	// SwitchResolver 允许宿主服务把已初始化的外部开关系统接入 tracing 开关解析。
	// 配置文件不会直接填充接口；workflow 使用 Switch 模式时必须由宿主服务设置。
	SwitchResolver SwitchResolver `json:"-" yaml:"-"`
}

Config 是业务进程启动 FlowLink runtime 所需的框架配置。

这个结构只描述启动期依赖，不包含单次请求的 params、workflowId 或 response。 宿主服务可以自己加载配置后填充它，也可以让 ConfigureFromFile 从 YAML 文件读取。

func LoadConfig 

func LoadConfig(path string, options ...ConfigLoadOption) (*Config, error)

LoadConfig 读取配置文件中的 flowlink 配置段。

本函数复用 switch-components 的环境解析和 YAML 配置解析器，但不使用其全局 ConfigManager，避免 plan 在宿主进程内清空或覆盖业务自己的配置加载状态。

func (*Config) Initial 

func (c *Config) Initial()

Initial 补齐配置默认值。

type ConfigLoadOption 

type ConfigLoadOption = configfile.Option

ConfigLoadOption 调整 FlowLink 文件配置加载行为。

func WithConfigDefaultEnv 

func WithConfigDefaultEnv(env string) ConfigLoadOption

WithConfigDefaultEnv 设置 APP_ENV 为空时使用的默认环境。

func WithConfigFilePattern 

func WithConfigFilePattern(pattern string) ConfigLoadOption

WithConfigFilePattern 设置目录模式下匹配的配置文件名。

type DebugHTTPClient 

type DebugHTTPClient struct {
	BaseURL    string
	Token      string
	HTTPClient *http.Client
}

DebugHTTPClient 是控制台或工具访问业务服务调试入口的轻量客户端。

它只负责按照 workerdebug 协议发送请求和解析统一响应包，不参与 workflow planning 或节点执行。真正执行始终发生在挂载 NewDebugHandler 的业务进程里。

func (*DebugHTTPClient) Execute 

func (c *DebugHTTPClient) Execute(ctx context.Context, req *workerdebug.DebugRequest) (*workerdebug.DebugResponse, error)

Execute 向业务服务发送一次 workflow 调试请求。

func (*DebugHTTPClient) Validate 

func (c *DebugHTTPClient) Validate() error

Validate 校验调试客户端具备最小可用配置。

type DebugHandlerOptions 

type DebugHandlerOptions struct {
	// Assembly 为空时使用进程默认 Runtime 或默认 Assembly。
	Assembly *Assembly
	// Token 必须显式配置，请求必须携带相同的 X-FlowLink-Debug-Token。
	Token string
	// AllowedOrigins 是浏览器请求的显式来源白名单。没有 Origin 的服务间请求不受此项影响。
	AllowedOrigins []string
}

DebugHandlerOptions 描述业务服务暴露调试入口时需要的配置。

type MemorySourceConfig 

type MemorySourceConfig = source.MemorySourceConfig

type MemoryWorkflowLoader 

type MemoryWorkflowLoader = source.MemoryWorkflowLoader

func NewMemoryWorkflowLoader 

func NewMemoryWorkflowLoader() *MemoryWorkflowLoader

NewMemoryWorkflowLoader 创建空的内存 workflow loader。

type PollingSourceConfig 

type PollingSourceConfig = source.PollingSourceConfig

type PollingWorkflowLoader 

type PollingWorkflowLoader = source.PollingWorkflowLoader

func NewPollingWorkflowLoader 

func NewPollingWorkflowLoader(opts *PollingWorkflowLoaderOptions) (*PollingWorkflowLoader, error)

NewPollingWorkflowLoader 创建绑定到单个 worker scope 的轮询 loader。

type PollingWorkflowLoaderOptions 

type PollingWorkflowLoaderOptions = source.PollingWorkflowLoaderOptions

type PollingWorkflowLoaderStatus 

type PollingWorkflowLoaderStatus = source.PollingWorkflowLoaderStatus

type PublishedWorkflowHead 

type PublishedWorkflowHead = source.PublishedWorkflowHead

type PublishedWorkflowRef 

type PublishedWorkflowRef = source.PublishedWorkflowRef

type PublishedWorkflowResolver 

type PublishedWorkflowResolver = source.PublishedWorkflowResolver

func StaticResolver 

func StaticResolver(workflowID string, version string) PublishedWorkflowResolver

StaticResolver 固定返回同一个 workflowId/version，适合单链路 worker。

type ResolverFunc 

type ResolverFunc = source.ResolverFunc

type Runtime 

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

Runtime 是业务服务启动期创建的 FlowLink 执行环境。

它持有进程级依赖：Assembly、已发布 workflow loader，以及可选的服务范围。 请求级数据仍然只放在 core.FlowContext 中。

func NewRuntime 

func NewRuntime(options RuntimeOptions) (*Runtime, error)

NewRuntime 创建一个可显式持有的 Runtime。

如果同一进程需要服务多个互相隔离的 FlowLink 环境，可以保存返回值并直接调用 Runtime.Execute；普通单环境服务使用 Configure 即可。

func NewRuntimeFromConfig 

func NewRuntimeFromConfig(ctx context.Context, config Config) (*Runtime, error)

NewRuntimeFromConfig 使用 FlowLink 配置创建显式 Runtime。

同一进程需要维护多个互相隔离的 FlowLink runtime 时，调用方可以保存返回值并直接调用 Runtime.Execute，而不是使用包级默认 runtime。

func NewRuntimeFromFile 

func NewRuntimeFromFile(ctx context.Context, path string, options ...ConfigLoadOption) (*Runtime, error)

NewRuntimeFromFile 从配置文件读取 FlowLink 配置，并返回显式 Runtime。

func (*Runtime) Debug 

func (r *Runtime) Debug(ctx context.Context, req *workerdebug.DebugRequest) (*workerdebug.DebugResponse, error)

Debug 执行一个调试请求，并返回控制台可直接展示的报告。

func (*Runtime) Execute 

func (r *Runtime) Execute(fc *core.FlowContext) (*core.WorkflowResult, error)

Execute 加载并执行当前 FlowContext 引用或启动期默认的已发布 workflow。

func (*Runtime) ExecutePlan 

func (r *Runtime) ExecutePlan(fc *core.FlowContext, plan *core.Plan) (*core.WorkflowResult, error)

ExecutePlan 执行预先构建好的 plan。

func (*Runtime) ExecuteRef 

func (r *Runtime) ExecuteRef(fc *core.FlowContext, ref *PublishedWorkflowRef) (*core.WorkflowResult, error)

ExecuteRef 加载并执行指定的已发布 workflow 引用。

func (*Runtime) ExecuteWorkflow 

func (r *Runtime) ExecuteWorkflow(fc *core.FlowContext, workflow *core.Workflow) (*core.WorkflowResult, error)

ExecuteWorkflow 执行调用方直接提供的 workflow 定义。

type RuntimeOptions 

type RuntimeOptions struct {
	// Assembly 负责节点注册、workflow 校验、planning 和 runtime 执行。
	Assembly *Assembly
	// Scope 限定本次执行允许加载的已发布 workflow 范围。
	Scope ServiceScope
	// Source 是已发布 workflow 来源。
	// 它可以是 AdminSource 返回的 HTTP 来源，也可以是内存、轮询缓存或业务自定义实现。
	Source WorkflowSource
	// Workflow 是单服务单链路时的默认 workflow 引用。
	// 为空时，Execute 要求请求 FlowContext 写入 workflowId。
	Workflow PublishedWorkflowRef
	// SwitchResolver 在执行前把 workflow.tracing.switchBinding 解析成最终 bool。
	// workflow 使用 Switch 模式时必须提供，否则该 workflow 不会执行。
	SwitchResolver SwitchResolver
}

RuntimeOptions 描述启动期 FlowLink 运行环境需要的依赖。

普通业务服务在启动时通过 Configure 传入一次。需要按请求覆盖运行依赖的高级场景， 仍可以继续配合 UseRuntime 使用。

func RuntimeOptionsFromConfig 

func RuntimeOptionsFromConfig(ctx context.Context, config Config) (RuntimeOptions, error)

RuntimeOptionsFromConfig 把配置结构转换成 plan 原生 RuntimeOptions。

公开这个转换点，是为了让高级调用方能在最终 Configure 前补充 Assembly 或替换 Source。

type ServiceScope 

type ServiceScope = source.ServiceScope

type SourceConfig 

type SourceConfig = source.SourceConfig

type SwitchResolveRequest 

type SwitchResolveRequest struct {
	SwitchName string
	Target     TraceSwitchTarget
	Scope      ServiceScope
	Workflow   *core.Workflow
}

SwitchResolveRequest 是 plan 层解析外部开关时传给 resolver 的上下文。

resolver 至少需要 SwitchName；Scope、Workflow 和 Target 用于业务接入更复杂的 开关系统时做隔离、统计或路由。

type SwitchResolver 

type SwitchResolver interface {
	ResolveSwitch(fc *core.FlowContext, request SwitchResolveRequest) (bool, error)
}

SwitchResolver 把外部开关名解析成最终 bool。

plan 只依赖这个抽象，不直接把 Switch SDK 调用写入 engine。Switch 模式要求解析成功， 否则本次 workflow 不会进入 engine。

type SwitchResolverFunc 

type SwitchResolverFunc func(fc *core.FlowContext, request SwitchResolveRequest) (bool, error)

SwitchResolverFunc 将普通函数适配成 SwitchResolver。

func (SwitchResolverFunc) ResolveSwitch 

func (fn SwitchResolverFunc) ResolveSwitch(fc *core.FlowContext, request SwitchResolveRequest) (bool, error)

ResolveSwitch 调用底层函数。

type TraceSwitchTarget 

type TraceSwitchTarget string

TraceSwitchTarget 表示 workflow.tracing 中由外部开关系统控制的追踪项。 

const (
	// TraceSwitchEnabled 对应 workflow.tracing.enabled，统一控制事件、日志和统计。
	TraceSwitchEnabled TraceSwitchTarget = "tracing"
)

type Worker 

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

Worker 是消费消息并执行已发布 workflow 的可复用 helper。

普通请求侧入口优先使用 Configure + Execute。当一个进程绑定固定 scope 和 workflow 来源， 并希望得到 adapter.Handler 形式包装时，Worker 仍然有价值。

func NewWorker 

func NewWorker(assembly *Assembly, source WorkflowSource, scope ServiceScope) (*Worker, error)

NewWorker 创建绑定单个 Assembly、单个 workflow 来源和单个服务范围的 Worker。

func (*Worker) Assembly 

func (w *Worker) Assembly() *Assembly

Assembly 返回当前 Worker 使用的 Assembly。

func (*Worker) ExecutePublished 

func (w *Worker) ExecutePublished(ctx context.Context, ref *PublishedWorkflowRef, envelope *core.Envelope) (*core.WorkflowResult, error)

ExecutePublished 加载并执行一个已发布 workflow 引用。

Worker 会在执行前把自身固定 scope 应用到 envelope。 envelope 中已经存在的字段会由 ServiceScope.ApplyEnvelope 保留。

func (*Worker) HandleMessage 

func (w *Worker) HandleMessage(ctx context.Context, msg adapter.Message, resolver PublishedWorkflowResolver) (*core.WorkflowResult, error)

HandleMessage 从 adapter 消息解析已发布 workflow 引用并执行。

func (*Worker) Handler 

func (w *Worker) Handler(resolver PublishedWorkflowResolver) adapter.Handler

Handler 将 Worker 适配成 adapter.Handler。

返回的 handler 只上报执行错误；如果调用方需要 WorkflowResult，应直接调用 HandleMessage。

func (*Worker) LoadPublished 

func (w *Worker) LoadPublished(ctx context.Context, ref *PublishedWorkflowRef) (*core.Workflow, error)

LoadPublished 加载 workflow，校验它属于当前 Worker scope，并在执行前完成 workflow 校验。

func (*Worker) Scope 

func (w *Worker) Scope() ServiceScope

Scope 返回当前 Worker 允许执行的服务范围。

func (*Worker) Source 

func (w *Worker) Source() WorkflowSource

Source 返回当前 Worker 使用的 workflow 来源。

type WorkerBootstrap 

type WorkerBootstrap = source.WorkerBootstrap

type WorkerConfig added in v0.2.0 

type WorkerConfig struct {
	Version                  string `json:"version" yaml:"version"`
	InstanceID               string `json:"instanceId" yaml:"instanceId"`
	CapabilityReportInterval string `json:"capabilityReportInterval" yaml:"capabilityReportInterval"`
	DisableCapabilityReport  bool   `json:"disableCapabilityReport" yaml:"disableCapabilityReport"`
}

WorkerConfig 配置业务 Worker 节点能力上报。

func (*WorkerConfig) Initial added in v0.2.0 

func (c *WorkerConfig) Initial()

type WorkerHTTPClient 

type WorkerHTTPClient = source.WorkerHTTPClient

type WorkflowSource 

type WorkflowSource = source.WorkflowSource

func AdminSource 

func AdminSource(baseURL string, token string) WorkflowSource

AdminSource 创建基于 flowlink-admin worker pull 接口的 workflow 来源。

func AdminSourceWithConfig 

func AdminSourceWithConfig(config AdminSourceConfig) WorkflowSource

AdminSourceWithConfig 使用完整配置创建 flowlink-admin workflow 来源。

func BuildWorkflowSource 

func BuildWorkflowSource(ctx context.Context, scope ServiceScope, config SourceConfig) (WorkflowSource, error)

BuildWorkflowSource 根据 SourceConfig 创建已发布 workflow 来源。

type WorkflowSourceFactory 

type WorkflowSourceFactory = source.WorkflowSourceFactory

Source Files

View all Source files

Directories

Path Synopsis
internal
configfile
Package planner 将用户保存的 workflow 定义转换为 runtime 可执行的 plan。
 Package planner 将用户保存的 workflow 定义转换为 runtime 可执行的 plan。
Package source 负责已发布 workflow 的来源协议和默认加载实现。
 Package source 负责已发布 workflow 的来源协议和默认加载实现。
Package switchsdk 把 Switch SDK 适配为 flowlink-plan 的 SwitchResolver。
 Package switchsdk 把 Switch SDK 适配为 flowlink-plan 的 SwitchResolver。

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.