Rust examples: move to SDK 1.44.0 and use the Join helper

examples/rust/Cargo.toml

@@ -6,7 +6,7 @@ publish = false
 rust-version = "1.97"
 
 [dependencies]
-dex-sdk = "=1.31.0"
+dex-sdk = "=1.44.0"
 fastrand = "2.3"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

examples/rust/README.md

@@ -87,7 +87,7 @@ remain split so their orchestration boundaries are visible.
 | Manual recovery | [`ManualRecoveryFlow`](src/patterns/intervention/flow.rs) | Exhausted retry recovery and manual Channel decision |
 | Static parallel Steps | [`StaticParallelStepsFlow`](src/patterns/parallel/parallel_step_flows.rs) | A fixed set of concurrent Step types |
 | Dynamic parallel Steps | [`DynamicParallelStepsFlow`](src/patterns/parallel/parallel_step_flows.rs) | A runtime-sized set of one Step type |
-| Await parallel Steps | [`AwaitParallelStepsFlow`](src/patterns/parallel/parallel_step_flows.rs) | Wait for every worker's Channel message |
+| Await parallel Steps | [`AwaitParallelStepsFlow`](src/patterns/parallel/parallel_step_flows.rs) | Join every worker's result |
 | First-win parallel Steps | [`FirstWinParallelStepsFlow`](src/patterns/parallel/parallel_step_flows.rs) | Keep the first result and cancel siblings |
 | Basic SubFlow fan-out | [`BasicParentFlow`](src/patterns/parallel_subflows/flow.rs) | Wait for half of the SubFlows and cancel the remaining IDs |
 | Long-lived SubFlow parent | [`AdvancedLongLiveParentFlow`](src/patterns/parallel_subflows/flow.rs) | Bounded workers, request Channel, stop Attribute, and SubFlow loop |

examples/rust/src/patterns/parallel/parallel_step_flows.rs

@@ -13,9 +13,10 @@
 // limitations under the License.
 
 use dex_sdk::{
-    Channel, Context, Flow, HandlerResult, PersistenceSchema, Step, StepDecision, StepList,
+    Context, Flow, HandlerResult, Join, JoinRound, PersistenceSchema, Step, StepDecision, StepList,
     StepMovement, Wait,
 };
+use serde::{Deserialize, Serialize};
 use std::{sync::LazyLock, thread, time::Duration};
 
 #[derive(Default)]
@@ -120,7 +121,7 @@ impl Flow for AwaitParallelStepsFlow {
             .and(&self.await_step)
     }
     fn persistence(&self) -> PersistenceSchema {
-        PersistenceSchema::new().channel(&COMPLETE)
+        PersistenceSchema::new().join(&COMPLETE)
     }
 }
 #[derive(Default)]
@@ -130,40 +131,58 @@ impl Step for AwaitInit {
     fn step_type(&self) -> &'static str {
         "InitStep"
     }
-    fn execute(&self, _: &mut Context, count: usize) -> HandlerResult<StepDecision> {
-        let mut movements = vec![StepMovement::to(&Await, count)];
-        movements.extend((0..count).map(|index| StepMovement::to(&AwaitWork, index)));
+    fn execute(&self, context: &mut Context, count: usize) -> HandlerResult<StepDecision> {
+        let branches = (0..count)
+            .map(|index| {
+                Box::new(move |round: &JoinRound| {
+                    StepMovement::to(
+                        &AwaitWork,
+                        WorkItem {
+                            index,
+                            round: round.clone(),
+                        },
+                    )
+                }) as Box<dyn FnOnce(&JoinRound) -> StepMovement>
+            })
+            .collect();
+        let movements =
+            COMPLETE.fork(context, branches, |round| StepMovement::to(&Await, round))?;
         Ok(StepDecision::go_to_many(movements))
     }
 }
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct WorkItem {
+    pub index: usize,
+    pub round: JoinRound,
+}
 #[derive(Default)]
 struct AwaitWork;
 impl Step for AwaitWork {
-    type Input = usize;
+    type Input = WorkItem;
     fn step_type(&self) -> &'static str {
         "DoWork"
     }
