Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280use std::future::Future;use std::net::SocketAddr;use std::pin::Pin;use std::sync::Arc;
use bytes::Bytes;use futures::stream::{Stream, StreamExt};use http::{HeaderMap, StatusCode};use thiserror::Error;use tokio_tungstenite::tungstenite::{ Bytes as WsBytes, Message as TungsteniteMessage, protocol::CloseFrame as TungsteniteClose, protocol::frame::coding::CloseCode as TungsteniteCloseCode,};use url::Url;
#[derive(Debug, Error)]pub enum NetworkError { #[error("connect: {0}")] Connect(String), #[error("timeout: {0}")] Timeout(String), #[error("redirect: {0}")] Redirect(String), #[error("transport: {0}")] Transport(String), #[error("body: {0}")] Body(String), #[error("protocol: {0}")] Protocol(String),}
pub struct HttpRequest { pub url: Url, pub headers: HeaderMap,}
pub type BodyStream = Pin<Box<dyn Stream<Item = Result<Bytes, NetworkError>> + Send + 'static>>;
pub struct HttpResponseHead { pub status: StatusCode, pub headers: HeaderMap, pub content_length: Option<u64>, pub body: BodyStream,}
pub type HttpResult = Result<HttpResponseHead, NetworkError>;pub type HttpResponseFuture = Pin<Box<dyn Future<Output = HttpResult> + Send + 'static>>;
pub trait HttpTransport: Send + Sync + 'static { fn execute(&self, request: HttpRequest) -> HttpResponseFuture;}
#[derive(Clone, Debug)]pub struct ReqwestHttp { client: reqwest::Client,}
impl ReqwestHttp { pub fn new(client: reqwest::Client) -> Self { Self { client } }
pub fn shared(client: reqwest::Client) -> Arc<dyn HttpTransport> { Arc::new(Self::new(client)) }}
impl HttpTransport for ReqwestHttp { fn execute(&self, request: HttpRequest) -> HttpResponseFuture { let client = self.client.clone(); Box::pin(async move { let resp = client .get(request.url) .headers(request.headers) .send() .await .map_err(map_reqwest)?; let status = resp.status(); let headers = resp.headers().clone(); let content_length = resp.content_length(); let body: BodyStream = Box::pin( resp.bytes_stream() .map(|chunk| chunk.map_err(|e| NetworkError::Body(e.to_string()))), ); Ok(HttpResponseHead { status, headers, content_length, body, }) }) }}
fn map_reqwest(err: reqwest::Error) -> NetworkError { let msg = err.to_string(); if err.is_timeout() { NetworkError::Timeout(msg) } else if err.is_connect() { NetworkError::Connect(msg) } else if err.is_redirect() { NetworkError::Redirect(msg) } else { NetworkError::Transport(msg) }}
#[derive(Clone, Debug)]pub enum WsMessage { Text(String), Binary(Bytes), Ping(Bytes), Pong(Bytes), Close { code: u16, reason: String },}
pub type WsSendFuture<'a> = Pin<Box<dyn Future<Output = Result<(), NetworkError>> + Send + 'a>>;pub type WsMessageFuture<'a> = Pin<Box<dyn Future<Output = Option<Result<WsMessage, NetworkError>>> + Send + 'a>>;
pub trait WsSink: Send + 'static { fn send<'a>(&'a mut self, message: WsMessage) -> WsSendFuture<'a>;}
pub trait WsStream: Send + 'static { fn next<'a>(&'a mut self) -> WsMessageFuture<'a>;}
pub struct WsConn { pub sink: Box<dyn WsSink>, pub stream: Box<dyn WsStream>,}
pub type WsConnectFuture = Pin<Box<dyn Future<Output = Result<WsConn, NetworkError>> + Send + 'static>>;
pub trait WsTransport: Send + Sync + 'static { fn connect(&self, url: Url) -> WsConnectFuture;}
pub type AddrGuard = Arc<dyn Fn(&[SocketAddr]) -> Result<(), NetworkError> + Send + Sync>;
#[derive(Clone, Copy, Debug, Default)]pub struct TungsteniteWs;
impl TungsteniteWs { pub fn shared() -> Arc<dyn WsTransport> { Arc::new(Self) }}
impl WsTransport for TungsteniteWs { fn connect(&self, url: Url) -> WsConnectFuture { Box::pin(async move { let url_str = url.as_str().to_owned(); let (ws, _resp) = tokio_tungstenite::connect_async(&url_str) .await .map_err(|e| NetworkError::Connect(e.to_string()))?; let (sink_inner, stream_inner) = futures::StreamExt::split(ws); let sink: Box<dyn WsSink> = Box::new(TungsteniteSink { inner: sink_inner }); let stream: Box<dyn WsStream> = Box::new(TungsteniteStream { inner: stream_inner, }); Ok(WsConn { sink, stream }) }) }}
pub struct GuardedWs { guard: AddrGuard,}
impl GuardedWs { pub fn shared(guard: AddrGuard) -> Arc<dyn WsTransport> { Arc::new(Self { guard }) }}
impl WsTransport for GuardedWs { fn connect(&self, url: Url) -> WsConnectFuture { let guard = self.guard.clone(); Box::pin(async move { let host = url .host_str() .ok_or_else(|| NetworkError::Connect("ws url missing host".to_owned()))? .to_owned(); let port = url .port_or_known_default() .ok_or_else(|| NetworkError::Connect("ws url missing port".to_owned()))?; let addrs: Vec<SocketAddr> = tokio::net::lookup_host((host.as_str(), port)) .await .map_err(|e| NetworkError::Connect(e.to_string()))? .collect(); guard(&addrs)?; let addr = addrs .into_iter() .next() .ok_or_else(|| NetworkError::Connect(format!("no addresses for {host}")))?; let tcp = tokio::net::TcpStream::connect(addr) .await .map_err(|e| NetworkError::Connect(e.to_string()))?; let (ws, _resp) = tokio_tungstenite::client_async_tls(url.as_str(), tcp) .await .map_err(|e| NetworkError::Connect(e.to_string()))?; let (sink_inner, stream_inner) = futures::StreamExt::split(ws); let sink: Box<dyn WsSink> = Box::new(TungsteniteSink { inner: sink_inner }); let stream: Box<dyn WsStream> = Box::new(TungsteniteStream { inner: stream_inner, }); Ok(WsConn { sink, stream }) }) }}
type TungsteniteWsStream = tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;
struct TungsteniteSink { inner: futures::stream::SplitSink<TungsteniteWsStream, TungsteniteMessage>,}
impl WsSink for TungsteniteSink { fn send<'a>(&'a mut self, message: WsMessage) -> WsSendFuture<'a> { Box::pin(async move { use futures::SinkExt; self.inner .send(message_to_tungstenite(message)) .await .map_err(|e| NetworkError::Transport(e.to_string())) }) }}
struct TungsteniteStream { inner: futures::stream::SplitStream<TungsteniteWsStream>,}
impl WsStream for TungsteniteStream { fn next<'a>(&'a mut self) -> WsMessageFuture<'a> { Box::pin(async move { let item = StreamExt::next(&mut self.inner).await?; Some( item.map_err(|e| NetworkError::Transport(e.to_string())) .and_then(message_from_tungstenite), ) }) }}
fn message_to_tungstenite(message: WsMessage) -> TungsteniteMessage { match message { WsMessage::Text(text) => TungsteniteMessage::Text(text.into()), WsMessage::Binary(bytes) => TungsteniteMessage::Binary(WsBytes::copy_from_slice(&bytes)), WsMessage::Ping(bytes) => TungsteniteMessage::Ping(WsBytes::copy_from_slice(&bytes)), WsMessage::Pong(bytes) => TungsteniteMessage::Pong(WsBytes::copy_from_slice(&bytes)), WsMessage::Close { code, reason } => TungsteniteMessage::Close(Some(TungsteniteClose { code: TungsteniteCloseCode::from(code), reason: reason.into(), })), }}
fn message_from_tungstenite(message: TungsteniteMessage) -> Result<WsMessage, NetworkError> { match message { TungsteniteMessage::Text(t) => Ok(WsMessage::Text(t.to_string())), TungsteniteMessage::Binary(b) => Ok(WsMessage::Binary(Bytes::copy_from_slice(&b))), TungsteniteMessage::Ping(b) => Ok(WsMessage::Ping(Bytes::copy_from_slice(&b))), TungsteniteMessage::Pong(b) => Ok(WsMessage::Pong(Bytes::copy_from_slice(&b))), TungsteniteMessage::Close(close) => { let (code, reason) = close .map(|c| (u16::from(c.code), c.reason.to_string())) .unwrap_or((1000, String::new())); Ok(WsMessage::Close { code, reason }) } TungsteniteMessage::Frame(_) => Err(NetworkError::Protocol( "tungstenite raw frame surfaced unexpectedly".to_owned(), )), }}