//! Mocking helpers for testing with futures::io types.

//!

//! Note that some of this code might be of general use, but for now

//! we're only trying it for testing.

use futures::channel::mpsc;

use futures::io::{AsyncRead, AsyncWrite};

use futures::sink::{Sink, SinkExt};

use futures::stream::Stream;

use std::io::{Error as IoError, ErrorKind, Result as IoResult};

use std::pin::Pin;

use std::task::{Context, Poll};

/// Channel capacity for our internal MPSC channels.

///

/// We keep this intentionally low to make sure that some blocking

/// will occur occur.

const CAPACITY: usize = 4;

/// Maximum size for a queued buffer on a local chunk.

///

/// This size is deliberately weird, to try to find errors.

const CHUNKSZ: usize = 213;

/// Construct a new pair of linked LocalStream objects.

///

/// Any bytes written to one will be readable on the other, and vice

/// versa.  These streams will behave more or less like a socketpair,

/// except without actually going through the operating system.

///

/// Note that this implementation is intended for testing only, and

/// isn't optimized.

/// One half of a pair of linked streams returned by [`stream_pair`].

//

// Implementation notes: linked streams are made out a pair of mpsc

// channels.  There's one channel for sending bytes in each direction.

// Bytes are sent as IoResult<Vec<u8>>: sending an error causes an error

// to occur on the other side.

pub struct LocalStream {

    /// The writing side of the channel that we use to implement this

    /// stream.

    ///

    /// The reading side is held by the other linked stream.

    w: mpsc::Sender<IoResult<Vec<u8>>>,

    /// The reading side of the channel that we use to implement this

    /// stream.

    ///

    /// The writing side is held by the other linked stream.

    r: mpsc::Receiver<IoResult<Vec<u8>>>,

    /// Bytes that we have read from `r` but not yet delivered.

    pending_bytes: Vec<u8>,

    /// Data about the other side of this stream's fake TLS certificate, if any.

    /// If this is present, I/O operations will fail with an error.

    ///

    /// How this is intended to work: things that return `LocalStream`s that could potentially

    /// be connected to a fake TLS listener should set this field. Then, a fake TLS wrapper

    /// type would clear this field (after checking its contents are as expected).

    ///

    /// FIXME(eta): this is a bit of a layering violation, but it's hard to do otherwise

    pub(crate) tls_cert: Option<Vec<u8>>,

}

/// Helper: pull bytes off the front of `pending_bytes` and put them

/// onto `buf.  Return the number of bytes moved.

impl AsyncRead for LocalStream {

            }

        }

}

impl AsyncWrite for LocalStream {

        }

        } else {

        };

        }

}

/// An error generated by [`LocalStream::send_err`].

#[non_exhaustive]

pub struct SyntheticError;

impl std::error::Error for SyntheticError {}

impl std::fmt::Display for SyntheticError {

}

impl LocalStream {

    /// Send an error to the other linked local stream.

    ///

    /// When the other stream reads this message, it will generate a

    /// [`std::io::Error`] with the provided `ErrorKind`.

}

#[cfg(test)]

mod test {

    #![allow(clippy::unwrap_used)]

    use super::*;

    use futures::io::{AsyncReadExt, AsyncWriteExt};

    use futures_await_test::async_test;

    use rand::thread_rng;

    use rand::Rng;

    #[async_test]

    async fn basic_rw() {

        let (mut s1, mut s2) = stream_pair();

        let mut text1 = vec![0_u8; 9999];

        thread_rng().fill(&mut text1[..]);

        let (v1, v2): (IoResult<()>, IoResult<()>) = futures::join!(

            async {

                for _ in 0_u8..10 {

                    s1.write_all(&text1[..]).await?;

                }

                s1.close().await?;

                Ok(())

            },

            async {

                let mut text2: Vec<u8> = Vec::new();

                let mut buf = [0_u8; 33];

                loop {

                    let n = s2.read(&mut buf[..]).await?;

                    if n == 0 {

                        break;

                    }

                    text2.extend(&buf[..n]);

                }

                for ch in text2[..].chunks(text1.len()) {

                    assert_eq!(ch, &text1[..]);

                }

                Ok(())

            }

        );

        v1.unwrap();

        v2.unwrap();

    }

    #[async_test]

    async fn send_error() {

        let (mut s1, mut s2) = stream_pair();

        let (v1, v2): (IoResult<()>, IoResult<()>) = futures::join!(

            async {

                s1.write_all(b"hello world").await?;

                s1.send_err(ErrorKind::PermissionDenied).await;

                Ok(())

            },

            async {

                let mut buf = [0_u8; 33];

                loop {

                    let n = s2.read(&mut buf[..]).await?;

                    if n == 0 {

                        break;

                    }

                }

                Ok(())

            }

        );

        v1.unwrap();

        let e = v2.err().unwrap();

        assert_eq!(e.kind(), ErrorKind::PermissionDenied);

        let synth = e.into_inner().unwrap();

        assert_eq!(synth.to_string(), "Synthetic error");

    }

    #[async_test]

    async fn drop_reader() {

        let (mut s1, s2) = stream_pair();

        let (v1, v2): (IoResult<()>, IoResult<()>) = futures::join!(

            async {

                for _ in 0_u16..1000 {

                    s1.write_all(&[9_u8; 9999]).await?;

                }

                Ok(())

            },

            async {

                drop(s2);

                Ok(())

            }

        );

        v2.unwrap();

        let e = v1.err().unwrap();

        assert_eq!(e.kind(), ErrorKind::BrokenPipe);

    }

}