62
// In steady state (i.e. absence of ongoing updates or signals) the wf should only have
63
// a single wft in the history.
64
>
func Workflow(ctx workflow.Context, unsafeWorkflowVersionGetter func() DeploymentWorkflowVersion, unsafeMaxVersion func() int, args *deploymentspb.WorkerDeploymentWorkflowArgs) error {
workflow.go
65
>
workflowRunner := &WorkflowRunner{
66
>
WorkerDeploymentWorkflowArgs: args,
67
>
workflowVersion: getWorkflowVersion(ctx, unsafeWorkflowVersionGetter),
68
>
a: nil,
69
>
logger: sdklog.With(workflow.GetLogger(ctx), "wf-namespace", args.NamespaceName),
70
>
metrics: workflow.GetMetricsHandler(ctx).WithTags(map[string]string{"namespace": args.NamespaceName}),
71
>
lock: workflow.NewMutex(ctx),
72
>
unsafeMaxVersion: unsafeMaxVersion,
73
>
signalHandler: &SignalHandler{
74
>
signalSelector: workflow.NewSelector(ctx),
75
>
},
76
>
}
77
>
78
>
return workflowRunner.run(ctx)
79
>
}
80
81
>
func getWorkflowVersion(ctx workflow.Context, unsafeWorkflowVersionGetter func() DeploymentWorkflowVersion) DeploymentWorkflowVersion {
workflow.go
82
>
if workflow.GetVersion(ctx, "workflowVersionAdded", workflow.DefaultVersion, 0) >= 0 {
84
>
err := workflow.MutableSideEffect(ctx, "workflowVersion",
85
>
func(_ workflow.Context) any { return unsafeWorkflowVersionGetter() },
86
func(a, b any) bool { return a == b }).
87
Get(&ver)
89
>
return ver
90
>
}
91
}
92
return 0