connection_utils/utils/

divide_channel.rs

use std::{pin::Pin, sync::Arc};

use anyhow::Result;
use tokio_util::codec::Framed;
use serde::{Serialize, Deserialize};
use cs_utils::{random_str, random_number};
use futures::{StreamExt, stream::{SplitStream, SplitSink}, SinkExt};
use tokio::{io::{duplex, WriteHalf, ReadHalf, AsyncReadExt, AsyncWriteExt, split, AsyncRead, AsyncWrite}, sync::{Mutex, watch::Receiver}};

use crate::{codecs::GenericCodec, Channel, TransportChannel, create_framed_stream};

#[derive(Serialize, Deserialize, Debug)]
pub enum LayerMessage {
    Data(u8, Vec<u8>),
    Close(u8),
}

async fn forward_reads<TAsyncDuplex: AsyncRead + AsyncWrite + Send + Unpin + ?Sized + 'static>(
    on_channel1_closed: Receiver<bool>,
    on_channel2_closed: Receiver<bool>,
    mut channel: SplitStream<Framed<Pin<Box<TAsyncDuplex>>, GenericCodec<LayerMessage>>>,
    mut child1: WriteHalf<Pin<Box<dyn Channel>>>,
    mut child2: WriteHalf<Pin<Box<dyn Channel>>>,
)-> Result<()> {
    let mut child1 = Pin::new(&mut child1);
    let mut child2 = Pin::new(&mut child2);

    loop {
        let message = match channel.next().await {
            Some(item) => item?,
            None => {
                // TODO: error out?

                return Ok(());
            },
        };

        match message {
            LayerMessage::Data(id, data) => {
                let child_channel = match id {
                    0 => &mut child1,
                    1 => &mut child2,
                    _ => panic!("Unknown channel number: {}.", id),
                };
        
                let on_channel_closed = match id {
                    0 => &on_channel1_closed,
                    1 => &on_channel2_closed,
                    _ => panic!("Unknown channel number: {}.", id),
                };
        
                if data.len() == 0 && *on_channel_closed.borrow()  {
                    continue;
                }
        
                let result = child_channel.write_all(&data[..]).await;
        
                if result.is_err() {
                    // println!("[{}][forward_reads]> failed to write {} bytes to channel {}", label, data.len(), id);
                }
            },
            LayerMessage::Close(id) => {
                let child_channel = match id {
                    0 => &mut child1,
                    1 => &mut child2,
                    _ => panic!("Unknown channel number: {}.", id),
                };

                child_channel.shutdown().await?;
            },
        };
    }
}

async fn read_channel<TAsyncDuplex: AsyncRead + AsyncWrite + Send + Unpin + ?Sized + 'static>(
    on_channel_closed: Receiver<bool>,
    channel_number: u8,
    mut channel: ReadHalf<Pin<Box<dyn Channel>>>,
    target_channel: Arc<Mutex<SplitSink<Framed<Pin<Box<TAsyncDuplex>>, GenericCodec<LayerMessage>>, LayerMessage>>>,
    _buffer_size: u32,
) -> Result<()> {
    let mut buf = [0; 4_096];

    loop {
        let bytes_read = channel.read(&mut buf).await?;
        let data = &buf[..bytes_read];

        let is_closed = *on_channel_closed.borrow();

        {
            let mut lock = target_channel.lock().await;

            if data.len() == 0 && is_closed {
                let _result = lock.send(LayerMessage::Close(channel_number)).await;

                return Ok(());
            }

            // send channel data message
            lock.send(LayerMessage::Data(channel_number, data.to_vec())).await?;
        }
    }
}

async fn forward_writes<TAsyncDuplex: AsyncRead + AsyncWrite + Send + Unpin + ?Sized + 'static>(
    on_channel1_closed: Receiver<bool>,
    on_channel2_closed: Receiver<bool>,
    channel: SplitSink<Framed<Pin<Box<TAsyncDuplex>>, GenericCodec<LayerMessage>>, LayerMessage>,
    child1: ReadHalf<Pin<Box<dyn Channel>>>,
    child2: ReadHalf<Pin<Box<dyn Channel>>>,
    buffer_size: u32,
) -> Result<()> {
    let channel = Arc::new(Mutex::new(channel));

    tokio::try_join!(
        Box::pin(read_channel(on_channel1_closed, 0, child1, Arc::clone(&channel), buffer_size)),
        Box::pin(read_channel(on_channel2_closed, 1, child2, Arc::clone(&channel), buffer_size)),
    )?;

    return Ok(());
}

