cs-utils 0.21.1

use std::pin::Pin;
use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, AsyncReadExt, split, WriteHalf, ReadHalf};

use crate::random_number;

pub async fn test_async_stream_sends_data<TAsyncDuplex: AsyncRead + AsyncWrite + Send + ?Sized + 'static>(
    mut channel: Pin<Box<TAsyncDuplex>>,
    test_data: String,
) -> Pin<Box<TAsyncDuplex>> {
    let mut i = 0;
    let data = test_data.as_bytes().to_vec();
    let test_data_len = test_data.len();

    while i < test_data.len() {
        let message_len = random_number(8..=(test_data_len / 10));
        
        let message_len = if i + message_len < data.len() {
            i + message_len
        } else {
            data.len()
        };

        let bytes_sent = channel
            .write(&data[i..message_len]).await
            .expect("Cannot send a message.");

        assert!(
            bytes_sent > 0,
            "No bytes sent.",
        );
            
        i += bytes_sent as usize;
    }

    return channel;
}

pub async fn test_async_stream_receives_data<TAsyncDuplex: AsyncRead + AsyncWrite + Send + ?Sized + 'static>(
    mut channel: Pin<Box<TAsyncDuplex>>,
    test_data: String,
) -> Pin<Box<TAsyncDuplex>> {
    let mut received_data = String::new();

    let mut data = [0; 1024];

    loop {
        let bytes_read = channel
            .read(&mut data).await
            .expect("Cannot receive message.");

        let message_str = std::str::from_utf8(&data[..bytes_read])
            .expect("Cannot parse UTF8 message.")
            .to_string();

        received_data = format!("{}{}", &received_data, message_str);

        if received_data.len() == test_data.len() {
            assert_eq!(
                received_data,
                test_data,
                "Sent and received data must match.",
            );

            break;
        }
    }

    return channel;
}

pub async fn test_async_stream_data_transfer<TAsyncDuplex: AsyncRead + AsyncWrite + Send + ?Sized + 'static>(
    channel1: Pin<Box<TAsyncDuplex>>,
    channel2: Pin<Box<TAsyncDuplex>>,
    test_data: String,
) -> (Pin<Box<TAsyncDuplex>>, Pin<Box<TAsyncDuplex>>) {
    let test_data1 = test_data.clone();
    let test_data2 = test_data.clone();

    return tokio::try_join!(
        tokio::spawn(test_async_stream_sends_data(channel1, test_data1)),
        tokio::spawn(test_async_stream_receives_data(channel2, test_data2)),
    ).unwrap();
}

pub async fn test_async_half_sends_data<TAsyncDuplex: AsyncRead + AsyncWrite + Send + Unpin + 'static>(
    mut channel: WriteHalf<TAsyncDuplex>,
    test_data: String,
) {
    let mut i = 0;
    let data = test_data.as_bytes().to_vec();
    let test_data_len = test_data.len();

    while i < test_data.len() {
        let message_len = random_number(8..=(test_data_len / 10));
        
        let message_len = if i + message_len < data.len() {
            i + message_len
        } else {
            data.len()
        };

        let bytes_sent = channel
            .write(&data[i..message_len]).await
            .expect("Cannot send a message.");

        assert!(
            bytes_sent > 0,
            "No bytes sent.",
        );
            
        i += bytes_sent as usize;
    }
}

pub async fn test_async_half_receives_data<TAsyncDuplex: AsyncRead + AsyncWrite + Send + Unpin + 'static>(
    mut channel: ReadHalf<TAsyncDuplex>,
    test_data: String,
) {
    let mut received_data = String::new();

    let mut data = [0; 1024];

    loop {
        let bytes_read = channel
            .read(&mut data).await
            .expect("Cannot receive message.");

        let message_str = std::str::from_utf8(&data[..bytes_read])
            .expect("Cannot parse UTF8 message.")
            .to_string();

        received_data = format!("{}{}", &received_data, message_str);

        if received_data.len() == test_data.len() {
            assert_eq!(
                received_data,
                test_data,
                "Sent and received data must match.",
            );

            break;
        }
    }
}

pub async fn test_async_stream_duplex<TAsyncDuplex: AsyncRead + AsyncWrite + Send + ?Sized + 'static>(
    channel1: Pin<Box<TAsyncDuplex>>,
    channel2: Pin<Box<TAsyncDuplex>>,
    test_data: String,
) {

    let (channel1_source, channel1_sink) = split(channel1);
    let (channel2_source, channel2_sink) = split(channel2);

    let reversed_test_data: String = test_data
        .chars()
        .rev()
        .collect();

    tokio::try_join!(
        tokio::spawn(test_async_half_sends_data(channel2_sink, test_data.clone())),
        tokio::spawn(test_async_half_receives_data(channel1_source, test_data.clone())),
        // for other direction, use reversed data string to send something different than `test_data`
        tokio::spawn(test_async_half_sends_data(channel1_sink, reversed_test_data.clone())),
        tokio::spawn(test_async_half_receives_data(channel2_source, reversed_test_data.clone())),
    ).unwrap();
}

/// Test an `AsyncRead + AsyncWrite` stream data transfer
/// 
/// ### Examples
/// 
/// ```
/// use tokio::io::duplex;
/// 
/// #[tokio::main]
/// async fn main() {
///     // either both `async` and `test` features must be enabled or the `all` one
///     #[cfg(any(all(feature = "async", feature = "test"), feature = "all"))]
///     {
///         use cs_utils::{futures::test::test_async_stream, random_str_rg};
///         // create stream to test
///         let (channel1, channel2) = duplex(4096);
///         
///         // create test data
///         let test_data = random_str_rg(1024..=25600);
///         
///         // test data transfer
///         test_async_stream(
///             Box::new(channel1),
///             Box::new(channel2),
///             test_data,
///         ).await;
///         
///         println!("👌 data transfer succeeded");
///     }
/// }
/// ```
pub async fn test_async_stream<TAsyncDuplex: AsyncRead + AsyncWrite + Send + Unpin + ?Sized + 'static>(
    channel1: Box<TAsyncDuplex>,
    channel2: Box<TAsyncDuplex>,
    test_data: String,
) {
    return test_async_stream_pinned(
        Pin::new(channel1),
        Pin::new(channel2),
        test_data,
    ).await;
}

pub async fn test_async_stream_pinned<TAsyncDuplex: AsyncRead + AsyncWrite + Send + ?Sized + 'static>(
    channel1: Pin<Box<TAsyncDuplex>>,
    channel2: Pin<Box<TAsyncDuplex>>,
    test_data: String,
) {
    // test `channel1` to `channel2` direction
    let (channel1, channel2) = test_async_stream_data_transfer(
        channel1,
        channel2,
        test_data.clone(),
    ).await;

    // test `channel2` to `channel1` direction
    let (channel2, channel1) = test_async_stream_data_transfer(
        channel2,
        channel1,
        test_data.clone(),
    ).await;

    // test bidirectional data transfer
    test_async_stream_duplex(
        channel1,
        channel2,
        test_data,
    ).await;
}