clientix_core/client/asynchronous/stream/
mod.rs

1pub mod sse;
2
3use std::net::SocketAddr;
4use std::pin::Pin;
5use std::task::{Context, Poll};
6use bytes::Bytes;
7use futures_core::Stream;
8use futures_util::{StreamExt, TryStreamExt};
9use http::{HeaderMap, StatusCode, Version};
10use reqwest::Url;
11use crate::client::asynchronous::stream::sse::ClientixSSEStream;
12use crate::client::response::ClientixResult;
13
14pub struct ClientixStream {
15    version: Version,
16    content_length: Option<u64>,
17    status: StatusCode,
18    url: Url,
19    remote_addr: Option<SocketAddr>,
20    headers: HeaderMap,
21    stream: Pin<Box<dyn Stream<Item = ClientixResult<Bytes>>>>,
22}
23
24impl ClientixStream {
25
26    pub fn new(
27        version: Version,
28        content_length: Option<u64>,
29        status: StatusCode,
30        url: Url,
31        remote_addr: Option<SocketAddr>,
32        headers: HeaderMap,
33        stream: impl Stream<Item = ClientixResult<Bytes>> + 'static
34    ) -> Self {
35        Self {
36            version,
37            content_length,
38            status,
39            url,
40            remote_addr,
41            headers,
42            stream: Box::pin(stream)
43        }
44    }
45    
46    pub fn sse(self) -> ClientixSSEStream<String> {
47        self.into()
48    }
49
50    pub fn version(&self) -> Version {
51        self.version
52    }
53
54    pub fn content_length(&self) -> Option<u64> {
55        self.content_length
56    }
57
58    pub fn status(&self) -> StatusCode {
59        self.status
60    }
61
62    pub fn url(&self) -> &Url {
63        &self.url
64    }
65
66    pub fn remote_addr(&self) -> Option<SocketAddr> {
67        self.remote_addr
68    }
69
70    pub fn headers(&self) -> &HeaderMap {
71        &self.headers
72    }
73
74    pub async fn execute<F>(mut self, mut handle: F) where F: FnMut(ClientixResult<Bytes>) {
75        while let Some(result) = self.stream.next().await {
76            handle(result);
77        }
78    }
79
80    pub async fn collect(self) -> ClientixResult<Vec<Bytes>> {
81        self.stream.try_collect().await
82    }
83    
84}
85
86impl Stream for ClientixStream {
87    type Item = ClientixResult<Bytes>;
88
89    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
90        self.stream.as_mut().poll_next(cx)
91    }
92}
clientix_core/client/asynchronous/stream/mod.rs

clientix_core/client/asynchronous/stream/
mod.rs
