//! Queues for stream messages.

//!

//! While these are technically "channels", we call them "queues" to indicate that they're mostly

//! just dumb pipes. They do some tracking (memquota and size), but nothing else. The higher-level

//! object is [`StreamReceiver`](crate::stream::raw::StreamReceiver) which tracks SENDME and END

//! messages. So the idea is that the "queue" (ex: [`StreamQueueReceiver`]) just holds data and the

//! "channel" (ex: `StreamReceiver`) adds the Tor logic.

//!

//! The main purpose of these types are so that we can count how many bytes of stream data are

//! stored for the stream. Ideally we'd use a channel type that tracks and reports this as part of

//! its implementation, but popular channel implementations don't seem to do that.

use std::fmt::Debug;

use std::pin::Pin;

use std::sync::{Arc, Mutex};

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

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

use tor_async_utils::SinkTrySend;

use tor_async_utils::peekable_stream::UnobtrusivePeekableStream;

use tor_async_utils::stream_peek::StreamUnobtrusivePeeker;

use tor_cell::relaycell::UnparsedRelayMsg;

use tor_memquota::mq_queue::{self, ChannelSpec, MpscSpec, MpscUnboundedSpec};

use tor_rtcompat::DynTimeProvider;

use crate::memquota::{SpecificAccount, StreamAccount};

// TODO(arti#534): remove these type aliases when we remove the "flowctl-cc" feature,

// and just use `MpscUnboundedSpec` everywhere

#[cfg(feature = "flowctl-cc")]

/// Alias for the memquota mpsc spec.

type Spec = MpscUnboundedSpec;

#[cfg(not(feature = "flowctl-cc"))]

/// Alias for the memquota mpsc spec.

type Spec = MpscSpec;

/// Create a new stream queue for incoming messages.

        cfg_if::cfg_if! {

            if #[cfg(not(feature = "flowctl-cc"))] {

                MpscSpec::new(size).new_mq(time_prov.clone(), memquota.as_raw_account())?

            } else {

            }

        }

    };

/// For testing purposes, create a stream queue wth a no-op memquota account and a fake time

/// provider.

#[cfg(test)]

    // The fake Account doesn't care about the data ages, so this will do.

    //

    // This would be wrong to use generally in tests, where we might want to mock time,

    // since we end up, here with totally *different* mocked time.

    // But it's OK here, and saves passing a runtime parameter into this function.

        #[cfg(not(feature = "flowctl-cc"))]

        size,

    )

/// The sending end of a channel of incoming stream messages.

#[derive(Debug)]

#[pin_project::pin_project]

pub(crate) struct StreamQueueSender {

    /// The inner sender.

    #[pin]

    sender: mq_queue::Sender<UnparsedRelayMsg, Spec>,

    /// Number of bytes within the queue.

    counter: Arc<Mutex<usize>>,

}

/// The receiving end of a channel of incoming stream messages.

#[derive(Debug)]

#[pin_project::pin_project]

pub(crate) struct StreamQueueReceiver {

    /// The inner receiver.

    ///

    /// We add the [`StreamUnobtrusivePeeker`] here so that peeked messages are included in

    /// `counter`.

    // TODO(arti#534): the possible extra msg held by the `StreamUnobtrusivePeeker` isn't tracked by

    // memquota

    #[pin]

    receiver: StreamUnobtrusivePeeker<mq_queue::Receiver<UnparsedRelayMsg, Spec>>,

    /// Number of bytes within the queue.

    counter: Arc<Mutex<usize>>,

}

impl StreamQueueSender {

    /// Get the approximate number of data bytes queued for this stream.

    ///

    /// As messages can be dequeued at any time, the return value may be larger than the actual

    /// number of bytes queued for this stream.

}

impl StreamQueueReceiver {

    /// Get the approximate number of data bytes queued for this stream.

    ///

    /// As messages can be enqueued at any time, the return value may be smaller than the actual

    /// number of bytes queued for this stream.

}

impl Sink<UnparsedRelayMsg> for StreamQueueSender {

    type Error = <mq_queue::Sender<UnparsedRelayMsg, MpscSpec> as Sink<UnparsedRelayMsg>>::Error;

        // This lock ensures that us sending the item and the counter increase are done

        // "atomically", so that the receiver doesn't see the item and try to decrement the

        // counter before we've incremented the counter, which could cause an underflow.

}

impl SinkTrySend<UnparsedRelayMsg> for StreamQueueSender {

    type Error =

        <mq_queue::Sender<UnparsedRelayMsg, MpscSpec> as SinkTrySend<UnparsedRelayMsg>>::Error;

        // See comments in `StreamQueueSender::start_send`.

}

impl Stream for StreamQueueReceiver {

    type Item = UnparsedRelayMsg;

        // This lock ensures that us receiving the item and the counter decrease are done

        // "atomically", so that the sender doesn't send a new item and try to increase the

        // counter before we've decreased the counter, which could cause an overflow.

        };

}

impl UnobtrusivePeekableStream for StreamQueueReceiver {

}

/// The `length` field of the message, or 0 if not a data message.

///

/// If the RELAY_DATA message had an invalid length field, we just ignore the message.

/// The receiver will find out eventually when it tries to parse the message.

/// We could return an error here, but for now I think it's best not to behave as if this

/// queue is performing any validation.

///

/// This is its own function so that all parts of the code use the same logic.