Tutorial3/4

3. Route within a Component

Publish status "1" orders as order-cancelled and route them to a second pipeline inside wms-loader.

Classify the event type in the extractor

The extractor stamps every event with the type order-placed. CloudEventSerializeStep also takes a type: function that picks the type per order. In erp-extractor/src/Composition/Composition.cs, find the WithSerializer call and add the type: argument:

.WithSerializer(new CloudEventSerializeStep<OrderPlaced, Context>(
    subject: order => order.OrderId,
    time: order => order.OrderDate,
    type: (order, _) => order.Status == "1" ? "order-cancelled" : Constants.EventType))

Constants.EventType is "order-placed". When type: is set, its value wins over the type the publisher is configured with. Both messages keep the OrderPlaced record as payload; the CloudEvent type is what tells them apart.

Declare the second message

Declare order-cancelled on the same topic, with the same record:

public static class Messages
{
    /// <summary>Cancelled orders, published on the orders topic with the same record as placed ones.</summary>
    public static readonly MessageRef<OrderPlaced> OrderCancelled =
        MessageRef<OrderPlaced>.Define("order-cancelled", topic: "orders");

    /// <summary>OrderPlaced messages, published on the orders topic.</summary>
    public static readonly MessageRef<OrderPlaced> OrderPlaced =
        MessageRef<OrderPlaced>.Define("order-placed", topic: "orders");
}

Then declare that the extractor publishes it, and that the WMS subscription handles it:

public void Define(SystemBuilder builder)
{
    builder.AddExtractor("erp-extractor")
        .From(Ports.ErpExtractorSource)
        .Publishes(Messages.OrderPlaced)
        .Publishes(Messages.OrderCancelled)
        .Calls(Services.Idempotency)
        .Calls(Services.BusinessIncidents);
    // Rendered as a declarative Dapr Subscription: one routing rule per route, in order,
    // and a default route for the rest of the topic. The subscription's first message
    // anchors it; the rest are also handled on the same channel.
    builder.AddLoader("wms-loader")
        .Subscribes(Messages.OrderPlaced,
            configure: sub => sub
                .AlsoHandles(Messages.OrderCancelled)
                .IgnoreOthers())
        .To(Ports.WmsLoaderDestination)
        .Calls(Services.Idempotency)
        .Calls(Services.BusinessIncidents);
}

A component may publish several messages on one topic, but each message only once.

Add the cancellation pipeline

Each pipeline lives in its own folder under src/Routes/. From the distribution-workspace root, copy the existing one:

cp -R wms-loader/src/Routes/OrderPlaced wms-loader/src/Routes/OrderCancelled

In the six copied files, change the namespace line from WmsLoader.Routes.OrderPlaced to WmsLoader.Routes.OrderCancelled. Leave using Payload = Contracts.OrderPlaced; as it is; the payload record is the same. Then set the message the route handles:

namespace WmsLoader.Routes.OrderCancelled;

/// <summary>
/// The message this loader subscribes to. Its CloudEvents type is the name
/// the system host's Messages.cs declares.
/// </summary>
public static class Route
{
    public const string Message = "order-cancelled";
}

Mark the WMS record as cancelled. Keep the signature; replace only the Status argument:

// Map the canonical order to the destination system's load record.
var output = new Out(
    OrderId: input.OrderId,
    CustomerId: input.CustomerId,
    LoadedAt: DateTimeOffset.UtcNow,
    Status: "Cancelled");

Register one route per message

Open wms-loader/src/Composition/Composition.cs. Add an alias below the existing ones:

using OrderPlacedRoute = WmsLoader.Routes.OrderPlaced;
using OrderPlacedPayload = Contracts.OrderPlaced;
using OrderCancelledRoute = WmsLoader.Routes.OrderCancelled;

In ConfigureServices, replace everything from the SendStep registration down to, but not including, return services;:

