Channel

View as MarkdownView as Markdown

A Channel is a durable first-in, first-out (FIFO) queue inside one Flow run. Steps and RPCs publish typed messages to it. A Step consumes messages by waiting on the Channel in WaitFor. Messages never move between Flow runs.

Flow run

STEPExampleStepA

RPCExampleRpc

publish messages

CHANNELrequests_queueFIFO queue·one wait consumes each message

consume

STEPExampleStepB

WaitForrequests_queue.for_one()

Executeprocess request

When to use a Channel

Use a Channel for events that a Step waits for, such as an approval, a user reply, or a new work item. Use an Attribute for state that you overwrite and read many times. Use a Stream for live progress that a user can afford to lose.

To wait for parallel branches to finish, use a Join. A Join is built on a ChannelMap, and it counts the branches and keys each round for you. See Await Parallel Steps.

Guarantees

• Dex stores every message durably with the Flow.
• Messages stay in arrival order.
• One wait consumes a message. Dex does not copy it to other waiting Steps.
• A Step can be canceled after its wait consumed messages but before Dex saves its Execute result. Then Dex puts the messages back at the front of their Channel, in their original order. Another wait can consume them.
• A canceled Execute may already have seen a returned message. So a message can reach Execute twice. Make message processing idempotent.
• When a Step is canceled because the Flow closes, Dex drops the messages instead. That happens on a StopFlow request, on a Flow timeout, and on a decision that closes the Flow with ForceComplete or ForceFail.
• Any other cancellation puts the messages back, even when the Flow closes later, for example after GracefulComplete. No Step consumes them after the close.
• When one Step method cancels a Step and publishes messages, the canceled Step never receives those messages.
• When several Steps wait on one Channel, the Step that started waiting first gets the first chance at new messages. A Step whose wait needs more messages than are pending is skipped. A later Step can then take them.
• So a Step that waits for several messages can wait forever while other Steps keep taking them one at a time. On one Channel, do not mix Steps that wait for one message with Steps that wait for several.
• A message can arrive before a Step starts waiting. Dex keeps it until a later wait consumes it.
• Dex publishes a message only when the Step method or RPC that published it succeeds. See Step guarantees.

Define, publish, and wait

A Channel has a stable name and a message type. Define it and add it to the Flow persistence schema.

In this example, the start Step waits for one message on the ApprovalMessages Channel, or for a Timer. The publishApprovalMessage RPC publishes the approval.

channel_flow.pyPython

12345678910111213141516171819202122232425262728293031323334

class ChannelWait(Step[int]):
    def __init__(
        self,
        approval_messages: Channel[str],
        queued_messages: Channel[str],
    ) -> None:
        self.approval_messages = approval_messages
        self.queued_messages = queued_messages

    def get_step_options(self) -> StepOptions:
        return StepOptions(execute_load_channels=(self.queued_messages,))

    def wait_for(self, context: Context, input: int) -> Wait:
        return Wait.any_of(
            self.approval_messages.for_one(),
            Timer.by_duration(timedelta(seconds=input)),
        )

    def execute(self, context: Context, input: int) -> StepDecision:
        pending_queued_messages = self.queued_messages.pending_messages(context)
        if pending_queued_messages:
            self.queued_messages.delete(context, pending_queued_messages[0].message_id)
            return graceful_complete(pending_queued_messages[0].value)
        if context.has_timer_fired():
            return graceful_complete("approval timed out")
        approval_message_values = self.approval_messages.results(context)
        return graceful_complete(approval_message_values[0])

class ChannelFlow(Flow[int]):
    approval_messages = Channel("ApprovalMessages", str)

    @rpc
    def publish_approval_message(self, context: Context) -> None:
        self.approval_messages.publish(context, "approved")

Example: examples/python/dex_examples/primitives/channel/channel_flow.py

Execute first takes a message that an RPC queued on a second Channel, QueuedMessages, if one is pending. See Manage pending messages. The Timer branch returns before it reads Channel results. If the approval wins, the Channel condition has consumed one message, so the first result is present. The full example also loads QueuedMessages in the Step options, registers the Channels in the persistence schema, and registers the RPC.

Wait and consume

A Channel condition in WaitFor waits until enough messages are queued. Dex then removes the selected messages in FIFO order and passes them to Execute as Channel results.

Condition	When it proceeds	What it consumes
ForOne	At least one message is queued.	Exactly one message.
ForN	At least the requested number is queued.	Exactly that number of messages.
AtLeast	At least the requested number is queued.	Every message currently available.
AtMost	Immediately, including when the queue is empty.	Up to the requested number of currently queued messages.
AtLeastAtMost	At least the lower bound is queued.	From the lower bound up to the upper bound of currently queued messages.

Publish messages

Inside a Step or RPC, publish with the handler Context. Code outside a Worker publishes by calling an RPC of the Flow. This keeps validation, authorization, and related state changes inside the Flow.

Manage pending messages

Each message gets an ID when Dex accepts it. A Step or RPC can load the pending messages, read them in FIFO order, and delete one by ID. Reading does not consume a message. Other Steps and RPCs can change the queue while a handler runs, so a loaded list can be stale.

Inside a Step or a non-transactional RPC, deleting a message that is gone does nothing. A transactional RPC fails instead, and Dex saves none of its changes. A direct delete from dexcli or Dex Web fails with a not-found error. See Pending messages.

ChannelMap

Use a ChannelMap when one Flow needs many independent queues, such as one per order, customer, or job. A ChannelMap has one name and one message type. Each instance key has its own FIFO queue.

Reference

The Channel page of each SDK reference lists the API names in that language.

Data primitive semantics covers wait conditions, pending messages, deletion, state loading, and ChannelMap keys in detail.