//! Declare DataStream, a type that wraps RawCellStream so as to be useful

//! for byte-oriented communication.

use crate::{Error, Result};

use tor_cell::relaycell::msg::EndReason;

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

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

use futures::Future;

#[cfg(feature = "tokio")]

use tokio_crate::io::ReadBuf;

#[cfg(feature = "tokio")]

use tokio_crate::io::{AsyncRead as TokioAsyncRead, AsyncWrite as TokioAsyncWrite};

#[cfg(feature = "tokio")]

use tokio_util::compat::{FuturesAsyncReadCompatExt, FuturesAsyncWriteCompatExt};

use std::fmt::Debug;

use std::io::Result as IoResult;

use std::pin::Pin;

use educe::Educe;

use crate::circuit::StreamTarget;

use crate::stream::StreamReader;

use tor_basic_utils::skip_fmt;

use tor_cell::relaycell::msg::{Data, RelayMsg};

use tor_error::internal;

/// An anonymized stream over the Tor network.

///

/// For most purposes, you can think of this type as an anonymized

/// TCP stream: it can read and write data, and get closed when it's done.

///

/// [`DataStream`] implements [`futures::io::AsyncRead`] and

/// [`futures::io::AsyncWrite`], so you can use it anywhere that those

/// traits are expected.

///

/// # Examples

///

/// Connecting to an HTTP server and sending a request, using

/// [`AsyncWriteExt::write_all`](futures::io::AsyncWriteExt::write_all):

///

/// ```ignore

/// let mut stream = tor_client.connect(("icanhazip.com", 80), None).await?;

///

/// use futures::io::AsyncWriteExt;

///

/// stream

///     .write_all(b"GET / HTTP/1.1\r\nHost: icanhazip.com\r\nConnection: close\r\n\r\n")

///     .await?;

///

/// // Flushing the stream is important; see below!

/// stream.flush().await?;

/// ```

///

/// Reading the result, using [`AsyncReadExt::read_to_end`](futures::io::AsyncReadExt::read_to_end):

///

/// ```ignore

/// use futures::io::AsyncReadExt;

///

/// let mut buf = Vec::new();

/// stream.read_to_end(&mut buf).await?;

///

/// println!("{}", String::from_utf8_lossy(&buf));

/// ```

///

/// # Usage with Tokio

///

/// If the `tokio` crate feature is enabled, this type also implements

/// [`tokio::io::AsyncRead`](tokio_crate::io::AsyncRead) and

/// [`tokio::io::AsyncWrite`](tokio_crate::io::AsyncWrite) for easier integration

/// with code that expects those traits.

///

/// # Remember to call `flush`!

///

/// DataStream buffers data internally, in order to write as few cells

/// as possible onto the network.  In order to make sure that your

/// data has actually been sent, you need to make sure that

/// [`AsyncWrite::poll_flush`] runs to completion: probably via

/// [`AsyncWriteExt::flush`](futures::io::AsyncWriteExt::flush).

///

/// # Splitting the type

///

/// This type is internally composed of a [`DataReader`] and a [`DataWriter`]; the

/// `DataStream::split` method can be used to split it into those two parts, for more

/// convenient usage with e.g. stream combinators.

// # Semver note

//

// Note that this type is re-exported as a part of the public API of

// the `arti-client` crate.  Any changes to its API here in

// `tor-proto` need to be reflected above.

pub struct DataStream {

    /// Underlying writer for this stream

    w: DataWriter,

    /// Underlying reader for this stream

    r: DataReader,

}

/// The write half of a [`DataStream`], implementing [`futures::io::AsyncWrite`].

///

/// See the [`DataStream`] docs for more information. In particular, note

/// that this writer requires `poll_flush` to complete in order to guarantee that

/// all data has been written.

///

/// # Usage with Tokio

///

/// If the `tokio` crate feature is enabled, this type also implements

/// [`tokio::io::AsyncWrite`](tokio_crate::io::AsyncWrite) for easier integration

/// with code that expects that trait.

// # Semver note

//

// Note that this type is re-exported as a part of the public API of

// the `arti-client` crate.  Any changes to its API here in

// `tor-proto` need to be reflected above.

pub struct DataWriter {

    /// Internal state for this writer

    ///

    /// This is stored in an Option so that we can mutate it in the

    /// AsyncWrite functions.  It might be possible to do better here,

    /// and we should refactor if so.

    state: Option<DataWriterState>,

}

/// The read half of a [`DataStream`], implementing [`futures::io::AsyncRead`].

///

/// See the [`DataStream`] docs for more information.

///

/// # Usage with Tokio

///

/// If the `tokio` crate feature is enabled, this type also implements

/// [`tokio::io::AsyncRead`](tokio_crate::io::AsyncRead) for easier integration

/// with code that expects that trait.

//

// # Semver note

//

// Note that this type is re-exported as a part of the public API of

// the `arti-client` crate.  Any changes to its API here in

// `tor-proto` need to be reflected above.

pub struct DataReader {

    /// Internal state for this reader.

    ///

    /// This is stored in an Option so that we can mutate it in

    /// poll_read().  It might be possible to do better here, and we

    /// should refactor if so.

    state: Option<DataReaderState>,

}

impl DataStream {

