Class Join<T>

java.lang.Object
io.superdurable.dex.PersistenceDefinition
io.superdurable.dex.Join<T>
Type Parameters:
T - the branch result type; each result is stored exactly as a ChannelMap<T> message
public final class Join<T> extends PersistenceDefinition
Synchronizes parallel branches of one Flow: a split Step starts the branches and one join Step waits for their results.

Use a Join whenever a Step must continue only after all, the first, or a number of parallel branches finish. Declare it like a Channel and register it in the Flow's PersistenceSchema. A Join is built on a ChannelMap with the same name: each round of the Join uses one ChannelMap instance, and each branch publishes its result there. Dex Web and pending-message reads show the Join as that ChannelMap. Its name must not equal the name of another Attribute, Channel, ChannelMap, or Stream of the Flow.

A round has four parts:

  1. Split. The split Step's Execute calls fork(Context, List, Function). It creates one JoinRound whose key is the split Step's execution ID and whose branch count is the number of branch builders. It calls each builder and the join builder with that round and returns the movements. Return them with StepDecision.goToMany(List).
  2. Branch. Each branch reports exactly one result per round. Its last Step returns endBranch(Context, JoinRound, T), or calls reportBranchResult(Context, JoinRound, T) when it must continue.
  3. Join. The join Step's WaitFor waits with forAllBranches(JoinRound), forFirstBranch(JoinRound), or forBranches(JoinRound, int). Its Execute reads the consumed results with getBranchResults(Context, JoinRound).
  4. Reset. When the join consumed fewer results than branches and the Flow loops, the next round's split Step waits with forRemainingBranches(JoinRound, int) before it forks again, and may read the late results with getRemainingBranchResults(Context, JoinRound). This blocks the next round until every branch of the previous round has reported.

Example: request three quotes per round, accept the first two, and repeat for three rounds. The example compiles as written. 

import io.superdurable.dex.Channel;
import io.superdurable.dex.Context;
import io.superdurable.dex.Flow;
import io.superdurable.dex.Join;
import io.superdurable.dex.JoinRound;
import io.superdurable.dex.PersistenceSchema;
import io.superdurable.dex.Step;
import io.superdurable.dex.StepDecision;
import io.superdurable.dex.StepList;
import io.superdurable.dex.StepMovement;
import io.superdurable.dex.Wait;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.function.Function;

public final class QuoteFlow implements Flow<QuoteFlow.QuoteRequest> {
    public static final class QuoteRequest {
        public int requiredQuotes = 2;
        public int roundNumber = 1;
        public int roundCount = 3;
        public JoinRound previousRound; // null in the first round
    }

    public static final class SupplierInput {
        public String supplier;
        public JoinRound round;
    }

    public static final class AcceptInput {
        public QuoteRequest request;
        public JoinRound round;
    }

    private final Join<String> quoteReceived = Join.define("quoteReceived", String.class);
    private final Channel<Void> supplierReleased = Channel.define("supplierReleased", Void.class);
    private final RequestQuotes requestQuotes = new RequestQuotes();
    private final QuoteFromSupplier quoteFromSupplier = new QuoteFromSupplier();
    private final AcceptQuotes acceptQuotes = new AcceptQuotes();

    public StepList<QuoteRequest> getSteps() {
        return StepList.startStep(requestQuotes).otherSteps(quoteFromSupplier, acceptQuotes);
    }

    public PersistenceSchema getPersistenceSchema() {
        return PersistenceSchema.of(quoteReceived, supplierReleased);
    }

    // Split: reset the previous round, then fork three branches and one join.
    final class RequestQuotes implements Step<QuoteRequest> {
        public Class<QuoteRequest> getInputType() {
            return QuoteRequest.class;
        }

        public Wait waitFor(Context context, QuoteRequest request) {
            if (request.previousRound == null) {
                return Wait.skipImmediately();
            }
            return Wait.until(quoteReceived.forRemainingBranches(
                    request.previousRound, request.requiredQuotes));
        }

