//! Declare the lowest level of stream: a stream that operates on raw

//! cells.

use std::pin::Pin;

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

use futures::Stream;

use pin_project::pin_project;

use tor_async_utils::peekable_stream::UnobtrusivePeekableStream;

use tor_cell::relaycell::{RelayCmd, UnparsedRelayMsg};

use tracing::debug;

use crate::congestion::sendme;

use crate::stream::StreamTarget;

use crate::stream::queue::StreamQueueReceiver;

use crate::{Error, Result};

/// The read part of a stream on a particular circuit.

///

/// This [`Stream`](Stream) will return incoming messages for this Tor stream, excluding flow control

/// related messages like SENDME, XON, and XOFF.

///

/// To avoid ambiguity, the following uses "stream" to refer to the `futures::Stream`, not the Tor

/// stream.

///

/// If the stream ends unexpectedly (before an END message), the stream will return an error.

/// After the stream returns an END message or an error, this stream will be "terminated" and future

/// [`poll_next`](Stream::poll_next) calls will return `None`.

// I think it would be better to *not* return an error if the stream ends before an END message is

// received, and just return `None`. The caller will know if it received an END message or not, so

// returning an error isn't very useful and is maybe unexpected.

#[derive(Debug)]

#[pin_project]

pub struct StreamReceiver {

    /// The underlying `StreamTarget` for this stream.

    ///

    /// A reader has this target in order to:

    ///   * Make the reactor send SENDME messages.

    ///   * Tell the reactor when there is a protocol error.

    ///   * Keep the stream alive at least until the StreamReceiver

    ///     is dropped.

    pub(crate) target: StreamTarget,

    /// Channel to receive stream messages from the reactor.

    #[pin]

    pub(crate) receiver: StreamQueueReceiver,

    /// Congestion control receive window for this stream.

    ///

    /// Having this here means we're only going to update it when the end consumer of this stream

    /// actually reads things, meaning we don't ask for more data until it's actually needed (as

    /// opposed to having the reactor assume we're always reading, and potentially overwhelm itself

    /// with having to buffer data).

    pub(crate) recv_window: sendme::StreamRecvWindow,

    /// Whether or not this stream has ended.

    pub(crate) ended: bool,

}

impl StreamReceiver {

    /// Try to read the next relay message from this stream.

            Poll::Ready(None) => {

                // The channel is indicating that it has terminated, likely from a dropped sender.

                // But if we're here, it means we never received an END cell.

                //

                // We don't know here if the sender was dropped because either:

                // - The stream map dropped its open entry (see `StreamMap::terminate()`).

                //   For example if `DataStream::close()` was called.

                // - The circuit closed and the entire stream map was dropped.

                //

                // We only know conclusively that the stream is closed.

            }

        };

                    // If the tunnel has closed, sending a message to the tunnel reactor may fail.

                    // But this is okay. We still want the user to be able to continue reading the

                    // remaining queued data for this stream, and if the tunnel has closed it

                    // wouldn't make sense to send a SENDME message anyways.

                } else {

                    // This error is unexpected. Let's return it to the user.

                }

    /// Shut down this stream.

    /// Is the stream currently empty?

    ///

    /// This method is inherently subject to race conditions. More data may arrive even before this

    /// method returns, so a result of `true` might have been outdated before the method even

    /// returned.

    ///

    /// This takes a `&mut` so that we can peek the stream.

    ///

    /// We provide an `is_empty` method rather than implementing [`UnobtrusivePeekableStream`]

    /// directly since `UnobtrusivePeekableStream` allows you to mutate the peeked item, which could

    /// break any accounting we do here in `StreamReceiver` (like stream sendme accounting). Also

    /// the stream types are incompatible (the inner receiver returns items of `UnparsedRelayMsg`,

    /// but this [`StreamReceiver`] returns items of `Result<UnparsedRelayMsg>`).

        // The `StreamQueueReceiver` gives us two ways of checking if the queue is empty:

        // `unobtrusive_peek().is_none()` and `approx_stream_bytes() == 0`. The peek seems like a

        // better approach, so we do that here.

        // TODO(arti#534): Should reconsider using `unobtrusive_peek()`. What we really want to know

        // is if there is more stream data in the queue. But peeking only tells us if there are more

        // messages. There could be more messages, but none of them data messages.

        // if the peek says that the stream is empty, assert that `approx_stream_bytes()` shows 0

        // bytes

        #[cfg(debug_assertions)]

}

impl Stream for StreamReceiver {

    type Item = Result<UnparsedRelayMsg>;

            // Prevent reading more messages from streams after they've ended. `None` indicates that

            // the stream is complete/terminated.

            }

                // We return the error, and future polls will return `None`.

            }

        }

}