use crate::transport::http;
use crate::{HttpBody, HttpRequest, HttpResponse};
use http_body_util::BodyExt;
use hyper::body::Bytes;
use hyper::header::{ACCEPT, CONTENT_TYPE};
use hyper::http::HeaderValue;
use hyper::{Method, Uri};
use jsonrpsee_core::BoxError;
use jsonrpsee_types::{Id, RequestSer};
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use tower::{Layer, Service};
#[derive(Debug, thiserror::Error)]
#[error("ProxyGetRequestLayer path must start with `/`, got `{0}`")]
pub struct InvalidPath(String);
#[derive(Debug, Clone)]
pub struct ProxyGetRequestLayer {
path: String,
method: String,
}
impl ProxyGetRequestLayer {
pub fn new(path: impl Into<String>, method: impl Into<String>) -> Result<Self, InvalidPath> {
let path = path.into();
if !path.starts_with('/') {
return Err(InvalidPath(path));
}
Ok(Self { path, method: method.into() })
}
}
impl<S> Layer<S> for ProxyGetRequestLayer {
type Service = ProxyGetRequest<S>;
fn layer(&self, inner: S) -> Self::Service {
ProxyGetRequest::new(inner, &self.path, &self.method)
.expect("Path already validated in ProxyGetRequestLayer; qed")
}
}
#[derive(Debug, Clone)]
pub struct ProxyGetRequest<S> {
inner: S,
path: Arc<str>,
method: Arc<str>,
}
impl<S> ProxyGetRequest<S> {
pub fn new(inner: S, path: &str, method: &str) -> Result<Self, InvalidPath> {
if !path.starts_with('/') {
return Err(InvalidPath(path.to_string()));
}
Ok(Self { inner, path: Arc::from(path), method: Arc::from(method) })
}
}
impl<S, B> Service<HttpRequest<B>> for ProxyGetRequest<S>
where
S: Service<HttpRequest, Response = HttpResponse>,
S::Response: 'static,
S::Error: Into<BoxError> + 'static,
S::Future: Send + 'static,
B: http_body::Body<Data = Bytes> + Send + 'static,
B::Data: Send,
B::Error: Into<BoxError>,
{
type Response = S::Response;
type Error = BoxError;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
#[inline]
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx).map_err(Into::into)
}
fn call(&mut self, mut req: HttpRequest<B>) -> Self::Future {
let modify = self.path.as_ref() == req.uri() && req.method() == Method::GET;
let req = if modify {
*req.method_mut() = Method::POST;
*req.uri_mut() = Uri::from_static("/");
req.headers_mut().insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
req.headers_mut().insert(ACCEPT, HeaderValue::from_static("application/json"));
let bytes = serde_json::to_vec(&RequestSer::borrowed(&Id::Number(0), &self.method, None))
.expect("Valid request; qed");
let body = HttpBody::from(bytes);
req.map(|_| body)
} else {
req.map(HttpBody::new)
};
let fut = self.inner.call(req);
let res_fut = async move {
let res = fut.await.map_err(|err| err.into())?;
if !modify {
return Ok(res);
}
let mut body = http_body_util::BodyStream::new(res.into_body());
let mut bytes = Vec::new();
while let Some(frame) = body.frame().await {
let data = frame?.into_data().map_err(|e| format!("{e:?}"))?;
bytes.extend(data);
}
#[derive(serde::Deserialize, Debug)]
struct RpcPayload<'a> {
#[serde(borrow)]
result: &'a serde_json::value::RawValue,
}
let response = if let Ok(payload) = serde_json::from_slice::<RpcPayload>(&bytes) {
http::response::ok_response(payload.result.to_string())
} else {
http::response::internal_error()
};
Ok(response)
};
Box::pin(res_fut)
}
}