rama-http 0.4.0

use std::convert::Infallible;

use rama_core::stream::json::JsonReadStream;
use rama_http_types::{BodyDataStream, header};

use crate::Request;
use crate::service::web::extract::{
    FromRequest, FromRequestBody, OptionalFromRequest, OptionalFromRequestBody,
};

pub type JsonLines<T> = JsonReadStream<T, BodyDataStream>;

impl<T> FromRequest for JsonLines<T>
where
    T: serde::de::DeserializeOwned + Send + Sync + 'static,
{
    type Rejection = Infallible;

    async fn from_request(req: Request) -> Result<Self, Self::Rejection> {
        let (parts, body) = req.into_parts();
        let future = <Self as FromRequestBody>::from_request_body(&parts, body);
        future.await
    }
}

impl<T> FromRequestBody for JsonLines<T>
where
    T: serde::de::DeserializeOwned + Send + Sync + 'static,
{
    fn from_request_body(
        _parts: &crate::request::Parts,
        body: crate::Body,
    ) -> impl Future<Output = Result<Self, Self::Rejection>> + Send + 'static {
        std::future::ready(Ok(body.into_json_stream()))
    }
}

impl<T> OptionalFromRequest for JsonLines<T>
where
    T: serde::de::DeserializeOwned + Send + Sync + 'static,
{
    type Rejection = Infallible;

    async fn from_request(req: Request) -> Result<Option<Self>, Self::Rejection> {
        let (parts, body) = req.into_parts();
        let future = <Self as OptionalFromRequestBody>::from_optional_request_body(&parts, body);
        future.await
    }
}

impl<T> OptionalFromRequestBody for JsonLines<T>
where
    T: serde::de::DeserializeOwned + Send + Sync + 'static,
{
    fn from_optional_request_body(
        parts: &crate::request::Parts,
        body: crate::Body,
    ) -> impl Future<Output = Result<Option<Self>, <Self as OptionalFromRequest>::Rejection>>
    + Send
    + 'static {
        let future = if parts.headers.get(header::CONTENT_TYPE).is_some() {
            Some(<Self as FromRequestBody>::from_request_body(parts, body))
        } else {
            None
        };

        async move {
            match future {
                Some(future) => future.await.map(Some),
                None => Ok(None),
            }
        }
    }
}
#[cfg(test)]
mod tests {
    use crate::service::{client::HttpClientExt as _, web::Router};

    use super::JsonLines;
    use rama_core::stream::StreamExt as _;
    use rama_http_types::StatusCode;
    use serde::{Deserialize, Serialize};

    #[derive(Serialize, Deserialize, PartialEq, Eq, Debug)]
    struct User {
        id: i32,
    }

    #[tokio::test]
    async fn extractor() {
        let app = Router::new().with_post("/", |mut stream: JsonLines<User>| async move {
            assert_eq!(stream.next().await.unwrap().unwrap(), User { id: 1 });
            assert_eq!(stream.next().await.unwrap().unwrap(), User { id: 2 });
            assert_eq!(stream.next().await.unwrap().unwrap(), User { id: 3 });

            // sources are downcastable to `serde_json::Error`
            let err = stream.next().await.unwrap().unwrap_err();
            let _: &serde_json::Error = err
                .source()
                .unwrap()
                .downcast_ref::<serde_json::Error>()
                .unwrap();
        });

        let res = app
            .post("http://example.com")
            .body(
                [
                    "{\"id\":1}",
                    "{\"id\":2}",
                    "{\"id\":3}",
                    // to trigger an error for source downcasting
                    "{\"id\":false}",
                ]
                .join("\n"),
            )
            .send()
            .await
            .unwrap();

        assert_eq!(res.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn optional_terminal_extractor_present() {
        let request = crate::Request::builder()
            .header(crate::header::CONTENT_TYPE, "application/x-ndjson")
            .body(crate::Body::from("{\"id\":42}\n"))
            .unwrap();

        let mut stream =
            <Option<JsonLines<User>> as crate::service::web::extract::FromRequest>::from_request(
                request,
            )
            .await
            .unwrap()
            .expect("content type makes optional JSON Lines present");

        assert_eq!(stream.next().await.unwrap().unwrap(), User { id: 42 });
        assert!(stream.next().await.is_none());
    }
}