tokio-process-tools 0.9.0

Correctness-focused async subprocess orchestration for Tokio: bounded output, multi-consumer streams, output detection, guaranteed cleanup and graceful termination.
Documentation 
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

//! Tokio-bound driver loops that step a [`StreamVisitor`] / [`AsyncStreamVisitor`] over a
//! [`Subscription`] until EOF, `Break`, cancellation, or a stream read error.
//!
//! The traits these loops drive are runtime-agnostic and live one level up at
//! [`crate::output_stream::visitor`]; the loops themselves are tokio-bound because they use
//! `tokio::select!` to race the next stream event against a `oneshot::Receiver` cancellation
//! token. The [`spawn_consumer_*`] helpers package the loop into a [`Consumer<S>`] handle by
//! `tokio::spawn`-ing the driver and wiring the cancellation channel.

use crate::StreamReadError;
use crate::output_stream::Next;
use crate::output_stream::Subscription;
use crate::output_stream::consumer::Consumer;
use crate::output_stream::event::StreamEvent;
use crate::output_stream::visitor::{AsyncStreamVisitor, StreamVisitor};
use tokio::sync::oneshot;

pub(crate) async fn consume_sync<S, V>(
    mut subscription: S,
    mut visitor: V,
    mut term_sig_rx: oneshot::Receiver<()>,
) -> Result<V::Output, StreamReadError>
where
    S: Subscription,
    V: StreamVisitor,
{
    loop {
        tokio::select! {
            out = subscription.next_event() => {
                match out {
                    Some(StreamEvent::Chunk(chunk)) => {
                        if visitor.on_chunk(chunk) == Next::Break {
                            break;
                        }
                    }
                    Some(StreamEvent::Gap) => visitor.on_gap(),
                    Some(StreamEvent::Eof) | None => {
                        visitor.on_eof();
                        break;
                    }
                    Some(StreamEvent::ReadError(err)) => return Err(err),
                }
            }
            _msg = &mut term_sig_rx => break,
        }
    }
    Ok(visitor.into_output())
}

pub(crate) async fn consume_async<S, V>(
    mut subscription: S,
    mut visitor: V,
    mut term_sig_rx: oneshot::Receiver<()>,
) -> Result<V::Output, StreamReadError>
where
    S: Subscription,
    V: AsyncStreamVisitor,
{
    loop {
        tokio::select! {
            out = subscription.next_event() => {
                match out {
                    Some(StreamEvent::Chunk(chunk)) => {
                        if visitor.on_chunk(chunk).await == Next::Break {
                            break;
                        }
                    }
                    Some(StreamEvent::Gap) => visitor.on_gap(),
                    Some(StreamEvent::Eof) | None => {
                        visitor.on_eof().await;
                        break;
                    }
                    Some(StreamEvent::ReadError(err)) => return Err(err),
                }
            }
            _msg = &mut term_sig_rx => break,
        }
    }
    Ok(visitor.into_output())
}

pub(crate) fn spawn_consumer_sync<S, V>(
    stream_name: &'static str,
    subscription: S,
    visitor: V,
) -> Consumer<V::Output>
where
    S: Subscription,
    V: StreamVisitor,
{
    let (term_sig_tx, term_sig_rx) = oneshot::channel::<()>();
    let driver = consume_sync(subscription, visitor, term_sig_rx);
    Consumer {
        stream_name,
        task: Some(tokio::spawn(driver)),
        task_termination_sender: Some(term_sig_tx),
    }
}

pub(crate) fn spawn_consumer_async<S, V>(
    stream_name: &'static str,
    subscription: S,
    visitor: V,
) -> Consumer<V::Output>
where
    S: Subscription,
    V: AsyncStreamVisitor,
{
    let (term_sig_tx, term_sig_rx) = oneshot::channel::<()>();
    let driver = consume_async(subscription, visitor, term_sig_rx);
    Consumer {
        stream_name,
        task: Some(tokio::spawn(driver)),
        task_termination_sender: Some(term_sig_tx),
    }
}