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.
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/main.cpp
auto dispatch_channel = options.add_client_server_channel (
sample_names_t::dispatch_route_channel);
dispatch_channel.server ()
.set_bind_host (host_from_tcp_endpoint (topology.dispatch_route_endpoint))
.listen (port_from_http_url (topology.dispatch_route_endpoint))
.add_handler_group ("dispatch");
dispatch_channel.client ();
options.add_client_server_channel (sample_names_t::tracking_route_channel).client ();
auto courier_mesh = options.add_route_mesh (sample_names_t::courier_actor_discovery);
courier_mesh.set_routing_id (zlink::routing_id_t::from (sample_names_t::dispatch_route_node));
courier_mesh.listen (topology.dispatch_spot_router_endpoint)
.channel (sample_names_t::courier_actor_discovery)
.client ();
courier_mesh.objects ().client ();
options.handlers ()
.group ("dispatch")
.add_send<assign_delivery_handler_t> ()
.add_send<offer_delivery_result_handler_t> ();
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/CourierActorNode/main.cpp
/* 배송원의 결정을 배차 쪽으로 돌려보내는 통로. */
options.add_client_server_channel (sample_names_t::dispatch_route_channel).client ();
auto actor_mesh = options.add_route_mesh (sample_names_t::courier_actor_discovery);
actor_mesh.set_routing_id (zlink::routing_id_t::from (instance_name));
actor_mesh.listen (spot_router_endpoint);
actor_mesh.channel (sample_names_t::courier_actor_discovery).server ();
actor_mesh.objects ()
.server ()
.add_entry_spot<courier_entry_spot_t, channel_client_t> ()
.add_actor_factory<courier_actor_t, courier_actor_factory_t> (
sample_names_t::courier_actor_type)
.disable_relocation ();
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/main.cpp
const auto request = payload.parse_json<bind_courier_session_req_t> ();
/* Global ActorId로 current owner를 찾거나 eligible node에 생성한다. Application은
* courier id에서 physical NodeRid를 계산하지 않는다. */
auto located = actors.get_or_create (sample_names_t::courier_actor_type,
request.courier_id,
ensure_courier_actor_req_t{request.courier_id});
if (!located) {
throw framework_exception_t (located.error_kind (),
located.error ()
? located.error ()->what ()
: "courier actor could not be located");
}
/* Ready 결과의 exact ActorRef는 Framework session bind에만 사용한다. Application
* message나 client reply에는 ActorRef와 physical route를 넣지 않는다. */
auto actor = co_await actors.bind_or_get (located.value ().ref ()).async ();
const auto actor_id = std::string (actor.actor_id ());
_bound_actors.insert (actor_id);
auto reply = co_await actor
.relay_request (bind_courier_session_req_t::packet_name,
zlink::message_t::from_json (
bind_courier_session_req_t{request.courier_id}))
.async ();
stream.reply_packet (reply).async ();
const std::string line = std::format ("deliverydispatch-courier bound courier={}\n",
request.courier_id);
std::cerr << line;
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.
Server/Dispatch/main.cpp
class create_delivery_http_handler_t
{
public:
using request_type = create_delivery_req_t;
using reply_type = create_delivery_res_t;
static constexpr const char *topic_name = "CreateDeliveryReq";
explicit create_delivery_http_handler_t (channel_client_t &channels) : _channels (channels) {}
create_delivery_res_t handle (const create_delivery_req_t &request)
{
/* 배차 투입은 응답 없는 one-way send(`AssignDeliveryMsg`)다. HTTP edge는 접수만
* 확인하고, 진행 상태는 Tracking 기록과 고객 stream push로 전달된다. */
_channels
.send (sample_names_t::dispatch_route_channel,
assign_delivery_msg_t{request.delivery_id,
request.customer_id,
request.pickup_address,
request.dropoff_address})
.async ();
const std::string line = std::format ("deliverydispatch api: created delivery={}\n",
request.delivery_id);
std::cerr << line;
return create_delivery_res_t{request.delivery_id};
}
private:
channel_client_t &_channels;
};
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/main.cpp
/* 첫 제안. Assigned를 기록하고 제안을 보낸 뒤 이 턴은 끝난다. */
task_t<void> start (const assign_delivery_msg_t &request)
{
const std::string assign_line = std::format (
"deliverydispatch dispatch: assign delivery={} customer={}\n",
request.delivery_id,
request.customer_id);
std::cerr << assign_line;
const auto &courier_id = _couriers.candidates ().front ();
co_await _statuses.publish (request, delivery_status_t::assigned, courier_id);
const auto attempt = _state.offer (request, 0, sample_timings_t::courier_decision_timeout);
co_await _offers.offer (request, courier_id, 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/main.cpp
task_t<void>
offer (const assign_delivery_msg_t &delivery, const std::string &courier_id, int attempt)
{
auto actor = co_await _directory.find (courier_id);
if (!actor) {
throw framework_exception_t (framework_error_kind_t::not_found,
"courier actor route was not found: " + courier_id);
}
co_await _actors
.send (actor->actor_id (),
offer_delivery_msg_t{courier_id,
delivery.delivery_id,
attempt,
delivery.pickup_address,
delivery.dropoff_address})
.async ();
}
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/CourierActorNode/main.cpp
task_t<void> offer_delivery (courier_actor_t &actor,
message_context_t &,
const offer_delivery_msg_t &message)
{
actor.offered_attempts[message.delivery_id] = message.attempt;
co_await actor.context ()
.bound_session ()
.send (offer_delivery_notify_t{message.courier_id,
message.delivery_id,
message.pickup_address,
message.dropoff_address})
.async ();
}
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/CourierActorNode/main.cpp
const auto offered = actor.offered_attempts.find (decision.delivery_id);
if (offered == actor.offered_attempts.end ()) {
const std::string line = std::format (
"deliverydispatch courier-actor: decision for an unknown offer delivery={}\n",
decision.delivery_id);
std::cerr << line;
return;
}
const auto attempt = offered->second;
actor.offered_attempts.erase (offered);
_channels
.send (sample_names_t::dispatch_route_channel,
offer_delivery_result_msg_t{decision.delivery_id,
decision.courier_id,
attempt,
decision.accepted,
decision.reason})
.async ();
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/main.cpp
for (auto &decision : pending.second) {
auto offer = state.settle (decision.delivery_id, decision.attempt);
if (!offer) {
const std::string stale_line = std::format (
"deliverydispatch-dispatch stale-decision-ignored delivery={} courier={} "
"attempt={}\n",
decision.delivery_id,
decision.courier_id,
decision.attempt);
std::cerr << stale_line;
continue;
}
_work.push_back (settle (worker, *offer, std::move (decision)));
}
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.
Server/Dispatch/main.cpp
void sweep (dispatch_state_t &state, dispatch_worker_t &worker)
{
reap ();
auto pending = state.take_pending ();
for (auto &assignment : pending.first)
_work.push_back (start (worker, std::move (assignment)));
for (auto &decision : pending.second) {
auto offer = state.settle (decision.delivery_id, decision.attempt);
if (!offer) {
const std::string stale_line = std::format (
"deliverydispatch-dispatch stale-decision-ignored delivery={} courier={} "
"attempt={}\n",
decision.delivery_id,
decision.courier_id,
decision.attempt);
std::cerr << stale_line;
continue;
}
_work.push_back (settle (worker, *offer, std::move (decision)));
}
for (const auto &offer : state.expired ()) {
const std::string expired_line = std::format (
"deliverydispatch dispatch: offer expired delivery={} attempt={}\n",
offer.request.delivery_id,
offer.attempt);
std::cerr << expired_line;
_work.push_back (reassign (worker, offer));
}
}
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/main.cpp
/* 시한이 지난 제안. 다음 후보로 재제안한다 — 제안 시한은 worker가 소유한다. */
task_t<void> reassign (const delivery_offer_t &offer)
{
const auto next_index = offer.candidate_index + 1;
if (next_index >= _couriers.candidates ().size ()) {
co_await _statuses.publish (
offer.request, delivery_status_t::failed, _couriers.candidates ().back ());
_state.close (offer.request.delivery_id);
const std::string failed_line = std::format (
"deliverydispatch-dispatch failed delivery={} reason=candidates-exhausted\n",
offer.request.delivery_id);
std::cerr << failed_line;
co_return;
}
const auto &courier_id = _couriers.candidates ()[next_index];
co_await _statuses.publish (offer.request, delivery_status_t::reassigned, courier_id);
const auto attempt = _state.offer (
offer.request, next_index, sample_timings_t::courier_decision_timeout);
co_await _offers.offer (offer.request, courier_id, 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/Handlers/tracking_handlers.hpp
_evidence.append (request);
auto actor_ref = co_await _actor_directory.find (request.customer_id);
if (!actor_ref) {
throw zlink::framework::framework_exception_t (
zlink::framework::framework_error_kind_t::not_found,
"customer actor route was not found");
}
_actors
.send (actor_ref->actor_id (),
delivery_status_updated_msg_t{request.delivery_id,
request.customer_id,
request.status,
request.courier_id,
request.occurred_at_unix_ms})
.async ();
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/main.cpp
void status_updated (customer_actor_t &actor,
message_context_t &,
const delivery_status_updated_msg_t &status)
{
auto customer_id = _sessions.customer_for_delivery (status.delivery_id);
if (!customer_id || *customer_id != actor.actor_id) {
const std::string ignored_line = std::format (
"deliverydispatch customer-entry: ignored status delivery={} actor={}\n",
status.delivery_id,
actor.actor_id);
std::cerr << ignored_line;
return;
}
actor.context ()
.bound_session ()
.send (delivery_status_notify_t{
status.delivery_id, status.status, status.courier_id, status.occurred_at_unix_ms})
.async ();
if (status.status == delivery_status_t::delivered) {
const std::string delivered_line = std::format (
"deliverydispatch-customer pushed status=Delivered delivery={}\n",
status.delivery_id);
std::cerr << delivered_line;
}
}
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.
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.
10. Related Documents¶
- Comparison with the other samples and how to choose: Picking a Sample
- Requirements, message contract and completion criteria: DeliveryDispatch scenario
- A layout that binds several Actors to one session: Reading Along: SupportChat