async fn forward<TAsyncDuplex: AsyncRead + AsyncWrite + Send + Unpin + ?Sized + 'static>(
    channel: Box<TAsyncDuplex>,
    child1: Box<dyn Channel>,
    child2: Box<dyn Channel>,
    buffer_size: u32,
) {
    let (
        sink,
        source,
    ) = create_framed_stream::<LayerMessage, _>(channel).split();

    let on_channel1_closed1 = child1.on_close();
    let on_channel2_closed1 = child2.on_close();

    let on_channel1_closed2 = child1.on_close();
    let on_channel2_closed2 = child2.on_close();

    let (child1_read, child1_write) = split(Pin::new(child1));
    let (child2_read, child2_write) = split(Pin::new(child2));

    tokio::join!(
        Box::pin(async move {
            match forward_reads(on_channel1_closed1, on_channel2_closed1, source, child1_write, child2_write).await {
                Ok(_) => {
                    // println!("[{}][forward][forward_reads]> succeed", label1);
                },
                Err(_error) => {
                    // println!("[{}][forward][forward_reads]> failed: {}", label1, error);

                    // panic!("[{}][forward][forward_reads]> failed {}", label1, error);
                },
            };
        }),
        Box::pin(async move {
            match forward_writes(on_channel1_closed2, on_channel2_closed2, sink, child1_read, child2_read, buffer_size).await {
                Ok(_) => {
                    // println!("[{}][forward][forward_writes]> succeed", label2);
                },
                Err(_error) => {
                    // println!("[{}][forward][forward_writes]> failed {}", label2, error);

                    // panic!("[{}][forward][forward_writes]> failed {}", label2, error);
                },
            };
        }),
    );
}

fn divide_impl<TAsyncDuplex: AsyncRead + AsyncWrite + Send + Unpin + ?Sized + 'static>(
    id: u16,
    label: String,
    buffer_size: u32,
    stream: Box<TAsyncDuplex>,
) -> (Box<dyn Channel>, Box<dyn Channel>) {
    let (child1_sink, child1_source) = duplex(buffer_size as usize);
    let (child2_sink, child2_source) = duplex(buffer_size as usize);

    let (
        child_channel1,
        child_channel2,
    ) = TransportChannel::new_pair(
        id,
        format!("{}-1", label),
        (Box::new(child1_sink), Box::new(child1_source)),
        buffer_size,
    );

    let (
        forward_channel1,
        forward_channel2,
    ) = TransportChannel::new_pair(
        id,
        format!("{}-2", label),
        (Box::new(child2_sink), Box::new(child2_source)),
        buffer_size,
    );

   tokio::spawn(
        forward(
            stream,
            child_channel1,
            forward_channel1,
            buffer_size,
        ),
    );

    return (child_channel2, forward_channel2);
}

/// Divides a channel into two.
pub fn divide_channel(
    channel: Box<dyn Channel>,
) -> (Box<dyn Channel>, Box<dyn Channel>) {
    let id = channel.id();
    let label = channel.label().to_string();
    let buffer_size = channel.buffer_size();

    return divide_impl(id, label, buffer_size, channel);
}

/// Divides a stream into two. Returns two channels since channels are more powerful while
/// still implement the same traits as the streams do.
pub fn divide_stream<TAsyncDuplex: AsyncRead + AsyncWrite + Send + Unpin + ?Sized + 'static>(
    stream: Box<TAsyncDuplex>,
) -> (Box<dyn Channel>, Box<dyn Channel>) {
    let id = random_number(0..=u16::MAX);
    let label = format!("stream-{}", random_str(4));

    return divide_impl(id, label, 4_096, stream);
}

#[cfg(test)]
mod tests {
    use rstest::rstest;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use cs_utils::{futures::wait_random, traits::Random};
    
    use crate::test::{test_stream, TestOptions};
    use crate::mocks::{channel_mock_pair, ChannelMockOptions};

    use super::{divide_channel, divide_stream};

    mod divide_channel {
        use super::*;

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[tokio::test]
        async fn transfers_data(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, local_channel2) = divide_channel(local_channel);
            let (remote_channel1, remote_channel2) = divide_channel(remote_channel);