        public StepDecision execute(Context context, QuoteRequest request) {
            List<Function<JoinRound, StepMovement<?>>> branches = new ArrayList<>();
            for (String supplier : Arrays.asList("A", "B", "C")) {
                branches.add(round -> {
                    SupplierInput branchInput = new SupplierInput();
                    branchInput.supplier = supplier;
                    branchInput.round = round;
                    return StepMovement.of(QuoteFromSupplier.class, branchInput);
                });
            }
            List<StepMovement<?>> movements = quoteReceived.fork(context, branches, round -> {
                AcceptInput joinInput = new AcceptInput();
                joinInput.request = request;
                joinInput.round = round;
                return StepMovement.of(AcceptQuotes.class, joinInput);
            });
            return StepDecision.goToMany(movements);
        }
    }

    // Branch: report exactly one result, then end.
    final class QuoteFromSupplier implements Step<SupplierInput> {
        public Class<SupplierInput> getInputType() {
            return SupplierInput.class;
        }

        public Wait waitFor(Context context, SupplierInput input) {
            return Wait.until(supplierReleased.forOne());
        }

        public StepDecision execute(Context context, SupplierInput input) {
            return quoteReceived.endBranch(context, input.round, input.supplier);
        }
    }

    // Join: accept the first requiredQuotes results, then start the next round.
    final class AcceptQuotes implements Step<AcceptInput> {
        public Class<AcceptInput> getInputType() {
            return AcceptInput.class;
        }

        public Wait waitFor(Context context, AcceptInput input) {
            return Wait.until(quoteReceived.forBranches(
                    input.round, input.request.requiredQuotes));
        }

        public StepDecision execute(Context context, AcceptInput input) {
            List<String> accepted = quoteReceived.getBranchResults(context, input.round);
            QuoteRequest request = input.request;
            if (request.roundNumber == request.roundCount) {
                return StepDecision.forceComplete(accepted);
            }
            QuoteRequest next = new QuoteRequest();
            next.requiredQuotes = request.requiredQuotes;
            next.roundNumber = request.roundNumber + 1;
            next.roundCount = request.roundCount;
            next.previousRound = input.round;
            return StepDecision.goTo(RequestQuotes.class, next);
        }
    }
}

For a run-time number of branches, map the input list to one builder per element; the branch count is the list length. A join with a deadline combines a Join condition with a Timer, such as Wait.anyOf(Timer.byDuration(timeout, "deadline"), join.forAllBranches(round, "all")).

Rules and traps:

  • Each branch reports exactly one result per round. A result is committed together with the Step method that reports it, so a retried method never reports twice, and a failed attempt reports nothing. A branch that reports twice is counted twice. A branch that never reports, for example because its failure route goes to a recovery Step that does not report, blocks an all-branches join forever. Report a failure result from the recovery Step, or give the join a Timer deadline.
  • A branch passes its round only to endBranch(Context, JoinRound, T) or reportBranchResult(Context, JoinRound, T). It never waits on a join condition. A multi-Step branch passes the round down to its last Step.
  • Exactly one join Step waits on a round. Two Steps waiting on the same round split its results between them.
  • Fork a Join at most once per Step invocation. A second fork(Context, List, Function) of the same Join in the same invocation throws. Use another Join, or forkWithRoundKey(String, List, Function) with distinct keys, for two independent joins.
  • An RPC has no Step execution ID, so it cannot call fork(Context, List, Function). Use forkWithRoundKey(String, List, Function) with an application-chosen key instead.
  • The receivedCount passed to forRemainingBranches(JoinRound, int) must equal the number of results the join consumed: the count of forBranches(JoinRound, int), or 1 for forFirstBranch(JoinRound).
  • Without a reset, late results stay in the round's ChannelMap instance until the Flow closes. They do not delay graceful completion. A Flow that loops must reset every round.
  • To cancel the remaining branches when the join fires, return the join's decision with StepDecision.withCancelingSiblingSteps(Class...) and the branch Step classes. Fork puts the branches and the join in one decision, so the selection covers exactly this round's branches. Later Steps of a multi-Step branch are not siblings; select them with StepDecision.withCancelingSteps(Class...), which is correct only when one round is active at a time. Rounds forked by an RPC through forkWithRoundKey(String, List, Function) share their parent with every other Step that RPC name started, so prefer withCancelingSteps there.
  • A branch canceled by a cancelling join never reports, so the next split cannot reset that round with forRemainingBranches(JoinRound, int); it would wait forever. Do not reset a cancelling join's round. Its late results stay until the Flow closes. To reset it anyway, have an RPC report a result for every canceled branch with reportBranchResult(Context, JoinRound, T) first.
  • A branch canceled from outside never reports, so an all-branches join keeps waiting. Include the join in the cancellation, have the canceling RPC report a result for the canceled branch with reportBranchResult(Context, JoinRound, T), or give the join a Timer deadline.
  • Canceling a join Step after its wait matched but before its Execute committed makes the round unable to fire again, unless the Dex Server returns a canceled Step's consumed messages to their Channel. Check the server change log for that behavior before recreating a join for the same round.