    /// Wrap raw stream reader and target parts as a DataStream.

    ///

    /// For non-optimistic stream, function `wait_for_connection`

    /// must be called after to make sure CONNECTED is received.

    /// Divide this DataStream into its constituent parts.

    /// Wait until a CONNECTED cell is received, or some other cell

    /// is received to indicate an error.

    ///

    /// Does nothing if this stream is already connected.

            } else {

                // This succeeds if the cell is CONNECTED, and fails otherwise.

            };

        } else {

        }

}

impl AsyncRead for DataStream {

}

#[cfg(feature = "tokio")]

impl TokioAsyncRead for DataStream {

}

impl AsyncWrite for DataStream {

}

#[cfg(feature = "tokio")]

impl TokioAsyncWrite for DataStream {

}

/// An enumeration for the state of a DataWriter.

///

/// We have to use an enum here because, for as long as we're waiting

/// for a flush operation to complete, the future returned by

/// `flush_cell()` owns the DataWriterImpl.

#[educe(Debug)]

enum DataWriterState {

    /// The writer has closed or gotten an error: nothing more to do.

    Closed,

    /// The writer is not currently flushing; more data can get queued

    /// immediately.

    Ready(DataWriterImpl),

    /// The writer is flushing a cell.

    Flushing(

        #[educe(Debug(method = "skip_fmt"))]

        Pin<Box<dyn Future<Output = (DataWriterImpl, Result<()>)> + Send>>,

    ),

}

/// Internal: the write part of a DataStream

#[educe(Debug)]

struct DataWriterImpl {

    /// The underlying StreamTarget object.

    s: StreamTarget,

    /// Buffered data to send over the connection.

    // TODO: this buffer is probably smaller than we want, but it's good

    // enough for now.  If we _do_ make it bigger, we'll have to change

    // our use of Data::split_from to handle the case where we can't fit

    // all the data.

    #[educe(Debug(method = "skip_fmt"))]

    buf: Box<[u8; Data::MAXLEN]>,

    /// Number of unflushed bytes in buf.

    n_pending: usize,

}

impl DataWriter {

    /// Helper for poll_flush() and poll_close(): Performs a flush, then

    /// closes the stream if should_close is true.

        // TODO: this whole function is a bit copy-pasted.

                    // Nothing to flush!

            }

            DataWriterState::Closed => {

            }

        };

            }

            }

            Poll::Pending => {

            }

        }

}

impl AsyncWrite for DataWriter {

            }

            DataWriterState::Closed => {

            }

        };

            }

            }

            Poll::Pending => {

            }

        }

}

#[cfg(feature = "tokio")]

impl TokioAsyncWrite for DataWriter {

}

impl DataWriterImpl {

    /// Try to flush the current buffer contents as a data cell.

        } else {

        };

    /// Add as many bytes as possible from `b` to our internal buffer;

    /// return the number we were able to add.

            // that is, len == 0

}

/// An enumeration for the state of a DataReader.

///

/// We have to use an enum here because, when we're waiting for

/// ReadingCell to complete, the future returned by `read_cell()` owns the

/// DataCellImpl.  If we wanted to store the future and the cell at the

/// same time, we'd need to make a self-referential structure, which isn't

/// possible in safe Rust AIUI.

#[educe(Debug)]

enum DataReaderState {

    /// In this state we have received an end cell or an error.

    Closed,

    /// In this state the reader is not currently fetching a cell; it

    /// either has data or not.

    Ready(DataReaderImpl),

    /// The reader is currently fetching a cell: this future is the

    /// progress it is making.

    ReadingCell(

        #[educe(Debug(method = "skip_fmt"))]

        Pin<Box<dyn Future<Output = (DataReaderImpl, Result<()>)> + Send>>,

    ),

}

/// Wrapper for the read part of a DataStream

#[educe(Debug)]

struct DataReaderImpl {

    /// The underlying StreamReader object.

    #[educe(Debug(method = "skip_fmt"))]

    s: StreamReader,

    /// If present, data that we received on this stream but have not

    /// been able to send to the caller yet.

    // TODO: This data structure is probably not what we want, but

    // it's good enough for now.

    #[educe(Debug(method = "skip_fmt"))]

    pending: Vec<u8>,

    /// Index into pending to show what we've already read.

    offset: usize,

    /// If true, we have received a CONNECTED cell on this stream.

    connected: bool,

}

impl AsyncRead for DataReader {

        loop {

                        // We read data into the buffer.  Tell the caller.

                }

                DataReaderState::Closed => {

                }

            };

            // We have a future that represents an in-progress read.

            // See if it can make progress.

                    } else {

                    };

                }

                Poll::Pending => {

                    // The future is pending; store it and tell the

                    // caller to get back to us later.

                }

            }

        }

}

#[cfg(feature = "tokio")]

impl TokioAsyncRead for DataReader {

}

impl DataReaderImpl {

    /// Pull as many bytes as we can off of self.pending, and return that

    /// number of bytes.

    /// Return true iff there are no buffered bytes here to yield

    /// Load self.pending with the contents of a new data cell.

    ///

    /// This function takes ownership of self so that we can avoid

    /// self-referential lifetimes.

            }

            }

            }

        };

    /// Add the data from `d` to the end of our pending bytes.

}