alice.pds.demo.boltless.dev
This repository has no description
1use std::future::Future;
2use std::net::SocketAddr;
3use std::pin::Pin;
4use std::sync::Arc;
5
6use bytes::Bytes;
7use futures::stream::{Stream, StreamExt};
8use http::{HeaderMap, StatusCode};
9use thiserror::Error;
10use tokio_tungstenite::tungstenite::{
11 Bytes as WsBytes, Message as TungsteniteMessage, protocol::CloseFrame as TungsteniteClose,
12 protocol::frame::coding::CloseCode as TungsteniteCloseCode,
13};
14use url::Url;
15
16#[derive(Debug, Error)]
17pub enum NetworkError {
18 #[error("connect: {0}")]
19 Connect(String),
20 #[error("timeout: {0}")]
21 Timeout(String),
22 #[error("redirect: {0}")]
23 Redirect(String),
24 #[error("transport: {0}")]
25 Transport(String),
26 #[error("body: {0}")]
27 Body(String),
28 #[error("protocol: {0}")]
29 Protocol(String),
30}
31
32pub struct HttpRequest {
33 pub url: Url,
34 pub headers: HeaderMap,
35}
36
37pub type BodyStream = Pin<Box<dyn Stream<Item = Result<Bytes, NetworkError>> + Send + 'static>>;
38
39pub struct HttpResponseHead {
40 pub status: StatusCode,
41 pub headers: HeaderMap,
42 pub content_length: Option<u64>,
43 pub body: BodyStream,
44}
45
46pub type HttpResult = Result<HttpResponseHead, NetworkError>;
47pub type HttpResponseFuture = Pin<Box<dyn Future<Output = HttpResult> + Send + 'static>>;
48
49pub trait HttpTransport: Send + Sync + 'static {
50 fn execute(&self, request: HttpRequest) -> HttpResponseFuture;
51}
52
53#[derive(Clone, Debug)]
54pub struct ReqwestHttp {
55 client: reqwest::Client,
56}
57
58impl ReqwestHttp {
59 pub fn new(client: reqwest::Client) -> Self {
60 Self { client }
61 }
62
63 pub fn shared(client: reqwest::Client) -> Arc<dyn HttpTransport> {
64 Arc::new(Self::new(client))
65 }
66}
67
68impl HttpTransport for ReqwestHttp {
69 fn execute(&self, request: HttpRequest) -> HttpResponseFuture {
70 let client = self.client.clone();
71 Box::pin(async move {
72 let resp = client
73 .get(request.url)
74 .headers(request.headers)
75 .send()
76 .await
77 .map_err(map_reqwest)?;
78 let status = resp.status();
79 let headers = resp.headers().clone();
80 let content_length = resp.content_length();
81 let body: BodyStream = Box::pin(
82 resp.bytes_stream()
83 .map(|chunk| chunk.map_err(|e| NetworkError::Body(e.to_string()))),
84 );
85 Ok(HttpResponseHead {
86 status,
87 headers,
88 content_length,
89 body,
90 })
91 })
92 }
93}
94
95/// Lets jacquard resolve identities over the workspace reqwest (0.13); jacquard's own
96/// `HttpClient` impl is against reqwest 0.12, which is built here without TLS.
97impl jacquard_common::http_client::HttpClient for ReqwestHttp {
98 type Error = reqwest::Error;
99
100 async fn send_http(
101 &self,
102 request: http::Request<Vec<u8>>,
103 ) -> Result<http::Response<Vec<u8>>, reqwest::Error> {
104 let (parts, body) = request.into_parts();
105 let mut req = self
106 .client
107 .request(parts.method, parts.uri.to_string())
108 .body(body);
109 for (name, value) in parts.headers.iter() {
110 req = req.header(name, value);
111 }
112
113 let resp = req.send().await?;
114 let mut builder = http::Response::builder().status(resp.status());
115 for (name, value) in resp.headers().iter() {
116 builder = builder.header(name, value);
117 }
118 let body = resp.bytes().await?.to_vec();
119 Ok(builder
120 .body(body)
121 .expect("response parts came from reqwest"))
122 }
123}
124
125fn map_reqwest(err: reqwest::Error) -> NetworkError {
126 let msg = err.to_string();
127 if err.is_timeout() {
128 NetworkError::Timeout(msg)
129 } else if err.is_connect() {
130 NetworkError::Connect(msg)
131 } else if err.is_redirect() {
132 NetworkError::Redirect(msg)
133 } else {
134 NetworkError::Transport(msg)
135 }
136}
137
138#[derive(Clone, Debug)]
139pub enum WsMessage {
140 Text(String),
141 Binary(Bytes),
142 Ping(Bytes),
143 Pong(Bytes),
144 Close { code: u16, reason: String },
145}
146
147pub type WsSendFuture<'a> = Pin<Box<dyn Future<Output = Result<(), NetworkError>> + Send + 'a>>;
148pub type WsMessageFuture<'a> =
149 Pin<Box<dyn Future<Output = Option<Result<WsMessage, NetworkError>>> + Send + 'a>>;
150
151pub trait WsSink: Send + 'static {
152 fn send<'a>(&'a mut self, message: WsMessage) -> WsSendFuture<'a>;
153}
154
155pub trait WsStream: Send + 'static {
156 fn next<'a>(&'a mut self) -> WsMessageFuture<'a>;
157}
158
159pub struct WsConn {
160 pub sink: Box<dyn WsSink>,
161 pub stream: Box<dyn WsStream>,
162}
163
164pub type WsConnectFuture =
165 Pin<Box<dyn Future<Output = Result<WsConn, NetworkError>> + Send + 'static>>;
166
167pub trait WsTransport: Send + Sync + 'static {
168 fn connect(&self, url: Url) -> WsConnectFuture;
169}
170
171pub type AddrGuard = Arc<dyn Fn(&[SocketAddr]) -> Result<(), NetworkError> + Send + Sync>;
172
173#[derive(Clone, Copy, Debug, Default)]
174pub struct TungsteniteWs;
175
176impl TungsteniteWs {
177 pub fn shared() -> Arc<dyn WsTransport> {
178 Arc::new(Self)
179 }
180}
181
182impl WsTransport for TungsteniteWs {
183 fn connect(&self, url: Url) -> WsConnectFuture {
184 Box::pin(async move {
185 let url_str = url.as_str().to_owned();
186 let (ws, _resp) = tokio_tungstenite::connect_async(&url_str)
187 .await
188 .map_err(|e| NetworkError::Connect(e.to_string()))?;
189 let (sink_inner, stream_inner) = futures::StreamExt::split(ws);
190 let sink: Box<dyn WsSink> = Box::new(TungsteniteSink { inner: sink_inner });
191 let stream: Box<dyn WsStream> = Box::new(TungsteniteStream {
192 inner: stream_inner,
193 });
194 Ok(WsConn { sink, stream })
195 })
196 }
197}
198
199pub struct GuardedWs {
200 guard: AddrGuard,
201}
202
203impl GuardedWs {
204 pub fn shared(guard: AddrGuard) -> Arc<dyn WsTransport> {
205 Arc::new(Self { guard })
206 }
207}
208
209impl WsTransport for GuardedWs {
210 fn connect(&self, url: Url) -> WsConnectFuture {
211 let guard = self.guard.clone();
212 Box::pin(async move {
213 let host = url
214 .host_str()
215 .ok_or_else(|| NetworkError::Connect("ws url missing host".to_owned()))?
216 .to_owned();
217 let port = url
218 .port_or_known_default()
219 .ok_or_else(|| NetworkError::Connect("ws url missing port".to_owned()))?;
220 let addrs: Vec<SocketAddr> = tokio::net::lookup_host((host.as_str(), port))
221 .await
222 .map_err(|e| NetworkError::Connect(e.to_string()))?
223 .collect();
224 guard(&addrs)?;
225 let addr = addrs
226 .into_iter()
227 .next()
228 .ok_or_else(|| NetworkError::Connect(format!("no addresses for {host}")))?;
229 let tcp = tokio::net::TcpStream::connect(addr)
230 .await
231 .map_err(|e| NetworkError::Connect(e.to_string()))?;
232 let (ws, _resp) = tokio_tungstenite::client_async_tls(url.as_str(), tcp)
233 .await
234 .map_err(|e| NetworkError::Connect(e.to_string()))?;
235 let (sink_inner, stream_inner) = futures::StreamExt::split(ws);
236 let sink: Box<dyn WsSink> = Box::new(TungsteniteSink { inner: sink_inner });
237 let stream: Box<dyn WsStream> = Box::new(TungsteniteStream {
238 inner: stream_inner,
239 });
240 Ok(WsConn { sink, stream })
241 })
242 }
243}
244
245type TungsteniteWsStream =
246 tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;
247
248struct TungsteniteSink {
249 inner: futures::stream::SplitSink<TungsteniteWsStream, TungsteniteMessage>,
250}
251
252impl WsSink for TungsteniteSink {
253 fn send<'a>(&'a mut self, message: WsMessage) -> WsSendFuture<'a> {
254 Box::pin(async move {
255 use futures::SinkExt;
256 self.inner
257 .send(message_to_tungstenite(message))
258 .await
259 .map_err(|e| NetworkError::Transport(e.to_string()))
260 })
261 }
262}
263
264struct TungsteniteStream {
265 inner: futures::stream::SplitStream<TungsteniteWsStream>,
266}
267
268impl WsStream for TungsteniteStream {
269 fn next<'a>(&'a mut self) -> WsMessageFuture<'a> {
270 Box::pin(async move {
271 let item = StreamExt::next(&mut self.inner).await?;
272 Some(
273 item.map_err(|e| NetworkError::Transport(e.to_string()))
274 .and_then(message_from_tungstenite),
275 )
276 })
277 }
278}
279
280fn message_to_tungstenite(message: WsMessage) -> TungsteniteMessage {
281 match message {
282 WsMessage::Text(text) => TungsteniteMessage::Text(text.into()),
283 WsMessage::Binary(bytes) => TungsteniteMessage::Binary(WsBytes::copy_from_slice(&bytes)),
284 WsMessage::Ping(bytes) => TungsteniteMessage::Ping(WsBytes::copy_from_slice(&bytes)),
285 WsMessage::Pong(bytes) => TungsteniteMessage::Pong(WsBytes::copy_from_slice(&bytes)),
286 WsMessage::Close { code, reason } => TungsteniteMessage::Close(Some(TungsteniteClose {
287 code: TungsteniteCloseCode::from(code),
288 reason: reason.into(),
289 })),
290 }
291}
292
293fn message_from_tungstenite(message: TungsteniteMessage) -> Result<WsMessage, NetworkError> {
294 match message {
295 TungsteniteMessage::Text(t) => Ok(WsMessage::Text(t.to_string())),
296 TungsteniteMessage::Binary(b) => Ok(WsMessage::Binary(Bytes::copy_from_slice(&b))),
297 TungsteniteMessage::Ping(b) => Ok(WsMessage::Ping(Bytes::copy_from_slice(&b))),
298 TungsteniteMessage::Pong(b) => Ok(WsMessage::Pong(Bytes::copy_from_slice(&b))),
299 TungsteniteMessage::Close(close) => {
300 let (code, reason) = close
301 .map(|c| (u16::from(c.code), c.reason.to_string()))
302 .unwrap_or((1000, String::new()));
303 Ok(WsMessage::Close { code, reason })
304 }
305 TungsteniteMessage::Frame(_) => Err(NetworkError::Protocol(
306 "tungstenite raw frame surfaced unexpectedly".to_owned(),
307 )),
308 }
309}