hypershell_tokio_components/providers/
stream.rs

1use core::convert::Infallible;
2use core::iter::Once;
3use core::marker::PhantomData;
4
5use cgp::extra::handler::{Handler, HandlerComponent};
6use cgp::prelude::*;
7use futures::AsyncRead as FuturesAsyncRead;
8use futures::io::Cursor;
9use futures::stream::Iter;
10use tokio::io::{AsyncRead as TokioAsyncRead, AsyncReadExt as _};
11use tokio_util::compat::{Compat, FuturesAsyncReadCompatExt, TokioAsyncReadCompatExt};
12use tokio_util::io::ReaderStream;
13
14use crate::types::{FuturesAsyncReadStream, FuturesStream, TokioAsyncReadStream};
15
16#[cgp_new_provider]
17impl<Context, Code, Input> Handler<Context, Code, Input> for HandleTokioAsyncReadToBytes
18where
19    Context: CanRaiseAsyncError<std::io::Error>,
20    Input: Send + TokioAsyncRead + Unpin,
21    Code: Send,
22{
23    type Output = Vec<u8>;
24
25    async fn handle(
26        _context: &Context,
27        _tag: PhantomData<Code>,
28        mut input: Input,
29    ) -> Result<Vec<u8>, Context::Error> {
30        let mut output = Vec::new();
31
32        input
33            .read_to_end(&mut output)
34            .await
35            .map_err(Context::raise_error)?;
36
37        Ok(output)
38    }
39}
40
41#[cgp_new_provider]
42impl<Context, Code, Input> Handler<Context, Code, Input> for HandleTokioAsyncReadToString
43where
44    Context: CanRaiseAsyncError<std::io::Error>,
45    Input: Send + TokioAsyncRead + Unpin,
46    Code: Send,
47{
48    type Output = String;
49
50    async fn handle(
51        _context: &Context,
52        _tag: PhantomData<Code>,
53        mut input: Input,
54    ) -> Result<String, Context::Error> {
55        let mut output = String::new();
56
57        input
58            .read_to_string(&mut output)
59            .await
60            .map_err(Context::raise_error)?;
61
62        Ok(output)
63    }
64}
65
66#[cgp_new_provider]
67impl<Context, Code, Input> Handler<Context, Code, Input> for HandleBytesToTokioAsyncRead
68where
69    Context: CanRaiseAsyncError<std::io::Error>,
70    Input: Send + AsRef<[u8]> + Unpin,
71    Code: Send,
72{
73    type Output = TokioAsyncReadStream<Compat<Cursor<Input>>>;
74
75    async fn handle(
76        _context: &Context,
77        _tag: PhantomData<Code>,
78        input: Input,
79    ) -> Result<TokioAsyncReadStream<Compat<Cursor<Input>>>, Context::Error> {
80        Ok(Cursor::new(input).compat().into())
81    }
82}
83
84#[cgp_new_provider]
85impl<Context, Code, Input> Handler<Context, Code, Input> for HandleBytesToStream
86where
87    Context: CanRaiseAsyncError<std::io::Error>,
88    Input: Send + AsRef<[u8]> + Unpin,
89    Code: Send,
90{
91    type Output = FuturesStream<Iter<Once<Result<Input, Infallible>>>>;
92
93    async fn handle(
94        _context: &Context,
95        _tag: PhantomData<Code>,
96        input: Input,
97    ) -> Result<Self::Output, Context::Error> {
98        Ok(futures::stream::iter(core::iter::once(Ok(input))).into())
99    }
100}
101
102#[cgp_new_provider]
103impl<Context, Code, Input> Handler<Context, Code, Input> for FuturesToTokioAsyncRead
104where
105    Context: HasAsyncErrorType,
106    Input: Send + FuturesAsyncRead + Unpin,
107    Code: Send,
108{
109    type Output = TokioAsyncReadStream<Compat<Input>>;
110
111    async fn handle(
112        _context: &Context,
113        _tag: PhantomData<Code>,
114        input: Input,
115    ) -> Result<TokioAsyncReadStream<Compat<Input>>, Context::Error> {
116        Ok(input.compat().into())
117    }
118}
119
120#[cgp_new_provider]
121impl<Context, Code, Input> Handler<Context, Code, Input> for TokioToFuturesAsyncRead
122where
123    Context: HasAsyncErrorType,
124    Input: Send + TokioAsyncRead + Unpin,
125    Code: Send,
126{
127    type Output = FuturesAsyncReadStream<Compat<Input>>;
128
129    async fn handle(
130        _context: &Context,
131        _tag: PhantomData<Code>,
132        input: Input,
133    ) -> Result<FuturesAsyncReadStream<Compat<Input>>, Context::Error> {
134        Ok(input.compat().into())
135    }
136}
137
138#[cgp_new_provider]
139impl<Context, Code, Input> Handler<Context, Code, Input> for WrapTokioAsyncRead
140where
141    Context: HasAsyncErrorType,
142    Input: Send + TokioAsyncRead + Unpin,
143    Code: Send,
144{
145    type Output = TokioAsyncReadStream<Input>;
146
147    async fn handle(
148        _context: &Context,
149        _tag: PhantomData<Code>,
150        input: Input,
151    ) -> Result<TokioAsyncReadStream<Input>, Context::Error> {
152        Ok(input.into())
153    }
154}
155
156#[cgp_new_provider]
157impl<Context, Code, Input> Handler<Context, Code, Input> for WrapFuturesAsyncRead
158where
159    Context: HasAsyncErrorType,
160    Input: Send + FuturesAsyncRead + Unpin,
161    Code: Send,
162{
163    type Output = FuturesAsyncReadStream<Input>;
164
165    async fn handle(
166        _context: &Context,
167        _tag: PhantomData<Code>,
168        input: Input,
169    ) -> Result<FuturesAsyncReadStream<Input>, Context::Error> {
170        Ok(input.into())
171    }
172}
173
174#[cgp_new_provider]
175impl<Context, Code, Input> Handler<Context, Code, Input> for AsyncReadToStream
176where
177    Context: HasAsyncErrorType,
178    Input: Send + TokioAsyncRead + Unpin,
179    Code: Send,
180{
181    type Output = FuturesStream<ReaderStream<Input>>;
182
183    async fn handle(
184        _context: &Context,
185        _tag: PhantomData<Code>,
186        input: Input,
187    ) -> Result<FuturesStream<ReaderStream<Input>>, Context::Error> {
188        Ok(ReaderStream::new(input).into())
189    }
190}
hypershell_tokio_components/providers/stream.rs

hypershell_tokio_components/providers/
stream.rs
