Skip to content

Reading Along: DeliveryDispatch

Guide Home | Previous: Reading Along: SupportChat | Next: Reading Along: ShoppingMall

View in another language — C++ · C#/.NET · Java · Kotlin · Node/TypeScript

What you get from this chapter

You can open the DeliveryDispatch sample in an editor and follow the code from a delivery request accepted over HTTP, through the offer to a courier and the reassignment on no answer, to the status pushed to the customer. The code in this chapter comes from the DeliveryDispatch sample in the per-language example repositories.

Picking a Sample introduced what this sample demonstrates. This chapter is what you read after that introduction — the roles and where their code lives, the message flow of the main scenarios, and, for each flow, the framework feature it uses and the chapter that explains it, in the order the source is laid out. This chapter explains the DeliveryDispatch sample's roles and code locations, its main message flows, and its run verification in source order. See the DeliveryDispatch scenario for requirements, message contracts, and verification criteria.

1. What This Sample Demonstrates

The customer creates a delivery over HTTP and receives its status over STREAM. The courier receives offers over STREAM and sends decisions. The servers in between keep no session map and no socket registry — the courier and the customer are each an Actor with an id, and offers and statuses are pushed to the session bound to that Actor. No handler holds an execution turn while waiting for the courier's answer; the offer deadline is a record that Dispatch manages.

↗ View larger

The flow this chapter follows is the courier binding → HTTP intake and the first offer → the offer push and the decision → reassignment on timeout → status recording and the customer push.

2. Roles and Where the Code Lives

Role Processes Owns Code
Dispatch 1 HTTP intake, candidate selection, offer records and deadlines, reassignment Server/Dispatch
CourierSession 1 Courier STREAM, courier Actor binding and relay Server/CourierSession
CourierActorNode 2 Courier Entry Spot and courier Actors Server/CourierActorNode
Tracking 1 Status event recording, forwarding to the customer Actor Server/Tracking
CustomerGateway 1 Customer STREAM, customer Actor binding and pushes Server/CustomerGateway
Client 1 The customer and courier scenarios with self-checks Client

Candidate order, deadlines and attempt validation belong to Dispatch's Application, and the status transitions to Tracking's Domain; neither references a framework type. Messages are set by the JSON contract in shared.

3. Server Configuration

Dispatch is an object client of the courier mesh, the server of the dispatch channel, and a client of the tracking channel. In the implementations whose HTTP handler sends the request over the dispatch channel, the same process is that channel's client as well.

Server/Dispatch/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/dispatch/DispatchServerApplication.kt

options.addClientServerChannel(SampleNames.CourierChannel).client()
// The courier's decision comes back here as its own one-way message, so dispatch has
// to be a channel server (common sample spec section 7.4).
val dispatchEndpoint = URI.create(SampleTopology.DispatchChannelEndpoint)
options.addClientServerChannel(SampleNames.DispatchChannel).client()
options
    .addClientServerChannel(SampleNames.DispatchChannel)
    .server()
    .setBindHost(dispatchEndpoint.host)
    .setAdvertiseHost(dispatchEndpoint.host)
    .listen(dispatchEndpoint.port)
    .addHandlerGroup(SampleNames.DispatchChannel)
options.addClientServerChannel(SampleNames.TrackingChannel).client()
val courierRoutes = options.addRouteMesh(SampleNames.CourierSpotMesh)
courierRoutes
    .listen(SampleTopology.DispatchSpotEndpoint)
    .setRoutingIdPrefix("delivery-dispatch")
courierRoutes.objects().client()

CourierActorNode registers the courier Entry Spot and the courier Actor factory as an object server. Both nodes provide the same type, so the framework picks the node an Actor is created on.

Server/CourierSpotNode/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/courierspotnode/CourierSpotNodeApplication.kt

val spotNode = options.addRouteMesh(SampleNames.CourierSpotMesh)
//  Fixed RID so Dispatch can name this courier node in its actor-route readiness row.
//  An auto-assigned RID cannot be named, and peer status carries no endpoint.
spotNode
    .listen(selected.spotEndpoint)
    .setRoutingId(
        RoutingId.from(
            if (SampleTopology.CourierNode == "node2") {
                SampleNames.CourierNode2
            } else {
                SampleNames.CourierNode1
            }
        )
    )
spotNode.objects().server().addEntrySpot<CourierEntrySpot>().addActorFactory<
    CourierActor,
    CourierActorFactory,
>(
    SampleNames.CourierActorType
) {
    disableRelocation()
}
// The courier's decision goes back to dispatch as its own one-way message, so this
// node needs a way to speak to the dispatch channel (common sample spec section 7.4).
options.addClientServerChannel(SampleNames.DispatchChannel).client()

The client and server roles of a channel are covered by Channel Messaging, and the difference between an object server and an object client by Spot.

4. Courier Binding

When the courier client connects over STREAM and asks to bind, CourierSession creates or finds the Actor by the courier id and binds it to the current session.

Server/CourierSession/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/couriersession/sessions/CourierSession.kt

