async-curl 0.7.0

use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;

use curl::easy::Easy2;
use curl::easy::Handler;
use curl::easy::WriteError;
use http::status::StatusCode;
use log::LevelFilter;
use tokio::runtime::Builder;
use tokio::sync::Mutex;
use wiremock::matchers::method;
use wiremock::matchers::path;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;

use crate::actor::Actor;
use crate::actor::CurlActor;
use crate::curl::AsyncCurl;

#[derive(Debug, Clone, Default)]
pub struct ResponseHandler {
    data: Option<Vec<u8>>,
}

impl Handler for ResponseHandler {
    /// This will store the response from the server
    /// to the data vector.
    fn write(&mut self, stream: &[u8]) -> Result<usize, WriteError> {
        if self.data.is_none() {
            self.data = Some(stream.to_vec());
        } else if let Some(ref mut data) = self.data {
            data.extend_from_slice(stream);
        }
        Ok(stream.len())
    }
}

impl ResponseHandler {
    /// Instantiation of the ResponseHandler
    /// and initialize the data vector.
    pub fn new() -> Self {
        Self::default()
    }

    /// This will give the data to the receiving variable
    pub fn take(&mut self) -> Option<Vec<u8>> {
        self.data.take()
    }
}

#[ctor::ctor(unsafe)]
fn setup_test_logger() {
    env_logger::Builder::from_env(
        env_logger::Env::default().default_filter_or("your_crate_name=trace"),
    )
    .filter_level(LevelFilter::Trace)
    .format_timestamp_nanos()
    .init();
}

async fn start_mock_server(
    node: &str,
    mock_body: String,
    mock_status_code: StatusCode,
) -> MockServer {
    let server = MockServer::start().await;
    Mock::given(method("GET"))
        .and(path(node))
        .respond_with(ResponseTemplate::new(mock_status_code).set_body_string(mock_body))
        .mount(&server)
        .await;
    server
}

#[tokio::test]
async fn test_async_requests() {
    const MOCK_BODY_RESPONSE: &str = r#"{"token":"12345"}"#;
    const MOCK_STATUS_CODE: StatusCode = StatusCode::OK;

    let server = start_mock_server(
        "/async-test",
        MOCK_BODY_RESPONSE.to_string(),
        StatusCode::OK,
    )
    .await;
    let url = format!("{}{}", server.uri(), "/async-test");

    let curl = CurlActor::new();
    let mut easy2 = Easy2::new(ResponseHandler::new());
    easy2.url(url.as_str()).unwrap();
    easy2.get(true).unwrap();

    let curl1 = curl.clone();
    let spawn1 = tokio::spawn(async move {
        let result = curl1.send_request(easy2).await;
        let mut result = result.unwrap();
        // Test response body
        assert_eq!(
            String::from_utf8_lossy(&result.get_mut().take().unwrap()),
            MOCK_BODY_RESPONSE.to_string()
        );

        // Test response status code
        let status_code = result.response_code().unwrap() as u16;

        assert_eq!(status_code, MOCK_STATUS_CODE.as_u16());
    });

    let mut easy2 = Easy2::new(ResponseHandler::new());
    easy2.url(url.as_str()).unwrap();
    easy2.get(true).unwrap();

    let spawn2 = tokio::spawn(async move {
        let result = curl.send_request(easy2).await;
        let mut result = result.unwrap();
        // Test response body
        assert_eq!(
            String::from_utf8_lossy(&result.get_mut().take().unwrap()),
            MOCK_BODY_RESPONSE.to_string()
        );

        // Test response status code
        let status_code = result.response_code().unwrap() as u16;
        assert_eq!(status_code, MOCK_STATUS_CODE.as_u16());
    });

    let (r1, r2) = tokio::join!(spawn1, spawn2);

    for result in [r1, r2] {
        if let Err(err) = result {
            if err.is_panic() {
                std::panic::resume_unwind(err.into_panic());
            } else {
                panic!("task was cancelled: {err}");
            }
        }
    }
}

#[tokio::test]
async fn test_error() {
    let url = "https://no-connection";

    let curl = CurlActor::new();

    let mut easy2 = Easy2::new(ResponseHandler::new());
    easy2.url(url).unwrap();
    easy2.get(true).unwrap();

    let result = curl.send_request(easy2).await;
    let _ = result.unwrap_err();
}

