Parallel, sequence and after

ProWorkflows needs a JoobQ Pro key. Without one every Core feature keeps working: the Pro routes of the SDK gateway answer 402 pro_required and the console's Pro routes are not served. Get a licence · Core and Pro · Pro features index

Node	Behaviour
parallel { ... }	Runs its children concurrently; its output is {child: output}. What follows waits for every branch
sequence { ... }	Runs its children in order; its output is {child: output}
after: [step, ...] on a step, wait_for or timer	Also waits for these earlier steps, on top of what precedes the node, so a parallel branch can wait for a sibling. An unknown step or a cycle is refused (422)

Crystal

workflow "conf-parallel-#{LANG}" do
  parallel do
    step "p1", One
    step "p2", Two
  end
  step "join", Join, input: {a: ref("steps.p1.output.v"), b: ref("steps.p2.output.v")}
end,

Python

wf("parallel").parallel(lambda b: b.step("p1", value(1)).step("p2", value(2)))
              .step("join", join, input={"a": output("p1", "v"), "b": output("p2", "v")}),

TypeScript

wf('parallel')
  .parallel((b) => { b.step('p1', value(1)).step('p2', value(2)) })
  .step('join', async (i: any) => ({ sum: i.a + i.b }), { input: { a: output('p1', 'v'), b: output('p2', 'v') } }),

Ruby

wf("parallel")
  .parallel { step("p1", ->(_i, _c) { { v: 1 } }); step("p2", ->(_i, _c) { { v: 2 } }) }
  .step("join", ->(i, _ctx) { { sum: i["a"] + i["b"] } }, input: { a: output("p1", "v"), b: output("p2", "v") }),

Go

wf("parallel").
	Parallel(func(b *joobq.WorkflowBlueprint[*App]) { b.Step("p1", value(1)).Step("p2", value(2)) }).
	Step("join", func(in map[string]any, ec *EC) (any, error) {
		return map[string]any{"sum": num(in["a"]) + num(in["b"])}, nil
	}, joobq.WithInput[*App](map[string]any{"a": joobq.Output("p1", "v"), "b": joobq.Output("p2", "v")})),

Rust

wf("parallel")
    .parallel(|b| b.step("p1", |_i: Value, _c: &Ctx| Ok(json!({"v": 1}))).step("p2", |_i: Value, _c: &Ctx| Ok(json!({"v": 2}))))
    .step_with("join", |i: Value, _ctx: &Ctx| Ok(json!({"sum": i["a"].as_i64().unwrap_or(0) + i["b"].as_i64().unwrap_or(0)})),
        StepOptions { input: json!({"a": output("p1", Some("v")), "b": output("p2", Some("v"))}), ..Default::default() }),

Java

wf("parallel")
    .parallel(b -> b.step("p1", (i, ctx) -> map("v", 1)).step("p2", (i, ctx) -> map("v", 2)))
    .step("join", (i, ctx) -> map("sum", num(m(i).get("a")) + num(m(i).get("b"))),
        StepOptions.input(map("a", DSL.output("p1", "v"), "b", DSL.output("p2", "v")))),

PHP

wf('parallel')
    ->parallel(fn ($b) => $b->step('p1', fn () => ['v' => 1])->step('p2', fn () => ['v' => 2]))
    ->step('join', fn ($i, $ctx) => ['sum' => $i['a'] + $i['b']], input: ['a' => DSL::output('p1', 'v'), 'b' => DSL::output('p2', 'v')]),

.NET

Wf("parallel")
    .Parallel(b => b.Step("p1", (i, c) => Task.FromResult<object?>(new { v = 1 })).Step("p2", (i, c) => Task.FromResult<object?>(new { v = 2 })))
    .Step("join", (i, ctx) => Task.FromResult<object?>(new { sum = Num(i?["a"]) + Num(i?["b"]) }), input: new { a = Dsl.Output("p1", "v"), b = Dsl.Output("p2", "v") }),

Tested against a real server in every language shown.

after:

In this blueprint a2 is a branch of the same parallel group as a1, but starts only when a1 has finished:

Crystal

workflow "conf-after-#{LANG}" do
  parallel do
    step "a1", Timed
    step "a2", Timed, after: ["a1"]
  end
end,

Python

wf("after").parallel(lambda b: b.step("a1", timed).step("a2", timed, after=["a1"])),

TypeScript

wf('after').parallel((b) => { b.step('a1', timed).step('a2', timed, { after: ['a1'] }) }),

Ruby

wf("after").parallel { step("a1", TIMED); step("a2", TIMED, after: ["a1"]) },

Go

wf("after").Parallel(func(b *joobq.WorkflowBlueprint[*App]) {
	b.Step("a1", timed).Step("a2", timed, joobq.WithAfter[*App]("a1"))
}),

Rust

wf("after").parallel(|b| b.step("a1", timed).step_with("a2", timed, StepOptions { after: vec!["a1".into()], ..Default::default() })),

Java

wf("after").parallel(b -> b.step("a1", Worker::timed).step("a2", Worker::timed, StepOptions.<App>none().after("a1"))),

PHP

wf('after')->parallel(fn ($b) => $b->step('a1', 'timed')->step('a2', 'timed', after: ['a1'])),

.NET

Wf("after").Parallel(b => b.Step("a1", Timed).Step("a2", Timed, after: new[] { "a1" })),

Tested against a real server in every language shown.

Each branch of a parallel group is a step task of its own and needs a free worker slot to overlap with the others, as on a Crystal worker.