Design and Implementation of a Code‑less Distributed Task Orchestration Platform for XES

The author, a member of the 4S tool team at Xueersi Online School, describes the challenges of traditional development workflows—frequent requirement changes, lengthy dev‑ops approvals, and heterogeneous code styles that hinder understanding and reuse.

To address these issues, the article proposes turning static compiled code (e.g., Go binaries) into a Code‑less, drag‑and‑drop orchestration model where business processes are represented as directed‑acyclic graphs (DAGs) that can be dynamically configured.

The core concept is to model a workflow as a state machine consisting of Nodes (tasks), Edges (conditional branches), and Messages (input/output payloads). Nodes encapsulate business functions, Edges encode boolean expressions similar to switch‑case logic, and Messages carry context, TTL, and tracing information.

The system architecture is divided into three modules: a Scheduler Engine, a Trigger component, and Business Workers. Workers implement SDK‑provided interfaces, the Scheduler handles dispatch and load‑balancing, and the Trigger initiates state‑machine execution. Etcd is used for distributed configuration and strong consistency.

Key data structures are illustrated with Go code:

type PlayBook struct {
    lock sync.Mutex `json:"-"`
    ttl ttl `json:"-"`
    ttlEnable bool `json:"-"`
    startAt string `json:"-"`
    validated bool `json:"-"`
    Id int
    States map[string]*Node
}

type Node struct {
    NodeName string
    NodeCode string
    Description string
    EdgesOfCondition edges
    Task *Task
}

type Edge struct {
    Description string
    NextNode string
    Express string
    Priority int
}

type Message struct {
    Meta MetaImpl `json:"Meta"`
    Task TaskImpl `json:"Task"`
    Context ContextImpl `json:"Context"`
}

The expression parser converts ASL‑style strings into executable conditions:

func (e *Edge) ParseAslExpress(message *Message) (result []*dsl.Expression, token dsl.Token) {
    if strings.Contains(e.Express, string(dsl.And)) {
        token = dsl.And
    } else if strings.Contains(e.Express, string(dsl.Or)) {
        token = dsl.Or
    } else {
        token = dsl.Default
    }
    conditions := strings.Split(e.Express, string(token))
    for _, condition := range conditions {
        var params []interface{}
        res := reg.FindAllString(condition, -1)
        for i, r := range res {
            condition = strings.Replace(condition, r, fmt.Sprintf("[param%v]", i), -1)
            value := jsonutil.GetKey(r, message.Context.GetLast(), message.Context.GetStore())
            params = append(params, value)
        }
        result = append(result, &dsl.Expression{Condition: condition, Params: params})
    }
    return
}

Two practical examples are provided. The first demonstrates a simple arithmetic workflow (ADD, SUB, MUL, DIV) using a custom worker SDK. The worker registration and event loop are shown:

package main
import (
    "context"
    "errors"
    logger "git.100tal.com/wangxiao_go_lib/xesLogger"
    "git.100tal.com/wangxiao_xuefu_rpa/xesFlow/core"
    "git.100tal.com/wangxiao_xuefu_rpa/xesFlow/pkg/workersdk"
    "git.100tal.com/wangxiao_xuefu_rpa/xesFlow/test/worker/task"
)
var TaskRegistry = make(map[string]Task)

type Task interface { Do(input interface{}) (output interface{}, err error) }

func init() {
    TaskRegistry["ADD"] = &task.AddTask{}
    TaskRegistry["SUB"] = &task.SubtractionTask{}
    TaskRegistry["DIV"] = &task.DivisionTask{}
    TaskRegistry["MUL"] = &task.MultiplyTask{}
}

func handler(msg *core.Message) (interface{}, error) {
    taskName := msg.Task.GetXrm().GetTaskName()
    t, ok := TaskRegistry[taskName]
    if !ok { return nil, errors.New("task not exist") }
    return t.Do(msg.Task.GetInput())
}

func main() {
    worker := workersdk.NewWorker(context.Background(), &workersdk.ConfigWorker{XesFlowServer:"http://127.0.0.1:8080", App:"my_test_worker", Token:"123456", HandlerFunc:handler, ConsumerCount:50})
    worker.IgnoreInfoEvent()
    worker.IgnoreSrvEvent()
    eventChan, err := worker.Register()
    if err != nil { logger.E("err happens", err) }
    var count int
    for {
        e := <-eventChan
        count++
        logger.I("main", "receiving server event: type:[%v], content:[%v], trace_id:%v count:%v", e.Type, e.Content, e.Id, count)
    }
}

The workflow can be triggered via SDK or plain HTTP:

var cli = triggersdk.NewHttpClient("http://api.xueersi.com/orchestration/")
cli.ExecuteStateMachine(&proto.TriggerStateMachineReq{Token:"loop", Namespace:"my_test_worker", StateMachineId:1, Data:3})

curl -H "Content-Type: application/json" -X POST -d '{"token":"ssss","state_machine_id":1,"namespace":"my_test_worker","data":100}' http://api.xueersi.com/orchestration/trigger/execute

The second case, “Fuxi data consumption layer”, shows how to split nodes, design triggers for data cleaning, and assemble messages for downstream processing, illustrated with additional architecture diagrams.

In conclusion, abstracting a project into an orchestrated DAG not only enforces stricter code standards and simplifies maintenance but also provides visualizable, reusable, and highly configurable business logic across the XES ecosystem.