#[tokio::test]
async fn test_concurrency_abort() {
    const MOCK_BODY_RESPONSE: &str = r#"{"token":"12345"}"#;
    let server = start_mock_server(
        "/async-test",
        MOCK_BODY_RESPONSE.to_string(),
        StatusCode::OK,
    )
    .await;
    let url = format!("{}{}", server.uri(), "/async-test");
    let check_cancelled = Arc::new(Mutex::new(true));
    let main_curl = CurlActor::new();

    let check_cancelled1 = check_cancelled.clone();
    let curl = main_curl.clone();
    let http_task = tokio::spawn(async move {
        let mut easy2 = Easy2::new(ResponseHandler::new());
        easy2.url(url.as_str()).unwrap();
        easy2.get(true).unwrap();
        log::trace!("[current_thread] HTTP task . . . .");
        let _ = curl.send_request(easy2).await.unwrap();
        let mut lock = check_cancelled1.lock().await;
        *lock = false;
        log::trace!("[current_thread] HTTP task completed");
    });

    let other_task = tokio::spawn(async move {
        for _n in 0..10 {
            log::trace!("[current_thread] Other task . . . .");
            tokio::time::sleep(Duration::from_millis(1)).await;
            log::trace!("[current_thread] Other task completed");
        }
    });

    let abort_task = tokio::spawn(async move {
        log::trace!("[current_thread] Aborting HTTP task . . . .");
        http_task.abort();
        log::trace!("[current_thread] HTTP task aborted");
    });

    let (other_task, abort_task) = tokio::join!(other_task, abort_task);
    other_task.unwrap();
    abort_task.unwrap();
    assert!(*check_cancelled.lock().await);
}

#[tokio::test]
async fn test_curl_builder() {
    const MOCK_BODY_RESPONSE: &str = r#"{"token":"12345"}"#;
    let server = start_mock_server(
        "/async-test",
        MOCK_BODY_RESPONSE.to_string(),
        StatusCode::OK,
    )
    .await;
    let url = format!("{}{}", server.uri(), "/async-test");

    let actor = CurlActor::new();
    let collector = ResponseHandler::new();

    let mut curl = AsyncCurl::new(actor, collector)
        .url(url.as_str())
        .unwrap()
        .finalize()
        .perform()
        .await
        .unwrap();

    log::trace!("{:?}", curl);
    let body = curl.get_mut().take();
    let status = curl.response_code().unwrap() as u16;

    assert_eq!(body, Some(MOCK_BODY_RESPONSE.as_bytes().to_vec()));
    assert_eq!(status, StatusCode::OK);
}

#[tokio::test]
async fn test_concurrency_abort_multi_threaded_runtime() {
    const MOCK_BODY_RESPONSE: &str = r#"{"token":"12345"}"#;
    let server = start_mock_server(
        "/async-test",
        MOCK_BODY_RESPONSE.to_string(),
        StatusCode::OK,
    )
    .await;
    let url = format!("{}{}", server.uri(), "/async-test");
    let check_cancelled = Arc::new(Mutex::new(true));
    let runtime = Builder::new_multi_thread().enable_all().build().unwrap();
    let main_curl = CurlActor::new_runtime(runtime);

    let check_cancelled1 = check_cancelled.clone();
    let curl = main_curl.clone();
    let http_task = tokio::spawn(async move {
        let mut easy2 = Easy2::new(ResponseHandler::new());
        easy2.url(url.as_str()).unwrap();
        easy2.get(true).unwrap();
        log::trace!("[multi_threaded] HTTP task . . . .");
        let _ = curl.send_request(easy2).await.unwrap();
        let mut lock = check_cancelled1.lock().await;
        *lock = false;
        log::trace!("[multi_threaded] HTTP task completed");
    });

    let other_task = tokio::spawn(async move {
        for _n in 0..10 {
            log::trace!("[multi_threaded] Other task . . . .");
            tokio::time::sleep(Duration::from_millis(1)).await;
            log::trace!("[multi_threaded] Other task completed");
        }
    });

    let abort_task = tokio::spawn(async move {
        log::trace!("[multi_threaded] Aborting HTTP task . . . .");
        http_task.abort();
        log::trace!("[multi_threaded] HTTP task aborted");
    });

    let (other_task, abort_task) = tokio::join!(other_task, abort_task);
    other_task.unwrap();
    abort_task.unwrap();
    assert!(*check_cancelled.lock().await);
}