-    fn execute(&self, context: &mut Context, _: usize) -> HandlerResult<StepDecision> {
+    fn execute(&self, context: &mut Context, item: WorkItem) -> HandlerResult<StepDecision> {
         thread::sleep(Duration::from_millis(fastrand::u64(50..500)));
-        COMPLETE.publish(context, ())?;
-        Ok(StepDecision::dead_end())
+        COMPLETE.end_branch(context, &item.round, item.index)
     }
 }
 #[derive(Default)]
 struct Await;
 impl Step for Await {
-    type Input = usize;
+    type Input = JoinRound;
     fn step_type(&self) -> &'static str {
         "Await"
     }
-    fn wait_for(&self, _: &mut Context, count: usize) -> HandlerResult<Wait> {
-        Ok(Wait::until(COMPLETE.for_n(count)))
+    fn wait_for(&self, _: &mut Context, round: JoinRound) -> HandlerResult<Wait> {
+        Ok(Wait::until(COMPLETE.for_all_branches(&round)))
     }
-    fn execute(&self, _: &mut Context, count: usize) -> HandlerResult<StepDecision> {
-        Ok(StepDecision::graceful_complete(count))
+    fn execute(&self, context: &mut Context, round: JoinRound) -> HandlerResult<StepDecision> {
+        let results = COMPLETE.branch_results(context, &round)?;
+        Ok(StepDecision::graceful_complete(results.len()))
     }
 }
-static COMPLETE: LazyLock<Channel<()>> = LazyLock::new(|| Channel::new("parallel-complete"));
+static COMPLETE: LazyLock<Join<usize>> = LazyLock::new(|| Join::new("parallel-complete"));
 
 #[derive(Default)]
 pub struct FirstWinParallelStepsFlow {

examples/rust/src/workflow_patterns.rs

@@ -31,7 +31,9 @@ pub mod triggers;
 use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use dex_sdk::{Client, HandlerError, HandlerResult, Registry, SdkResult};
+use dex_sdk::{
+    Client, HandlerError, HandlerResult, JoinRound, Registry, SdkResult, Step, StepMovement,
+};
 use serde::{Deserialize, Serialize};
 
 /// The time one task held a shared resource, in Unix epoch milliseconds.
@@ -116,9 +118,15 @@ pub(crate) fn sorted<T: Ord>(mut values: Vec<T>) -> Vec<T> {
     values
 }
 
-/// Returns the ChannelMap instance key that holds the signals of `round`.
-pub(crate) fn round_instance(round: u32) -> String {
-    format!("round-{round}")
+/// Builds one branch movement of a Join round.
+pub(crate) type JoinBranchBuilder = Box<dyn FnOnce(&JoinRound) -> StepMovement>;
+
+/// Returns a branch builder that starts `step` with the round as its input.
+pub(crate) fn round_branch<BranchStep>(step: BranchStep) -> JoinBranchBuilder
+where
+    BranchStep: Step<Input = JoinRound> + 'static,
+{
+    Box::new(move |round: &JoinRound| StepMovement::to(&step, round.clone()))
 }
 
 pub(crate) fn current_unix_millis() -> HandlerResult<i64> {

examples/rust/src/workflow_patterns/and_join_and_threads.rs

@@ -18,12 +18,12 @@
 use std::sync::LazyLock;
 
 use dex_sdk::{
-    Channel, Context, Flow, HandlerError, HandlerResult, PersistenceSchema, Rpc, RpcList, Step,
-    StepDecision, StepList, StepMovement, Wait,
+    Channel, Context, Flow, HandlerError, HandlerResult, Join, JoinRound, PersistenceSchema, Rpc,
+    RpcList, Step, StepDecision, StepList, StepMovement, Wait,
 };
 use serde::{Deserialize, Serialize};
 
-use super::sorted;
+use super::{JoinBranchBuilder, sorted};
 
 static INVOICE_RECEIVED: LazyLock<Channel<String>> =
     LazyLock::new(|| Channel::new("invoiceReceived"));
@@ -147,10 +147,18 @@ fn single_condition_result(context: &Context, channel: &Channel<String>) -> Hand
         .ok_or_else(|| HandlerError::new("MissingChannelMessage", "the join consumed no message"))
 }
 
-static PARCEL_PACKED: LazyLock<Channel<usize>> = LazyLock::new(|| Channel::new("parcelPacked"));
+static PARCEL_PACKED: LazyLock<Join<usize>> = LazyLock::new(|| Join::new("parcelPacked"));
 
-/// WCP-42 Thread Split and WCP-41 Thread Merge: GoToMany starts several
-/// executions of one Step type, and the merge waits for one message each.
+/// One parcel of a shipment, packed in its own thread.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ParcelPacking {
+    pub index: usize,
+    pub round: JoinRound,
+}
+
+/// WCP-42 Thread Split and WCP-41 Thread Merge: a Join forks several
+/// executions of one Step type, and the merge waits for one result each.
 #[derive(Default)]
 pub struct ThreadSplitMergeFlow {
     split_shipment: SplitShipment,
@@ -168,7 +176,7 @@ impl Flow for ThreadSplitMergeFlow {
     }
 
     fn persistence(&self) -> PersistenceSchema {
-        PersistenceSchema::new().channel(&PARCEL_PACKED)
+        PersistenceSchema::new().join(&PARCEL_PACKED)
     }
 }
 
@@ -180,44 +188,52 @@ impl Step for SplitShipment {
 
     fn execute(
         &self,
-        _context: &mut Context,
+        context: &mut Context,
         parcel_count: Self::Input,
     ) -> HandlerResult<StepDecision> {
-        let parcels = (0..parcel_count).map(|index| StepMovement::to(&PackParcel, index));
-        Ok(StepDecision::go_to_many(
-            parcels.chain([StepMovement::to(&MergeParcels, parcel_count)]),
-        ))
+        let parcels = (0..parcel_count)
+            .map(|index| {
+                Box::new(move |round: &JoinRound| {
+                    StepMovement::to(
+                        &PackParcel,
+                        ParcelPacking {
+                            index,
+                            round: round.clone(),
+                        },
+                    )
+                }) as JoinBranchBuilder
+            })
+            .collect();
+        let movements = PARCEL_PACKED.fork(context, parcels, |round| {
+            StepMovement::to(&MergeParcels, round)
+        })?;
+        Ok(StepDecision::go_to_many(movements))
     }
 }
 
 #[derive(Default)]
 struct PackParcel;
 
 impl Step for PackParcel {
-    type Input = usize;
+    type Input = ParcelPacking;
 
-    fn execute(&self, context: &mut Context, index: Self::Input) -> HandlerResult<StepDecision> {
-        PARCEL_PACKED.publish(context, index)?;
-        Ok(StepDecision::dead_end())
+    fn execute(&self, context: &mut Context, parcel: Self::Input) -> HandlerResult<StepDecision> {
+        PARCEL_PACKED.end_branch(context, &parcel.round, parcel.index)
     }
 }
 
 #[derive(Default)]
 struct MergeParcels;
 
 impl Step for MergeParcels {
-    type Input = usize;
+    type Input = JoinRound;
 
-    fn wait_for(&self, _context: &mut Context, parcel_count: Self::Input) -> HandlerResult<Wait> {
-        Ok(Wait::until(PARCEL_PACKED.for_n(parcel_count)))
+    fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult<Wait> {
+        Ok(Wait::until(PARCEL_PACKED.for_all_branches(&round)))
     }
 
-    fn execute(
-        &self,
-        context: &mut Context,
-        _parcel_count: Self::Input,
-    ) -> HandlerResult<StepDecision> {
-        let packed_parcels = PARCEL_PACKED.condition_results(context)?;
+    fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult<StepDecision> {
+        let packed_parcels = PARCEL_PACKED.branch_results(context, &round)?;
         Ok(StepDecision::graceful_complete(sorted(packed_parcels)))
     }
 }