117 lines
4.1 KiB
Go
117 lines
4.1 KiB
Go
// workflow-query 发布有限等待流程，按精确业务键和固定版本分页，并按控制版本结清示例实例。
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"time"
|
|
|
|
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
|
)
|
|
|
|
func main() {
|
|
if err := run(); err != nil {
|
|
fmt.Fprintln(os.Stderr, err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
func run() error {
|
|
raw, err := os.ReadFile(os.Getenv("ORBIT_CREDENTIALS_FILE"))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var credentials struct {
|
|
Token string `json:"token"`
|
|
}
|
|
if json.Unmarshal(raw, &credentials) != nil || credentials.Token == "" {
|
|
return errors.New("invalid private credentials file")
|
|
}
|
|
client, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: credentials.Token})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
operation, err := orbit.NewWorkerMessageID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
spec := orbit.WorkflowSpecification{SchemaVersion: 1, InputType: orbit.WorkflowType{Kind: "NULL"}, TimeoutMS: 120000, FailurePolicy: "REVIEW", Body: orbit.WorkflowBlock{
|
|
Steps: []orbit.WorkflowStep{{Key: "receipt", Kind: "SIGNAL", Signal: &orbit.WorkflowSignal{Name: "object.ready", PayloadVersion: "v1", Correlation: orbit.WorkflowValue{Kind: "LITERAL", Literal: json.RawMessage(`"` + operation + `"`)}, PayloadType: orbit.WorkflowType{Kind: "BOOLEAN"}, TimeoutMS: 120000}}},
|
|
OutputType: orbit.WorkflowType{Kind: "BOOLEAN"}, Output: orbit.WorkflowRef("receipt"),
|
|
}}
|
|
d, err := client.CreateWorkflowDefinition(ctx, operation, orbit.CreateWorkflowDefinitionRequest{Name: "query." + operation})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err = client.PublishWorkflowVersion(ctx, d.Definition.ID, operation, orbit.PublishWorkflowVersionRequest{Version: "v1", Specification: spec, Reason: "publish query example"}); err != nil {
|
|
return err
|
|
}
|
|
key := "对象 /+% é " + operation + " "
|
|
var instances []orbit.WorkflowInstance
|
|
for i, business := range []string{key, key, key[:len(key)-1]} {
|
|
out, err := client.StartWorkflow(ctx, fmt.Sprintf("%s-%d", operation, i), orbit.StartWorkflowRequest{DefinitionID: d.Definition.ID, Version: "v1", Input: json.RawMessage(`null`), BusinessKey: business})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
instances = append(instances, out.Instance)
|
|
}
|
|
from, before := instances[0].CreatedAt, instances[2].CreatedAt.Add(time.Microsecond)
|
|
query := orbit.ListWorkflowsRequest{DefinitionID: d.Definition.ID, Version: "v1", BusinessKey: key, State: "WAITING", CreatedFrom: &from, CreatedBefore: &before, Limit: 1}
|
|
seen := map[string]bool{}
|
|
pages := 0
|
|
for {
|
|
page, err := client.ListWorkflows(ctx, query)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
pages++
|
|
for _, v := range page.Items {
|
|
if v.BusinessKey != key || v.DefinitionID != d.Definition.ID || v.Version != "v1" || seen[v.ID] {
|
|
return errors.New("workflow query changed filters or repeated an item")
|
|
}
|
|
seen[v.ID] = true
|
|
}
|
|
if page.NextCursor == "" {
|
|
break
|
|
}
|
|
query.Cursor = page.NextCursor
|
|
}
|
|
if len(seen) != 2 || pages != 2 || !seen[instances[0].ID] || !seen[instances[1].ID] {
|
|
return errors.New("expected exactly two matching workflow instances")
|
|
}
|
|
fmt.Printf("workflow_query_verified=true matched=%d pages=%d exact_trailing_space=true\n", len(seen), pages)
|
|
// 取消记录保留，固定定义版本和审计也保留；失败时不要直接删除运行记录。
|
|
for _, v := range instances {
|
|
current, err := client.GetWorkflow(ctx, v.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err = client.ControlWorkflow(ctx, v.ID, operation+"-cleanup", orbit.WorkflowControlRequest{Operation: "CANCEL", ExpectedControlRevision: current.ControlRevision, Reason: "query example cleanup"}); err != nil {
|
|
return err
|
|
}
|
|
for {
|
|
current, err = client.GetWorkflow(ctx, v.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if current.State == "CANCELLED" {
|
|
break
|
|
}
|
|
if current.State == "NEEDS_REVIEW" {
|
|
return errors.New("cleanup requires review")
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-time.After(50 * time.Millisecond):
|
|
}
|
|
}
|
|
}
|
|
fmt.Println("workflow_query_cleanup=CANCELLED count=3")
|
|
return nil
|
|
}
|