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
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
#[cfg(all(test, feature = "hotpath"))]
pub mod tests {
use std::process::Command;
use hotpath::json::{JsonChannelsList, JsonReport};
// cargo run -p test-channels-std --example basic_std --features hotpath
#[test]
fn test_basic_output() {
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-channels-std",
"--example",
"basic_std",
"--features",
"hotpath",
])
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"Command failed with status: {}",
output.status
);
assert!(!output.stderr.is_empty(), "Stderr is empty");
let all_expected = ["Actor 1", "unbounded-channel", "bounded[10]"];
let stdout = String::from_utf8_lossy(&output.stdout);
for expected in all_expected {
assert!(
stdout.contains(expected),
"Expected:\n{expected}\n\nGot:\n{stdout}",
);
}
}
// cargo run -p test-channels-std --example basic_json_std --features hotpath
#[test]
fn test_basic_json_output() {
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-channels-std",
"--example",
"basic_json_std",
"--features",
"hotpath",
])
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"Command failed with status: {}",
output.status
);
let all_expected = ["\"label\": \"unbounded\"", "\"label\": \"bounded\""];
let stdout = String::from_utf8_lossy(&output.stdout);
for expected in all_expected {
assert!(
stdout.contains(expected),
"Expected:\n{expected}\n\nGot:\n{stdout}",
);
}
}
// cargo run -p test-channels-std --example closed_std --features hotpath
#[test]
fn test_closed_channels_output() {
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-channels-std",
"--example",
"closed_std",
"--features",
"hotpath",
])
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"Command failed with status: {}",
output.status
);
}
// cargo run -p test-channels-std --example iter_std --features hotpath
#[test]
fn test_iter_output() {
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-channels-std",
"--example",
"iter_std",
"--features",
"hotpath",
])
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"Command failed with status: {}",
output.status
);
let stdout = String::from_utf8_lossy(&output.stdout);
let all_expected = [
"Actor 1",
"Actor 1-2",
"Actor 1-3",
"bounded",
"bounded-2",
"bounded-3",
];
for expected in all_expected {
assert!(
stdout.contains(expected),
"Expected:\n{expected}\n\nGot:\n{stdout}",
);
}
}
// cargo run -p test-channels-std --example slow_consumer_std --features hotpath
#[test]
fn test_slow_consumer_no_panic() {
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-channels-std",
"--example",
"slow_consumer_std",
"--features",
"hotpath",
])
.output()
.expect("Failed to execute command");
let stdout = String::from_utf8_lossy(&output.stdout);
let stderr = String::from_utf8_lossy(&output.stderr);
assert!(
output.status.success(),
"Command failed with status: {}\nStdout:\n{}\nStderr:\n{}",
output.status,
stdout,
stderr
);
assert!(
stdout.contains("Slow consumer example completed!"),
"Expected completion message not found.\nOutput:\n{}",
stdout
);
}
// HOTPATH_METRICS_PORT=6770 TEST_SLEEP_SECONDS=10 cargo run -p test-channels-std --example basic_std --features hotpath
#[test]
fn test_data_endpoints() {
use std::{thread::sleep, time::Duration};
let mut child = Command::new("cargo")
.args([
"run",
"-p",
"test-channels-std",
"--example",
"basic_std",
"--features",
"hotpath",
])
.env("HOTPATH_METRICS_PORT", "6770")
.env("TEST_SLEEP_SECONDS", "10")
.spawn()
.expect("Failed to spawn command");
let mut json_text = String::new();
let mut last_error = None;
let all_expected = ["basic_std.rs", "unbounded-channel", "Actor 1"];
for _attempt in 0..12 {
sleep(Duration::from_millis(750));
match ureq::get("http://localhost:6770/channels").call() {
Ok(mut response) => {
json_text = response
.body_mut()
.read_to_string()
.expect("Failed to read response body");
last_error = None;
if all_expected.iter().all(|e| json_text.contains(e)) {
break;
}
}
Err(e) => {
last_error = Some(format!("Request error: {}", e));
}
}
}
if let Some(error) = last_error {
let _ = child.kill();
panic!("Failed after 12 retries: {}", error);
}
for expected in all_expected {
assert!(
json_text.contains(expected),
"Expected:\n{expected}\n\nGot:\n{json_text}",
);
}
let channels: JsonChannelsList =
serde_json::from_str(&json_text).expect("Failed to parse channels JSON");
if let Some(channel) = channels.data.first() {
let logs_url = format!("http://localhost:6770/channels/{}/logs", channel.id);
let response = ureq::get(&logs_url)
.call()
.expect("Failed to call /channels/:id/logs endpoint");
assert_eq!(
response.status(),
200,
"Expected status 200 for /channels/:id/logs endpoint"
);
}
let _ = child.kill();
let _ = child.wait();
}
// The report is followed by trailing log lines, so we locate the report's
// opening brace and read just the first JSON value from that point.
fn parse_channels(stdout: &str) -> JsonChannelsList {
let json_start = stdout.find('{').expect("No JSON report in output");
let report: JsonReport = serde_json::Deserializer::from_str(&stdout[json_start..])
.into_iter::<JsonReport>()
.next()
.expect("No JSON value in output")
.expect("Failed to parse JSON report");
report.channels.expect("No channels section in report")
}
fn run_example(name: &str) -> String {
let output = Command::new("cargo")
.args([
"run",
"-p",
"test-channels-std",
"--example",
name,
"--features",
"hotpath",
])
.output()
.expect("Failed to execute command");
assert!(
output.status.success(),
"Command failed with status: {}",
output.status
);
String::from_utf8_lossy(&output.stdout).into_owned()
}
// Bounded std requires `capacity`, which the macro accepts in any argument
// position. Every order of `capacity`/`label`/`log` must compile and register a channel.
//
// cargo run -p test-channels-std --example wrap_arg_orders_std --features hotpath
#[test]
fn test_capacity_arg_orders() {
let stdout = run_example("wrap_arg_orders_std");
let channels = parse_channels(&stdout);
for label in ["a", "b", "c", "d", "e", "f", "g", "h"] {
channels
.data
.iter()
.find(|c| c.label == label)
.unwrap_or_else(|| panic!("channel {label:?} not found"));
}
}
// The self-tracked queue counter reports the exact depth (50 messages parked,
// none received), where a forwarder would drain immediately and report ~0.
//
// cargo run -p test-channels-std --example wrap_std --features hotpath
#[test]
fn test_exact_queue_depth() {
let stdout = run_example("wrap_std");
let channels = parse_channels(&stdout);
let entry = channels
.data
.iter()
.find(|c| c.label == "wrap-queue")
.expect("wrap-queue channel not found");
assert_eq!(entry.sent_count, 50, "expected 50 sends");
assert_eq!(
entry.received_count, 0,
"expected 0 receives at report time"
);
assert_eq!(
entry.queue_size,
Some(50),
"expected exact queue depth of 50"
);
assert_eq!(
entry.max_queue_size,
Some(50),
"expected max queue depth of 50"
);
}
// Unbounded: every message sent and drained, queue back to zero with the
// high-water mark preserved.
//
// cargo run -p test-channels-std --example wrap_unbounded_std --features hotpath
#[test]
fn test_unbounded_sent_received() {
let stdout = run_example("wrap_unbounded_std");
let channels = parse_channels(&stdout);
let entry = channels
.data
.iter()
.find(|c| c.label == "wrap-unbounded")
.expect("wrap-unbounded channel not found");
assert_eq!(entry.sent_count, 200, "expected 200 sends");
assert_eq!(entry.received_count, 200, "expected 200 receives");
assert_eq!(entry.queue_size, Some(0), "expected drained queue");
assert_eq!(
entry.max_queue_size,
Some(200),
"expected max queue depth of 200"
);
}
// A producer racing a consumer on an unbounded channel must never underflow
// the depth counter (counting happens before each publish). `run_example` already
// asserts the process exited successfully - in debug builds an underflow would
// panic the consumer thread and fail that check. Here we additionally assert the
// counter never wrapped: a release-build underflow would surface as an absurd
// queue length, so `received <= sent` and a bounded `max_queue_size` confirm sanity.
//
// cargo run -p test-channels-std --example wrap_concurrent_std --features hotpath
#[test]
fn test_concurrent_no_underflow() {
let stdout = run_example("wrap_concurrent_std");
let channels = parse_channels(&stdout);
let entry = channels
.data
.iter()
.find(|c| c.label == "wrap-concurrent")
.expect("wrap-concurrent channel not found");
assert!(
entry.received_count <= entry.sent_count,
"received ({}) must not exceed sent ({})",
entry.received_count,
entry.sent_count
);
assert!(
entry.max_queue_size.unwrap_or(0) <= entry.sent_count as usize,
"max queue ({:?}) is absurd - the depth counter underflowed and wrapped",
entry.max_queue_size
);
}
// Dropping the single receiver while the sender is alive must mark the channel
// closed. std receivers are not Clone, so there is no clone-count path.
//
// cargo run -p test-channels-std --example wrap_closed_std --features hotpath
#[test]
fn test_receiver_dropped_closes() {
let stdout = run_example("wrap_closed_std");
let channels = parse_channels(&stdout);
let entry = channels
.data
.iter()
.find(|c| c.label == "recv-dropped")
.expect("recv-dropped channel not found");
assert_eq!(
entry.state.as_deref(),
Some("closed"),
"expected closed state after receiver drop"
);
}
}