services.AddSingleton<SendStep<OrderPlacedRoute.Out, Context>>(sp =>
    new OrderPlacedRoute.Sender(
        sp.GetRequiredKeyedService<IFileAdapter>(Constants.DestinationPort)));
services.AddSingleton<SendStep<OrderCancelledRoute.Out, Context>>(sp =>
    new OrderCancelledRoute.Sender(
        sp.GetRequiredKeyedService<IFileAdapter>(Constants.DestinationPort)));

// One pipeline per message: the framework matches each event's CloudEvent type to a route.
services.AddLoader(
    loader => loader.Unrouted = UnroutedPolicy.Ack,
    routes => routes
        .On<OrderPlacedPayload, OrderPlacedRoute.Out, Context>(OrderPlacedRoute.Route.Message,
            (builder, _) => builder
                .WithDeserializer(new OrderPlacedRoute.Deserializer())
                .WithValidator(new OrderPlacedRoute.Validator())
                .WithIdempotency()
                .WithTransformer(new OrderPlacedRoute.Transformer())
                .WithSenderFromServices()
                .WithBusinessIncidents(BusinessId, BusinessId),
            NewContext)
        .On<OrderPlacedPayload, OrderCancelledRoute.Out, Context>(OrderCancelledRoute.Route.Message,
            (builder, _) => builder
                .WithDeserializer(new OrderCancelledRoute.Deserializer())
                .WithValidator(new OrderCancelledRoute.Validator())
                .WithIdempotency()
                .WithTransformer(new OrderCancelledRoute.Transformer())
                .WithSenderFromServices()
                .WithBusinessIncidents(BusinessId, BusinessId),
            NewContext));

Then add the context factory the routes share, next to BusinessId at the bottom of the class:

private static Context NewContext(Dictionary<string, string> metadata, bool isRetry) => new(metadata, isRetry);

What the routed AddLoader does:

On<TInput, TOutput, TCtx>(eventType, …) registers one pipeline for one CloudEvent type, matched exactly. Each route has its own steps and its own Out record.
Each route keeps its own idempotency records, so a cancellation is not treated as a duplicate of the same order placed earlier.
Unrouted = UnroutedPolicy.Ack acknowledges a message that no route handles. The System’s subscription already decides what reaches the loader, so this matches its IgnoreOthers().

See the routing rule

Validate the topology and render the deployment files without starting the host:

cd order-notifications
task check
task generate
cat out/components/wms-loader-subscription.yaml

task check ends with:

ok: 'order-notifications' is valid (2 components).

The build before it prints warning ASPIRE010 about AspireUseCliBundle=false. The host does not use the Aspire CLI bundle; ignore the warning.

The subscription file is the routing rule the sidecar applies:

apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: "wms-loader-subscription"
spec:
  pubsubname: "order-notifications-pubsub"
  topic: "orders"
  routes:
    rules:
    - match: "event.type == 'order-placed'"
      path: "/order-placed"
    - match: "event.type == 'order-cancelled'"
      path: "/order-cancelled"
    default: "/unhandled"
scopes:
- "wms-loader"

The sidecar delivers both messages to wms-loader. Inside the loader, the framework hands each one to the pipeline registered for its type.

Run it

Clear the WMS folder so it holds only this run’s output, seed the source folder, and start the host:

rm -rf test/wms-loader-destination
task seed
task run

Check it

In a second terminal, from order-notifications/:

ls test/wms-loader-destination
cat test/wms-loader-destination/ORD-1002.order-cancelled.json

ORD-1001.order-placed.json
ORD-1002.order-cancelled.json
{"OrderId":"ORD-1002","CustomerId":"CUST-42","LoadedAt":"2026-10-08T14:07:48.965025+00:00","Status":"Cancelled"}

The extractor published ORD-1002 as order-cancelled, and the loader ran it through the OrderCancelled pipeline, which set Status to "Cancelled".

Stop the host with Ctrl+C before chapter 4.