1use scope::metrics::{Metric, Sample};
38use std::collections::VecDeque;
39use std::fmt;
40use std::io::{self, Write};
41use std::sync::atomic::{AtomicBool, Ordering};
42use std::sync::{Arc, Mutex};
43use std::time::Instant;
44
45type Sink = Arc<Mutex<dyn Write + Send>>;
47
48#[derive(Debug, Clone, Default)]
53pub struct RunProgress {
54 pub run: u64,
56 pub total: Option<u64>,
58 pub next_at: Option<Instant>,
60}
61
62pub type ProgressHook = Arc<Mutex<RunProgress>>;
64
65#[derive(Debug, Clone)]
67pub struct RunRecord {
68 pub run: u64,
70 pub at: chrono::DateTime<chrono::Utc>,
72 pub sample: Sample,
74 pub snapshot: String,
76 pub view: Option<Arc<SummaryView>>,
78}
79
80#[derive(Debug, Clone)]
82pub struct SummaryView {
83 pub summary: scope::market::MarketSummary,
84 pub venue: String,
86}
87
88pub type ViewHook = Arc<Mutex<Option<SummaryView>>>;
90
91#[derive(Debug, Default)]
93pub struct Series {
94 pub metrics: Vec<Metric>,
96 pub runs: VecDeque<RunRecord>,
98 pub dropped: u64,
100}
101
102pub type SeriesHook = Arc<Mutex<Series>>;
104
105pub struct Ask {
108 pub prompt: String,
109 pub options: Vec<String>,
110 pub reply: tokio::sync::oneshot::Sender<Option<usize>>,
111}
112
113pub type Asker = Arc<Mutex<Option<Ask>>>;
116
117pub const MAX_SERIES_RUNS: usize = 500;
119
120#[derive(Clone)]
124pub struct Output {
125 out: Sink,
126 err: Sink,
127 progress: bool,
128 hook: Option<ProgressHook>,
129 series: Option<SeriesHook>,
130 tui: bool,
131 prompts: bool,
133 asker: Option<Asker>,
135 view: Option<ViewHook>,
137}
138
139impl Output {
140 pub fn stdio() -> Self {
143 Self {
144 out: Arc::new(Mutex::new(io::stdout())),
145 err: Arc::new(Mutex::new(io::stderr())),
146 progress: true,
147 hook: None,
148 series: None,
149 tui: false,
150 prompts: true,
151 asker: None,
152 view: None,
153 }
154 }
155
156 pub fn capture() -> (Self, Captured) {
161 let out = Arc::new(Mutex::new(Vec::new()));
162 let err = Arc::new(Mutex::new(Vec::new()));
163 let captured = Captured {
164 data: out.clone(),
165 out: out.clone(),
166 err: err.clone(),
167 trimmed: Arc::default(),
168 };
169 let output = Self {
170 out,
171 err,
172 progress: false,
173 hook: None,
174 series: None,
175 tui: false,
176 prompts: false,
177 asker: None,
178 view: None,
179 };
180 (output, captured)
181 }
182
183 pub fn capture_merged() -> (Self, Captured) {
191 let buf = Arc::new(Mutex::new(Vec::new()));
192 let data = Arc::new(Mutex::new(Vec::new()));
193 let captured = Captured {
194 out: buf.clone(),
195 err: buf.clone(),
196 data: data.clone(),
197 trimmed: Arc::default(),
198 };
199 let output = Self {
200 out: Arc::new(Mutex::new(Tee {
201 merged: buf.clone(),
202 data,
203 })),
204 err: buf,
205 progress: false,
206 hook: Some(ProgressHook::default()),
207 series: Some(SeriesHook::default()),
208 tui: true,
209 prompts: false,
210 asker: None,
211 view: Some(ViewHook::default()),
212 };
213 (output, captured)
214 }
215
216 pub fn teed(&self, limit: usize) -> (Self, Captured) {
230 let buf = Arc::new(Mutex::new(Vec::new()));
231 let data = Arc::new(Mutex::new(Vec::new()));
232 let trimmed = Arc::new(AtomicBool::new(false));
233 let captured = Captured {
234 out: buf.clone(),
235 err: buf.clone(),
236 data: data.clone(),
237 trimmed: trimmed.clone(),
238 };
239 let copy = |real: &Sink, data: Option<Arc<Mutex<Vec<u8>>>>| -> Sink {
240 Arc::new(Mutex::new(TeeSink {
241 real: real.clone(),
242 merged: buf.clone(),
243 data,
244 limit,
245 trimmed: trimmed.clone(),
246 }))
247 };
248 let output = Self {
249 out: copy(&self.out, Some(data.clone())),
250 err: copy(&self.err, None),
251 progress: self.progress,
252 hook: None,
253 series: None,
254 tui: true,
255 prompts: self.prompts,
256 asker: None,
257 view: None,
258 };
259 (output, captured)
260 }
261
262 #[cfg(test)]
264 pub(crate) fn from_writers(
265 out: impl Write + Send + 'static,
266 err: impl Write + Send + 'static,
267 ) -> Self {
268 Self {
269 out: Arc::new(Mutex::new(out)),
270 err: Arc::new(Mutex::new(err)),
271 progress: false,
272 hook: None,
273 series: None,
274 tui: false,
275 prompts: true,
276 asker: None,
277 view: None,
278 }
279 }
280
281 pub fn progress_enabled(&self) -> bool {
283 self.progress
284 }
285
286 pub fn without_prompts(mut self) -> Self {
289 self.prompts = false;
290 self
291 }
292
293 pub fn with_asker(mut self, asker: Asker) -> Self {
296 self.asker = Some(asker);
297 self
298 }
299
300 pub fn asker(&self) -> Option<Asker> {
302 self.asker.clone()
303 }
304
305 pub fn prompts_allowed(&self) -> bool {
307 self.prompts
308 }
309
310 pub fn is_tui(&self) -> bool {
313 self.tui
314 }
315
316 pub fn progress_hook(&self) -> Option<ProgressHook> {
319 self.hook.clone()
320 }
321
322 pub fn series_hook(&self) -> Option<SeriesHook> {
325 self.series.clone()
326 }
327
328 pub fn view_hook(&self) -> Option<ViewHook> {
330 self.view.clone()
331 }
332
333 pub fn record_view(&self, view: SummaryView) -> bool {
336 let Some(slot) = &self.view else {
337 return false;
338 };
339 *slot.lock().unwrap_or_else(|e| e.into_inner()) = Some(view);
340 true
341 }
342
343 pub fn record_run(&self, metrics: &[Metric], record: RunRecord) -> bool {
346 let Some(series) = &self.series else {
347 return false;
348 };
349 let mut series = lock(series);
350 if series.metrics.is_empty() {
351 series.metrics = metrics.to_vec();
352 }
353 series.runs.push_back(record);
354 while series.runs.len() > MAX_SERIES_RUNS {
355 series.runs.pop_front();
356 series.dropped += 1;
357 }
358 true
359 }
360
361 pub fn update_progress(&self, change: impl FnOnce(&mut RunProgress)) {
363 if let Some(hook) = &self.hook {
364 change(&mut lock(hook));
365 }
366 }
367
368 pub fn write_out(&self, args: fmt::Arguments<'_>) -> io::Result<()> {
375 emit(&self.out, args)
376 }
377
378 pub fn write_err(&self, args: fmt::Arguments<'_>) -> io::Result<()> {
384 emit(&self.err, args)
385 }
386
387 pub fn note(&self, args: fmt::Arguments<'_>) {
391 let _ = emit(&self.err, args);
392 }
393
394 pub fn note_saved(&self, what: &str, path: impl fmt::Display) {
396 crate::note!(self, "{what} saved to: {path}");
397 }
398}
399
400#[derive(Clone)]
402pub struct Captured {
403 data: Arc<Mutex<Vec<u8>>>,
405 out: Arc<Mutex<Vec<u8>>>,
406 err: Arc<Mutex<Vec<u8>>>,
407 trimmed: Arc<AtomicBool>,
409}
410
411impl Captured {
412 pub fn trimmed(&self) -> bool {
415 self.trimmed.load(Ordering::Relaxed)
416 }
417
418 pub fn out(&self) -> String {
420 String::from_utf8_lossy(&lock(&self.out)).into_owned()
421 }
422
423 pub fn data(&self) -> String {
426 String::from_utf8_lossy(&lock(&self.data)).into_owned()
427 }
428
429 pub fn err(&self) -> String {
431 String::from_utf8_lossy(&lock(&self.err)).into_owned()
432 }
433
434 pub fn out_sink(&self) -> scope::diag::Sink {
437 self.out.clone()
438 }
439}
440
441struct Tee {
444 merged: Arc<Mutex<Vec<u8>>>,
445 data: Arc<Mutex<Vec<u8>>>,
446}
447
448impl Write for Tee {
449 fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
450 lock(&self.merged).extend_from_slice(buf);
451 lock(&self.data).extend_from_slice(buf);
452 Ok(buf.len())
453 }
454 fn flush(&mut self) -> io::Result<()> {
455 Ok(())
456 }
457}
458
459struct TeeSink {
462 real: Sink,
463 merged: Arc<Mutex<Vec<u8>>>,
464 data: Option<Arc<Mutex<Vec<u8>>>>,
466 limit: usize,
467 trimmed: Arc<AtomicBool>,
468}
469
470impl Write for TeeSink {
471 fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
474 for copy in std::iter::once(&self.merged).chain(self.data.as_ref()) {
475 let mut copy = lock(copy);
476 copy.extend_from_slice(buf);
477 if keep_tail(&mut copy, self.limit) {
478 self.trimmed.store(true, Ordering::Relaxed);
479 }
480 }
481 lock(&self.real).write_all(buf)?;
482 Ok(buf.len())
483 }
484 fn flush(&mut self) -> io::Result<()> {
485 lock(&self.real).flush()
486 }
487}
488
489fn keep_tail(buf: &mut Vec<u8>, limit: usize) -> bool {
495 if buf.len() <= limit {
496 return false;
497 }
498 let cut = buf.len() - limit / 2;
499 let start = buf[cut..]
500 .iter()
501 .position(|&b| b == b'\n')
502 .map(|i| cut + i + 1)
503 .filter(|&start| start < buf.len())
504 .unwrap_or_else(|| whole_cut(buf, cut));
505 buf.drain(..start);
506 true
507}
508
509fn whole_cut(buf: &[u8], cut: usize) -> usize {
513 let mut i = 0;
514 while i < cut {
515 i = crate::cli::ansi::escape_end(buf, i).unwrap_or(i + 1);
516 }
517 let continuation = |b: u8| b & 0b1100_0000 == 0b1000_0000;
518 buf[i.min(buf.len())..]
519 .iter()
520 .position(|&b| !continuation(b))
521 .map_or(buf.len(), |n| i + n)
522}
523
524fn lock<T: ?Sized>(m: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
527 m.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
528}
529
530fn emit(sink: &Mutex<dyn Write + Send>, args: fmt::Arguments<'_>) -> io::Result<()> {
535 lock(sink).write_fmt(args)
536}
537
538#[macro_export]
540macro_rules! outln {
541 ($o:expr) => {
542 $o.write_out(format_args!("\n"))
543 };
544 ($o:expr, $($arg:tt)*) => {
545 $o.write_out(format_args!("{}\n", format_args!($($arg)*)))
546 };
547}
548
549#[macro_export]
551macro_rules! out {
552 ($o:expr, $($arg:tt)*) => {
553 $o.write_out(format_args!($($arg)*))
554 };
555}
556
557#[macro_export]
559macro_rules! errln {
560 ($o:expr) => {
561 $o.write_err(format_args!("\n"))
562 };
563 ($o:expr, $($arg:tt)*) => {
564 $o.write_err(format_args!("{}\n", format_args!($($arg)*)))
565 };
566}
567
568#[macro_export]
571macro_rules! note {
572 ($o:expr) => {
573 $o.note(format_args!("\n"))
574 };
575 ($o:expr, $($arg:tt)*) => {
576 $o.note(format_args!("{}\n", format_args!($($arg)*)))
577 };
578}
579
580#[macro_export]
582macro_rules! err {
583 ($o:expr, $($arg:tt)*) => {
584 $o.write_err(format_args!($($arg)*))
585 };
586}
587
588#[cfg(test)]
593pub(crate) mod tests {
594 use super::*;
595
596 #[test]
597 fn test_capture_separates_data_from_diagnostics() {
598 let (o, cap) = Output::capture();
600 outln!(o, "{{\"ok\":{}}}", true).unwrap();
601 errln!(o, "Warning: {}", "partial data").unwrap();
602 assert_eq!(cap.out(), "{\"ok\":true}\n");
603 assert_eq!(cap.err(), "Warning: partial data\n");
604 }
605
606 #[test]
607 fn test_a_note_goes_to_the_diagnostic_channel_and_never_fails() {
608 let (o, cap) = Output::capture();
611 note!(o, "Fetching {}...", "USDC");
612 assert_eq!(cap.err(), "Fetching USDC...\n");
613 assert_eq!(cap.out(), "");
614 let closed = Output::from_writers(Vec::new(), ClosedPipe);
615 note!(closed, "dropped");
616 }
617
618 #[test]
619 fn test_note_saved_is_one_line_on_the_diagnostic_channel() {
620 let (o, cap) = Output::capture();
621 o.note_saved("Report", "r.md");
622 assert_eq!(cap.err(), "Report saved to: r.md\n");
623 assert_eq!(cap.out(), "");
624 }
625
626 #[test]
627 fn test_clones_share_writers() {
628 let (o, cap) = Output::capture();
630 let o2 = o.clone();
631 out!(o, "a").unwrap();
632 out!(o2, "b").unwrap();
633 outln!(o).unwrap();
634 assert_eq!(cap.out(), "ab\n");
635 }
636
637 pub(crate) struct ClosedPipe;
639
640 impl Write for ClosedPipe {
641 fn write(&mut self, _: &[u8]) -> io::Result<usize> {
642 Err(io::Error::new(io::ErrorKind::BrokenPipe, "closed"))
643 }
644 fn flush(&mut self) -> io::Result<()> {
645 Ok(())
646 }
647 }
648
649 #[test]
650 fn test_merged_capture_keeps_data_apart_from_diagnostics() {
651 let (o, cap) = Output::capture_merged();
652 outln!(o, "data").unwrap();
653 errln!(o, "warn").unwrap();
654 assert_eq!(cap.data(), "data\n");
655 assert_eq!(cap.out(), "data\nwarn\n");
656 }
657
658 #[test]
659 fn test_failed_write_is_returned_not_dropped() {
660 let o = Output::from_writers(ClosedPipe, ClosedPipe);
662 let e = outln!(o, "data").unwrap_err();
663 assert_eq!(e.kind(), io::ErrorKind::BrokenPipe);
664 assert_eq!(
665 errln!(o, "warn").unwrap_err().kind(),
666 io::ErrorKind::BrokenPipe
667 );
668 }
669
670 #[test]
671 fn test_only_the_tui_capture_is_tui() {
672 assert!(Output::capture_merged().0.is_tui());
673 assert!(!Output::capture().0.is_tui());
674 assert!(!Output::stdio().is_tui());
675 assert!(Output::stdio().teed(1).0.is_tui());
676 }
677
678 #[test]
679 fn test_capture_merged_keeps_write_order() {
680 let (o, cap) = Output::capture_merged();
683 outln!(o, "row 1").unwrap();
684 errln!(o, "warning").unwrap();
685 outln!(o, "row 2").unwrap();
686 assert_eq!(cap.out(), "row 1\nwarning\nrow 2\n");
687 }
688
689 #[test]
690 fn test_teed_output_reaches_the_terminal_and_keeps_a_copy() {
691 let (terminal, screen) = Output::capture();
694 let (o, copy) = terminal.teed(1024);
695 outln!(o, "row").unwrap();
696 errln!(o, "warn").unwrap();
697 assert_eq!(screen.out(), "row\n");
698 assert_eq!(screen.err(), "warn\n");
699 assert_eq!(copy.out(), "row\nwarn\n");
700 assert_eq!(copy.data(), "row\n");
701 assert!(!copy.trimmed());
702 assert!(o.is_tui());
705 assert_eq!(o.progress_enabled(), terminal.progress_enabled());
706 assert!(Output::stdio().teed(1).0.progress_enabled());
707 assert!(o.progress_hook().is_none());
708 }
709
710 #[test]
711 fn test_teed_copy_keeps_whole_lines_from_the_end_within_its_limit() {
712 let (terminal, screen) = Output::capture();
714 let (o, copy) = terminal.teed(100);
715 for i in 0..1000 {
716 outln!(o, "line {i:04}").unwrap();
717 }
718 assert!(copy.trimmed());
719 let kept = copy.out();
720 assert!(kept.len() <= 100, "{}", kept.len());
721 assert!(kept.starts_with("line "), "whole lines: {kept:?}");
722 assert!(kept.ends_with("line 0999\n"), "{kept:?}");
723 assert!(copy.data().ends_with("line 0999\n"));
724 assert_eq!(screen.out().lines().count(), 1000, "the terminal got all");
725 }
726
727 #[test]
728 fn test_teed_copy_keeps_the_tail_of_one_long_line() {
729 let (terminal, _screen) = Output::capture();
732 let (o, copy) = terminal.teed(100);
733 let blob: String = (0..300)
734 .map(|i| char::from(b'a' + (i % 26) as u8))
735 .collect();
736 out!(o, "{blob}").unwrap();
737 assert!(copy.trimmed());
738 let kept = copy.out();
739 assert!(!kept.is_empty() && kept.len() <= 100, "{}", kept.len());
740 assert!(blob.ends_with(&kept), "the end of the line: {kept:?}");
741 }
742
743 #[test]
744 fn test_trim_of_one_line_splits_no_character_and_no_escape() {
745 let mut chars = "é".repeat(60).into_bytes(); assert!(keep_tail(&mut chars, 102)); assert!(String::from_utf8(chars.clone()).is_ok(), "{chars:?}");
750 assert_eq!(chars.len(), 50);
751 let mut colored = b"x".repeat(60);
752 colored.extend_from_slice(b"\x1b[38;5;196mred");
753 assert!(keep_tail(&mut colored, 20));
755 assert_eq!(colored, b"red");
756 let mut link = b"y".repeat(60);
757 link.extend_from_slice(b"\x1b]8;;https://example.com/a/long/path\x1b\\text");
758 let cut_at = link.len() - 40 / 2;
760 assert!(cut_at > 60 && cut_at < 98, "{cut_at}");
761 assert!(keep_tail(&mut link, 40));
762 assert_eq!(link, b"text");
763 }
764
765 #[test]
766 fn test_teed_copy_keeps_text_the_terminal_refused() {
767 let (o, copy) = Output::from_writers(ClosedPipe, ClosedPipe).teed(64);
769 assert!(outln!(o, "data").is_err());
770 assert_eq!(copy.out(), "data\n");
771 }
772
773 #[test]
774 fn test_capture_disables_progress() {
775 let (o, _) = Output::capture();
777 assert!(!o.progress_enabled());
778 assert!(Output::stdio().progress_enabled());
779 }
780}