Since:
1.44.0

  • Method Details

    • define

      public static <T> Join<T> define(String name, Class<T> resultType)
      Defines a typed Join.

      Declare the Join once, typically as a final field of the Flow, and register it in the Flow's PersistenceSchema. The name is durable: renaming it strands the branch results of open Flows.

      Type Parameters:
      T - the branch result type
      Parameters:
      name - the stable Join name, which is also the name of the underlying ChannelMap; must not be blank or contain /
      resultType - the concrete Java class of each branch result
      Returns:
      the Join definition
      Throws:
      IllegalArgumentException - if name is null, blank, or contains /
      NullPointerException - if resultType is null

    • getName

      public String getName()
      Returns the durable name of this Join, which is also the name of its ChannelMap.
      Returns:
      the Join name

    • getChannelMap

      public ChannelMap<T> getChannelMap()
      Returns the ChannelMap that stores this Join's branch results.

      Use it to inspect a round's pending results with the existing ChannelMap API, passing the round's JoinRound.getRoundKey() as the instance key. For example, an RPC can load the round's instance with RPCInvokeOptions.Builder.addLoadChannelMapInstance(ChannelMap, String), then read ChannelMap.pendingMessages(Context, String) and ChannelMap.size(Context, String). Late results that no reset consumed stay there until the Flow closes.

      The returned object is the definition registered for this Join, so every API that checks the registered ChannelMap accepts it. Never publish to it or delete from it directly. Report branch results only with endBranch(Context, JoinRound, T) or reportBranchResult(Context, JoinRound, T); a direct publication or deletion breaks the exactly-once result count of the round. 

      List<ChannelMessage<String>> pending =
        quoteReceived.getChannelMap().pendingMessages(context, round.getRoundKey());
      Returns:
      the registered ChannelMap with the same name as this Join
      Since:
      1.44.0

    • fork

      public List<StepMovement<?>> fork(Context context, List<? extends Function<JoinRound, ? extends StepMovement<?>>> branchBuilders, Function<JoinRound, ? extends StepMovement<?>> joinBuilder)
      Starts a new round from a split Step and builds its branch and join movements.

      The round key is the current Step execution ID, which is unique in the Flow run and stays the same across retries. The branch count is branchBuilders.size(). Fork calls every branch builder in order and then the join builder, each with the same JoinRound, and returns the movements in that order. Pass them to StepDecision.goToMany(List), optionally together with other movements. Fork itself makes no server call.

      An empty builder list is allowed. Its round has zero branches, and forAllBranches(JoinRound) on it is satisfied immediately.

      See Join for a complete example.

      Parameters:
      context - the Step invocation context
      branchBuilders - one builder per branch; each receives the round and returns the branch's first movement, whose input should carry the round
      joinBuilder - receives the round and returns the join Step's movement, whose input should carry the round
      Returns:
      the branch movements in builder order, followed by the join movement
      Throws:
      NullPointerException - if an argument, a builder, or a built movement is null
      IllegalStateException - if called outside a Step invocation, such as from an RPC, or if this Join was already forked in the same invocation
      IllegalArgumentException - if context is not a Dex invocation context

    • forkWithRoundKey

      public List<StepMovement<?>> forkWithRoundKey(String roundKey, List<? extends Function<JoinRound, ? extends StepMovement<?>>> branchBuilders, Function<JoinRound, ? extends StepMovement<?>> joinBuilder)
      Starts a new round with an application-chosen key and builds its branch and join movements.

      Use it in an RPC, which has no Step execution ID, or when a migrated Flow must keep the instance keys of a hand-written join, such as round-3. The key must be unique among the rounds of this Join that can be active at the same time; two active rounds with the same key share their results. Otherwise it behaves like fork(Context, List, Function). In an RPC, return the movements with RPCResult.of(Object, List).

      Parameters:
      roundKey - the round key, used as the ChannelMap instance key; must not be blank or contain /
      branchBuilders - one builder per branch; each receives the round and returns the branch's first movement
      joinBuilder - receives the round and returns the join Step's movement
      Returns:
      the branch movements in builder order, followed by the join movement
      Throws:
      IllegalArgumentException - if roundKey is null, blank, or contains /
      NullPointerException - if a builder list, a builder, or a built movement is null

    • endBranch

      public StepDecision endBranch(Context context, JoinRound round, T result)
      Reports this branch's result and ends the branch.

      Return the decision from the last Step of a branch. It publishes result to the round and returns StepDecision.deadEnd(). The publication commits together with this Execute, so it counts exactly once. Report from the recovery Step too when the branch's failure route skips this call, or the join may wait forever.

      Parameters:
      context - the Step invocation context
      round - the round received in the branch Step's input
      result - the branch result; it may carry the branch identity when the join needs to know which branch finished
      Returns:
      a dead-end decision
      Throws:
      IllegalArgumentException - if the round has a blank key, a key containing /, or a negative branch count, or if this Join is not registered in the Flow
      ValueMappingException - if the result cannot be encoded

    • reportBranchResult

      public void reportBranchResult(Context context, JoinRound round, T result)
      Reports this branch's result without ending the branch.

      Use it when a branch continues after reporting, or from an RPC that reports on behalf of a branch it canceled. The result is published to the round's ChannelMap instance and commits together with the current method. Each branch reports exactly once per round.

      Parameters:
      context - the Step or RPC invocation context
      round - the round of the branch
      result - the branch result
      Throws:
      IllegalArgumentException - if the round has a blank key, a key containing /, or a negative branch count, or if this Join is not registered in the Flow
      ValueMappingException - if the result cannot be encoded

    • forAllBranches

      public Condition forAllBranches(JoinRound round)
      Creates a condition satisfied when every branch of the round has reported.

      Return it from the join Step's WaitFor, alone or inside Wait.allOf(Condition...), Wait.anyOf(Condition...), or Wait.anyCombinationOf(ConditionCombination...). It consumes exactly round.getBranchCount() results. A round with zero branches is satisfied immediately.

      Parameters:
      round - the round received in the join Step's input
      Returns:
      the condition
      Throws:
      IllegalArgumentException - if the round has a blank key, a key containing /, or a negative branch count

    • forAllBranches

      public Condition forAllBranches(JoinRound round, String conditionId)
      Creates a named condition satisfied when every branch of the round has reported.
      Parameters:
      round - the round received in the join Step's input
      conditionId - the condition ID used to identify the condition in a combined wait, or null
      Returns:
      the condition
      Throws:
      IllegalArgumentException - if the round has a blank key, a key containing /, or a negative branch count
      See Also:

    • forFirstBranch

      public Condition forFirstBranch(JoinRound round)
      Creates a condition satisfied by the first branch result of the round.

      It consumes exactly one result: the earliest committed one, even when several are already waiting. Use it for a discriminator. Put the branch identity in the result when the join must know which branch won. Reset with forRemainingBranches(round, 1) when the Flow loops.

      A round with zero branches has no first result, so this method throws instead of returning a condition that would wait forever. When a round may be empty, check JoinRound.getBranchCount() first.

      Parameters:
      round - the round received in the join Step's input
      Returns:
      the condition
      Throws:
      IllegalArgumentException - if the round has a blank key, a key containing /, or fewer than one branch

    • forFirstBranch

      public Condition forFirstBranch(JoinRound round, String conditionId)
      Creates a named condition satisfied by the first branch result of the round.
      Parameters:
      round - the round received in the join Step's input
      conditionId - the condition ID used to identify the condition in a combined wait, or null
      Returns:
      the condition
      Throws:
      IllegalArgumentException - if the round has a blank key, a key containing /, or fewer than one branch
      See Also:

    • forBranches

      public Condition forBranches(JoinRound round, int count)
      Creates a condition satisfied when count branches of the round have reported.

      Use it for a partial join (n of m). It consumes exactly count results in commit order. Reset with forRemainingBranches(round, count) when the Flow loops.

      Parameters:
      round - the round received in the join Step's input
      count - the number of results to wait for and consume; from 1 to the branch count
      Returns:
      the condition
      Throws:
      IllegalArgumentException - if the round has a blank key, a key containing /, or a negative branch count, or if count is outside 1 to the branch count

    • forBranches

      public Condition forBranches(JoinRound round, int count, String conditionId)
      Creates a named condition satisfied when count branches of the round have reported.
      Parameters:
      round - the round received in the join Step's input
      count - the number of results to wait for and consume; from 1 to the branch count
      conditionId - the condition ID used to identify the condition in a combined wait, or null
      Returns:
      the condition
      Throws:
      IllegalArgumentException - if the round has a blank key, a key containing /, or a negative branch count, or if count is outside 1 to the branch count
      See Also:

    • getBranchResults

      public List<T> getBranchResults(Context context, JoinRound round)
      Returns the branch results consumed by the join condition, in commit order.

      Call it from the join Step's Execute after forAllBranches(JoinRound), forFirstBranch(JoinRound), or forBranches(JoinRound, int) was satisfied. When the condition was not satisfied, for example because a Timer won an Wait.anyOf(Condition...), the list is empty.

      Parameters:
      context - the Step invocation context
      round - the round received in the join Step's input
      Returns:
      the immutable typed results
      Throws:
      IllegalArgumentException - if the round has a blank key, a key containing /, or a negative branch count, or if this Join is not registered in the Flow

    • forRemainingBranches

      public Condition forRemainingBranches(JoinRound round, int receivedCount)
      Creates a condition that waits for and consumes the late results of a round: the reset.

      Return it from the next round's split Step's WaitFor, with the previous round and the number of results its join consumed. It waits for the remaining round.getBranchCount() - receivedCount results and consumes them, so the next round starts only after every branch of the previous round reported, and the previous round leaves no messages behind. When nothing remains, it is satisfied immediately. In the first round, which has no previous round, return Wait.skipImmediately() instead.

      Count only the results the join consumed. The reset waits for every other branch of the round, so it never completes while a canceled branch has no reported result. Do not reset the round of a cancelling join unless an RPC reported a result for each canceled branch.

      Parameters:
      round - the previous round
      receivedCount - the number of results already consumed from that round, which is the join's count, or 1 for a first-branch join; from 0 to the branch count
      Returns:
      the condition
      Throws:
      IllegalArgumentException - if the round has a blank key, a key containing /, or a negative branch count, or if receivedCount is outside 0 to the branch count

    • forRemainingBranches

      public Condition forRemainingBranches(JoinRound round, int receivedCount, String conditionId)
      Creates a named condition that waits for and consumes the late results of a round.
      Parameters:
      round - the previous round
      receivedCount - the number of results already consumed from that round; from 0 to the branch count
      conditionId - the condition ID used to identify the condition in a combined wait, or null
      Returns:
      the condition
      Throws:
      IllegalArgumentException - if the round has a blank key, a key containing /, or a negative branch count, or if receivedCount is outside 0 to the branch count
      See Also:

    • getRemainingBranchResults

      public List<T> getRemainingBranchResults(Context context, JoinRound round)
      Returns the late results consumed by forRemainingBranches(JoinRound, int), in commit order.

      Call it from the Execute of the Step that waited for the reset, usually the next round's split Step. It reads the same ChannelMap instance as getBranchResults(Context, JoinRound); the separate name states at the call site that these results arrived after the join fired.

      Parameters:
      context - the Step invocation context
      round - the previous round
      Returns:
      the immutable typed late results
      Throws:
      IllegalArgumentException - if the round has a blank key, a key containing /, or a negative branch count, or if this Join is not registered in the Flow