#[tokio::test]
async fn test_new_runtime_with_capacity() {
    const MOCK_BODY_RESPONSE: &str = r#"{"token":"12345"}"#;
    let server = start_mock_server(
        "/capacity-test",
        MOCK_BODY_RESPONSE.to_string(),
        StatusCode::OK,
    )
    .await;
    let url = format!("{}{}", server.uri(), "/capacity-test");

    let runtime = Builder::new_multi_thread().enable_all().build().unwrap();
    let curl = CurlActor::new_runtime_with_capacity(runtime, 4);

    let mut easy2 = Easy2::new(ResponseHandler::new());
    easy2.url(url.as_str()).unwrap();
    easy2.get(true).unwrap();

    let result = curl.send_request(easy2).await;
    let mut result = result.unwrap();

    assert_eq!(
        String::from_utf8_lossy(&result.get_mut().take().unwrap()),
        MOCK_BODY_RESPONSE.to_string()
    );
    let status_code = result.response_code().unwrap() as u16;
    assert_eq!(status_code, StatusCode::OK.as_u16());
}

async fn start_mock_server_with_delay(
    node: &str,
    mock_body: String,
    mock_status_code: StatusCode,
    delay: Duration,
) -> MockServer {
    let server = MockServer::start().await;
    Mock::given(method("GET"))
        .and(path(node))
        .respond_with(
            ResponseTemplate::new(mock_status_code)
                .set_delay(delay)
                .set_body_string(mock_body),
        )
        .mount(&server)
        .await;
    server
}

