rs-wait-valid-req 0.1.0

use std::io;
use std::sync::Arc;

use std::collections::BTreeMap;

pub use form_urlencoded;

use axum::http;

pub use http::Uri;

use http::StatusCode;

pub fn create_uri_parser(keys: Vec<String>) -> impl Fn(&Uri) -> BTreeMap<String, String> {
    move |uri: &Uri| {
        form_urlencoded::parse(uri.query().unwrap_or("").as_bytes())
            .into_iter()
            .filter(|(k, _)| keys.contains(&k.to_string()))
            .map(|(k, v)| (k.into_owned(), v.into_owned()))
            .collect()
    }
}

pub async fn send_converted<F, T>(
    uri: &Uri,
    f: &F,
    sender: &tokio::sync::mpsc::Sender<T>,
) -> Result<(), io::Error>
where
    T: Send + Sync + 'static,
    F: Fn(&Uri) -> Result<T, io::Error>,
{
    let t: T = f(uri)?;
    sender.try_send(t).map_err(io::Error::other)?;
    Ok(())
}

pub trait UriToParsed: Sync + Send + 'static {
    type Item: Sync + Send + 'static;
    fn convert(&self, u: &Uri) -> Result<Self::Item, io::Error>;
}

impl<U> UriToParsed for Arc<U>
where
    U: UriToParsed,
{
    type Item = U::Item;
    fn convert(&self, u: &Uri) -> Result<Self::Item, io::Error> {
        let original: &U = self;
        original.convert(u)
    }
}

pub async fn wait_valid_req_forever<T, U>(addr: String, u2p: U) -> Result<T, io::Error>
where
    T: Sync + Send + 'static,
    U: UriToParsed<Item = T> + Clone,
{
    let (finish_tx, mut finish_rx) = tokio::sync::mpsc::channel::<()>(1);
    let (tx, mut rx) = tokio::sync::mpsc::channel::<T>(1);

    let server_handle = tokio::spawn(async move {
        let router = axum::Router::new().route(
            "/",
            axum::routing::get(move |uri: Uri| {
                let tx_clone = tx.clone();
                let finish_tx_clone = finish_tx.clone();
                async move {
                    let f = |u: &Uri| u2p.convert(u);
                    match send_converted(&uri, &f, &tx_clone).await {
                        Ok(_) => {}
                        Err(_) => {
                            return (
                                StatusCode::INTERNAL_SERVER_ERROR,
                                "unable to handle the request",
                            );
                        }
                    };

                    match finish_tx_clone.try_send(()) {
                        Ok(_) => (StatusCode::OK, "ok"),
                        Err(_) => (
                            StatusCode::INTERNAL_SERVER_ERROR,
                            "unable to send the finish notification",
                        ),
                    }
                }
            }),
        );

        let listener = tokio::net::TcpListener::bind(&addr).await;
        match listener {
            Ok(listener) => {
                axum::serve(listener, router)
                    .with_graceful_shutdown(async move {
                        finish_rx.recv().await;
                    })
                    .await
            }
            Err(e) => Err(e),
        }
    });

    // Wait for the server to start or fail
    server_handle
        .await
        .map_err(|e| io::Error::other(format!("server task panicked: {}", e)))?
        .map_err(|e| io::Error::other(format!("server failed to start: {}", e)))?;

    rx.recv().await.ok_or(io::Error::other("no response got"))
}