1use std::any::Any;
29use std::cell::{Cell, RefCell};
30use std::io::Write;
31use std::panic::{self, AssertUnwindSafe};
32use std::sync::mpsc::{self, RecvTimeoutError};
33use std::sync::{Arc, Mutex};
34use std::thread::{self, JoinHandle};
35use std::time::Duration;
36
37use crate::console::{Console, ConsoleOptions};
38use crate::control::Control;
39use crate::file_proxy::{FileProxy, ProxyBuffer, ProxyStream, ProxyTarget};
40use crate::live_render::VerticalOverflow;
41use crate::protocol::Renderable;
42use crate::segment::Segment;
43
44#[derive(Debug, Clone, Copy, PartialEq)]
48pub struct LiveOptions {
49 pub transient: bool,
51 pub screen: bool,
53 pub redirect_stdout: bool,
56 pub redirect_stderr: bool,
58 pub vertical_overflow: VerticalOverflow,
61 pub refresh_on_start: bool,
65}
66
67impl Default for LiveOptions {
68 fn default() -> Self {
69 LiveOptions {
70 transient: false,
71 screen: false,
72 redirect_stdout: true,
73 redirect_stderr: true,
74 vertical_overflow: VerticalOverflow::Ellipsis,
75 refresh_on_start: true,
76 }
77 }
78}
79
80type Snapshot = Arc<Vec<Segment>>;
83
84struct LiveEntry {
97 content: Mutex<Option<Snapshot>>,
100 frame: Mutex<Frame>,
104}
105
106#[derive(Clone)]
108enum Slot {
109 Own,
111 Other(Snapshot),
113}
114
115struct Frame {
118 shape: Option<(usize, usize)>,
120 vertical_overflow: VerticalOverflow,
121 slots: Vec<Slot>,
124 generation: u64,
126 stale: bool,
129 hooked: bool,
131 started: bool,
132 transient: bool,
133 alt_screen: bool,
134 interactive: bool,
136}
137
138fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
140 mutex
141 .lock()
142 .unwrap_or_else(std::sync::PoisonError::into_inner)
143}
144
145impl LiveEntry {
146 fn new(interactive: bool) -> Self {
147 LiveEntry {
148 content: Mutex::new(None),
149 frame: Mutex::new(Frame {
150 shape: None,
151 vertical_overflow: VerticalOverflow::Ellipsis,
152 slots: vec![Slot::Own],
153 generation: 0,
154 stale: false,
155 hooked: false,
156 started: false,
157 transient: false,
158 alt_screen: false,
159 interactive,
160 }),
161 }
162 }
163
164 fn frame(&self) -> std::sync::MutexGuard<'_, Frame> {
165 lock(&self.frame)
166 }
167
168 fn content(&self) -> Option<Snapshot> {
169 lock(&self.content).clone()
170 }
171
172 fn set_content(&self, content: Snapshot) {
173 *lock(&self.content) = Some(content);
174 }
175}
176
177fn entry_of(item: &Arc<dyn Any + Send + Sync>) -> Option<Arc<LiveEntry>> {
179 item.clone().downcast::<LiveEntry>().ok()
180}
181
182fn group_slots(stack: &[Arc<dyn Any + Send + Sync>], first: &Arc<LiveEntry>) -> Vec<Slot> {
185 stack
186 .iter()
187 .filter_map(entry_of)
188 .filter_map(|entry| {
189 if Arc::ptr_eq(&entry, first) {
190 Some(Slot::Own)
191 } else {
192 entry.content().map(Slot::Other)
193 }
194 })
195 .collect()
196}
197
198impl Frame {
199 fn set_slots(&mut self, slots: Vec<Slot>) {
200 self.slots = slots;
201 self.generation += 1;
202 }
203
204 fn draws(&self) -> bool {
207 self.hooked && (self.interactive || (!self.started && !self.transient))
208 }
209
210 fn arrange(&mut self, rendered: &str, live: &str, shape: (usize, usize)) -> String {
213 let reset = if !self.interactive {
214 String::new()
215 } else if self.alt_screen {
216 Control::home().as_str().to_string()
217 } else {
218 crate::live_render::position_cursor(self.shape)
219 .as_str()
220 .to_string()
221 };
222 self.shape = Some(shape);
223 format!("{reset}{rendered}{live}")
224 }
225
226 fn draw(&mut self, console: &Console, own: &Own<'_>, rendered: &str) -> String {
230 if !self.draws() {
231 return rendered.to_string();
232 }
233 let (live, shape, _) = render_frame(
234 console,
235 own,
236 &self.slots,
237 self.alt_screen,
238 self.vertical_overflow,
239 );
240 self.arrange(rendered, &live, shape)
241 }
242}
243
244enum Own<'a> {
246 Render(&'a dyn Renderable),
248 Snapshot(Option<Snapshot>),
250}
251
252struct FrameContent<'a> {
256 own: &'a Own<'a>,
257 slots: &'a [Slot],
258 alt_screen: bool,
259 stash: RefCell<Option<Snapshot>>,
260}
261
262struct FrameGroup<'a, 'b>(&'b FrameContent<'a>);
264
265impl Renderable for FrameGroup<'_, '_> {
266 fn rich_render(&self, console: &Console, options: &ConsoleOptions) -> Vec<Segment> {
267 let content = self.0;
268 let mut segments = Vec::new();
269 let mut first = true;
270 for slot in content.slots {
271 let rendered = match (slot, content.own) {
272 (Slot::Own, Own::Render(renderable)) => {
273 let rendered = console.render(*renderable, Some(options));
274 *content.stash.borrow_mut() = Some(Arc::new(rendered.clone()));
275 rendered
276 }
277 (Slot::Own, Own::Snapshot(snapshot)) => {
278 snapshot.as_deref().cloned().unwrap_or_default()
279 }
280 (Slot::Other(snapshot), _) => snapshot.as_ref().clone(),
281 };
282 if rendered.is_empty() {
284 continue;
285 }
286 if !first {
287 segments.push(Segment::line());
288 }
289 first = false;
290 segments.extend(rendered);
291 }
292 segments
293 }
294}
295
296impl Renderable for FrameContent<'_> {
297 fn rich_render(&self, console: &Console, options: &ConsoleOptions) -> Vec<Segment> {
298 if self.alt_screen {
299 crate::screen::render_screen(Some(&FrameGroup(self)), None, console, options)
300 } else {
301 FrameGroup(self).rich_render(console, options)
302 }
303 }
304}
305
306struct FrameRender<'a> {
308 content: FrameContent<'a>,
309 vertical_overflow: VerticalOverflow,
310 shape: Cell<(usize, usize)>,
311}
312
313impl Renderable for FrameRender<'_> {
314 fn rich_render(&self, console: &Console, options: &ConsoleOptions) -> Vec<Segment> {
315 let (segments, shape) = crate::live_render::render(
316 &self.content,
317 None,
318 self.vertical_overflow,
319 console,
320 options,
321 );
322 self.shape.set(shape);
323 segments
324 }
325
326 fn print_end_line(&self, _options: &ConsoleOptions) -> bool {
327 false
328 }
329}
330
331fn render_frame(
334 console: &Console,
335 own: &Own<'_>,
336 slots: &[Slot],
337 alt_screen: bool,
338 vertical_overflow: VerticalOverflow,
339) -> (String, (usize, usize), Option<Snapshot>) {
340 let render = FrameRender {
341 content: FrameContent {
342 own,
343 slots,
344 alt_screen,
345 stash: RefCell::new(None),
346 },
347 vertical_overflow,
348 shape: Cell::new((0, 0)),
349 };
350 let live = console.render_to_string(&render);
351 let shape = render.shape.get();
352 (live, shape, render.content.stash.into_inner())
353}
354
355pub struct Live<W: Write> {
365 console: Console,
366 writer: W,
367 content: Box<dyn Renderable>,
369 entry: Arc<LiveEntry>,
372 started: bool,
373 hooked: bool,
375 nested: bool,
378 transient: bool,
379 screen: bool,
380 alt_screen: bool,
382 redirect_stdout: bool,
383 redirect_stderr: bool,
384 vertical_overflow: VerticalOverflow,
385 stdout_buffer: ProxyBuffer,
386 stderr_buffer: ProxyBuffer,
387 get_renderable: Option<GetRenderable>,
390 pending: Option<Box<dyn Renderable>>,
393}
394
395type GetRenderable = Box<dyn Fn() -> Box<dyn Renderable>>;
397
398impl<W: Write> Live<W> {
399 pub fn new(renderable: Box<dyn Renderable>, console: Console, writer: W) -> Self {
402 let interactive = console.is_terminal() && !console.is_dumb_terminal();
403 Live {
404 console,
405 writer,
406 content: renderable,
407 entry: Arc::new(LiveEntry::new(interactive)),
408 started: false,
409 hooked: false,
410 nested: false,
411 transient: false,
412 screen: false,
413 alt_screen: false,
414 redirect_stdout: true,
415 redirect_stderr: true,
416 vertical_overflow: VerticalOverflow::Ellipsis,
417 stdout_buffer: ProxyBuffer::default(),
418 stderr_buffer: ProxyBuffer::default(),
419 get_renderable: None,
420 pending: None,
421 }
422 }
423
424 pub fn get_renderable(mut self, get: impl Fn() -> Box<dyn Renderable> + 'static) -> Self {
429 self.content = get();
430 self.get_renderable = Some(Box::new(get));
431 self
432 }
433
434 pub fn options(self, options: &LiveOptions) -> Self {
436 self.transient(options.transient)
437 .screen(options.screen)
438 .redirect_stdout(options.redirect_stdout)
439 .redirect_stderr(options.redirect_stderr)
440 .vertical_overflow(options.vertical_overflow)
441 }
442
443 pub fn transient(mut self, transient: bool) -> Self {
446 self.transient = transient;
447 self.entry.frame().transient = self.is_transient();
448 self
449 }
450
451 pub fn screen(mut self, screen: bool) -> Self {
455 self.screen = screen;
456 self.entry.frame().transient = self.is_transient();
457 self
458 }
459
460 pub fn redirect_stdout(mut self, redirect: bool) -> Self {
464 self.redirect_stdout = redirect;
465 self
466 }
467
468 pub fn redirect_stderr(mut self, redirect: bool) -> Self {
470 self.redirect_stderr = redirect;
471 self
472 }
473
474 pub fn vertical_overflow(mut self, vertical_overflow: VerticalOverflow) -> Self {
478 self.vertical_overflow = vertical_overflow;
479 self.entry.frame().vertical_overflow = vertical_overflow;
480 self
481 }
482
483 pub fn is_started(&self) -> bool {
486 self.started
487 }
488
489 pub fn console(&self) -> &Console {
491 &self.console
492 }
493
494 fn is_transient(&self) -> bool {
496 self.transient || self.screen
497 }
498
499 fn is_dumb_terminal(&self) -> bool {
501 self.console.is_dumb_terminal()
502 }
503
504 fn is_interactive(&self) -> bool {
506 self.console.is_terminal() && !self.console.is_dumb_terminal()
507 }
508
509 fn write_str(&mut self, text: &str) {
510 if !text.is_empty() {
511 let _ = self.writer.write_all(text.as_bytes());
512 }
513 }
514
515 fn control(&mut self, control: &Control) {
517 if !self.is_dumb_terminal() {
518 self.write_str(control.as_str());
519 }
520 }
521
522 fn show_cursor(&mut self, show: bool) {
524 if self.console.is_terminal() {
525 self.control(&Control::show_cursor(show));
526 }
527 }
528
529 fn set_alt_screen(&mut self, enable: bool) -> bool {
531 if self.console.is_terminal() && !self.console.legacy_windows() {
532 self.control(&Control::alt_screen(enable));
533 return true;
534 }
535 false
536 }
537
538 fn first(&self) -> Option<Arc<LiveEntry>> {
541 let stack = self.console.live_stack();
542 let first = entry_of(stack.first()?)?;
543 (!Arc::ptr_eq(&first, &self.entry)).then_some(first)
544 }
545
546 pub fn start(&mut self) {
549 self.start_with(true);
550 }
551
552 pub fn start_with(&mut self, refresh: bool) {
556 if self.started {
557 return;
558 }
559 self.started = true;
560 self.entry.frame().started = true;
561 if !self.console.set_live(self.entry.clone()) {
565 self.nested = true;
566 self.publish();
567 return;
568 }
569 if self.screen {
570 self.alt_screen = self.set_alt_screen(true);
571 }
572 self.show_cursor(false);
573 self.stdout_buffer = ProxyBuffer::default();
574 self.stderr_buffer = ProxyBuffer::default();
575 self.hooked = true;
576 {
577 let mut frame = self.entry.frame();
578 frame.hooked = true;
579 frame.alt_screen = self.alt_screen;
580 }
581 if refresh {
582 self.refresh();
583 }
584 }
585
586 pub fn update(&mut self, renderable: Box<dyn Renderable>) {
589 self.update_with(renderable, true);
590 }
591
592 pub fn update_with(&mut self, renderable: Box<dyn Renderable>, refresh: bool) {
596 self.get_renderable = None;
597 self.pending = Some(renderable);
598 if refresh {
599 self.refresh();
600 } else if self.started && (self.nested || self.console.live_stack().len() > 1) {
601 self.publish();
603 }
604 }
605
606 fn take_content(&mut self) {
608 if let Some(get) = &self.get_renderable {
609 self.content = get();
610 } else if let Some(renderable) = self.pending.take() {
611 self.content = renderable;
612 }
613 }
614
615 fn publish(&mut self) {
618 let alt_screen = match self.first() {
619 Some(first) => first.frame().alt_screen,
620 None => self.alt_screen,
621 };
622 let renderable = self.pending.as_deref().unwrap_or(self.content.as_ref());
623 let (_, _, stash) = render_frame(
624 &self.console,
625 &Own::Render(renderable),
626 &[Slot::Own],
627 alt_screen,
628 self.vertical_overflow,
629 );
630 if let Some(stash) = stash {
631 self.entry.set_content(stash);
632 }
633 }
634
635 pub fn refresh(&mut self) {
642 self.take_content();
644 if self.nested {
645 self.publish();
647 self.refresh_first();
648 return;
649 }
650 let stack = self.console.live_stack();
651 let slots = match stack.first().and_then(entry_of) {
652 Some(first) if Arc::ptr_eq(&first, &self.entry) => group_slots(&stack, &self.entry),
653 _ => vec![Slot::Own],
654 };
655 if self.is_interactive() || (!self.started && !self.is_transient()) {
656 self.emit_with("", Some(slots));
657 } else {
658 let mut frame = self.entry.frame();
659 frame.set_slots(slots);
660 frame.stale = false;
661 }
662 }
663
664 fn tick(&mut self) {
668 if self.nested {
669 self.take_content();
670 self.publish();
671 } else {
672 self.refresh();
673 }
674 }
675
676 fn refresh_first(&mut self) {
679 let stack = self.console.live_stack();
680 let Some(first) = stack.first().and_then(entry_of) else {
683 return;
684 };
685 if Arc::ptr_eq(&first, &self.entry) {
686 return;
687 }
688 let slots = group_slots(&stack, &first);
689 let own = Own::Snapshot(first.content());
690 let output = {
691 let mut frame = first.frame();
692 frame.set_slots(slots);
693 frame.stale = true;
694 frame.draw(&self.console, &own, "")
695 };
696 self.write_str(&output);
697 }
698
699 pub fn print(&mut self, renderable: &dyn Renderable) {
703 let rendered = self.console.capture(|console| console.print(renderable));
704 self.emit(&rendered);
705 }
706
707 pub fn print_str(&mut self, content: &str) {
710 let rendered = self.console.capture(|console| console.print_str(content));
711 self.emit(&rendered);
712 }
713
714 #[track_caller]
719 pub fn log(
720 &mut self,
721 objects: &[crate::console::LogObject<'_>],
722 options: &crate::console::LogOptions<'_>,
723 ) -> crate::errors::Result<()> {
724 let caller = std::panic::Location::caller();
725 self.log_at(objects, options, caller.file(), caller.line())
726 }
727
728 pub fn log_at(
731 &mut self,
732 objects: &[crate::console::LogObject<'_>],
733 options: &crate::console::LogOptions<'_>,
734 file: &str,
735 line: u32,
736 ) -> crate::errors::Result<()> {
737 let mut result = Ok(());
738 let rendered = self.console.capture(|console| {
739 result = console.log_at(objects, options, file, line);
740 });
741 self.emit(&rendered);
742 result
743 }
744
745 fn emit(&mut self, rendered: &str) {
748 self.emit_with(rendered, None);
749 }
750
751 fn emit_with(&mut self, rendered: &str, slots: Option<Vec<Slot>>) {
754 let entry = self.entry.clone();
755 if !self.hooked {
756 if let Some(slots) = slots {
757 entry.frame().set_slots(slots);
758 }
759 self.emit_through_first(rendered);
760 return;
761 }
762 let (slots, generation, draws, stale) = {
763 let mut frame = entry.frame();
764 if let Some(slots) = slots {
765 frame.set_slots(slots);
766 frame.stale = false;
767 }
768 let stale = std::mem::take(&mut frame.stale);
769 (frame.slots.clone(), frame.generation, frame.draws(), stale)
770 };
771 if stale {
772 self.take_content();
774 }
775 if !draws {
776 self.write_str(rendered);
777 return;
778 }
779 let (live, shape, stash) = render_frame(
782 &self.console,
783 &Own::Render(self.content.as_ref()),
784 &slots,
785 self.alt_screen,
786 self.vertical_overflow,
787 );
788 if let (Some(stash), None) = (&stash, &self.pending) {
789 entry.set_content(stash.clone());
790 }
791 let mut frame = entry.frame();
792 let output = if frame.generation == generation {
793 frame.arrange(rendered, &live, shape)
794 } else {
795 let own = Own::Snapshot(stash.or_else(|| entry.content()));
798 frame.draw(&self.console, &own, rendered)
799 };
800 if !output.is_empty() {
801 let _ = self.writer.write_all(output.as_bytes());
802 }
803 }
804
805 fn emit_through_first(&mut self, rendered: &str) {
808 let output = match self.first() {
809 Some(first) => {
810 let own = Own::Snapshot(first.content());
811 let mut frame = first.frame();
812 frame.draw(&self.console, &own, rendered)
813 }
814 None => rendered.to_string(),
815 };
816 self.write_str(&output);
817 }
818
819 pub fn stop(&mut self) {
823 if !self.started {
824 return;
825 }
826 self.started = false;
827 self.entry.frame().started = false;
828 self.console.clear_live();
829 if self.nested {
830 self.nested = false;
833 if !self.is_transient() {
834 self.take_content();
835 let content = self.content.as_ref();
836 let rendered = self.console.capture(|console| console.print(content));
837 self.emit_through_first(&rendered);
838 }
839 let _ = self.writer.flush();
840 return;
841 }
842 self.vertical_overflow = VerticalOverflow::Visible;
844 self.entry.frame().vertical_overflow = VerticalOverflow::Visible;
845 let alt_screen = self.alt_screen;
846 if !alt_screen {
847 self.refresh();
848 }
849 for stream in [ProxyStream::Stdout, ProxyStream::Stderr] {
853 if let Some(pending) = self.proxy_buffer(stream).flush() {
854 self.print_str(&pending);
855 }
856 }
857 self.hooked = false;
858 let entry = self.entry.clone();
859 let mut frame = entry.frame();
860 frame.hooked = false;
861 let height = frame.shape.map_or(0, |(_, height)| height);
864 if !alt_screen && self.console.is_terminal() && height > 0 {
865 self.write_str("\n");
866 }
867 self.show_cursor(true);
868 if alt_screen {
869 self.set_alt_screen(false);
870 }
871 if self.is_transient() && !alt_screen {
872 let restore = crate::live_render::restore_cursor(frame.shape);
873 self.control(&restore);
874 }
875 self.alt_screen = false;
877 frame.alt_screen = false;
878 drop(frame);
879 let _ = self.writer.flush();
880 }
881
882 fn abort(&mut self) {
886 if !self.started && !self.hooked {
890 return;
891 }
892 if self.started {
893 self.console.clear_live();
895 }
896 self.started = false;
897 {
898 let mut frame = self.entry.frame();
899 frame.started = false;
900 frame.hooked = false;
901 }
902 if std::mem::take(&mut self.nested) {
903 return;
905 }
906 self.hooked = false;
907 if self.console.is_terminal() {
908 let _ = write!(self.writer, "\n{}", Control::show_cursor(true).as_str());
909 if self.alt_screen {
910 let _ = write!(self.writer, "{}", Control::alt_screen(false).as_str());
911 }
912 let _ = self.writer.flush();
913 }
914 self.alt_screen = false;
915 self.entry.frame().alt_screen = false;
916 }
917
918 fn redirecting(&self, stream: ProxyStream) -> bool {
922 let enabled = match stream {
923 ProxyStream::Stdout => self.redirect_stdout,
924 ProxyStream::Stderr => self.redirect_stderr,
925 };
926 self.started && enabled && self.console.is_terminal()
927 }
928
929 fn proxy_buffer(&mut self, stream: ProxyStream) -> &mut ProxyBuffer {
930 match stream {
931 ProxyStream::Stdout => &mut self.stdout_buffer,
932 ProxyStream::Stderr => &mut self.stderr_buffer,
933 }
934 }
935
936 pub fn stdout(&mut self) -> FileProxy<&mut Self, std::io::Stdout> {
940 FileProxy::new(self, ProxyStream::Stdout, std::io::stdout())
941 }
942
943 pub fn stderr(&mut self) -> FileProxy<&mut Self, std::io::Stderr> {
945 FileProxy::new(self, ProxyStream::Stderr, std::io::stderr())
946 }
947
948 pub fn file_proxy<F: Write>(
951 &mut self,
952 stream: ProxyStream,
953 file: F,
954 ) -> FileProxy<&mut Self, F> {
955 FileProxy::new(self, stream, file)
956 }
957
958 pub fn writer(&self) -> &W {
960 &self.writer
961 }
962
963 pub fn into_writer(self) -> W {
965 self.writer
966 }
967}
968
969impl<W: Write> ProxyTarget for &mut Live<W> {
970 fn proxy_write(&mut self, stream: ProxyStream, bytes: &[u8]) -> bool {
971 if !self.redirecting(stream) {
972 return false;
973 }
974 if let Some(text) = self.proxy_buffer(stream).write(bytes) {
975 self.print(&text);
976 }
977 true
978 }
979
980 fn proxy_flush(&mut self, stream: ProxyStream) -> bool {
981 if !self.redirecting(stream) {
982 return false;
983 }
984 if let Some(pending) = self.proxy_buffer(stream).flush() {
985 self.print_str(&pending);
986 }
987 true
988 }
989}
990
991type LiveJob<W> = Box<dyn FnOnce(&mut Live<W>) + Send>;
993
994enum LiveMessage<W: Write> {
996 Update(Box<dyn Renderable + Send>),
998 Refresh,
1000 Run(LiveJob<W>),
1003 Stop,
1005}
1006
1007impl<W: Write + Send + 'static> Live<W> {
1008 pub fn spawn(
1014 renderable: Box<dyn Renderable + Send>,
1015 console: Console,
1016 writer: W,
1017 refresh_per_second: f64,
1018 ) -> AutoLive<W> {
1019 Live::spawn_options(
1020 renderable,
1021 console,
1022 writer,
1023 refresh_per_second,
1024 LiveOptions::default(),
1025 )
1026 }
1027
1028 pub fn spawn_with(
1031 renderable: Box<dyn Renderable + Send>,
1032 console: Console,
1033 writer: W,
1034 refresh_per_second: f64,
1035 transient: bool,
1036 ) -> AutoLive<W> {
1037 Live::spawn_options(
1038 renderable,
1039 console,
1040 writer,
1041 refresh_per_second,
1042 LiveOptions {
1043 transient,
1044 ..LiveOptions::default()
1045 },
1046 )
1047 }
1048
1049 pub fn spawn_options(
1051 renderable: Box<dyn Renderable + Send>,
1052 console: Console,
1053 writer: W,
1054 refresh_per_second: f64,
1055 options: LiveOptions,
1056 ) -> AutoLive<W> {
1057 Live::spawn_inner(
1058 move |console, writer| Live::new(renderable, console, writer),
1059 console,
1060 writer,
1061 refresh_per_second,
1062 options,
1063 )
1064 }
1065
1066 pub fn spawn_get_renderable(
1070 get: impl Fn() -> Box<dyn Renderable> + Send + 'static,
1071 console: Console,
1072 writer: W,
1073 refresh_per_second: f64,
1074 options: LiveOptions,
1075 ) -> AutoLive<W> {
1076 Live::spawn_inner(
1077 move |console, writer| {
1078 Live::new(Box::new(crate::text::Text::new("")), console, writer).get_renderable(get)
1079 },
1080 console,
1081 writer,
1082 refresh_per_second,
1083 options,
1084 )
1085 }
1086
1087 fn spawn_inner(
1088 make: impl FnOnce(Console, W) -> Live<W> + Send + 'static,
1089 console: Console,
1090 writer: W,
1091 refresh_per_second: f64,
1092 options: LiveOptions,
1093 ) -> AutoLive<W> {
1094 let (sender, receiver) = mpsc::channel::<LiveMessage<W>>();
1095 let (started, wait_started) = mpsc::channel::<()>();
1096 let interval = Duration::from_secs_f64(1.0 / refresh_per_second.max(f64::MIN_POSITIVE));
1097 let terminal = console.is_terminal();
1098 let handle = thread::spawn(move || {
1099 let mut live = make(console, writer).options(&options);
1102 let outcome = panic::catch_unwind(AssertUnwindSafe(|| {
1107 live.start_with(options.refresh_on_start);
1108 let _ = started.send(());
1109 loop {
1110 match receiver.recv_timeout(interval) {
1111 Ok(LiveMessage::Update(renderable)) => live.update(renderable),
1112 Ok(LiveMessage::Refresh) => live.refresh(),
1113 Err(RecvTimeoutError::Timeout) => live.tick(),
1114 Ok(LiveMessage::Run(job)) => job(&mut live),
1115 Ok(LiveMessage::Stop) | Err(RecvTimeoutError::Disconnected) => {
1117 live.stop();
1118 break;
1119 }
1120 }
1121 }
1122 }));
1123 let failure = outcome.err().map(|payload| {
1124 live.abort();
1125 panic_message(payload.as_ref())
1126 });
1127 (live.into_writer(), failure)
1128 });
1129 let _ = wait_started.recv();
1133 AutoLive {
1134 handle: LiveHandle {
1135 sender: Some(sender),
1136 redirect_stdout: options.redirect_stdout && terminal,
1137 redirect_stderr: options.redirect_stderr && terminal,
1138 holder: None,
1139 },
1140 thread: Some(handle),
1141 }
1142 }
1143}
1144
1145pub struct LiveHandle<W: Write> {
1150 sender: Option<mpsc::Sender<LiveMessage<W>>>,
1151 redirect_stdout: bool,
1152 redirect_stderr: bool,
1153 holder: Option<Holder>,
1156}
1157
1158pub(crate) type Holder = Arc<Mutex<Option<thread::ThreadId>>>;
1160
1161impl<W: Write> Clone for LiveHandle<W> {
1162 fn clone(&self) -> Self {
1163 LiveHandle {
1164 sender: self.sender.clone(),
1165 redirect_stdout: self.redirect_stdout,
1166 redirect_stderr: self.redirect_stderr,
1167 holder: self.holder.clone(),
1168 }
1169 }
1170}
1171
1172impl<W: Write + Send + 'static> LiveHandle<W> {
1173 pub(crate) fn detached() -> Self {
1176 LiveHandle {
1177 sender: None,
1178 redirect_stdout: false,
1179 redirect_stderr: false,
1180 holder: None,
1181 }
1182 }
1183
1184 pub(crate) fn with_holder(mut self, holder: Holder) -> Self {
1187 self.holder = Some(holder);
1188 self
1189 }
1190
1191 pub(crate) fn run(&self, job: impl FnOnce(&mut Live<W>) + Send + 'static) -> bool {
1196 let Some(sender) = &self.sender else {
1197 return false;
1198 };
1199 let held = self.holder.as_ref().is_some_and(|holder| {
1200 *holder
1201 .lock()
1202 .unwrap_or_else(|poisoned| poisoned.into_inner())
1203 == Some(thread::current().id())
1204 });
1205 if held {
1206 return sender.send(LiveMessage::Run(Box::new(job))).is_ok();
1207 }
1208 let (done, wait) = mpsc::channel();
1209 let job = move |live: &mut Live<W>| {
1210 job(live);
1211 let _ = done.send(());
1212 };
1213 if sender.send(LiveMessage::Run(Box::new(job))).is_err() {
1214 return false;
1215 }
1216 let _ = wait.recv();
1219 true
1220 }
1221
1222 pub fn print(&self, renderable: Box<dyn Renderable + Send>) {
1224 self.run(move |live| live.print(renderable.as_ref()));
1225 }
1226
1227 pub fn print_str(&self, content: impl Into<String>) {
1229 let content = content.into();
1230 self.run(move |live| live.print_str(&content));
1231 }
1232
1233 #[track_caller]
1237 pub fn log(
1238 &self,
1239 objects: &[crate::console::LogObject<'_>],
1240 options: &crate::console::LogOptions<'_>,
1241 ) -> crate::errors::Result<()> {
1242 let caller = std::panic::Location::caller();
1243 self.log_at(objects, options, caller.file(), caller.line())
1244 }
1245
1246 pub fn log_at(
1248 &self,
1249 objects: &[crate::console::LogObject<'_>],
1250 options: &crate::console::LogOptions<'_>,
1251 file: &str,
1252 line: u32,
1253 ) -> crate::errors::Result<()> {
1254 let call = crate::console::OwnedLog::new(objects, options, file, line);
1255 let result = Arc::new(Mutex::new(Ok(())));
1256 let slot = result.clone();
1257 self.run(move |live| {
1258 let mut outcome = Ok(());
1259 let rendered = live.console.capture(|console| outcome = call.log(console));
1260 live.emit(&rendered);
1261 *slot.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) = outcome;
1262 });
1263 let mut result = result
1264 .lock()
1265 .unwrap_or_else(|poisoned| poisoned.into_inner());
1266 std::mem::replace(&mut *result, Ok(()))
1267 }
1268
1269 fn redirecting(&self, stream: ProxyStream) -> bool {
1270 match stream {
1271 ProxyStream::Stdout => self.redirect_stdout,
1272 ProxyStream::Stderr => self.redirect_stderr,
1273 }
1274 }
1275
1276 pub fn file_proxy<F: Write>(&self, stream: ProxyStream, file: F) -> FileProxy<Self, F> {
1278 FileProxy::new(self.clone(), stream, file)
1279 }
1280}
1281
1282impl<W: Write + Send + 'static> ProxyTarget for LiveHandle<W> {
1283 fn proxy_write(&mut self, stream: ProxyStream, bytes: &[u8]) -> bool {
1284 if !self.redirecting(stream) {
1285 return false;
1286 }
1287 let bytes = bytes.to_vec();
1288 self.run(move |live| {
1289 (&mut *live).proxy_write(stream, &bytes);
1290 })
1291 }
1292
1293 fn proxy_flush(&mut self, stream: ProxyStream) -> bool {
1294 if !self.redirecting(stream) {
1295 return false;
1296 }
1297 self.run(move |live| {
1298 (&mut *live).proxy_flush(stream);
1299 })
1300 }
1301}
1302
1303pub struct AutoLive<W: Write + Send + 'static> {
1306 handle: LiveHandle<W>,
1307 thread: Option<JoinHandle<(W, Option<String>)>>,
1308}
1309
1310#[derive(Debug)]
1313pub struct LivePanic<W> {
1314 pub writer: W,
1317 pub message: String,
1319}
1320
1321fn panic_message(payload: &(dyn Any + Send)) -> String {
1323 if let Some(message) = payload.downcast_ref::<&str>() {
1324 (*message).to_string()
1325 } else if let Some(message) = payload.downcast_ref::<String>() {
1326 message.clone()
1327 } else {
1328 "<non-string panic payload>".to_string()
1329 }
1330}
1331
1332impl<W: Write + Send + 'static> AutoLive<W> {
1333 pub fn update(&self, renderable: Box<dyn Renderable + Send>) {
1335 self.send(LiveMessage::Update(renderable));
1336 }
1337
1338 pub fn refresh(&self) {
1340 self.send(LiveMessage::Refresh);
1341 }
1342
1343 fn send(&self, message: LiveMessage<W>) -> bool {
1344 match &self.handle.sender {
1345 Some(sender) => sender.send(message).is_ok(),
1346 None => false,
1347 }
1348 }
1349
1350 pub fn print(&self, renderable: Box<dyn Renderable + Send>) {
1352 self.handle.print(renderable);
1353 }
1354
1355 pub fn print_str(&self, content: impl Into<String>) {
1358 self.handle.print_str(content);
1359 }
1360
1361 #[track_caller]
1363 pub fn log(
1364 &self,
1365 objects: &[crate::console::LogObject<'_>],
1366 options: &crate::console::LogOptions<'_>,
1367 ) -> crate::errors::Result<()> {
1368 let caller = std::panic::Location::caller();
1369 self.handle
1370 .log_at(objects, options, caller.file(), caller.line())
1371 }
1372
1373 pub fn handle(&self) -> LiveHandle<W> {
1376 self.handle.clone()
1377 }
1378
1379 pub fn stdout(&self) -> FileProxy<LiveHandle<W>, std::io::Stdout> {
1382 self.handle
1383 .file_proxy(ProxyStream::Stdout, std::io::stdout())
1384 }
1385
1386 pub fn stderr(&self) -> FileProxy<LiveHandle<W>, std::io::Stderr> {
1388 self.handle
1389 .file_proxy(ProxyStream::Stderr, std::io::stderr())
1390 }
1391
1392 pub fn file_proxy<F: Write>(
1395 &self,
1396 stream: ProxyStream,
1397 file: F,
1398 ) -> FileProxy<LiveHandle<W>, F> {
1399 self.handle.file_proxy(stream, file)
1400 }
1401
1402 pub fn refresh_wait(&self) {
1406 let (done, wait) = mpsc::channel();
1407 let job = move |live: &mut Live<W>| {
1408 live.refresh();
1409 let _ = done.send(());
1410 };
1411 if self.send(LiveMessage::Run(Box::new(job))) {
1412 let _ = wait.recv();
1413 }
1414 }
1415
1416 pub fn stop(self) -> W {
1422 match self.try_stop() {
1423 Ok(writer) => writer,
1424 Err(failure) => failure.writer,
1425 }
1426 }
1427
1428 pub fn try_stop(mut self) -> Result<W, LivePanic<W>> {
1431 self.send(LiveMessage::Stop);
1432 let handle = self
1433 .thread
1434 .take()
1435 .expect("thread handle present until stop/drop");
1436 let (writer, failure) = match handle.join() {
1439 Ok(result) => result,
1440 Err(payload) => panic::resume_unwind(payload),
1441 };
1442 match failure {
1443 None => Ok(writer),
1444 Some(message) => Err(LivePanic { writer, message }),
1445 }
1446 }
1447}
1448
1449impl<W: Write + Send + 'static> Drop for AutoLive<W> {
1450 fn drop(&mut self) {
1451 if let Some(handle) = self.thread.take() {
1453 self.send(LiveMessage::Stop);
1454 let _ = handle.join();
1455 }
1456 }
1457}
1458
1459#[cfg(test)]
1460mod tests {
1461 use super::*;
1462 use crate::color::ColorSystem;
1463 use crate::text::Text;
1464
1465 fn console() -> Console {
1466 Console::builder()
1467 .force_terminal(true)
1468 .color_system(Some(ColorSystem::Truecolor))
1469 .width(20)
1470 .no_color(false)
1471 .build()
1472 }
1473
1474 #[test]
1475 fn redirected_live_emits_only_the_final_frame() {
1476 let console = Console::builder().force_terminal(false).width(20).build();
1477 let mut live = Live::new(Box::new(Text::new("first")), console, Vec::<u8>::new());
1478 live.start();
1479 live.update(Box::new(Text::new("last")));
1480 live.refresh();
1481 assert!(live.writer().is_empty());
1482 live.stop();
1483 assert_eq!(live.writer(), b"last");
1484 }
1485
1486 #[test]
1487 fn manual_refresh_stream_matches_upstream() {
1488 let mut live = Live::new(
1489 Box::new(Text::new("frame one")),
1490 console(),
1491 Vec::<u8>::new(),
1492 );
1493 live.start();
1494 live.update(Box::new(Text::new("frame two")));
1495 live.update(Box::new(Text::new("frame three")));
1496 live.stop();
1497
1498 let expected = "\x1b[?25lframe one\r\x1b[2Kframe two\r\x1b[2Kframe three\r\x1b[2Kframe three\n\x1b[?25h";
1501 assert_eq!(String::from_utf8(live.writer().clone()).unwrap(), expected);
1502 }
1503
1504 #[test]
1505 fn auto_refresh_thread_produces_the_same_stream() {
1506 let auto = Live::spawn(
1511 Box::new(Text::new("frame one")),
1512 console(),
1513 Vec::<u8>::new(),
1514 0.1,
1515 );
1516 auto.update(Box::new(Text::new("frame two")));
1517 auto.update(Box::new(Text::new("frame three")));
1518 let output = auto.stop();
1519
1520 let expected = "\x1b[?25lframe one\r\x1b[2Kframe two\r\x1b[2Kframe three\r\x1b[2Kframe three\n\x1b[?25h";
1521 assert_eq!(String::from_utf8(output).unwrap(), expected);
1522 }
1523
1524 struct PanicsLater {
1526 renders: std::sync::Arc<std::sync::atomic::AtomicUsize>,
1527 fail_after: usize,
1528 }
1529
1530 impl Renderable for PanicsLater {
1531 fn rich_render(
1532 &self,
1533 console: &Console,
1534 options: &crate::console::ConsoleOptions,
1535 ) -> Vec<crate::segment::Segment> {
1536 use std::sync::atomic::Ordering;
1537 if self.renders.fetch_add(1, Ordering::SeqCst) >= self.fail_after {
1538 panic!("render failed");
1539 }
1540 Text::new("ok").rich_render(console, options)
1541 }
1542 }
1543
1544 #[test]
1545 fn a_final_refresh_that_panics_still_restores_the_cursor() {
1546 let auto = Live::spawn(
1550 Box::new(PanicsLater {
1551 renders: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
1552 fail_after: 1,
1553 }),
1554 console(),
1555 Vec::<u8>::new(),
1556 0.1,
1557 );
1558 let (output, error) = match auto.try_stop() {
1559 Ok(output) => (output, None),
1560 Err(failure) => (failure.writer, Some(failure.message)),
1561 };
1562 let output = String::from_utf8(output).unwrap();
1563 assert!(
1564 output.ends_with("\x1b[?25h"),
1565 "cursor left hidden: {output:?}"
1566 );
1567 assert_eq!(error.as_deref(), Some("render failed"));
1568 }
1569
1570 #[test]
1571 fn a_panicking_render_still_restores_the_cursor() {
1572 let renders = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
1576 let auto = Live::spawn(
1577 Box::new(PanicsLater {
1578 renders: renders.clone(),
1579 fail_after: 1,
1580 }),
1581 console(),
1582 Vec::<u8>::new(),
1583 0.1,
1584 );
1585 auto.refresh_wait(); let (output, error) = match auto.try_stop() {
1587 Ok(output) => (output, None),
1588 Err(failure) => (failure.writer, Some(failure.message)),
1589 };
1590 let output = String::from_utf8(output).unwrap();
1591 assert!(output.starts_with("\x1b[?25lok"), "{output:?}");
1592 assert!(
1593 output.ends_with("\x1b[?25h"),
1594 "cursor left hidden: {output:?}"
1595 );
1596 assert_eq!(error.as_deref(), Some("render failed"));
1597
1598 let auto = Live::spawn(
1600 Box::new(PanicsLater {
1601 renders: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
1602 fail_after: 1,
1603 }),
1604 console(),
1605 Vec::<u8>::new(),
1606 0.1,
1607 );
1608 auto.refresh();
1609 let output = String::from_utf8(auto.stop()).unwrap();
1610 assert!(
1611 output.ends_with("\x1b[?25h"),
1612 "cursor left hidden: {output:?}"
1613 );
1614 }
1615
1616 #[test]
1617 fn dropping_the_handle_finalizes_the_display() {
1618 let console = console();
1621 let auto = Live::spawn(Box::new(Text::new("only")), console, Vec::<u8>::new(), 0.1);
1623 drop(auto); }
1628 #[derive(Clone, Default)]
1630 struct SharedSink(Arc<Mutex<Vec<u8>>>);
1631
1632 impl Write for SharedSink {
1633 fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
1634 self.0.lock().unwrap().extend_from_slice(bytes);
1635 Ok(bytes.len())
1636 }
1637
1638 fn flush(&mut self) -> std::io::Result<()> {
1639 Ok(())
1640 }
1641 }
1642
1643 impl SharedSink {
1644 fn output(&self) -> String {
1645 String::from_utf8(self.0.lock().unwrap().clone()).unwrap()
1646 }
1647 }
1648
1649 fn within_time(scenario: impl FnOnce() + Send + 'static) {
1652 let (done, wait) = mpsc::channel();
1653 let worker = thread::spawn(move || {
1654 scenario();
1655 let _ = done.send(());
1656 });
1657 match wait.recv_timeout(Duration::from_secs(30)) {
1658 Ok(()) => worker.join().unwrap(),
1659 Err(RecvTimeoutError::Disconnected) => {
1660 if let Err(payload) = worker.join() {
1661 panic::resume_unwind(payload);
1662 }
1663 }
1664 Err(RecvTimeoutError::Timeout) => panic!("nested displays deadlocked"),
1665 }
1666 }
1667
1668 fn spawn_text(console: &Console, sink: &SharedSink, content: &str) -> AutoLive<SharedSink> {
1669 Live::spawn(
1670 Box::new(Text::new(content)),
1671 console.clone(),
1672 sink.clone(),
1673 500.0,
1674 )
1675 }
1676
1677 #[test]
1678 fn a_nested_display_is_drawn_inside_the_first() {
1679 let console = console();
1680 let sink = SharedSink::default();
1681 let mut outer = Live::new(Box::new(Text::new("outer")), console.clone(), sink.clone());
1682 outer.start();
1683 let mut inner = Live::new(Box::new(Text::new("inner")), console.clone(), sink.clone());
1684 inner.start();
1685 inner.refresh();
1686 assert!(
1687 sink.output().ends_with("outer\ninner"),
1688 "{:?}",
1689 sink.output()
1690 );
1691 inner.stop();
1692 outer.stop();
1693 assert!(console.clone().set_live(Arc::new(())));
1696 console.clear_live();
1697 }
1698
1699 #[test]
1700 fn threaded_displays_start_and_stop_in_either_order() {
1701 for inner_first in [true, false] {
1702 within_time(move || {
1703 let console = console();
1704 let sink = SharedSink::default();
1705 let outer = spawn_text(&console, &sink, "outer");
1706 let inner = spawn_text(&console, &sink, "inner");
1707 for step in 0..50 {
1708 inner.update(Box::new(Text::new(format!("inner {step}"))));
1709 outer.refresh();
1710 inner.refresh_wait();
1711 inner.print_str("printed");
1712 }
1713 if inner_first {
1714 inner.stop();
1715 outer.stop();
1716 } else {
1717 outer.stop();
1718 inner.stop();
1719 }
1720 let output = sink.output();
1721 assert!(output.contains("outer\ninner 49"), "{output:?}");
1722 assert!(output.ends_with("\x1b[?25h") || output.ends_with('\n'));
1723 });
1724 }
1725 }
1726
1727 #[test]
1728 fn a_nested_display_stops_while_the_first_refreshes() {
1729 within_time(|| {
1730 let console = console();
1731 let sink = SharedSink::default();
1732 let outer = spawn_text(&console, &sink, "outer");
1733 for round in 0..200 {
1734 let mut inner = Live::new(
1735 Box::new(Text::new(format!("inner {round}"))),
1736 console.clone(),
1737 sink.clone(),
1738 );
1739 inner.start();
1740 inner.refresh();
1741 let nested = spawn_text(&console, &sink, "nested");
1742 nested.refresh_wait();
1743 drop(nested);
1744 inner.stop();
1745 }
1746 outer.stop();
1747 assert!(console.set_live(Arc::new(())));
1749 console.clear_live();
1750 });
1751 }
1752
1753 #[test]
1754 fn nested_progress_refreshes_from_inside_the_first_progress() {
1755 within_time(|| {
1759 let console = console();
1760 let sink = SharedSink::default();
1761 let first =
1762 crate::progress::Progress::new().start(console.clone(), sink.clone(), 500.0);
1763 let task = first.add_task("first", 10.0, 0.0);
1764 let nested =
1765 crate::progress::Progress::new().start(console.clone(), sink.clone(), 500.0);
1766 let inner_task = nested.add_task("nested", 10.0, 0.0);
1767 for _ in 0..20 {
1768 first.with(|progress| {
1769 progress.advance(task, 0.5);
1770 nested.advance(inner_task, 0.5);
1771 nested.refresh();
1772 });
1773 thread::sleep(Duration::from_millis(1));
1774 }
1775 nested.stop();
1776 first.stop();
1777 let output = sink.output();
1778 assert!(output.contains("first"), "{output:?}");
1779 assert!(output.contains("nested"), "{output:?}");
1780 });
1781 }
1782}