1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
//! Inline route dispatcher — the camel-core side of the
//! [`InlineRouteDispatcher`] capability (direct-inline-dispatch Task 2.2).
//!
//! [`RouteInlineDispatcher`] mirrors the envelope drain path in
//! [`route_controller_trait`](super::route_controller_trait) stage for stage:
//! one pipeline snapshot per dispatch (ADR-0004 atomic-swap discipline), the
//! startup-cohort gate, `ready_with_backoff`, the `CANCEL_TOKEN` scope, and
//! `DrainGuard` accounting — except the Exchange is handed over in-process
//! instead of through the consumer channel.
//!
//! Logging: the dispatcher emits nothing on normal operation. The producer
//! that invokes it (camel-direct) owns the b′ emission for dispatch results.
//! Error paths resolve with `CamelError` values only.
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use camel_api::{CamelError, Exchange};
use camel_component_api::{InFlightClaim, InlineRouteDispatcher};
use tokio_util::sync::CancellationToken;
use tower::Service;
use crate::lifecycle::adapters::pipeline_runtime::SharedPipeline;
use crate::lifecycle::adapters::route_compiler::CANCEL_TOKEN;
use crate::lifecycle::adapters::route_helpers::{DrainGuard, ready_with_backoff};
use crate::lifecycle::cohort_activation::CohortActivationGate;
/// Completed hops between fairness yields.
const HOP_YIELD_INTERVAL: u32 = 32;
/// Interior-mutable dispatcher state, shared by every dispatch future.
struct DispatcherState {
route_id: String,
/// Pipeline swap source — the same `Arc<ArcSwap<..>>` the envelope
/// drain path loads its snapshot from.
pipeline: SharedPipeline,
/// The route's `pipeline_cancel` child token for this boot.
cancel: CancellationToken,
/// Drain counter shared with the envelope path; `DrainGuard` decrements
/// exactly once on every dispatch exit path.
drain_in_flight: Arc<AtomicU64>,
/// FIFO admission permit serializing concurrent producers through the
/// pipeline.
admission: Arc<tokio::sync::Mutex<()>>,
/// Startup-cohort barrier — the same gate the envelope drain sites park
/// on, so the barrier covers the inline topology too.
cohort: Arc<CohortActivationGate>,
/// Fairness yield counter, cumulative across ALL dispatches through this
/// dispatcher.
hop_budget: AtomicU32,
/// Context-global accepted-not-completed counter (drainclaim): each
/// dispatch mints one claim held across the pipeline call. `None` on
/// test harnesses built without a controller counter.
in_flight: Option<Arc<AtomicU64>>,
/// Test-only count of times the `yield_now` fairness site fired.
#[cfg(test)]
yields: AtomicU32,
}
/// Capability published on the [`ConsumerContext`](camel_component_api::ConsumerContext)
/// for non-Concurrent route topologies: runs an Exchange straight through the
/// route pipeline (request-reply) without a channel round-trip.
///
/// Constructed once per boot at the publication site in
/// [`route_controller_trait`](super::route_controller_trait), before the
/// consumer spawns (and therefore before any `mark_ready`).
pub(crate) struct RouteInlineDispatcher {
state: Arc<DispatcherState>,
}
impl std::fmt::Debug for RouteInlineDispatcher {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RouteInlineDispatcher")
.field("route_id", &self.state.route_id)
.finish()
}
}
impl RouteInlineDispatcher {
pub(crate) fn new(
route_id: String,
pipeline: SharedPipeline,
cancel: CancellationToken,
drain_in_flight: Arc<AtomicU64>,
cohort: Arc<CohortActivationGate>,
in_flight: Option<Arc<AtomicU64>>,
) -> Self {
Self {
state: Arc::new(DispatcherState {
route_id,
pipeline,
cancel,
drain_in_flight,
admission: Arc::new(tokio::sync::Mutex::new(())),
cohort,
hop_budget: AtomicU32::new(0),
in_flight,
#[cfg(test)]
yields: AtomicU32::new(0),
}),
}
}
#[cfg(test)]
fn hop_budget_for_test(&self) -> u32 {
self.state.hop_budget.load(Ordering::Relaxed)
}
#[cfg(test)]
fn yields_for_test(&self) -> u32 {
self.state.yields.load(Ordering::Relaxed)
}
#[cfg(test)]
fn admission_for_test(&self) -> &Arc<tokio::sync::Mutex<()>> {
&self.state.admission
}
}
impl InlineRouteDispatcher for RouteInlineDispatcher {
fn dispatch(
&self,
exchange: Exchange,
) -> Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send + 'static>> {
let state = Arc::clone(&self.state);
// drainclaim: mint at the acceptance boundary (the dispatch call
// itself). The claim moves into the boxed future and drops when
// the future completes, is dropped (producer abort), or panics.
let claim = state.in_flight.as_ref().map(InFlightClaim::attach);
Box::pin(async move {
// Named binding, not `claim`: an `async move` block only
// captures what it references — this line is what moves the
// claim in and holds it for the whole body.
let _in_flight_claim = claim;
// Drain accounting starts BEFORE the operation: the guard's Drop
// runs exactly once on every exit path — producer cancellation
// (future drop), consumer cancellation, success, or error.
let _drain_guard = DrainGuard::new(Arc::clone(&state.drain_in_flight));
// ONE snapshot for the whole call (ADR-0004 atomic-swap
// discipline; mirrors the envelope drain path).
let mut pipe = state.pipeline.load().processor.clone_inner();
let admission = Arc::clone(&state.admission);
let cohort = Arc::clone(&state.cohort);
let operation_cancel = state.cancel.clone();
// claimfamily (rc-hllkk): split a sibling claim onto the
// exchange so residency inside pipeline-embedded stash sites
// (resequencer buffers, aggregator buckets) stays counted
// after this future completes. Taken back from an in-band Ok
// result below — stash emissions escape with theirs.
let mut exchange = exchange;
exchange.in_flight_claim = _in_flight_claim.as_ref().map(InFlightClaim::split);
// Operation, strictly ordered: admission → cohort gate →
// readiness → scoped pipeline call. Dropping this future on the
// cancel arm below drops every stage and releases the admission
// permit.
let operation = async move {
// FIFO serialization of concurrent producers.
let _admission = admission.lock().await;
// Park until the startup cohort opens (level-triggered, same
// mechanism as the envelope drain sites).
let mut cohort_rx = cohort.subscribe();
let _ = cohort_rx.wait_for(|open| *open).await;
ready_with_backoff(&mut pipe, &operation_cancel).await?;
CANCEL_TOKEN
.scope(operation_cancel, async move { pipe.call(exchange).await })
.await
};
// Consumer-cancel wins ties: the biased select polls the cancel
// arm first.
let mut result = tokio::select! {
biased;
_ = state.cancel.cancelled() => Err(CamelError::ConsumerStopping),
result = operation => result,
};
// claimfamily: reclaim the sibling from an in-band result (the
// exchange completed inside this dispatch) so release stays at
// future end; a stash emission escaped with its claim.
if let Ok(ref mut ex) = result {
ex.in_flight_claim = None;
}
if result.is_ok() {
let prev = state.hop_budget.fetch_add(1, Ordering::Relaxed);
if (prev + 1).is_multiple_of(HOP_YIELD_INTERVAL) {
#[cfg(test)]
state.yields.fetch_add(1, Ordering::Relaxed);
tokio::task::yield_now().await;
}
}
result
})
}
}
#[cfg(test)]
#[path = "inline_dispatcher_tests.rs"]
mod tests;