private suspend fun handleBindCourierSessionReq(
    dispatch: ZLinkSessionDispatchContext,
    payload: ZLinkMessage,
) {
    val request = payload.decode<BindCourierSessionReq>()
    val actorRef = findOrEnsureActor(request.courierId)
    val actor =
        sessionContext.actors().find(actorRef.actorId).orElse(null)
            ?: sessionContext.actors().bind(actorRef).await()
    val snapshot = ActorRefSnapshot.from(actorRef)
    actor
        .kotlin()
        .relay(
            dispatch,
            ZLinkMessage.of(
                BindCourierSessionReq(
                    courierId = request.courierId,
                    actor = snapshot,
                    sessionRoute = sessionContext.sessionId(),
                )
            ),
        )
        .await()
    sessionContext
        .client()
        .kotlin()
        .reply(BindCourierSessionRes(request.courierId, snapshot, sessionContext.sessionId()))
        .await()
    println("deliverydispatch-courier bound courier=${request.courierId}")
}

The call that obtains the Actor does not decide which CourierActorNode creates it. Binding is covered by Session and Actor.

5. HTTP Intake and the First Offer

The customer's HTTP request arrives at Dispatch's endpoint. The handler hands the request to the worker and replies at once — some implementations do so with a one-way send on the dispatch channel, others through an in-process queue.

↗ View larger

Server/Dispatch/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/dispatch/DispatchHttpServer.kt

private fun handleCreateDelivery(exchange: HttpExchange) {
    if (exchange.requestMethod != "POST") {
        exchange.writeJson(405, "")
        return
    }
    val request = json.readValue(exchange.requestBody, CreateDeliveryReq::class.java)
    val assign =
        AssignDeliveryMsg(
            deliveryId = request.deliveryId,
            customerId = request.customerId,
            pickupAddress = request.pickupAddress,
            dropoffAddress = request.dropoffAddress,
        )
    runBlocking { channels.kotlin().sendToChannel(SampleNames.DispatchChannel, assign).await() }
    exchange.writeJson(200, json.writeValueAsString(CreateDeliveryRes(request.deliveryId)))
}

The worker picks the first candidate, reports the Assigned status to Tracking, records the offer with its attempt and deadline, and sends the offer to the courier Actor. The turn that sends the offer ends there.

Server/Dispatch/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/dispatch/DispatchWorker.kt

/** The first offer. Records it, sends it, and returns — nobody is left waiting. */
suspend fun dispatch(request: AssignDeliveryMsg) {
    val courierId = candidates[0]
    publishStatus(request, DeliveryStatus.Assigned, courierId)
    val attempt = offers.offer(request, 0, SampleTimings.CourierDecisionTimeout)
    offer(request, courierId, attempt)
}

The offer is an Actor direct send addressed by the courier id. It is not a request, so nothing waits for a reply.

Server/Dispatch/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/dispatch/DispatchWorker.kt

/** The offer is a one-way send: the turn that sends it ends right there. */
private suspend fun offer(request: AssignDeliveryMsg, courierId: String, attempt: Int) {
    kotlinActors
        .sendToActor(
            courierId,
            OfferDeliveryMsg(
                courierId = courierId,
                deliveryId = request.deliveryId,
                attempt = attempt,
                pickupAddress = request.pickupAddress,
                dropoffAddress = request.dropoffAddress,
            ),
        )
        .await()
}

Calling an Actor by id is covered by Actor, and the difference between send and request by Channel Messaging.

6. The Offer Push and the Decision

The offer arrives at the courier Actor's Entry Spot handler. The Actor remembers the attempt and pushes an offer notify to its bound session.

Server/CourierSpotNode/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/courierspotnode/CourierActor.kt

/** Pushes the offer and returns. The courier takes as long as it takes. */
suspend fun offer(offer: OfferDeliveryMsg) {
    synchronized(offeredAttempts) { offeredAttempts[offer.deliveryId] = offer.attempt }
    actorContext
        .boundSession()
        .kotlin()
        .send(
            OfferDeliveryNotify(
                courierId = offer.courierId,
                deliveryId = offer.deliveryId,
                pickupAddress = offer.pickupAddress,
                dropoffAddress = offer.dropoffAddress,
            )
        )
        .await()
}

The courier's decision comes in over the same connection and is relayed to the Actor. The handler attaches the remembered attempt to the decision and sends it one-way to the dispatch channel. The node neither judges the decision nor times it.

Server/CourierSpotNode/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/courierspotnode/spots/handlers/CourierDecisionActorHandler.kt

val attempt = actor.takeOfferedAttempt(message.deliveryId)
if (attempt == null) {
    System.err.println(
        "deliverydispatch courier-actor: decision for an unknown offer " +
            "delivery=${message.deliveryId} courier=${actor.actorId()}"
    )
    return
}

kotlinChannels
    .sendToChannel(
        SampleNames.DispatchChannel,
        OfferDeliveryResultMsg(
            deliveryId = message.deliveryId,
            courierId = message.courierId,
            attempt = attempt,
            accepted = message.accepted,
            reason = message.reason,
        ),
    )
    .await()