// Ensure other futures can run while waiting for a curl request to complete,
// even if the curl request takes a long time to complete.
#[tokio::test(flavor = "current_thread")]
async fn test_async_concurrency_should_not_block() {
    const MOCK_BODY_RESPONSE: &str = r#"{"token":"12345"}"#;

    let server = start_mock_server_with_delay(
        "/async-test",
        MOCK_BODY_RESPONSE.to_string(),
        StatusCode::OK,
        Duration::from_millis(500), // Keep the request in-flight while verifying other futures make progress.
    )
    .await;

    let url = format!("{}{}", server.uri(), "/async-test");

    let curl = CurlActor::new();

    let http_task = tokio::spawn(async move {
        let mut easy2 = Easy2::new(ResponseHandler::new());
        easy2.url(&url).unwrap();
        easy2.get(true).unwrap();

        curl.send_request(easy2).await.unwrap();
    });

    let progress = Arc::new(AtomicUsize::new(0));
    let progress_clone = progress.clone();

    let ticker = tokio::spawn(async move {
        let start = tokio::time::Instant::now();

        while start.elapsed() <= Duration::from_millis(500) {
            progress_clone.fetch_add(1, Ordering::Relaxed);
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    });
    ticker.await.unwrap();

    let ticks = progress.load(Ordering::Relaxed);

    assert!(
        ticks >= 30,
        "executor appears blocked: only {} ticks executed in 500ms",
        ticks
    );
    println!("Ticks executed in 500ms: {}", ticks);

    http_task.await.unwrap();
}

// Ensure other futures can run while waiting for a curl request to complete,
// even if the curl request takes a long time to complete.
#[tokio::test(flavor = "multi_thread")]
async fn test_async_concurrency_should_not_block_multi_thread() {
    const MOCK_BODY_RESPONSE: &str = r#"{"token":"12345"}"#;

    let server = start_mock_server_with_delay(
        "/async-test",
        MOCK_BODY_RESPONSE.to_string(),
        StatusCode::OK,
        Duration::from_millis(500), // Keep the request in-flight while verifying other futures make progress.
    )
    .await;

    let url = format!("{}{}", server.uri(), "/async-test");
    let runtime = Builder::new_multi_thread().enable_all().build().unwrap();
    let curl = CurlActor::new_runtime(runtime);

    let http_task = tokio::spawn(async move {
        let mut easy2 = Easy2::new(ResponseHandler::new());
        easy2.url(&url).unwrap();
        easy2.get(true).unwrap();

        curl.send_request(easy2).await.unwrap();
    });

    let progress = Arc::new(AtomicUsize::new(0));
    let progress_clone = progress.clone();

    let ticker = tokio::spawn(async move {
        let start = tokio::time::Instant::now();

        while start.elapsed() <= Duration::from_millis(500) {
            progress_clone.fetch_add(1, Ordering::Relaxed);
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    });
    ticker.await.unwrap();

    let ticks = progress.load(Ordering::Relaxed);

    assert!(
        ticks >= 30,
        "executor appears blocked: only {} ticks executed in 500ms",
        ticks
    );
    println!("Ticks executed in 500ms: {}", ticks);

    http_task.await.unwrap();
}

// Ensure other futures can run while waiting for a curl request to complete,
// even if the curl request takes a long time to complete.
#[tokio::test(flavor = "current_thread")]
async fn test_async_concurrency_should_not_block_using_easy2() {
    const MOCK_BODY_RESPONSE: &str = r#"{"token":"12345"}"#;

    let server = start_mock_server_with_delay(
        "/async-test",
        MOCK_BODY_RESPONSE.to_string(),
        StatusCode::OK,
        Duration::from_millis(500), // Keep the request in-flight while verifying other futures make progress.
    )
    .await;

    let url = format!("{}{}", server.uri(), "/async-test");

    let curl = CurlActor::new();

    let http_task = tokio::spawn(async move {
        let mut easy2 = Easy2::new(ResponseHandler::new());
        easy2.url(&url).unwrap();
        easy2.get(true).unwrap();

        curl.perform_easy2(easy2).await.unwrap();
    });

    let progress = Arc::new(AtomicUsize::new(0));
    let progress_clone = progress.clone();

    let ticker = tokio::spawn(async move {
        let start = tokio::time::Instant::now();

        while start.elapsed() <= Duration::from_millis(500) {
            progress_clone.fetch_add(1, Ordering::Relaxed);
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    });
    ticker.await.unwrap();

    let ticks = progress.load(Ordering::Relaxed);

    assert!(
        ticks >= 30,
        "executor appears blocked: only {} ticks executed in 500ms",
        ticks
    );
    println!("Ticks executed in 500ms: {}", ticks);

    http_task.await.unwrap();
}

// Ensure other futures can run while waiting for a curl request to complete,
// even if the curl request takes a long time to complete.
#[tokio::test(flavor = "multi_thread")]
async fn test_async_concurrency_should_not_block_multi_thread_using_easy2() {
    const MOCK_BODY_RESPONSE: &str = r#"{"token":"12345"}"#;

    let server = start_mock_server_with_delay(
        "/async-test",
        MOCK_BODY_RESPONSE.to_string(),
        StatusCode::OK,
        Duration::from_millis(500), // Keep the request in-flight while verifying other futures make progress.
    )
    .await;

    let url = format!("{}{}", server.uri(), "/async-test");
    let runtime = Builder::new_multi_thread().enable_all().build().unwrap();
    let curl = CurlActor::new_runtime(runtime);

    let http_task = tokio::spawn(async move {
        let mut easy2 = Easy2::new(ResponseHandler::new());
        easy2.url(&url).unwrap();
        easy2.get(true).unwrap();

        curl.perform_easy2(easy2).await.unwrap();
    });

    let progress = Arc::new(AtomicUsize::new(0));
    let progress_clone = progress.clone();

    let ticker = tokio::spawn(async move {
        let start = tokio::time::Instant::now();

        while start.elapsed() <= Duration::from_millis(500) {
            progress_clone.fetch_add(1, Ordering::Relaxed);
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    });
    ticker.await.unwrap();

    let ticks = progress.load(Ordering::Relaxed);

    assert!(
        ticks >= 30,
        "executor appears blocked: only {} ticks executed in 500ms",
        ticks
    );
    println!("Ticks executed in 500ms: {}", ticks);

    http_task.await.unwrap();
}

#[tokio::test]
async fn test_async_requests_using_easy2_perform() {
    const MOCK_BODY_RESPONSE: &str = r#"{"token":"12345"}"#;
    const MOCK_STATUS_CODE: StatusCode = StatusCode::OK;

    let server = start_mock_server(
        "/async-test",
        MOCK_BODY_RESPONSE.to_string(),
        StatusCode::OK,
    )
    .await;
    let url = format!("{}{}", server.uri(), "/async-test");

    let curl = CurlActor::new();
    let mut easy2 = Easy2::new(ResponseHandler::new());
    easy2.url(url.as_str()).unwrap();
    easy2.get(true).unwrap();

    let curl1 = curl.clone();
    let spawn1 = tokio::spawn(async move {
        let result = curl1.perform_easy2(easy2).await;
        let mut result = result.unwrap();
        // Test response body
        assert_eq!(
            String::from_utf8_lossy(&result.get_mut().take().unwrap()),
            MOCK_BODY_RESPONSE.to_string()
        );

        // Test response status code
        let status_code = result.response_code().unwrap() as u16;

        assert_eq!(status_code, MOCK_STATUS_CODE.as_u16());
    });

    let mut easy2 = Easy2::new(ResponseHandler::new());
    easy2.url(url.as_str()).unwrap();
    easy2.get(true).unwrap();

    let spawn2 = tokio::spawn(async move {
        let result = curl.perform_easy2(easy2).await;
        let mut result = result.unwrap();
        // Test response body
        assert_eq!(
            String::from_utf8_lossy(&result.get_mut().take().unwrap()),
            MOCK_BODY_RESPONSE.to_string()
        );

        // Test response status code
        let status_code = result.response_code().unwrap() as u16;
        assert_eq!(status_code, MOCK_STATUS_CODE.as_u16());
    });

    let (r1, r2) = tokio::join!(spawn1, spawn2);

    for result in [r1, r2] {
        if let Err(err) = result {
            if err.is_panic() {
                std::panic::resume_unwind(err.into_panic());
            } else {
                panic!("task was cancelled: {err}");
            }
        }
    }
}

#[tokio::test]
async fn test_error_perform_easy2() {
    let url = "https://no-connection";

    let curl = CurlActor::new();

    let mut easy2 = Easy2::new(ResponseHandler::new());
    easy2.url(url).unwrap();
    easy2.get(true).unwrap();

    let result = curl.perform_easy2(easy2).await;
    let _ = result.unwrap_err();
}

#[tokio::test]
async fn test_concurrency_abort_using_perform_easy2() {
    const MOCK_BODY_RESPONSE: &str = r#"{"token":"12345"}"#;
    let server = start_mock_server(
        "/async-test",
        MOCK_BODY_RESPONSE.to_string(),
        StatusCode::OK,
    )
    .await;
    let url = format!("{}{}", server.uri(), "/async-test");
    let check_cancelled = Arc::new(Mutex::new(true));
    let main_curl = CurlActor::new();

    let check_cancelled1 = check_cancelled.clone();
    let curl = main_curl.clone();
    let http_task = tokio::spawn(async move {
        let mut easy2 = Easy2::new(ResponseHandler::new());
        easy2.url(url.as_str()).unwrap();
        easy2.get(true).unwrap();
        log::trace!("[current_thread] HTTP task . . . .");
        let _ = curl.perform_easy2(easy2).await.unwrap();
        let mut lock = check_cancelled1.lock().await;
        *lock = false;
        log::trace!("[current_thread] HTTP task completed");
    });

    let other_task = tokio::spawn(async move {
        for _n in 0..10 {
            log::trace!("[current_thread] Other task . . . .");
            tokio::time::sleep(Duration::from_millis(1)).await;
            log::trace!("[current_thread] Other task completed");
        }
    });

    let abort_task = tokio::spawn(async move {
        log::trace!("[current_thread] Aborting HTTP task . . . .");
        http_task.abort();
        log::trace!("[current_thread] HTTP task aborted");
    });

    let (other_task, abort_task) = tokio::join!(other_task, abort_task);
    other_task.unwrap();
    abort_task.unwrap();
    assert!(*check_cancelled.lock().await);
}

#[tokio::test]
async fn test_concurrency_abort_multi_threaded_runtime_using_perform_easy2() {
    const MOCK_BODY_RESPONSE: &str = r#"{"token":"12345"}"#;
    let server = start_mock_server(
        "/async-test",
        MOCK_BODY_RESPONSE.to_string(),
        StatusCode::OK,
    )
    .await;
    let url = format!("{}{}", server.uri(), "/async-test");
    let check_cancelled = Arc::new(Mutex::new(true));
    let runtime = Builder::new_multi_thread().enable_all().build().unwrap();
    let main_curl = CurlActor::new_runtime(runtime);

    let check_cancelled1 = check_cancelled.clone();
    let curl = main_curl.clone();
    let http_task = tokio::spawn(async move {
        let mut easy2 = Easy2::new(ResponseHandler::new());
        easy2.url(url.as_str()).unwrap();
        easy2.get(true).unwrap();
        log::trace!("[multi_threaded] HTTP task . . . .");
        let _ = curl.perform_easy2(easy2).await.unwrap();
        let mut lock = check_cancelled1.lock().await;
        *lock = false;
        log::trace!("[multi_threaded] HTTP task completed");
    });

    let other_task = tokio::spawn(async move {
        for _n in 0..10 {
            log::trace!("[multi_threaded] Other task . . . .");
            tokio::time::sleep(Duration::from_millis(1)).await;
            log::trace!("[multi_threaded] Other task completed");
        }
    });

    let abort_task = tokio::spawn(async move {
        log::trace!("[multi_threaded] Aborting HTTP task . . . .");
        http_task.abort();
        log::trace!("[multi_threaded] HTTP task aborted");
    });

    let (other_task, abort_task) = tokio::join!(other_task, abort_task);
    other_task.unwrap();
    abort_task.unwrap();
    assert!(*check_cancelled.lock().await);
}