use std::{collections::HashMap, fmt::Debug};
use bytes::Bytes;
use futures::TryStreamExt as _;
mod smol;
pub type BytesStream<E> = ccutils::futures::BoxedStream<Result<bytes::Bytes, E>>;
pub struct Request {
uri: String,
headers: HashMap<String, String>,
body: Option<Bytes>,
}
impl Request {
pub fn from_uri(uri: impl Into<String>) -> Request {
Self {
uri: uri.into(),
headers: Default::default(),
body: None,
}
}
pub fn body(mut self, body: impl Into<Bytes>) -> Self {
self.body = Some(body.into());
self
}
pub fn header(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
self.headers.insert(key.into(), value.into());
self
}
#[cfg(feature = "json")]
pub fn json(self, value: &impl serde::Serialize) -> Result<Self, serde_json::Error> {
Ok(self
.header("Content-Type", "application/json")
.body(Bytes::from(serde_json::to_vec(&value)?)))
}
}
pub trait HttpClientInterface {
type Response: Response;
fn wget(&self, request: Request) -> impl Future<Output = Self::Response> + Send;
fn wpost(&self, request: Request) -> impl Future<Output = Self::Response> + Send;
}
pub trait Response {
type StreamError: Debug + 'static;
fn into_stream(self) -> BytesStream<Self::StreamError>;
fn status(&self) -> http::StatusCode;
}
pub async fn get_all<TResponse>(resp: TResponse) -> Result<bytes::Bytes, TResponse::StreamError>
where
TResponse: Response,
{
resp.into_stream()
.try_fold(bytes::BytesMut::new(), |mut acc, chunk| {
acc.extend_from_slice(&chunk);
futures::future::ready(Ok(acc))
})
.await
.map(bytes::BytesMut::freeze)
}
#[cfg(any(
feature = "futures_runtime",
feature = "bevy_runtime",
feature = "bevy_runtime_018"
))]
mod smol_implementation {
use bytes::Bytes;
use futures::TryStreamExt as _;
use http_body_util::{BodyStream, Full};
pub struct Response {
response: hyper::Response<hyper::body::Incoming>,
}
impl super::Response for Response {
type StreamError = <hyper::body::Incoming as hyper::body::Body>::Error;
fn into_stream(self) -> super::BytesStream<Self::StreamError> {
ccutils::futures::pin_stream(
BodyStream::new(self.response.into_body())
.and_then(|frame| async move { Ok(frame.into_data().unwrap()) }),
)
}
fn status(&self) -> http::StatusCode {
self.response.status()
}
}
fn fill_request(
mut req: hyper::http::request::Builder,
yreq: super::Request,
) -> (hyper::http::request::Builder, Option<Bytes>) {
for (k, v) in yreq.headers.into_iter() {
req = req.header(k, v);
}
(req, yreq.body)
}
pub(crate) async fn wget_impl<RT>(this: &RT, mut yreq: super::Request) -> Response
where
RT: crate::TaskInterface,
{
let req = hyper::Request::get(std::mem::take(&mut yreq.uri));
let (req, _) = fill_request(req, yreq);
let response =
super::smol::fetch(this, req.body(Full::<Bytes>::new(Bytes::new())).unwrap())
.await
.unwrap();
Response { response }
}
pub(crate) async fn wpost_impl<RT>(this: &RT, mut yreq: super::Request) -> Response
where
RT: crate::TaskInterface,
{
let req = hyper::Request::post(std::mem::take(&mut yreq.uri));
let (req, body) = fill_request(req, yreq);
let body = body.unwrap_or_default();
let response = super::smol::fetch(this, req.body(Full::<Bytes>::new(body)).unwrap())
.await
.unwrap();
Response { response }
}
}
#[cfg(feature = "futures_runtime")]
mod futures_runtime_implementation {
use super::smol_implementation::*;
impl super::HttpClientInterface for crate::futures::Runtime {
type Response = Response;
fn wget(&self, request: super::Request) -> impl Future<Output = Self::Response> {
wget_impl(self, request)
}
fn wpost(&self, request: super::Request) -> impl Future<Output = Self::Response> {
wpost_impl(self, request)
}
}
}
#[cfg(feature = "tokio_runtime")]
mod tokio_implementation {
use bytes::Bytes;
use reqwest::RequestBuilder;
pub struct Response(reqwest::Response);
impl super::Response for Response {
type StreamError = reqwest::Error;
fn into_stream(self) -> super::BytesStream<Self::StreamError> {
ccutils::futures::pin_stream(self.0.bytes_stream())
}
fn status(&self) -> http::StatusCode {
self.0.status()
}
}
fn fill_request(req: RequestBuilder, yreq: super::Request) -> (RequestBuilder, Option<Bytes>) {
let mut req = req;
for (k, v) in yreq.headers.into_iter() {
req = req.header(k, v);
}
(req, yreq.body)
}
impl super::HttpClientInterface for crate::tokio::Runtime {
type Response = Response;
async fn wget(&self, yreq: super::Request) -> Self::Response {
let client = reqwest::Client::new();
let rb = client.get(&yreq.uri);
let (rb, _) = fill_request(rb, yreq);
Response(rb.send().await.unwrap())
}
async fn wpost(&self, yreq: super::Request) -> Self::Response {
let client = reqwest::Client::new();
let rb = client.post(&yreq.uri);
let (mut rb, body) = fill_request(rb, yreq);
if let Some(b) = body {
rb = rb.body(b);
}
Response(rb.send().await.unwrap())
}
}
}
#[cfg(any(feature = "bevy_runtime", feature = "bevy_runtime_018"))]
mod bevy_implementation {
use std::ops::Deref;
#[cfg(feature = "bevy_runtime")]
use bevy_tasks::TaskPool;
#[cfg(feature = "bevy_runtime_018")]
use bevy_tasks_018::TaskPool;
use super::smol_implementation::*;
impl<TP> super::HttpClientInterface for crate::bevy::Runtime<TP>
where
TP: Deref<Target = TaskPool> + Sync + 'static,
{
type Response = Response;
fn wget(&self, request: super::Request) -> impl Future<Output = Self::Response> {
wget_impl(self, request)
}
fn wpost(&self, request: super::Request) -> impl Future<Output = Self::Response> {
wpost_impl(self, request)
}
}
}
#[cfg(test)]
mod test {
use crate::{http::HttpClientInterface, prelude::*};
fn test_http_client_full_runtime<RT: CreationInterface + HttpClientInterface>(port: u16) {
let rt = RT::new(Config::new().prefix("test")).unwrap();
crate::common::http::test_http_client(&rt, port);
}
#[cfg(feature = "futures_runtime")]
#[test]
fn test_futures() {
test_http_client_full_runtime::<crate::futures::Runtime>(8000);
}
#[cfg(feature = "tokio_runtime")]
#[test]
fn test_tokio() {
test_http_client_full_runtime::<crate::tokio::Runtime>(8001);
}
}