hypershell_tokio_components/providers/
streaming_exec.rs

1use core::marker::PhantomData;
2
3use cgp::extra::handler::{CanHandle, Handler, HandlerComponent};
4use cgp::prelude::*;
5use hypershell_components::dsl::StreamingExec;
6use tokio::io::{AsyncRead, Empty, copy, empty};
7use tokio::process::{Child, ChildStdout};
8use tokio::spawn;
9use tokio_util::either::Either;
10
11use crate::dsl::CoreExec;
12
13#[cgp_new_provider]
14impl<Context, CommandPath, Args, Input> Handler<Context, StreamingExec<CommandPath, Args>, Input>
15    for HandleStreamingExec
16where
17    Context: CanHandle<CoreExec<CommandPath, Args>, (), Output = Child>
18        + CanRaiseAsyncError<std::io::Error>,
19    CommandPath: Send,
20    Args: Send,
21    Input: Send + Unpin + AsyncRead + 'static,
22{
23    type Output = Either<ChildStdout, Empty>;
24
25    async fn handle(
26        context: &Context,
27        _tag: PhantomData<StreamingExec<CommandPath, Args>>,
28        mut input: Input,
29    ) -> Result<Either<ChildStdout, Empty>, Context::Error> {
30        let mut child = context.handle(PhantomData, ()).await?;
31
32        if let Some(mut stdin) = child.stdin.take() {
33            spawn(async move {
34                let _ = copy(&mut input, &mut stdin).await;
35            });
36        }
37
38        let output = match child.stdout.take() {
39            Some(stdout) => Either::Left(stdout),
40            None => Either::Right(empty()),
41        };
42
43        Ok(output)
44    }
45}

hypershell_tokio_components/providers/streaming_exec.rs

hypershell_tokio_components/providers/
streaming_exec.rs