            tokio::try_join!(
                tokio::spawn(async move {
                    return test_stream(
                        local_channel1,
                        remote_channel1,
                        TestOptions::random()
                            .with_data_len(test_data_size),
                    ).await;
                }),
                tokio::spawn(async move {
                    return test_stream(
                        local_channel2,
                        remote_channel2,
                        TestOptions::random()
                            .with_data_len(test_data_size),
                    ).await;
                }),
            ).unwrap();
        }

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[tokio::test]
        async fn transfers_data_if_second_channel_is_not_used(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, _local_channel2) = divide_channel(local_channel);
            let (remote_channel1, _remote_channel2) = divide_channel(remote_channel);

            test_stream(
                local_channel1,
                remote_channel1,
                TestOptions::random()
                    .with_data_len(test_data_size),
            ).await;
        }

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[tokio::test]
        async fn transfers_data_if_first_channel_is_not_used(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (_local_channel1, local_channel2) = divide_channel(local_channel);
            let (_remote_channel1, remote_channel2) = divide_channel(remote_channel);

            test_stream(
                local_channel2,
                remote_channel2,
                TestOptions::random()
                    .with_data_len(test_data_size),
            ).await;
        }

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[tokio::test]
        async fn transfers_data_if_second_channel_dropped(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, local_channel2) = divide_channel(local_channel);
            let (remote_channel1, remote_channel2) = divide_channel(remote_channel);

            drop(local_channel2);
            drop(remote_channel2);

            test_stream(
                local_channel1,
                remote_channel1,
                TestOptions::random()
                    .with_data_len(test_data_size),
            ).await;
        }

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[tokio::test]
        async fn transfers_data_if_first_channel_dropped(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, local_channel2) = divide_channel(local_channel);
            let (remote_channel1, remote_channel2) = divide_channel(remote_channel);

            drop(local_channel1);
            drop(remote_channel1);

            test_stream(
                local_channel2,
                remote_channel2,
                TestOptions::random()
                    .with_data_len(test_data_size),
            ).await;
        }

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[tokio::test]
        async fn transfers_data_if_second_channel_shut_down(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, mut local_channel2) = divide_channel(local_channel);
            let (remote_channel1, mut remote_channel2) = divide_channel(remote_channel);

            local_channel2.shutdown().await.unwrap();
            remote_channel2.shutdown().await.unwrap();

            test_stream(
                local_channel1,
                remote_channel1,
                TestOptions::random()
                    .with_data_len(test_data_size),
            ).await;
        }

        #[rstest]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[case(32_768)]
        #[tokio::test]
        async fn transfers_data_if_second_channel_shut_down_during_transfer(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, local_channel2) = divide_channel(local_channel);
            let (remote_channel1, remote_channel2) = divide_channel(remote_channel);

            tokio::join!(
                Box::pin(async move {
                    test_stream(
                        local_channel1,
                        remote_channel1,
                        TestOptions::random()
                            .with_data_len(test_data_size),
                    ).await;
                }),
                Box::pin(async move {
                    // transfer a bit of data
                    let (mut local_channel2, mut remote_channel2) = test_stream(
                        local_channel2,
                        remote_channel2,
                        TestOptions::random()
                            .with_data_len(128),
                    ).await;

                    // shutdown the second channel
                    tokio::join!(
                        Box::pin(async move {
                            wait_random(1..5).await;
                            local_channel2.shutdown().await.unwrap();
                        }),
                        Box::pin(async move {
                            wait_random(1..5).await;
                            remote_channel2.shutdown().await.unwrap();
                        }),
                    );
                }),
            );
        }

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[case(32_768)]
        #[tokio::test]
        async fn transfers_data_if_first_channel_shut_down(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (mut local_channel1, local_channel2) = divide_channel(local_channel);
            let (mut remote_channel1, remote_channel2) = divide_channel(remote_channel);

            local_channel1.shutdown().await.unwrap();
            remote_channel1.shutdown().await.unwrap();

            test_stream(
                local_channel2,
                remote_channel2,
                TestOptions::random()
                    .with_data_len(test_data_size),
            ).await;
        }

        #[rstest]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[case(32_768)]
        #[tokio::test]
        async fn transfers_data_if_first_channel_shut_down_during_transfer(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, local_channel2) = divide_channel(local_channel);
            let (remote_channel1,remote_channel2) = divide_channel(remote_channel);

            tokio::join!(
                Box::pin(async move {
                    // transfer a bit of data
                    let (mut local_channel1, mut remote_channel1) = test_stream(
                        local_channel1,
                        remote_channel1,
                        TestOptions::random()
                            .with_data_len(128),
                    ).await;

                    // shutdown the first channel
                    tokio::join!(
                        Box::pin(async move {
                            wait_random(1..5).await;
                            local_channel1.shutdown().await.unwrap();
                        }),
                        Box::pin(async move {
                            wait_random(1..5).await;
                            remote_channel1.shutdown().await.unwrap();
                        }),
                    );
                }),
                Box::pin(async move {
                    test_stream(
                        local_channel2,
                        remote_channel2,
                        TestOptions::random()
                            .with_data_len(test_data_size),
                    ).await;
                }),
            );
        }

        mod shutdown {
            use super::*;

            #[rstest]
            #[case(128)]
            #[case(256)]
            #[case(512)]
            #[case(1_024)]
            #[case(2_048)]
            #[case(4_096)]
            #[case(8_192)]
            #[case(16_384)]
            #[tokio::test]
            async fn first_channel_closes_channel_if_remote_counterpart_is_closed(
                #[case] test_data_size: usize,
            ) {
                let (local_channel, remote_channel) = channel_mock_pair(
                    ChannelMockOptions::random().with_label("local"),
                    ChannelMockOptions::random().with_label("remote"),
                );

                let (local_channel1, local_channel2) = divide_channel(local_channel);
                let (remote_channel1, remote_channel2) = divide_channel(remote_channel);

                let (
                    mut local_channel1,
                    mut remote_channel1,
                ) = test_stream(
                    local_channel1,
                    remote_channel1,
                    TestOptions::random()
                        .with_data_len(test_data_size),
                ).await;

                tokio::join!(
                    // test that the second channel is not affected by shutting down of the first one
                    Box::pin(async move {
                        wait_random(1..=10).await;

                        return test_stream(
                            local_channel2,
                            remote_channel2,
                            TestOptions::random()
                                .with_data_len(2 * test_data_size),
                        ).await
                    }),
                    // transfer data in the first channel and then shut it down
                    // after shutting down, test that the remote counterpart receives `EOF` message
                    Box::pin(async move {
                        tokio::join!(
                            Box::pin(async move {
                                wait_random(5..=25).await;
                
                                local_channel1.shutdown().await.unwrap();

                                assert!(
                                    local_channel1.write("test".as_bytes()).await.is_err(),
                                    "Must fail to write to closed local channel.",
                                );

                                assert!(
                                    local_channel1.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                            Box::pin(async move {
                                let mut buf = [0; 4096];
                
                                let bytes_read = remote_channel1.read(&mut buf)
                                    .await.unwrap();
                
                                assert_eq!(
                                    bytes_read,
                                    0,
                                    "Must be 0 bytes read if remote channel is closed.",
                                );

                                assert!(
                                    remote_channel1.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                        );
                    }),
                );
            }

            #[rstest]
            #[case(128)]
            #[case(256)]
            #[case(512)]
            #[case(1_024)]
            #[case(2_048)]
            #[case(4_096)]
            #[case(8_192)]
            #[case(16_384)]
            #[tokio::test]
            async fn second_channel_closes_channel_if_remote_counterpart_is_closed(
                #[case] test_data_size: usize,
            ) {
                let (local_channel, remote_channel) = channel_mock_pair(
                    ChannelMockOptions::random().with_label("local"),
                    ChannelMockOptions::random().with_label("remote"),
                );

                let (local_channel1, local_channel2) = divide_channel(local_channel);
                let (remote_channel1, remote_channel2) = divide_channel(remote_channel);

                let (
                    mut local_channel2,
                    mut remote_channel2,
                ) = test_stream(
                    local_channel2,
                    remote_channel2,
                    TestOptions::random()
                        .with_data_len(test_data_size),
                ).await;

                tokio::join!(
                    // test that the first channel is not affected by shutting down of the first one
                    Box::pin(async move {
                        wait_random(1..=10).await;

                        return test_stream(
                            local_channel1,
                            remote_channel1,
                            TestOptions::random()
                                .with_data_len(2 * test_data_size),
                        ).await
                    }),
                    // transfer data in the second channel and then shut it down
                    // after shutting down, test that the remote counterpart receives `EOF` message
                    Box::pin(async move {
                        tokio::join!(
                            Box::pin(async move {
                                wait_random(5..=25).await;
                
                                local_channel2.shutdown().await.unwrap();

                                assert!(
                                    local_channel2.write("test".as_bytes()).await.is_err(),
                                    "Must fail to write to closed local channel.",
                                );

                                assert!(
                                    local_channel2.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                            Box::pin(async move {
                                let mut buf = [0; 4096];
                
                                let bytes_read = remote_channel2.read(&mut buf)
                                    .await.unwrap();
                
                                assert_eq!(
                                    bytes_read,
                                    0,
                                    "Must be 0 bytes read if remote channel is closed.",
                                );

                                assert!(
                                    remote_channel2.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                        );
                    }),
                );
            }

            #[rstest]
            #[case(128)]
            #[case(256)]
            #[case(512)]
            #[case(1_024)]
            #[case(2_048)]
            #[case(4_096)]
            #[case(8_192)]
            #[case(16_384)]
            #[tokio::test]
            async fn both_channels_close_channel_if_remote_counterpart_is_closed(
                #[case] test_data_size: usize,
            ) {
                let (local_channel, remote_channel) = channel_mock_pair(
                    ChannelMockOptions::random().with_label("local"),
                    ChannelMockOptions::random().with_label("remote"),
                );

                let (local_channel1, local_channel2) = divide_channel(local_channel);
                let (remote_channel1, remote_channel2) = divide_channel(remote_channel);

                tokio::join!(
                    // transfer data in the first channel and then shut it down
                    // after shutting down, test that the remote counterpart receives `EOF` message
                    Box::pin(async move {
                        wait_random(1..=25).await;

                        let (
                            mut local_channel1,
                            mut remote_channel1,
                        ) = test_stream(
                            local_channel1,
                            remote_channel1,
                            TestOptions::random()
                                .with_data_len(test_data_size),
                        ).await;

                        tokio::join!(
                            Box::pin(async move {
                                wait_random(1..=25).await;
                
                                local_channel1.shutdown().await.unwrap();

                                assert!(
                                    local_channel1.write(cs_utils::random_str_rg(16..64).as_bytes()).await.is_err(),
                                    "Must fail to write to closed local channel.",
                                );

                                assert!(
                                    local_channel1.write("test".as_bytes()).await.is_err(),
                                    "Must fail to write to closed local channel.",
                                );

                                assert!(
                                    local_channel1.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                            Box::pin(async move {
                                let mut buf = [0; 4096];
                
                                let bytes_read = remote_channel1.read(&mut buf)
                                    .await.unwrap();
                
                                assert_eq!(
                                    bytes_read,
                                    0,
                                    "Must be 0 bytes read if remote channel is closed.",
                                );

                                assert!(
                                    remote_channel1.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                        );
                    }),
                    // transfer data in the second channel and then shut it down
                    // after shutting down, test that the remote counterpart receives `EOF` message
                    Box::pin(async move {
                        wait_random(1..=25).await;

                        let (
                            mut local_channel2,
                            mut remote_channel2,
                        ) = test_stream(
                            local_channel2,
                            remote_channel2,
                            TestOptions::random()
                                .with_data_len(test_data_size),
                        ).await;

                        tokio::join!(
                            Box::pin(async move {
                                wait_random(1..=25).await;
                
                                local_channel2.shutdown().await.unwrap();

                                assert!(
                                    local_channel2.write("test".as_bytes()).await.is_err(),
                                    "Must fail to write to closed local channel.",
                                );

                                assert!(
                                    local_channel2.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                            Box::pin(async move {
                                let mut buf = [0; 4096];
                
                                let bytes_read = remote_channel2.read(&mut buf)
                                    .await.unwrap();
                
                                assert_eq!(
                                    bytes_read,
                                    0,
                                    "Must be 0 bytes read if remote channel is closed.",
                                );

                                assert!(
                                    remote_channel2.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                        );
                    }),
                );
            }
        }
    }

    mod divide_stream {
        use super::*;

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[tokio::test]
        async fn transfers_data(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, local_channel2) = divide_stream(local_channel);
            let (remote_channel1, remote_channel2) = divide_stream(remote_channel);

            tokio::try_join!(
                tokio::spawn(async move {
                    return test_stream(
                        local_channel1,
                        remote_channel1,
                        TestOptions::random()
                            .with_data_len(test_data_size),
                    ).await;
                }),
                tokio::spawn(async move {
                    return test_stream(
                        local_channel2,
                        remote_channel2,
                        TestOptions::random()
                            .with_data_len(test_data_size),
                    ).await;
                }),
            ).unwrap();
        }

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[tokio::test]
        async fn transfers_data_if_second_channel_is_not_used(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, _local_channel2) = divide_stream(local_channel);
            let (remote_channel1, _remote_channel2) = divide_stream(remote_channel);

            test_stream(
                local_channel1,
                remote_channel1,
                TestOptions::random()
                    .with_data_len(test_data_size),
            ).await;
        }

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[tokio::test]
        async fn transfers_data_if_first_channel_is_not_used(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (_local_channel1, local_channel2) = divide_stream(local_channel);
            let (_remote_channel1, remote_channel2) = divide_stream(remote_channel);

            test_stream(
                local_channel2,
                remote_channel2,
                TestOptions::random()
                    .with_data_len(test_data_size),
            ).await;
        }

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[tokio::test]
        async fn transfers_data_if_second_channel_dropped(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, local_channel2) = divide_stream(local_channel);
            let (remote_channel1, remote_channel2) = divide_stream(remote_channel);

            drop(local_channel2);
            drop(remote_channel2);

            test_stream(
                local_channel1,
                remote_channel1,
                TestOptions::random()
                    .with_data_len(test_data_size),
            ).await;
        }

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[tokio::test]
        async fn transfers_data_if_first_channel_dropped(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, local_channel2) = divide_stream(local_channel);
            let (remote_channel1, remote_channel2) = divide_stream(remote_channel);

            drop(local_channel1);
            drop(remote_channel1);

            test_stream(
                local_channel2,
                remote_channel2,
                TestOptions::random()
                    .with_data_len(test_data_size),
            ).await;
        }

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[tokio::test]
        async fn transfers_data_if_second_channel_shut_down(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, mut local_channel2) = divide_stream(local_channel);
            let (remote_channel1, mut remote_channel2) = divide_stream(remote_channel);

            local_channel2.shutdown().await.unwrap();
            remote_channel2.shutdown().await.unwrap();

            test_stream(
                local_channel1,
                remote_channel1,
                TestOptions::random()
                    .with_data_len(test_data_size),
            ).await;
        }

        #[rstest]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[case(32_768)]
        #[tokio::test]
        async fn transfers_data_if_second_channel_shut_down_during_transfer(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, local_channel2) = divide_stream(local_channel);
            let (remote_channel1, remote_channel2) = divide_stream(remote_channel);

            tokio::join!(
                Box::pin(async move {
                    test_stream(
                        local_channel1,
                        remote_channel1,
                        TestOptions::random()
                            .with_data_len(test_data_size),
                    ).await;
                }),
                Box::pin(async move {
                    // transfer a bit of data
                    let (mut local_channel2, mut remote_channel2) = test_stream(
                        local_channel2,
                        remote_channel2,
                        TestOptions::random()
                            .with_data_len(128),
                    ).await;

                    // shutdown the second channel
                    tokio::join!(
                        Box::pin(async move {
                            wait_random(1..5).await;
                            local_channel2.shutdown().await.unwrap();
                        }),
                        Box::pin(async move {
                            wait_random(1..5).await;
                            remote_channel2.shutdown().await.unwrap();
                        }),
                    );
                }),
            );
        }

        #[rstest]
        #[case(128)]
        #[case(256)]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[case(32_768)]
        #[tokio::test]
        async fn transfers_data_if_first_channel_shut_down(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (mut local_channel1, local_channel2) = divide_stream(local_channel);
            let (mut remote_channel1, remote_channel2) = divide_stream(remote_channel);

            local_channel1.shutdown().await.unwrap();
            remote_channel1.shutdown().await.unwrap();

            test_stream(
                local_channel2,
                remote_channel2,
                TestOptions::random()
                    .with_data_len(test_data_size),
            ).await;
        }

        #[rstest]
        #[case(512)]
        #[case(1_024)]
        #[case(2_048)]
        #[case(4_096)]
        #[case(8_192)]
        #[case(16_384)]
        #[case(32_768)]
        #[tokio::test]
        async fn transfers_data_if_first_channel_shut_down_during_transfer(
            #[case] test_data_size: usize,
        ) {
            let (local_channel, remote_channel) = channel_mock_pair(
                ChannelMockOptions::random().with_label("local"),
                ChannelMockOptions::random().with_label("remote"),
            );

            let (local_channel1, local_channel2) = divide_stream(local_channel);
            let (remote_channel1,remote_channel2) = divide_stream(remote_channel);

            tokio::join!(
                Box::pin(async move {
                    // transfer a bit of data
                    let (mut local_channel1, mut remote_channel1) = test_stream(
                        local_channel1,
                        remote_channel1,
                        TestOptions::random()
                            .with_data_len(128),
                    ).await;

                    // shutdown the first channel
                    tokio::join!(
                        Box::pin(async move {
                            wait_random(1..5).await;
                            local_channel1.shutdown().await.unwrap();
                        }),
                        Box::pin(async move {
                            wait_random(1..5).await;
                            remote_channel1.shutdown().await.unwrap();
                        }),
                    );
                }),
                Box::pin(async move {
                    test_stream(
                        local_channel2,
                        remote_channel2,
                        TestOptions::random()
                            .with_data_len(test_data_size),
                    ).await;
                }),
            );
        }

        mod shutdown {
            use super::*;

            #[rstest]
            #[case(128)]
            #[case(256)]
            #[case(512)]
            #[case(1_024)]
            #[case(2_048)]
            #[case(4_096)]
            #[case(8_192)]
            #[case(16_384)]
            #[tokio::test]
            async fn first_channel_closes_channel_if_remote_counterpart_is_closed(
                #[case] test_data_size: usize,
            ) {
                let (local_channel, remote_channel) = channel_mock_pair(
                    ChannelMockOptions::random().with_label("local"),
                    ChannelMockOptions::random().with_label("remote"),
                );

                let (local_channel1, local_channel2) = divide_stream(local_channel);
                let (remote_channel1, remote_channel2) = divide_stream(remote_channel);

                let (
                    mut local_channel1,
                    mut remote_channel1,
                ) = test_stream(
                    local_channel1,
                    remote_channel1,
                    TestOptions::random()
                        .with_data_len(test_data_size),
                ).await;

                tokio::join!(
                    // test that the second channel is not affected by shutting down of the first one
                    Box::pin(async move {
                        wait_random(1..=10).await;

                        return test_stream(
                            local_channel2,
                            remote_channel2,
                            TestOptions::random()
                                .with_data_len(2 * test_data_size),
                        ).await
                    }),
                    // transfer data in the first channel and then shut it down
                    // after shutting down, test that the remote counterpart receives `EOF` message
                    Box::pin(async move {
                        tokio::join!(
                            Box::pin(async move {
                                wait_random(5..=25).await;
                
                                local_channel1.shutdown().await.unwrap();

                                assert!(
                                    local_channel1.write("test".as_bytes()).await.is_err(),
                                    "Must fail to write to closed local channel.",
                                );

                                assert!(
                                    local_channel1.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                            Box::pin(async move {
                                let mut buf = [0; 4096];
                
                                let bytes_read = remote_channel1.read(&mut buf)
                                    .await.unwrap();
                
                                assert_eq!(
                                    bytes_read,
                                    0,
                                    "Must be 0 bytes read if remote channel is closed.",
                                );

                                assert!(
                                    remote_channel1.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                        );
                    }),
                );
            }

            #[rstest]
            #[case(128)]
            #[case(256)]
            #[case(512)]
            #[case(1_024)]
            #[case(2_048)]
            #[case(4_096)]
            #[case(8_192)]
            #[case(16_384)]
            #[tokio::test]
            async fn second_channel_closes_channel_if_remote_counterpart_is_closed(
                #[case] test_data_size: usize,
            ) {
                let (local_channel, remote_channel) = channel_mock_pair(
                    ChannelMockOptions::random().with_label("local"),
                    ChannelMockOptions::random().with_label("remote"),
                );

                let (local_channel1, local_channel2) = divide_stream(local_channel);
                let (remote_channel1, remote_channel2) = divide_stream(remote_channel);

                let (
                    mut local_channel2,
                    mut remote_channel2,
                ) = test_stream(
                    local_channel2,
                    remote_channel2,
                    TestOptions::random()
                        .with_data_len(test_data_size),
                ).await;

                tokio::join!(
                    // test that the first channel is not affected by shutting down of the first one
                    Box::pin(async move {
                        wait_random(1..=10).await;

                        return test_stream(
                            local_channel1,
                            remote_channel1,
                            TestOptions::random()
                                .with_data_len(2 * test_data_size),
                        ).await
                    }),
                    // transfer data in the second channel and then shut it down
                    // after shutting down, test that the remote counterpart receives `EOF` message
                    Box::pin(async move {
                        tokio::join!(
                            Box::pin(async move {
                                wait_random(5..=25).await;
                
                                local_channel2.shutdown().await.unwrap();

                                assert!(
                                    local_channel2.write("test".as_bytes()).await.is_err(),
                                    "Must fail to write to closed local channel.",
                                );

                                assert!(
                                    local_channel2.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                            Box::pin(async move {
                                let mut buf = [0; 4096];
                
                                let bytes_read = remote_channel2.read(&mut buf)
                                    .await.unwrap();
                
                                assert_eq!(
                                    bytes_read,
                                    0,
                                    "Must be 0 bytes read if remote channel is closed.",
                                );

                                assert!(
                                    remote_channel2.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                        );
                    }),
                );
            }

            #[rstest]
            #[case(128)]
            #[case(256)]
            #[case(512)]
            #[case(1_024)]
            #[case(2_048)]
            #[case(4_096)]
            #[case(8_192)]
            #[case(16_384)]
            #[tokio::test]
            async fn both_channels_close_channel_if_remote_counterpart_is_closed(
                #[case] test_data_size: usize,
            ) {
                let (local_channel, remote_channel) = channel_mock_pair(
                    ChannelMockOptions::random().with_label("local"),
                    ChannelMockOptions::random().with_label("remote"),
                );

                let (local_channel1, local_channel2) = divide_stream(local_channel);
                let (remote_channel1, remote_channel2) = divide_stream(remote_channel);

                tokio::join!(
                    // transfer data in the first channel and then shut it down
                    // after shutting down, test that the remote counterpart receives `EOF` message
                    Box::pin(async move {
                        wait_random(1..=25).await;

                        let (
                            mut local_channel1,
                            mut remote_channel1,
                        ) = test_stream(
                            local_channel1,
                            remote_channel1,
                            TestOptions::random()
                                .with_data_len(test_data_size),
                        ).await;

                        tokio::join!(
                            Box::pin(async move {
                                wait_random(1..=25).await;
                
                                local_channel1.shutdown().await.unwrap();

                                assert!(
                                    local_channel1.write(cs_utils::random_str_rg(16..64).as_bytes()).await.is_err(),
                                    "Must fail to write to closed local channel.",
                                );

                                assert!(
                                    local_channel1.write("test".as_bytes()).await.is_err(),
                                    "Must fail to write to closed local channel.",
                                );

                                assert!(
                                    local_channel1.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                            Box::pin(async move {
                                let mut buf = [0; 4096];
                
                                let bytes_read = remote_channel1.read(&mut buf)
                                    .await.unwrap();
                
                                assert_eq!(
                                    bytes_read,
                                    0,
                                    "Must be 0 bytes read if remote channel is closed.",
                                );

                                assert!(
                                    remote_channel1.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                        );
                    }),
                    // transfer data in the second channel and then shut it down
                    // after shutting down, test that the remote counterpart receives `EOF` message
                    Box::pin(async move {
                        wait_random(1..=25).await;

                        let (
                            mut local_channel2,
                            mut remote_channel2,
                        ) = test_stream(
                            local_channel2,
                            remote_channel2,
                            TestOptions::random()
                                .with_data_len(test_data_size),
                        ).await;

                        tokio::join!(
                            Box::pin(async move {
                                wait_random(1..=25).await;
                
                                local_channel2.shutdown().await.unwrap();

                                assert!(
                                    local_channel2.write("test".as_bytes()).await.is_err(),
                                    "Must fail to write to closed local channel.",
                                );

                                assert!(
                                    local_channel2.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                            Box::pin(async move {
                                let mut buf = [0; 4096];
                
                                let bytes_read = remote_channel2.read(&mut buf)
                                    .await.unwrap();
                
                                assert_eq!(
                                    bytes_read,
                                    0,
                                    "Must be 0 bytes read if remote channel is closed.",
                                );

                                assert!(
                                    remote_channel2.is_closed(),
                                    "Channel must be closed.",
                                );
                            }),
                        );
                    }),
                );
            }
        }
    }
}