Dispatch's handler applies a decision only when its attempt equals the current attempt on record. A decision from the previous courier that arrives after a reassignment is ignored as stale.

Server/Dispatch/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/dispatch/handlers/OfferDeliveryResultHandler.kt

val offer = offers.settle(message.deliveryId, message.attempt)
if (offer == null) {
    println(
        "deliverydispatch-dispatch stale-decision-ignored " +
            "delivery=${message.deliveryId} courier=${message.courierId} " +
            "attempt=${message.attempt}"
    )
    return
}

println(
    "deliverydispatch dispatch: decision delivery=${message.deliveryId} " +
        "courier=${message.courierId} attempt=${message.attempt} accepted=${message.accepted}"
)
worker.settle(offer, message.accepted, message.reason)

The handler kinds are covered by Handlers and Message Processing.

7. Reassignment on Timeout

The deadline is checked by Dispatch's sweeper. Each period it takes the expired offers and calls reassignment; no handler waits for a courier.

↗ View larger

Server/Dispatch/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/dispatch/OfferDeadlineSweeper.kt

private suspend fun sweep() {
    for (offer in offers.takeExpired()) {
        println(
            "deliverydispatch dispatch: offer expired delivery=${offer.request.deliveryId} " +
                "attempt=${offer.attempt}"
        )
        try {
            worker.reassign(offer)
        } catch (error: RuntimeException) {
            // The sweeper comes back next tick; one failed reassign must not stop it.
            System.err.println(
                "deliverydispatch dispatch: reassign failed " +
                    "delivery=${offer.request.deliveryId}: ${error.message}"
            )
        }
    }
}

Reassignment repeats the same procedure with the next candidate under a new attempt. When no candidate is left, the delivery ends in the Failed status.

Server/Dispatch/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/dispatch/DispatchWorker.kt

suspend fun reassign(offer: DeliveryOffer) {
    val nextIndex = offer.candidateIndex + 1
    if (nextIndex >= candidates.size) {
        println(
            "deliverydispatch-dispatch failed delivery=${offer.request.deliveryId} " +
                "reason=candidates-exhausted"
        )
        publishStatus(offer.request, DeliveryStatus.Failed, candidates.last())
        offers.close(offer.request.deliveryId)
        return
    }

    val courierId = candidates[nextIndex]
    publishStatus(offer.request, DeliveryStatus.Reassigned, courierId)
    val attempt = offers.offer(offer.request, nextIndex, SampleTimings.CourierDecisionTimeout)
    offer(offer.request, courierId, attempt)
}

The sweeper is a periodic job of the host, not a framework timer — the deadline is not the state of any Spot. The form to use when periodic execution inside a Spot is needed is covered by Timers and Workers.

8. Status Recording and the Customer Push

A status change reaches Tracking as a request on the tracking channel. Tracking records the evidence and forwards it to CustomerGateway with an Actor direct send addressed by the customer id.

Server/Tracking/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/tracking/handlers/DeliveryStatusChangedHandler.kt

evidenceStore.append(request)
if (request.status.name == "Delivered") {
    println("deliverydispatch-tracking status=Delivered delivery=${request.deliveryId}")
}
val actorRef =
    actorRefs.find(request.customerId).await().orElseThrow {
        IllegalStateException("customer actor not found: ${request.customerId}")
    }
kotlinActors
    .sendToActor(
        actorRef.actorId,
        DeliveryStatusUpdatedMsg(
            deliveryId = request.deliveryId,
            customerId = request.customerId,
            status = request.status,
            courierId = request.courierId,
            occurredAt = request.occurredAt,
        ),
    )
    .await()

The customer Actor's Entry Spot handler receives it and pushes a status notify to the bound session. When the customer reconnects, a new session is bound to the same Actor, so the push goes to the new connection.

Server/CustomerGateway/src/main/kotlin/systems/zlink/samples/kotlin/deliverydispatch/server/customergateway/spots/handlers/DeliveryStatusUpdatedHandler.kt

override suspend fun handle(
    entrySpot: CustomerEntrySpot,
    actor: CustomerActor,
    context: ZLinkMessageContext,
    message: DeliveryStatusUpdatedMsg,
) {
    actor.push(
        DeliveryStatusNotify(
            deliveryId = message.deliveryId,
            status = message.status,
            courierId = message.courierId,
            occurredAt = message.occurredAt,
        )
    )
    if (message.status.name == "Delivered") {
        println(
            "deliverydispatch-customer pushed status=Delivered " +
                "delivery=${message.deliveryId}"
        )
    }
}

The path to a bound session and the refresh after reconnecting are covered by How Session Binding Works.

9. Running and Verifying

One runner starts the Redis container, the server processes and the client scenario together.

framework/languages/java/samples/kotlin/DeliveryDispatch/run_sample.sh

The client asserts the order of the status notifies and their DeliveryId in both the normal dispatch flow and the timeout reassignment flow. The checks and the exact log strings are set by the DeliveryDispatch scenario.