alloy_transport_http/
hyper_transport.rs

1use crate::{Http, HttpConnect};
2use alloy_json_rpc::{RequestPacket, ResponsePacket};
3use alloy_transport::{
4    utils::guess_local_url, BoxTransport, TransportConnect, TransportError, TransportErrorKind,
5    TransportFut, TransportResult,
6};
7use http_body_util::{BodyExt, Full};
8use hyper::{
9    body::{Bytes, Incoming},
10    header, Request, Response,
11};
12use hyper_util::client::legacy::Error;
13use std::{future::Future, marker::PhantomData, pin::Pin, task};
14use tower::Service;
15use tracing::{debug, debug_span, trace, Instrument};
16
17#[cfg(feature = "hyper-tls")]
18type Hyper = hyper_util::client::legacy::Client<
19    hyper_tls::HttpsConnector<hyper_util::client::legacy::connect::HttpConnector>,
20    http_body_util::Full<::hyper::body::Bytes>,
21>;
22
23#[cfg(not(feature = "hyper-tls"))]
24type Hyper = hyper_util::client::legacy::Client<
25    hyper_util::client::legacy::connect::HttpConnector,
26    http_body_util::Full<::hyper::body::Bytes>,
27>;
28
29/// A [`hyper`] based transport client.
30pub type HyperTransport = Http<HyperClient>;
31
32impl HyperTransport {
33    /// Create a new [`HyperTransport`] with the given URL and default hyper client.
34    pub fn new_hyper(url: url::Url) -> Self {
35        let client = HyperClient::new();
36        Self::with_client(client, url)
37    }
38}
39
40/// A [hyper] based client that can be used with tower layers.
41#[derive(Clone, Debug)]
42pub struct HyperClient<B = Full<Bytes>, S = Hyper> {
43    service: S,
44    _pd: PhantomData<B>,
45}
46
47/// Alias for [`Response<Incoming>`]
48pub type HyperResponse = Response<Incoming>;
49
50/// Alias for pinned box future that results in [`HyperResponse`]
51pub type HyperResponseFut<T = HyperResponse, E = Error> =
52    Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'static>>;
53
54impl HyperClient {
55    /// Create a new [HyperClient] with the given URL and default hyper client.
56    pub fn new() -> Self {
57        let executor = hyper_util::rt::TokioExecutor::new();
58
59        #[cfg(feature = "hyper-tls")]
60        let service = hyper_util::client::legacy::Client::builder(executor)
61            .build(hyper_tls::HttpsConnector::new());
62
63        #[cfg(not(feature = "hyper-tls"))]
64        let service =
65            hyper_util::client::legacy::Client::builder(executor).build_http::<Full<Bytes>>();
66        Self { service, _pd: PhantomData }
67    }
68}
69
70impl Default for HyperClient {
71    fn default() -> Self {
72        Self::new()
73    }
74}
75
76impl<B, S> HyperClient<B, S> {
77    /// Create a new [HyperClient] with the given URL and service.
78    pub const fn with_service(service: S) -> Self {
79        Self { service, _pd: PhantomData }
80    }
81}
82
83impl<B, S, ResBody> Http<HyperClient<B, S>>
84where
85    S: Service<Request<B>, Response = Response<ResBody>> + Clone + Send + Sync + 'static,
86    S::Future: Send,
87    S::Error: std::error::Error + Send + Sync + 'static,
88    B: From<Vec<u8>> + Send + 'static + Clone,
89    ResBody: BodyExt + Send + 'static,
90    ResBody::Error: std::error::Error + Send + Sync + 'static,
91    ResBody::Data: Send,
92{
93    async fn do_hyper(self, req: RequestPacket) -> TransportResult<ResponsePacket> {
94        debug!(count = req.len(), "sending request packet to server");
95        let ser = req.serialize().map_err(TransportError::ser_err)?;
96        // convert the Box<RawValue> into a hyper request<B>
97        let body = ser.get().as_bytes().to_owned().into();
98
99        let req = hyper::Request::builder()
100            .method(hyper::Method::POST)
101            .uri(self.url.as_str())
102            .header(header::CONTENT_TYPE, header::HeaderValue::from_static("application/json"))
103            .body(body)
104            .expect("request parts are invalid");
105
106        let mut service = self.client.service;
107        let resp = service.call(req).await.map_err(TransportErrorKind::custom)?;
108
109        let status = resp.status();
110
111        debug!(%status, "received response from server");
112
113        // Unpack data from the response body. We do this regardless of
114        // the status code, as we want to return the error in the body
115        // if there is one.
116        let body = resp.into_body().collect().await.map_err(TransportErrorKind::custom)?.to_bytes();
117
118        debug!(bytes = body.len(), "retrieved response body. Use `trace` for full body");
119        trace!(body = %String::from_utf8_lossy(&body), "response body");
120
121        if !status.is_success() {
122            return Err(TransportErrorKind::http_error(
123                status.as_u16(),
124                String::from_utf8_lossy(&body).into_owned(),
125            ));
126        }
127
128        // Deserialize a Box<RawValue> from the body. If deserialization fails, return
129        // the body as a string in the error. The conversion to String
130        // is lossy and may not cover all the bytes in the body.
131        serde_json::from_slice(&body)
132            .map_err(|err| TransportError::deser_err(err, String::from_utf8_lossy(body.as_ref())))
133    }
134}
135
136impl TransportConnect for HttpConnect<HyperTransport> {
137    fn is_local(&self) -> bool {
138        guess_local_url(self.url.as_str())
139    }
140
141    async fn get_transport(&self) -> Result<BoxTransport, TransportError> {
142        Ok(BoxTransport::new(Http::with_client(HyperClient::new(), self.url.clone())))
143    }
144}
145
146impl<B, S> Service<RequestPacket> for Http<HyperClient<B, S>>
147where
148    S: Service<Request<B>, Response = HyperResponse> + Clone + Send + Sync + 'static,
149    S::Future: Send,
150    S::Error: std::error::Error + Send + Sync + 'static,
151    B: From<Vec<u8>> + Send + 'static + Clone + Sync,
152{
153    type Response = ResponsePacket;
154    type Error = TransportError;
155    type Future = TransportFut<'static>;
156
157    #[inline]
158    fn poll_ready(&mut self, _cx: &mut task::Context<'_>) -> task::Poll<Result<(), Self::Error>> {
159        // `hyper` always returns `Ok(())`.
160        task::Poll::Ready(Ok(()))
161    }
162
163    #[inline]
164    fn call(&mut self, req: RequestPacket) -> Self::Future {
165        let this = self.clone();
166        let span = debug_span!("HyperTransport", url = %this.url);
167        Box::pin(this.do_hyper(req).instrument(span))
168    }
169}
alloy_transport_http/hyper_transport.rs

alloy_transport_http/
hyper_transport.rs
