// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
// option. This file may not be copied, modified, or distributed
// except according to those terms.

// Building a stream of ordered bytes to give the application from a series of
// incoming STREAM frames.

use std::{
    cell::RefCell,
    cmp::max,
    collections::BTreeMap,
    fmt::Debug,
    mem,
    rc::{Rc, Weak},
    time::{Duration, Instant},
};

use neqo_common::{Buffer, Role, qtrace};
use smallvec::SmallVec;
use strum::Display;

use crate::{
    AppError, Error, Res,
    events::ConnectionEvents,
    fc::ReceiverFlowControl,
    frame::FrameType,
    packet,
    recovery::{self, StreamRecoveryToken},
    send_stream::SendStreams,
    stats::FrameStats,
    stream_id::StreamId,
};

#[derive(Debug, Default)]
pub struct RecvStreams {
    streams: BTreeMap<StreamId, RecvStream>,
    keep_alive: Weak<()>,
    /// Set when any stream has ended; cleared by `remove_ended`.
    has_ended: bool,
}

impl RecvStreams {
    pub fn write_frames<B: Buffer>(
        &mut self,
        builder: &mut packet::Builder<B>,
        tokens: &mut recovery::Tokens,
        stats: &mut FrameStats,
        now: Instant,
        rtt: Duration,
    ) {
        for stream in self.streams.values_mut() {
            stream.write_frame(builder, tokens, stats, now, rtt);
            if builder.is_full() {
                return;
            }
        }
    }

    pub fn insert(&mut self, id: StreamId, stream: RecvStream) {
        self.streams.insert(id, stream);
    }

    #[allow(
        clippy::allow_attributes,
        clippy::missing_errors_doc,
        reason = "OK here."
    )]
    pub fn get_mut(&mut self, id: StreamId) -> Res<&mut RecvStream> {
        self.streams.get_mut(&id).ok_or(Error::InvalidStreamId)
    }

    #[allow(
        clippy::allow_attributes,
        clippy::missing_errors_doc,
        reason = "OK here."
    )]
    pub fn keep_alive(&mut self, id: StreamId, k: bool) -> Res<()> {
        let self_ka = &mut self.keep_alive;
        let s = self.streams.get_mut(&id).ok_or(Error::InvalidStreamId)?;
        s.keep_alive = k.then(|| {
            self_ka.upgrade().unwrap_or_else(|| {
                let r = Rc::new(());
                *self_ka = Rc::downgrade(&r);
                r
            })
        });
        Ok(())
    }

    #[must_use]
    pub fn need_keep_alive(&self) -> bool {
        self.keep_alive.strong_count() > 0
    }

    pub fn clear(&mut self) {
        self.streams.clear();
        self.has_ended = false;
    }

    pub(crate) const fn set_ended(&mut self, ended: bool) {
        self.has_ended |= ended;
    }

    /// Read from a stream, noting when it ends.
    ///
    /// # Errors
    /// When the stream does not exist or has no more data.
    pub fn read(&mut self, stream_id: StreamId, data: &mut [u8]) -> Res<(usize, bool)> {
        let (n, fin) = self.get_mut(stream_id)?.read(data)?;
        self.set_ended(fin);
        Ok((n, fin))
    }

    /// Stop sending on a stream, noting when it ends.
    ///
    /// # Errors
    /// When the stream does not exist.
    pub fn stop_sending(&mut self, stream_id: StreamId, err: AppError) -> Res<()> {
        let ended = self.get_mut(stream_id)?.stop_sending(err);
        self.set_ended(ended);
        Ok(())
    }

    /// Reset a stream, noting if it ended.
    ///
    /// # Errors
    /// When flow control is violated.
    pub fn reset(
        &mut self,
        stream_id: StreamId,
        application_error_code: AppError,
        final_size: u64,
    ) -> Res<()> {
        if let Ok(rs) = self.get_mut(stream_id) {
            let ended = rs.reset(application_error_code, final_size)?;
            self.set_ended(ended);
        }
        Ok(())
    }

    /// Note whether a stop-sending ack ended the stream.
    pub fn stop_sending_acked(&mut self, stream_id: StreamId) {
        if let Ok(rs) = self.get_mut(stream_id) {
            let ended = rs.stop_sending_acked();
            self.set_ended(ended);
        }
    }

    pub fn remove_ended(&mut self, send_streams: &SendStreams, role: Role) -> (u64, u64) {
        if !self.has_ended {
            return (0, 0);
        }
        self.has_ended = false;
        // Note: retained ended bidi streams (send counterpart alive) will be re-flagged
        // when their send side is removed via `cleanup_closed_streams`.
        let mut removed_bidi = 0;
        let mut removed_uni = 0;
        self.streams.retain(|id, s| {
            let dead = s.is_ended() && (id.is_uni() || !send_streams.exists(*id));
            if dead && id.is_remote_initiated(role) {
                if id.is_bidi() {
                    removed_bidi += 1;
                } else {
                    removed_uni += 1;
                }
            }
            !dead
        });

        (removed_bidi, removed_uni)
    }
}

/// Holds data not yet read by application. Orders and dedupes data ranges
/// from incoming STREAM frames.
#[derive(Debug, Default)]
pub struct RxStreamOrderer {
    data_ranges: BTreeMap<u64, Vec<u8>>, // (start_offset, data)
    retired: u64,                        // Number of bytes the application has read
    received: u64,                       // The number of bytes stored in `data_ranges`
}

impl RxStreamOrderer {
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Process an incoming stream frame off the wire. This may result in data
    /// being available to upper layers if frame is not out of order (ooo) or
    /// if the frame fills a gap.
    /// # Panics
    /// Only when `u64` values cannot be converted to `usize`, which only
    /// happens on 32-bit machines that hold far too much data at the same time.
    pub fn inbound_frame(&mut self, mut new_start: u64, mut new_data: &[u8])&nbsp;{
        qtrace!("Inbound data offset={new_start} len={}", new_data.len());

        // Get entry before where new entry would go, so we can see if we already
        // have the new bytes.
        // Avoid copies and duplicated data.
        let new_end = new_start + u64::try_from(new_data.len()).expect("usize fits in u64");

        if new_end <= self.retired {
            // Range already read by application, this frame is very late and unneeded.
            return;
        }

        if new_start < self.retired {
            new_data =
                &new_data[usize::try_from(self.retired - new_start).expect("u64 fits in usize")..];
            new_start = self.retired;
        }

        if new_data.is_empty() {
            // No data to insert
            return;
        }

        let extend = if let Some((&prev_start, prev_vec)) =
            self.data_ranges.range_mut(..=new_start).next_back()
        {
            let prev_end = prev_start + u64::try_from(prev_vec.len()).expect("usize fits in u64");
            if new_end > prev_end {
                // PPPPPP    ->  PPPPPP
                //   NNNNNN            NN
                // NNNNNNNN            NN
                // Add a range containing only new data
                // (In-order frames will take this path, with no overlap)
                let overlap = prev_end.saturating_sub(new_start);
                qtrace!("New frame {new_start}-{new_end} received, overlap: {overlap}");
                new_start += overlap;
                new_data = &new_data[usize::try_from(overlap).expect("u64 fits in usize")..];
                // If it is small enough, extend the previous buffer.
                // This can't always extend, because otherwise the buffer could end up
                // growing indefinitely without being released.
                prev_vec.len() < 4096 && prev_end == new_start
            } else {
                // PPPPPP    ->  PPPPPP
                //   NNNN
                // NNNN
                // Do nothing
                qtrace!("Dropping frame with already-received range {new_start}-{new_end}");
                return;
            }
        } else {
            qtrace!("New frame {new_start}-{new_end} received");
            false
        };

        let mut to_add = new_data;
        if self
            .data_ranges
            .last_entry()
            .is_some_and(|e| *e.key() >= new_start)
        {
            // Is this at the end (common case)?  If so, nothing to do in this block
            // Common case:
            //  PPPPPP        -> PPPPPP
            //        NNNNNNN          NNNNNNN
            // or
            //  PPPPPP             -> PPPPPP
            //             NNNNNNN               NNNNNNN
            //
            // Not the common case, handle possible overlap with next entries
            //  PPPPPP       AAA      -> PPPPPP
            //        NNNNNNN                  NNNNNNN
            // or
            //  PPPPPP     AAAA      -> PPPPPP     AAAA
            //        NNNNNNN                 NNNNN
            // or (this is where to_remove is used)
            //  PPPPPP    AA       -> PPPPPP
            //        NNNNNNN               NNNNNNN

            let mut to_remove = SmallVec::<[_; 8]>::new();

            for (&next_start, next_data) in self.data_ranges.range_mut(new_start..) {
                let next_end =
                    next_start + u64::try_from(next_data.len()).expect("usize fits in u64");
                let overlap = new_end.saturating_sub(next_start);
                if overlap == 0 {
                    // Fills in the hole, exactly (probably common)
                    break;
                } else if next_end >= new_end {
                    qtrace!(
                        "New frame {new_start}-{new_end} overlaps with next frame by {overlap}, truncating"
                    );
                    let truncate_to =
                        new_data.len() - usize::try_from(overlap).expect("u64 fits in usize");
                    to_add = &new_data[..truncate_to];
                    break;
                }
                qtrace!(
                    "New frame {new_start}-{new_end} spans entire next frame {next_start}-{next_end}, replacing"
                );
                to_remove.push(next_start);
                // Continue, since we may have more overlaps
            }

            for start in to_remove {
                self.data_ranges.remove(&start);
            }
        }

        if !to_add.is_empty() {
            self.received += u64::try_from(to_add.len()).expect("usize fits in u64");
            if extend {
                if let Some((_, buf)) = self.data_ranges.range_mut(..=new_start).next_back() {
                    buf.extend_from_slice(to_add);
                }
            } else {
                self.data_ranges.insert(new_start, to_add.to_vec());
            }
        }
    }

    /// Are any bytes readable?
    #[must_use]
    pub fn data_ready(&self) -> bool {
        self.data_ranges
            .keys()
            .next()
            .is_some_and(|&start| start <= self.retired)
    }

    /// How many bytes are readable?
    fn bytes_ready(&self) -> usize {
        let mut prev_end = self.retired;
        self.data_ranges
            .iter()
            .map(|(start_offset, data)| {
                // All ranges don't overlap but we could have partially
                // retired some of the first entry's data.
                let data_len = data.len() as u64 - self.retired.saturating_sub(*start_offset);
                (start_offset, data_len)
            })
            .take_while(|(start_offset, data_len)| {
                if **start_offset <= prev_end {
                    prev_end += data_len;
                    true
                } else {
                    false
                }
            })
            // Accumulate, but saturate at usize::MAX.
            .fold(0, |acc: usize, (_, data_len)| {
                acc.saturating_add(usize::try_from(data_len).unwrap_or(usize::MAX))
            })
    }

    /// Bytes read by the application.
    #[must_use]
    pub const fn retired(&self) -> u64 {
        self.retired
    }

    #[must_use]
    pub const fn received(&self) -> u64 {
        self.received
    }

    /// Data bytes buffered. Could be more than `bytes_readable` if there are
    /// ranges missing.
    fn buffered(&self) -> u64 {
        self.data_ranges
            .iter()
            .map(|(&start, data)| data.len() as u64 - (self.retired.saturating_sub(start)))
            .sum()
    }

    /// Copy received data (if any) into the buffer. Returns bytes copied.
    fn read(&mut self, buf: &mut [u8]) -> usize {
        qtrace!("Reading {} bytes, {} available", buf.len(), self.buffered());
        let mut copied = 0;

        for (&range_start, range_data) in &mut self.data_ranges {
            let mut keep = false;
            if self.retired >= range_start {
                // Frame data has new contiguous bytes.
                let copy_offset = usize::try_from(max(range_start, self.retired) - range_start)
                    .expect("u64 fits in usize");
                assert!(range_data.len() >= copy_offset);
                let available = range_data.len() - copy_offset;
                let space = buf.len() - copied;
                let copy_bytes = if available > space {
                    keep = true;
                    space
                } else {
                    available
                };

                if copy_bytes > 0 {
                    let copy_slc = &range_data[copy_offset..copy_offset + copy_bytes];
                    buf[copied..copied + copy_bytes].copy_from_slice(copy_slc);
                    copied += copy_bytes;
                    self.retired += u64::try_from(copy_bytes).expect("usize fits in u64");
                }
            } else {
                // The data in the buffer isn't contiguous.
                keep = true;
            }
            if keep {
                let mut keep = self.data_ranges.split_off(&range_start);
                mem::swap(&mut self.data_ranges, &mut keep);
                return copied;
            }
        }

        self.data_ranges.clear();
        copied
    }

    /// Extend the given Vector with any available data.
    pub fn read_to_end(&mut self, buf: &mut Vec<u8>) -> usize {
        let orig_len = buf.len();
        buf.resize(orig_len + self.bytes_ready(), 0);
        self.read(&mut buf[orig_len..])
    }
}

/// QUIC receiving states, based on -transport 3.2.
#[derive(Debug, Display)]
// Because a dead_code warning is easier than clippy::unused_self, see https://github.com/rust-lang/rust/issues/68408
enum RecvStreamState {
    Recv {
        fc: ReceiverFlowControl<StreamId>,
        session_fc: Rc<RefCell<ReceiverFlowControl<()>>>,
        recv_buf: RxStreamOrderer,
    },
    SizeKnown {
        fc: ReceiverFlowControl<StreamId>,
        session_fc: Rc<RefCell<ReceiverFlowControl<()>>>,
        recv_buf: RxStreamOrderer,
    },
    DataRecvd {
        fc: ReceiverFlowControl<StreamId>,
        session_fc: Rc<RefCell<ReceiverFlowControl<()>>>,
        recv_buf: RxStreamOrderer,
    },
    DataRead {
        final_received: u64,
        final_read: u64,
    },
    AbortReading {
        fc: ReceiverFlowControl<StreamId>,
        session_fc: Rc<RefCell<ReceiverFlowControl<()>>>,
        final_size_reached: bool,
        frame_needed: bool,
        err: AppError,
        final_received: u64,
        final_read: u64,
    },
    WaitForReset {
        fc: ReceiverFlowControl<StreamId>,
        session_fc: Rc<RefCell<ReceiverFlowControl<()>>>,
        final_received: u64,
        final_read: u64,
    },
    ResetRecvd {
        final_received: u64,
        final_read: u64,
    },
    // Defined by spec but we don't use it: ResetRead
}

impl RecvStreamState {
    fn new(
        max_bytes: u64,
        stream_id: StreamId,
        session_fc: Rc<RefCell<ReceiverFlowControl<()>>>,
    ) -> Self {
        Self::Recv {
            fc: ReceiverFlowControl::new(stream_id, max_bytes),
            recv_buf: RxStreamOrderer::new(),
            session_fc,
        }
    }

    const fn recv_buf(&self) -> Option<&RxStreamOrderer> {
        match self {
            Self::Recv { recv_buf, .. }
            | Self::SizeKnown { recv_buf, .. }
            | Self::DataRecvd { recv_buf, .. } => Some(recv_buf),
            Self::DataRead { .. }
            | Self::AbortReading { .. }
            | Self::WaitForReset { .. }
            | Self::ResetRecvd { .. } => None,
        }
    }

    fn flow_control_consume_data(&mut self, consumed: u64, fin: bool) -> Res<()> {
        let (fc, session_fc, final_size_reached, retire_data) = match self {
            Self::Recv { fc, session_fc, .. } => (fc, session_fc, false, false),
            Self::WaitForReset { fc, session_fc, .. } => (fc, session_fc, false, true),
            Self::SizeKnown { fc, session_fc, .. } | Self::DataRecvd { fc, session_fc, .. } => {
                (fc, session_fc, true, false)
            }
            Self::AbortReading {
                fc,
                session_fc,
                final_size_reached,
                ..
            } => {
                let old_final_size_reached = *final_size_reached;
                *final_size_reached |= fin;
                (fc, session_fc, old_final_size_reached, true)
            }
            Self::DataRead { .. } | Self::ResetRecvd { .. } => {
                return Ok(());
            }
        };

        // Check final size:
        let final_size_ok = match (fin, final_size_reached) {
            (true, true) => consumed == fc.consumed(),
            (false, true) => consumed <= fc.consumed(),
            (true, false) => consumed >= fc.consumed(),
            (false, false) => true,
        };

        if !final_size_ok {
            return Err(Error::FinalSize);
        }

        let new_bytes_consumed = fc.set_consumed(consumed)?;
        session_fc.borrow_mut().consume(new_bytes_consumed)?;
        if retire_data {
            // Let's also retire this data since the stream has been aborted
            RecvStream::flow_control_retire_data(fc.consumed() - fc.retired(), fc, session_fc);
        }
        Ok(())
    }
}

// See https://www.w3.org/TR/webtransport/#receive-stream-stats
#[derive(Debug, Clone, Copy)]
pub struct Stats {
    // An indicator of progress on how many of the server application’s bytes
    // intended for this stream have been received so far.
    // Only sequential bytes up to, but not including, the first missing byte,
    // are counted. This number can only increase.
    pub bytes_received: u64,
    // The total number of bytes the application has successfully read from this
    // stream. This number can only increase, and is always less than or equal
    // to bytes_received.
    pub bytes_read: u64,
}

impl Stats {
    #[must_use]
    pub const fn new(bytes_received: u64, bytes_read: u64) -> Self {
        Self {
            bytes_received,
            bytes_read,
        }
    }

    #[must_use]
    pub const fn bytes_received(&self) -> u64 {
        self.bytes_received
    }

    #[must_use]
    pub const fn bytes_read(&self) -> u64 {
        self.bytes_read
    }
}

/// Implement a QUIC receive stream.
#[derive(Debug)]
pub struct RecvStream {
    stream_id: StreamId,
    state: RecvStreamState,
    conn_events: ConnectionEvents,
    keep_alive: Option<Rc<()>>,
}

impl RecvStream {
    pub fn new(
        stream_id: StreamId,
        max_stream_data: u64,
        session_fc: Rc<RefCell<ReceiverFlowControl<()>>>,
        conn_events: ConnectionEvents,
    ) -> Self {
        Self {
            stream_id,
            state: RecvStreamState::new(max_stream_data, stream_id, session_fc),
            conn_events,
            keep_alive: None,
        }
    }

    fn set_state(&mut self, new_state: RecvStreamState) {
        debug_assert_ne!(
            mem::discriminant(&self.state),
            mem::discriminant(&new_state)
        );
        qtrace!(
            "RecvStream {} state {} -> {new_state}",
            self.stream_id.as_u64(),
            self.state
        );

        match new_state {
            // Receiving all data, or receiving or requesting RESET_STREAM
            // is cause to stop keepalives.
            RecvStreamState::DataRecvd { .. }
            | RecvStreamState::AbortReading { .. }
            | RecvStreamState::ResetRecvd { .. } => {
                self.keep_alive = None;
            }
            // Once all the data is read, generate an event.
            RecvStreamState::DataRead { .. } => {
                self.conn_events.recv_stream_complete(self.stream_id);
            }
            _ => {}
        }

        self.state = new_state;
    }

    #[must_use]
    pub const fn stats(&self) -> Stats {
        match &self.state {
            RecvStreamState::Recv { recv_buf, .. }
            | RecvStreamState::SizeKnown { recv_buf, .. }
            | RecvStreamState::DataRecvd { recv_buf, .. } => {
                let received = recv_buf.received();
                let read = recv_buf.retired();
                Stats::new(received, read)
            }
            RecvStreamState::AbortReading {
                final_received,
                final_read,
                ..
            }
            | RecvStreamState::WaitForReset {
                final_received,
                final_read,
                ..
            }
            | RecvStreamState::DataRead {
                final_received,
                final_read,
            }
            | RecvStreamState::ResetRecvd {
                final_received,
                final_read,
            } => {
                let received = *final_received;
                let read = *final_read;
                Stats::new(received, read)
            }
        }
    }

    /// # Errors
    /// When the incoming data violates flow control limits.
    /// # Panics
    /// Only when `u64` values are so big that they can't fit in a `usize`, which
    /// only happens on a 32-bit machine that has far too much unread data.
    pub fn inbound_stream_frame(&mut self, fin: bool, offset: u64, data: &[u8]) -> Res<()> {
        // We should post a DataReadable event only once when we change from no-data-ready to
        // data-ready. Therefore remember the state before processing a new frame.
        let already_data_ready = self.data_ready();
        let new_end = offset + u64::try_from(data.len())?;

        self.state.flow_control_consume_data(new_end, fin)?;

        match &mut self.state {
            RecvStreamState::Recv {
                recv_buf,
                fc,
                session_fc,
            } => {
                recv_buf.inbound_frame(offset, data);
                if fin {
                    let all_recv =
                        fc.consumed() == recv_buf.retired() + recv_buf.bytes_ready() as u64;
                    let buf = mem::replace(recv_buf, RxStreamOrderer::new());
                    let fc_copy = mem::take(fc);
                    let session_fc_copy = mem::take(session_fc);
                    if all_recv {
                        self.set_state(RecvStreamState::DataRecvd {
                            fc: fc_copy,
                            session_fc: session_fc_copy,
                            recv_buf: buf,
                        });
                    } else {
                        self.set_state(RecvStreamState::SizeKnown {
                            fc: fc_copy,
                            session_fc: session_fc_copy,
                            recv_buf: buf,
                        });
                    }
                }
            }
            RecvStreamState::SizeKnown {
                recv_buf,
                fc,
                session_fc,
            } => {
                recv_buf.inbound_frame(offset, data);
                if fc.consumed() == recv_buf.retired() + recv_buf.bytes_ready() as u64 {
                    let buf = mem::replace(recv_buf, RxStreamOrderer::new());
                    let fc_copy = mem::take(fc);
                    let session_fc_copy = mem::take(session_fc);
                    self.set_state(RecvStreamState::DataRecvd {
                        fc: fc_copy,
                        session_fc: session_fc_copy,
                        recv_buf: buf,
                    });
                }
            }
            RecvStreamState::DataRecvd { .. }
            | RecvStreamState::DataRead { .. }
            | RecvStreamState::AbortReading { .. }
            | RecvStreamState::WaitForReset { .. }
            | RecvStreamState::ResetRecvd { .. } => {
                qtrace!("data received when we are in state {}", self.state);
            }
        }

        if !already_data_ready && (self.data_ready() || self.needs_to_inform_app_about_fin()) {
            self.conn_events.recv_stream_readable(self.stream_id);
        }

        Ok(())
    }

    /// # Errors
    /// When the reset occurs at an invalid point.
    ///
    /// # Returns
    /// `true` when the stream transitions to `ResetRecvd` (ended).
    /// `false` if the stream is already in a terminal state and the reset is a no-op.
    pub fn reset(&mut self, application_error_code: AppError, final_size: u64) -> Res<bool> {
        self.state.flow_control_consume_data(final_size, true)?;
        match &mut self.state {
            RecvStreamState::Recv {
                fc,
                session_fc,
                recv_buf,
            }
            | RecvStreamState::SizeKnown {
                fc,
                session_fc,
                recv_buf,
            } => {
                // make flow control consumes new data that not really exist.
                Self::flow_control_retire_data(final_size - fc.retired(), fc, session_fc);
                self.conn_events
                    .recv_stream_reset(self.stream_id, application_error_code);
                let received = recv_buf.received();
                let read = recv_buf.retired();
                self.set_state(RecvStreamState::ResetRecvd {
                    final_received: received,
                    final_read: read,
                });
                Ok(true)
            }
            RecvStreamState::AbortReading {
                fc,
                session_fc,
                final_received,
                final_read,
                ..
            }
            | RecvStreamState::WaitForReset {
                fc,
                session_fc,
                final_received,
                final_read,
            } => {
                // make flow control consumes new data that not really exist.
                Self::flow_control_retire_data(final_size - fc.retired(), fc, session_fc);
                self.conn_events
                    .recv_stream_reset(self.stream_id, application_error_code);
                let received = *final_received;
                let read = *final_read;
                self.set_state(RecvStreamState::ResetRecvd {
                    final_received: received,
                    final_read: read,
                });
                Ok(true)
            }
            _ => Ok(false), // Ignore reset if in DataRecvd, DataRead, or ResetRecvd
        }
    }

    fn flow_control_retire_data(
        new_read: u64,
        fc: &mut ReceiverFlowControl<StreamId>,
        session_fc: &Rc<RefCell<ReceiverFlowControl<()>>>,
    ) {
        if new_read > 0 {
            fc.add_retired(new_read);
            session_fc.borrow_mut().add_retired(new_read);
        }
    }

    /// Send a flow control update.
    /// This is used when a peer declares that they are blocked.
    /// This sends `MAX_STREAM_DATA` if there is any increase possible.
    pub const fn send_flowc_update(&mut self) {
        if let RecvStreamState::Recv { fc, .. } = &mut self.state {
            fc.send_flowc_update();
        }
    }

    pub const fn set_stream_max_data(&mut self, max_data: u64) {
        if let RecvStreamState::Recv { fc, .. } = &mut self.state {
            fc.set_max_active(max_data);
        }
    }

    #[must_use]
    pub const fn is_ended(&self) -> bool {
        matches!(
            self.state,
            RecvStreamState::ResetRecvd { .. } | RecvStreamState::DataRead { .. }
        )
    }

    // App got all data but did not get the fin signal.
    const fn needs_to_inform_app_about_fin(&self) -> bool {
        matches!(self.state, RecvStreamState::DataRecvd { .. })
    }

    fn data_ready(&self) -> bool {
        self.state
            .recv_buf()
            .is_some_and(RxStreamOrderer::data_ready)
    }

    /// # Errors
    /// `NoMoreData` if data and fin bit were previously read by the application.
    pub fn read(&mut self, buf: &mut [u8]) -> Res<(usize, bool)> {
        let data_recvd_state = matches!(self.state, RecvStreamState::DataRecvd { .. });
        match &mut self.state {
            RecvStreamState::Recv {
                recv_buf,
                fc,
                session_fc,
            }
            | RecvStreamState::SizeKnown {
                recv_buf,
                fc,
                session_fc,
                ..
            }
            | RecvStreamState::DataRecvd {
                recv_buf,
                fc,
                session_fc,
            } => {
                let bytes_read = recv_buf.read(buf);
                Self::flow_control_retire_data(u64::try_from(bytes_read)?, fc, session_fc);
                let fin_read = if data_recvd_state {
                    if recv_buf.buffered() == 0 {
                        let received = recv_buf.received();
                        let read = recv_buf.retired();
                        self.set_state(RecvStreamState::DataRead {
                            final_received: received,
                            final_read: read,
                        });
                        true
                    } else {
                        false
                    }
                } else {
                    false
                };
                Ok((bytes_read, fin_read))
            }
            RecvStreamState::DataRead { .. }
            | RecvStreamState::AbortReading { .. }
            | RecvStreamState::WaitForReset { .. }
            | RecvStreamState::ResetRecvd { .. } => Err(Error::NoMoreData),
        }
    }

    /// # Returns
    /// `true` if the stream transitions to `DataRead` (ended).
    /// `false` if the stream transitions to `AbortReading` or was already
    /// in a terminal or aborting state.
    #[must_use]
    pub fn stop_sending(&mut self, err: AppError) -> bool {
        qtrace!("stop_sending called when in state {}", self.state);
        match &mut self.state {
            RecvStreamState::Recv {
                fc,
                session_fc,
                recv_buf,
            }
            | RecvStreamState::SizeKnown {
                fc,
                session_fc,
                recv_buf,
            } => {
                // Retire data
                Self::flow_control_retire_data(fc.consumed() - fc.retired(), fc, session_fc);
                let fc_copy = mem::take(fc);
                let session_fc_copy = mem::take(session_fc);
                let received = recv_buf.received();
                let read = recv_buf.retired();
                self.set_state(RecvStreamState::AbortReading {
                    fc: fc_copy,
                    session_fc: session_fc_copy,
                    final_size_reached: matches!(self.state, RecvStreamState::SizeKnown { .. }),
                    frame_needed: true,
                    err,
                    final_received: received,
                    final_read: read,
                });
                false
            }
            RecvStreamState::DataRecvd {
                fc,
                session_fc,
                recv_buf,
            } => {
                Self::flow_control_retire_data(fc.consumed() - fc.retired(), fc, session_fc);
                let received = recv_buf.received();
                let read = recv_buf.retired();
                self.set_state(RecvStreamState::DataRead {
                    final_received: received,
                    final_read: read,
                });
                true
            }
            RecvStreamState::DataRead { .. }
            | RecvStreamState::AbortReading { .. }
            | RecvStreamState::WaitForReset { .. }
            | RecvStreamState::ResetRecvd { .. } => false,
        }
    }

    /// Maybe write a `MAX_STREAM_DATA` frame.
    pub fn write_frame<B: Buffer>(
        &mut self,
        builder: &mut packet::Builder<B>,
        tokens: &mut recovery::Tokens,
        stats: &mut FrameStats,
        now: Instant,
        rtt: Duration,
    ) {
        match &mut self.state {
            // Maybe send MAX_STREAM_DATA
            RecvStreamState::Recv { fc, .. } => fc.write_frames(builder, tokens, stats, now, rtt),
            // Maybe send STOP_SENDING
            RecvStreamState::AbortReading {
                frame_needed, err, ..
            } if *frame_needed
                && builder.write_varint_frame(&[
                    FrameType::StopSending.into(),
                    self.stream_id.as_u64(),
                    *err,
                ]) =>
            {
                tokens.push(recovery::Token::Stream(StreamRecoveryToken::StopSending {
                    stream_id: self.stream_id,
                }));
                stats.stop_sending += 1;
                *frame_needed = false;
            }
            _ => {}
        }
    }

    pub const fn max_stream_data_lost(&mut self, maximum_data: u64) {
        if let RecvStreamState::Recv { fc, .. } = &mut self.state {
            fc.frame_lost(maximum_data);
        }
    }

    pub const fn stop_sending_lost(&mut self) {
        if let RecvStreamState::AbortReading { frame_needed, .. } = &mut self.state {
            *frame_needed = true;
        }
    }

    /// # Returns
    /// `true` if the stream transitions to `ResetRecvd` (ended) because
    /// the final size was already known.
    #[must_use]
    pub fn stop_sending_acked(&mut self) -> bool {
        if let RecvStreamState::AbortReading {
            fc,
            session_fc,
            final_size_reached,
            final_received,
            final_read,
            ..
        } = &mut self.state
        {
            let received = *final_received;
            let read = *final_read;
            if *final_size_reached {
                // We already know the final_size of the stream therefore we
                // do not need to wait for RESET.
                self.set_state(RecvStreamState::ResetRecvd {
                    final_received: received,
                    final_read: read,
                });
                return true;
            }
            let fc_copy = mem::take(fc);
            let session_fc_copy = mem::take(session_fc);
            self.set_state(RecvStreamState::WaitForReset {
                fc: fc_copy,
                session_fc: session_fc_copy,
                final_received: received,
                final_read: read,
            });
        }
        false
    }

    #[cfg(test)]
    #[must_use]
    pub const fn has_frames_to_write(&self) -> bool {
        if let RecvStreamState::Recv { fc, .. } = &self.state {
            fc.frame_needed()
        } else {
            false
        }
    }

    #[cfg(test)]
    #[must_use]
    pub const fn fc(&self) -> Option<&ReceiverFlowControl<StreamId>> {
        match &self.state {
            RecvStreamState::Recv { fc, .. }
            | RecvStreamState::SizeKnown { fc, .. }
            | RecvStreamState::DataRecvd { fc, .. }
            | RecvStreamState::AbortReading { fc, .. }
            | RecvStreamState::WaitForReset { fc, .. } => Some(fc),
            _ => None,
        }
    }
}

#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests {
    use std::{cell::RefCell, fmt::Debug, ops::Range, rc::Rc, time::Duration};

    use neqo_common::{Encoder, qtrace};
    use test_fixture::now;

    use super::RecvStream;
    use crate::{
        ConnectionEvents, Error, INITIAL_LOCAL_MAX_STREAM_DATA, StreamId,
        fc::{ReceiverFlowControl, WINDOW_UPDATE_FRACTION},
        packet, recovery,
        recv_stream::RxStreamOrderer,
        stats::FrameStats,
    };

    const SESSION_WINDOW: usize = 1024;

    fn recv_ranges(ranges: &[Range<u64>], available: usize) {
        const ZEROES: &[u8] = &[0; 100];
        qtrace!("recv_ranges {ranges:?}");

        let mut s = RxStreamOrderer::default();
        for r in ranges {
            let data = &ZEROES[..usize::try_from(r.end - r.start).unwrap()];
            s.inbound_frame(r.start, data);
        }

        let mut buf = [0xff; 100];
        let mut total_recvd = 0;
        loop {
            let recvd = s.read(&mut buf[..]);
            qtrace!("recv_ranges read {recvd}");
            total_recvd += recvd;
            if recvd == 0 {
                assert_eq!(total_recvd, available);
                break;
            }
        }
    }

    /// A buffer of exactly 4096 bytes has reached the extension limit and must not be extended.
    #[test]
    fn inbound_frame_no_extend_at_4096() {
        let mut s = RxStreamOrderer::default();
        // Fill to the extend threshold.
        s.inbound_frame(0, &[0u8; 4096]);
        assert_eq!(s.data_ranges[&0].len(), 4096);
        // The next byte must not be merged; the threshold has been reached.
        s.inbound_frame(4096, &[1u8]);
        assert_eq!(
            s.data_ranges.len(),
            2,
            "a 4096-byte buffer must not be extended further"
        );
    }

    /// A buffer of 4095 bytes IS extended when the next frame is contiguous.
    #[test]
    fn inbound_frame_extends_below_4096() {
        let mut s = RxStreamOrderer::default();
        s.inbound_frame(0, &[0u8; 4095]);
        s.inbound_frame(4095, &[1u8]);
        assert_eq!(s.data_ranges.len(), 1);
        assert_eq!(s.data_ranges[&0].len(), 4096);
    }

    /// Reading exactly `available` bytes frees the range so the next read can proceed.
    #[test]
    fn read_exact_available_removes_range() {
        let mut s = RxStreamOrderer::default();
        s.inbound_frame(0, &[1u8; 5]);
        s.inbound_frame(5, &[2u8; 5]);

        let mut buf = [0u8; 5];
        assert_eq!(s.read(&mut buf), 5);
        assert_eq!(buf, [1u8; 5]);
        assert_eq!(s.read(&mut buf), 5);
        assert_eq!(buf, [2u8; 5]);
    }

    #[test]
    #[expect(
        clippy::single_range_in_vec_init,
        reason = "Because that lint makes no sense here."
    )]
    fn recv_noncontiguous() {
        // Non-contiguous with the start, no data available.
        recv_ranges(&[10..20], 0);
    }

    /// Overlaps with the start of a 10..20 range of bytes.
    #[test]
    fn recv_overlap_start() {
        // Overlap the start, with a larger new value.
        // More overlap than not.
        recv_ranges(&[10..20, 4..18, 0..4], 20);
        // Overlap the start, with a larger new value.
        // Less overlap than not.
        recv_ranges(&[10..20, 2..15, 0..2], 20);
        // Overlap the start, with a smaller new value.
        // More overlap than not.
        recv_ranges(&[10..20, 8..14, 0..8], 20);
        // Overlap the start, with a smaller new value.
        // Less overlap than not.
        recv_ranges(&[10..20, 6..13, 0..6], 20);

        // Again with some of the first range split in two.
        recv_ranges(&[10..11, 11..20, 4..18, 0..4], 20);
        recv_ranges(&[10..11, 11..20, 2..15, 0..2], 20);
        recv_ranges(&[10..11, 11..20, 8..14, 0..8], 20);
        recv_ranges(&[10..11, 11..20, 6..13, 0..6], 20);

        // Again with a gap in the first range.
        recv_ranges(&[10..11, 12..20, 4..18, 0..4], 20);
        recv_ranges(&[10..11, 12..20, 2..15, 0..2], 20);
        recv_ranges(&[10..11, 12..20, 8..14, 0..8], 20);
        recv_ranges(&[10..11, 12..20, 6..13, 0..6], 20);
    }

    /// Overlaps with the end of a 10..20 range of bytes.
    #[test]
    fn recv_overlap_end() {
        // Overlap the end, with a larger new value.
        // More overlap than not.
        recv_ranges(&[10..20, 12..25, 0..10], 25);
        // Overlap the end, with a larger new value.
        // Less overlap than not.
        recv_ranges(&[10..20, 17..33, 0..10], 33);
        // Overlap the end, with a smaller new value.
        // More overlap than not.
        recv_ranges(&[10..20, 15..21, 0..10], 21);
        // Overlap the end, with a smaller new value.
        // Less overlap than not.
        recv_ranges(&[10..20, 17..25, 0..10], 25);

        // Again with some of the first range split in two.
        recv_ranges(&[10..19, 19..20, 12..25, 0..10], 25);
        recv_ranges(&[10..19, 19..20, 17..33, 0..10], 33);
        recv_ranges(&[10..19, 19..20, 15..21, 0..10], 21);
        recv_ranges(&[10..19, 19..20, 17..25, 0..10], 25);

        // Again with a gap in the first range.
        recv_ranges(&[10..18, 19..20, 12..25, 0..10], 25);
        recv_ranges(&[10..18, 19..20, 17..33, 0..10], 33);
        recv_ranges(&[10..18, 19..20, 15..21, 0..10], 21);
        recv_ranges(&[10..18, 19..20, 17..25, 0..10], 25);
    }

    /// Complete overlaps with the start of a 10..20 range of bytes.
    #[test]
    fn recv_overlap_complete() {
        // Complete overlap, more at the end.
        recv_ranges(&[10..20, 9..23, 0..9], 23);
        // Complete overlap, more at the start.
        recv_ranges(&[10..20, 3..23, 0..3], 23);
        // Complete overlap, to end.
        recv_ranges(&[10..20, 5..20, 0..5], 20);
        // Complete overlap, from start.
        recv_ranges(&[10..20, 10..27, 0..10], 27);
        // Complete overlap, from 0 and more.
        recv_ranges(&[10..20, 0..23], 23);

        // Again with the first range split in two.
        recv_ranges(&[10..14, 14..20, 9..23, 0..9], 23);
        recv_ranges(&[10..14, 14..20, 3..23, 0..3], 23);
        recv_ranges(&[10..14, 14..20, 5..20, 0..5], 20);
        recv_ranges(&[10..14, 14..20, 10..27, 0..10], 27);
        recv_ranges(&[10..14, 14..20, 0..23], 23);

        // Again with the a gap in the first range.
        recv_ranges(&[10..13, 14..20, 9..23, 0..9], 23);
        recv_ranges(&[10..13, 14..20, 3..23, 0..3], 23);
        recv_ranges(&[10..13, 14..20, 5..20, 0..5], 20);
        recv_ranges(&[10..13, 14..20, 10..27, 0..10], 27);
        recv_ranges(&[10..13, 14..20, 0..23], 23);
    }

    /// An overlap with no new bytes.
    #[test]
    fn recv_overlap_duplicate() {
        recv_ranges(&[10..20, 11..12, 0..10], 20);
        recv_ranges(&[10..20, 10..15, 0..10], 20);
        recv_ranges(&[10..20, 14..20, 0..10], 20);
        // Now with the first range split.
        recv_ranges(&[10..14, 14..20, 10..15, 0..10], 20);
        recv_ranges(&[10..15, 16..20, 21..25, 10..25, 0..10], 25);
    }

    /// Reading exactly one chunk works, when the next chunk starts immediately.
    #[test]
    fn stop_reading_at_chunk() {
        const CHUNK_SIZE: usize = 10;
        const EXTRA_SIZE: usize = 3;
        let mut s = RxStreamOrderer::new();

        // Add three chunks.
        s.inbound_frame(0, &[0; CHUNK_SIZE]);
        let offset = u64::try_from(CHUNK_SIZE).unwrap();
        s.inbound_frame(offset, &[0; EXTRA_SIZE]);
        let offset = u64::try_from(CHUNK_SIZE + EXTRA_SIZE).unwrap();
        s.inbound_frame(offset, &[0; EXTRA_SIZE]);

        // Read, providing only enough space for the first.
        let mut buf = [0; 100];
        let count = s.read(&mut buf[..CHUNK_SIZE]);
        assert_eq!(count, CHUNK_SIZE);
        let count = s.read(&mut buf[..]);
        assert_eq!(count, EXTRA_SIZE * 2);
    }

    #[test]
    fn recv_overlap_while_reading() {
        let mut s = RxStreamOrderer::new();

        // Add a chunk
        s.inbound_frame(0, &[0; 150]);
        assert_eq!(s.data_ranges[&0].len(), 150);
        // Read, providing only enough space for the first 100.
        let mut buf = [0; 100];
        let count = s.read(&mut buf[..]);
        assert_eq!(count, 100);
        assert_eq!(s.retired, 100);

        // Add a second frame that overlaps.
        // This shouldn't truncate the first frame, as we're already
        // Reading from it.
        s.inbound_frame(120, &[0; 60]);
        assert_eq!(s.data_ranges[&0].len(), 180);
        // Read second part of first frame and all of the second frame
        let count = s.read(&mut buf[..]);
        assert_eq!(count, 80);
    }

    /// Reading exactly one chunk works, when there is a gap.
    #[test]
    fn stop_reading_at_gap() {
        const CHUNK_SIZE: usize = 10;
        const EXTRA_SIZE: usize = 3;
        let mut s = RxStreamOrderer::new();

        // Add three chunks.
        s.inbound_frame(0, &[0; CHUNK_SIZE]);
        let offset = u64::try_from(CHUNK_SIZE + EXTRA_SIZE).unwrap();
        s.inbound_frame(offset, &[0; EXTRA_SIZE]);

        // Read, providing only enough space for the first chunk.
        let mut buf = [0; 100];
        let count = s.read(&mut buf[..CHUNK_SIZE]);
        assert_eq!(count, CHUNK_SIZE);

        // Now fill the gap and ensure that everything can be read.
        let offset = u64::try_from(CHUNK_SIZE).unwrap();
        s.inbound_frame(offset, &[0; EXTRA_SIZE]);
        let count = s.read(&mut buf[..]);
        assert_eq!(count, EXTRA_SIZE * 2);
    }

    /// Reading exactly one chunk works, when there is a gap.
    #[test]
    fn stop_reading_in_chunk() {
        const CHUNK_SIZE: usize = 10;
        const EXTRA_SIZE: usize = 3;
        let mut s = RxStreamOrderer::new();

        // Add two chunks.
        s.inbound_frame(0, &[0; CHUNK_SIZE]);
        let offset = u64::try_from(CHUNK_SIZE).unwrap();
        s.inbound_frame(offset, &[0; EXTRA_SIZE]);

        // Read, providing only enough space for some of the first chunk.
        let mut buf = [0; 100];
        let count = s.read(&mut buf[..CHUNK_SIZE - EXTRA_SIZE]);
        assert_eq!(count, CHUNK_SIZE - EXTRA_SIZE);

        let count = s.read(&mut buf[..]);
        assert_eq!(count, EXTRA_SIZE * 2);
    }

    /// Read one byte at a time.
    #[test]
    fn read_byte_at_a_time() {
        const CHUNK_SIZE: usize = 10;
        const EXTRA_SIZE: usize = 3;
        let mut s = RxStreamOrderer::new();

        // Add two chunks.
        s.inbound_frame(0, &[0; CHUNK_SIZE]);
        let offset = u64::try_from(CHUNK_SIZE).unwrap();
        s.inbound_frame(offset, &[0; EXTRA_SIZE]);

        let mut buf = [0; 1];
        for _ in 0..CHUNK_SIZE + EXTRA_SIZE {
            let count = s.read(&mut buf[..]);
            assert_eq!(count, 1);
        }
        assert_eq!(0, s.read(&mut buf[..]));
    }

    fn check_stats(stream: &RecvStream, expected_received: u64, expected_read: u64) {
        let stream_stats = stream.stats();
        assert_eq!(expected_received, stream_stats.bytes_received());
        assert_eq!(expected_read, stream_stats.bytes_read());
    }

    #[test]
    fn stream_rx() {
        let conn_events = ConnectionEvents::default();

        let mut s = RecvStream::new(
            StreamId::from(567),
            1024,
            Rc::new(RefCell::new(ReceiverFlowControl::new((), 1024 * 1024))),
            conn_events,
        );

        // test receiving a contig frame and reading it works
        s.inbound_stream_frame(false, 0, &[1; 10]).unwrap();
        assert!(s.data_ready());
        check_stats(&s, 10, 0);

        let mut buf = vec![0u8; 100];
        assert_eq!(s.read(&mut buf).unwrap(), (10, false));
        assert_eq!(s.state.recv_buf().unwrap().retired(), 10);
        assert_eq!(s.state.recv_buf().unwrap().buffered(), 0);

        check_stats(&s, 10, 10);

        // test receiving a noncontig frame
        s.inbound_stream_frame(false, 12, &[2; 12]).unwrap();
        assert!(!s.data_ready());
        assert_eq!(s.read(&mut buf).unwrap(), (0, false));
        assert_eq!(s.state.recv_buf().unwrap().retired(), 10);
        assert_eq!(s.state.recv_buf().unwrap().buffered(), 12);

        check_stats(&s, 22, 10);

        // another frame that overlaps the first
        s.inbound_stream_frame(false, 14, &[3; 8]).unwrap();
        assert!(!s.data_ready());
        assert_eq!(s.state.recv_buf().unwrap().retired(), 10);
        assert_eq!(s.state.recv_buf().unwrap().buffered(), 12);

        check_stats(&s, 22, 10);

        // fill in the gap, but with a FIN
        s.inbound_stream_frame(true, 10, &[4; 6]).unwrap_err();
        assert!(!s.data_ready());
        assert_eq!(s.read(&mut buf).unwrap(), (0, false));
        assert_eq!(s.state.recv_buf().unwrap().retired(), 10);
        assert_eq!(s.state.recv_buf().unwrap().buffered(), 12);

        check_stats(&s, 22, 10);

        // fill in the gap
        s.inbound_stream_frame(false, 10, &[5; 10]).unwrap();
        assert!(s.data_ready());
        assert_eq!(s.state.recv_buf().unwrap().retired(), 10);
        assert_eq!(s.state.recv_buf().unwrap().buffered(), 14);

        check_stats(&s, 24, 10);

        // a legit FIN
        s.inbound_stream_frame(true, 24, &[6; 18]).unwrap();
        assert_eq!(s.state.recv_buf().unwrap().retired(), 10);
        assert_eq!(s.state.recv_buf().unwrap().buffered(), 32);
        assert!(s.data_ready());
        assert_eq!(s.read(&mut buf).unwrap(), (32, true));

        check_stats(&s, 42, 42);

        // Stream now no longer readable (is in DataRead state)
        s.read(&mut buf).unwrap_err();
    }

    fn check_chunks(s: &RxStreamOrderer, expected: &[(u64, usize)]) {
        assert_eq!(s.data_ranges.len(), expected.len());
        for ((start, buf), (expected_start, expected_len)) in s.data_ranges.iter().zip(expected) {
            assert_eq!((*start, buf.len()), (*expected_start, *expected_len));
        }
    }

    // Test deduplication when the new data is at the end.
    #[test]
    fn stream_rx_dedupe_tail() {
        let mut s = RxStreamOrderer::new();

        s.inbound_frame(0, &[1; 6]);
        check_chunks(&s, &[(0, 6)]);

        // New data that overlaps entirely (starting from the head), is ignored.
        s.inbound_frame(0, &[2; 3]);
        check_chunks(&s, &[(0, 6)]);

        // New data that overlaps at the tail has any new data appended.
        s.inbound_frame(2, &[3; 6]);
        check_chunks(&s, &[(0, 8)]);

        // New data that overlaps entirely (up to the tail), is ignored.
        s.inbound_frame(4, &[4; 4]);
        check_chunks(&s, &[(0, 8)]);

        // New data that overlaps, starting from the beginning is appended too.
        s.inbound_frame(0, &[5; 10]);
        check_chunks(&s, &[(0, 10)]);

        // New data that is entirely subsumed is ignored.
        s.inbound_frame(2, &[6; 2]);
        check_chunks(&s, &[(0, 10)]);

        let mut buf = [0; 16];
        assert_eq!(s.read(&mut buf[..]), 10);
        assert_eq!(buf[..10], [1, 1, 1, 1, 1, 1, 3, 3, 5, 5]);
    }

    /// When chunks are added before existing data, they aren't merged.
    #[test]
    fn stream_rx_dedupe_head() {
        let mut s = RxStreamOrderer::new();

        s.inbound_frame(1, &[6; 6]);
        check_chunks(&s, &[(1, 6)]);

        // Insertion before an existing chunk causes truncation of the new chunk.
        s.inbound_frame(0, &[7; 6]);
        check_chunks(&s, &[(0, 1), (1, 6)]);

        // Perfect overlap with existing slices has no effect.
        s.inbound_frame(0, &[8; 7]);
        check_chunks(&s, &[(0, 1), (1, 6)]);

        let mut buf = [0; 16];
        assert_eq!(s.read(&mut buf[..]), 7);
        assert_eq!(buf[..7], [7, 6, 6, 6, 6, 6, 6]);
    }

    #[test]
    fn stream_rx_dedupe_new_tail() {
        let mut s = RxStreamOrderer::new();

        s.inbound_frame(1, &[6; 6]);
        check_chunks(&s, &[(1, 6)]);

        // Insertion before an existing chunk causes truncation of the new chunk.
        s.inbound_frame(0, &[7; 6]);
        check_chunks(&s, &[(0, 1), (1, 6)]);

        // New data at the end causes the tail to be added to the first chunk,
        // replacing later chunks entirely.
        s.inbound_frame(0, &[9; 8]);
        check_chunks(&s, &[(0, 8)]);

        let mut buf = [0; 16];
        assert_eq!(s.read(&mut buf[..]), 8);
        assert_eq!(buf[..8], [7, 9, 9, 9, 9, 9, 9, 9]);
    }

    #[test]
    fn stream_rx_dedupe_replace() {
        let mut s = RxStreamOrderer::new();

        s.inbound_frame(2, &[6; 6]);
        check_chunks(&s, &[(2, 6)]);

        // Insertion before an existing chunk causes truncation of the new chunk.
        s.inbound_frame(1, &[7; 6]);
        check_chunks(&s, &[(1, 1), (2, 6)]);

        // New data at the start and end replaces all the slices.
        s.inbound_frame(0, &[9; 10]);
        check_chunks(&s, &[(0, 10)]);

        let mut buf = [0; 16];
        assert_eq!(s.read(&mut buf[..]), 10);
        assert_eq!(buf[..10], [9; 10]);
    }

    #[test]
    fn trim_retired() {
        let mut s = RxStreamOrderer::new();

        let mut buf = [0; 18];
        s.inbound_frame(0, &[1; 10]);

        // Partially read slices are retained.
        assert_eq!(s.read(&mut buf[..6]), 6);
        check_chunks(&s, &[(0, 10)]);

        // Partially read slices are kept and so are added to.
        s.inbound_frame(3, &buf[..10]);
        check_chunks(&s, &[(0, 13)]);

        // Wholly read pieces are dropped.
        assert_eq!(s.read(&mut buf[..]), 7);
        assert!(s.data_ranges.is_empty());

        // New data that overlaps with retired data is trimmed.
        s.inbound_frame(0, &buf[..]);
        check_chunks(&s, &[(13, 5)]);
    }

    #[test]
    fn stream_flowc_update() {
        let mut s = create_stream(1024 * INITIAL_LOCAL_MAX_STREAM_DATA as u64);
        let mut buf = vec![0u8; INITIAL_LOCAL_MAX_STREAM_DATA + 100]; // Make it overlarge

        assert!(!s.has_frames_to_write());
        let big_buf = vec![0; INITIAL_LOCAL_MAX_STREAM_DATA];
        s.inbound_stream_frame(false, 0, &big_buf).unwrap();
        assert!(!s.has_frames_to_write());
        assert_eq!(
            s.read(&mut buf).unwrap(),
            (INITIAL_LOCAL_MAX_STREAM_DATA, false)
        );
        assert!(!s.data_ready());

        // flow msg generated!
        assert!(s.has_frames_to_write());

        // consume it
        let mut builder =
            packet::Builder::short(Encoder::default(), false, None::<&[u8]>, packet::LIMIT);
        let mut token = recovery::Tokens::new();
        s.write_frame(
            &mut builder,
            &mut token,
            &mut FrameStats::default(),
            now(),
            Duration::from_millis(100),
        );

        // it should be gone
        assert!(!s.has_frames_to_write());
    }

    fn create_stream(session_fc: u64) -> RecvStream {
        let conn_events = ConnectionEvents::default();
        RecvStream::new(
            StreamId::from(67),
            INITIAL_LOCAL_MAX_STREAM_DATA as u64,
            Rc::new(RefCell::new(ReceiverFlowControl::new((), session_fc))),
            conn_events,
        )
    }

    #[test]
    fn stream_max_stream_data() {
        let mut s = create_stream(1024 * INITIAL_LOCAL_MAX_STREAM_DATA as u64);
        assert!(!s.has_frames_to_write());
        let big_buf = vec![0; INITIAL_LOCAL_MAX_STREAM_DATA];
        s.inbound_stream_frame(false, 0, &big_buf).unwrap();
        s.inbound_stream_frame(false, INITIAL_LOCAL_MAX_STREAM_DATA as u64, &[1; 1])
            .unwrap_err();
    }

    #[test]
    fn stream_orderer_bytes_ready() {
        let mut rx_ord = RxStreamOrderer::new();

        rx_ord.inbound_frame(0, &[1; 6]);
        assert_eq!(rx_ord.bytes_ready(), 6);
        assert_eq!(rx_ord.buffered(), 6);
        assert_eq!(rx_ord.retired(), 0);

        // read some so there's an offset into the first frame
        let mut buf = [0u8; 10];
        rx_ord.read(&mut buf[..2]);
        assert_eq!(rx_ord.bytes_ready(), 4);
        assert_eq!(rx_ord.buffered(), 4);
        assert_eq!(rx_ord.retired(), 2);

        // an overlapping frame
        rx_ord.inbound_frame(5, &[2; 6]);
        assert_eq!(rx_ord.bytes_ready(), 9);
        assert_eq!(rx_ord.buffered(), 9);
        assert_eq!(rx_ord.retired(), 2);

        // a noncontig frame
        rx_ord.inbound_frame(20, &[3; 6]);
        assert_eq!(rx_ord.bytes_ready(), 9);
        assert_eq!(rx_ord.buffered(), 15);
        assert_eq!(rx_ord.retired(), 2);

        // an old frame
        rx_ord.inbound_frame(0, &[4; 2]);
        assert_eq!(rx_ord.bytes_ready(), 9);
        assert_eq!(rx_ord.buffered(), 15);
        assert_eq!(rx_ord.retired(), 2);
    }

    #[test]
    fn no_stream_flowc_event_after_exiting_recv() {
        let mut s = create_stream(1024 * INITIAL_LOCAL_MAX_STREAM_DATA as u64);
        let mut buf = vec![0; INITIAL_LOCAL_MAX_STREAM_DATA];
        // Write from buf at first.
        s.inbound_stream_frame(false, 0, &buf).unwrap();
        // Then read into it.
        s.read(&mut buf).unwrap();
        assert!(s.has_frames_to_write());
        s.inbound_stream_frame(true, INITIAL_LOCAL_MAX_STREAM_DATA as u64, &[])
            .unwrap();
        assert!(!s.has_frames_to_write());
    }

    fn create_stream_with_fc(
        session_fc: Rc<RefCell<ReceiverFlowControl<()>>>,
        fc_limit: u64,
    ) -> RecvStream {
        RecvStream::new(
            StreamId::from(567),
            fc_limit,
            session_fc,
            ConnectionEvents::default(),
        )
    }

    fn create_stream_session_flow_control() -> (RecvStream, Rc<RefCell<ReceiverFlowControl<()>>>) {
        static_assertions::const_assert!(INITIAL_LOCAL_MAX_STREAM_DATA > SESSION_WINDOW);
        let session_fc = Rc::new(RefCell::new(ReceiverFlowControl::new(
            (),
            u64::try_from(SESSION_WINDOW).unwrap(),
        )));
        (
            create_stream_with_fc(Rc::clone(&session_fc), INITIAL_LOCAL_MAX_STREAM_DATA as u64),
            session_fc,
        )
    }

    #[test]
    fn session_flow_control() {
        let (mut s, session_fc) = create_stream_session_flow_control();

        s.inbound_stream_frame(false, 0, &[0; SESSION_WINDOW])
            .unwrap();
        assert!(!session_fc.borrow().frame_needed());
        // The buffer is big enough to hold SESSION_WINDOW, this will make sure that we always
        // read everything from he stream.
        let mut buf = [0; 2 * SESSION_WINDOW];
        s.read(&mut buf).unwrap();
        assert!(session_fc.borrow().frame_needed());
        // consume it
        let mut builder =
            packet::Builder::short(Encoder::default(), false, None::<&[u8]>, packet::LIMIT);
        let mut token = recovery::Tokens::new();
        session_fc.borrow_mut().write_frames(
            &mut builder,
            &mut token,
            &mut FrameStats::default(),
            now(),
            Duration::from_millis(100),
        );

        // Switch to SizeKnown state
        s.inbound_stream_frame(true, 2 * u64::try_from(SESSION_WINDOW).unwrap() - 1, &[e='color: green'>0])
            .unwrap();
        assert!(!session_fc.borrow().frame_needed());
        // Receive new data that can be read.
        s.inbound_stream_frame(
            false,
            u64::try_from(SESSION_WINDOW).unwrap(),
            &[0; SESSION_WINDOW / 2 + 1],
        )
        .unwrap();
        assert!(!session_fc.borrow().frame_needed());
        s.read(&mut buf).unwrap();
        assert!(session_fc.borrow().frame_needed());
        // consume it
        let mut builder =
            packet::Builder::short(Encoder::default(), false, None::<&[u8]>, packet::LIMIT);
        let mut token = recovery::Tokens::new();
        session_fc.borrow_mut().write_frames(
            &mut builder,
            &mut token,
            &mut FrameStats::default(),
            now(),
            Duration::from_millis(100),
        );

        // Test DataRecvd state
        let session_fc = Rc::new(RefCell::new(ReceiverFlowControl::new(
            (),
            u64::try_from(SESSION_WINDOW).unwrap(),
        )));
        let mut s = RecvStream::new(
            StreamId::from(567),
            INITIAL_LOCAL_MAX_STREAM_DATA as u64,
            Rc::clone(&session_fc),
            ConnectionEvents::default(),
        );

        s.inbound_stream_frame(true, 0, &[0; SESSION_WINDOW])
            .unwrap();
        assert!(!session_fc.borrow().frame_needed());
        s.read(&mut buf).unwrap();
        assert!(session_fc.borrow().frame_needed());
    }

    #[test]
    fn session_flow_control_reset() {
        let (mut s, session_fc) = create_stream_session_flow_control();

        s.inbound_stream_frame(false, 0, &[0; SESSION_WINDOW / 2])
            .unwrap();
        assert!(!session_fc.borrow().frame_needed());

        s.reset(Error::None.code(), u64::try_from(SESSION_WINDOW).unwrap())
            .unwrap();
        assert!(session_fc.borrow().frame_needed());
    }

    fn check_fc<T: Debug>(fc: &ReceiverFlowControl<T>, consumed: u64, retired: u64) {
        assert_eq!(fc.consumed(), consumed);
        assert_eq!(fc.retired(), retired);
    }

    /// Test consuming the flow control in `RecvStreamState::Recv`
    #[test]
    fn fc_state_recv_1() {
        const SW: u64 = 1024;
        const SW_US: usize = 1024;
        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new((), SW)));
        let mut s = create_stream_with_fc(Rc::clone(&fc), SW * 3 / 4);

        check_fc(&fc.borrow(), 0, 0);
        check_fc(s.fc().unwrap(), 0, 0);

        s.inbound_stream_frame(false, 0, &[0; SW_US / 4]).unwrap();

        check_fc(&fc.borrow(), SW / 4, 0);
        check_fc(s.fc().unwrap(), SW / 4, 0);
    }

    /// Test consuming the flow control in `RecvStreamState::Recv`
    /// with multiple streams
    #[test]
    fn fc_state_recv_2() {
        const SW: u64 = 1024;
        const SW_US: usize = 1024;
        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new((), SW)));
        let mut s1 = create_stream_with_fc(Rc::clone(&fc), SW * 3 / 4);
        let mut s2 = create_stream_with_fc(Rc::clone(&fc), SW * 3 / 4);

        check_fc(&fc.borrow(), 0, 0);
        check_fc(s1.fc().unwrap(), 0, 0);
        check_fc(s2.fc().unwrap(), 0, 0);

        s1.inbound_stream_frame(false, 0, &[0; SW_US / 4]).unwrap();

        check_fc(&fc.borrow(), SW / 4, 0);
        check_fc(s1.fc().unwrap(), SW / 4, 0);
        check_fc(s2.fc().unwrap(), 0, 0);

        s2.inbound_stream_frame(false, 0, &[0; SW_US / 4]).unwrap();

        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s1.fc().unwrap(), SW / 4, 0);
        check_fc(s2.fc().unwrap(), SW / 4, 0);
    }

    /// Test retiring the flow control in `RecvStreamState::Recv`
    /// with multiple streams
    #[test]
    fn fc_state_recv_3() {
        const SW: u64 = 1024;
        const SW_US: usize = 1024;
        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new((), SW)));
        let mut s1 = create_stream_with_fc(Rc::clone(&fc), SW * 3 / 4);
        let mut s2 = create_stream_with_fc(Rc::clone(&fc), SW * 3 / 4);

        check_fc(&fc.borrow(), 0, 0);
        check_fc(s1.fc().unwrap(), 0, 0);
        check_fc(s2.fc().unwrap(), 0, 0);

        s1.inbound_stream_frame(false, 0, &[0; SW_US / 4]).unwrap();
        s2.inbound_stream_frame(false, 0, &[0; SW_US / 4]).unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s1.fc().unwrap(), SW / 4, 0);
        check_fc(s2.fc().unwrap(), SW / 4, 0);

        // Read data
        let mut buf = [1; SW_US];
        assert_eq!(s1.read(&mut buf).unwrap(), (SW_US / 4, false));
        check_fc(&fc.borrow(), SW / 2, SW / 4);
        check_fc(s1.fc().unwrap(), SW / 4, SW / 4);
        check_fc(s2.fc().unwrap(), SW / 4, 0);

        assert_eq!(s2.read(&mut buf).unwrap(), (SW_US / 4, false));
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s1.fc().unwrap(), SW / 4, SW / 4);
        check_fc(s2.fc().unwrap(), SW / 4, SW / 4);

        // Read when there is no more date to be read will not change fc.
        assert_eq!(s1.read(&mut buf).unwrap(), (0, false));
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s1.fc().unwrap(), SW / 4, SW / 4);
        check_fc(s2.fc().unwrap(), SW / 4, SW / 4);

        // Receiving more data on a stream.
        s1.inbound_stream_frame(false, SW / 4, &[0; SW_US / 4])
            .unwrap();
        check_fc(&fc.borrow(), SW * 3 / 4, SW / 2);
        check_fc(s1.fc().unwrap(), SW / 2, SW / 4);
        check_fc(s2.fc().unwrap(), SW / 4, SW / 4);

        // Read data
        assert_eq!(s1.read(&mut buf).unwrap(), (SW_US / 4, false));
        check_fc(&fc.borrow(), SW * 3 / 4, SW * 3 / 4);
        check_fc(s1.fc().unwrap(), SW / 2, SW / 2);
        check_fc(s2.fc().unwrap(), SW / 4, SW / 4);
    }

    /// Test consuming the flow control in `RecvStreamState::Recv` - duplicate data
    #[test]
    fn fc_state_recv_4() {
        const SW: u64 = 1024;
        const SW_US: usize = 1024;
        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new((), SW)));
        let mut s = create_stream_with_fc(Rc::clone(&fc), SW * 3 / 4);

        check_fc(&fc.borrow(), 0, 0);
        check_fc(s.fc().unwrap(), 0, 0);

        s.inbound_stream_frame(false, 0, &[0; SW_US / 4]).unwrap();

        check_fc(&fc.borrow(), SW / 4, 0);
        check_fc(s.fc().unwrap(), SW / 4, 0);

        // Receiving duplicate frames (already consumed data) will not cause an error or
        // change fc.
        s.inbound_stream_frame(false, 0, &[0; SW_US / 8]).unwrap();
        check_fc(&fc.borrow(), SW / 4, 0);
        check_fc(s.fc().unwrap(), SW / 4, 0);
    }

    /// Test consuming the flow control in `RecvStreamState::Recv` - filling a gap in the
    /// data stream.
    #[test]
    fn fc_state_recv_5() {
        const SW: u64 = 1024;
        const SW_US: usize = 1024;
        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new((), SW)));
        let mut s = create_stream_with_fc(Rc::clone(&fc), SW * 3 / 4);

        // Receive out of order data.
        s.inbound_stream_frame(false, SW / 8, &[0; SW_US / 8])
            .unwrap();
        check_fc(&fc.borrow(), SW / 4, 0);
        check_fc(s.fc().unwrap(), SW / 4, 0);

        // Filling in the gap will not change fc.
        s.inbound_stream_frame(false, 0, &[0; SW_US / 8]).unwrap();
        check_fc(&fc.borrow(), SW / 4, 0);
        check_fc(s.fc().unwrap(), SW / 4, 0);
    }

    /// Test consuming the flow control in `RecvStreamState::Recv` - receiving frame past
    /// the flow control will cause an error.
    #[test]
    fn fc_state_recv_6() {
        const SW: u64 = 1024;
        const SW_US: usize = 1024;
        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new((), SW)));
        let mut s = create_stream_with_fc(Rc::clone(&fc), SW * 3 / 4);

        // Receiving frame past the flow control will cause an error.
        assert_eq!(
            s.inbound_stream_frame(false, 0, &[0; SW_US * 3 / 4 + 1]),
            Err(Error::FlowControl)
        );
    }

    /// Test that the flow controls will send updates.
    #[expect(
        clippy::too_many_lines,
        clippy::cast_possible_truncation,
        reason = "This is test code."
    )]
    #[test]
    fn fc_state_recv_7() {
        const CONNECTION_WINDOW: u64 = 1024;
        const CONNECTION_WINDOW_US: usize = CONNECTION_WINDOW as usize;

        const STREAM_WINDOW: u64 = CONNECTION_WINDOW / 2;
        const STREAM_WINDOW_US: usize = STREAM_WINDOW as usize;

        const WINDOW_UPDATE_FRACTION_US: usize = WINDOW_UPDATE_FRACTION as usize;

        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new(
            (),
            CONNECTION_WINDOW,
        )));
        let mut s = create_stream_with_fc(Rc::clone(&fc), STREAM_WINDOW);

        check_fc(&fc.borrow(), 0, 0);
        check_fc(s.fc().unwrap(), 0, 0);

        // Receive data up to but not over the fc update trigger point.
        s.inbound_stream_frame(false, 0, &[0; STREAM_WINDOW_US / WINDOW_UPDATE_FRACTION_US])
            .unwrap();
        let mut buf = [1; CONNECTION_WINDOW_US];
        assert_eq!(
            s.read(&mut buf).unwrap(),
            (STREAM_WINDOW_US / WINDOW_UPDATE_FRACTION_US, false)
        );
        check_fc(
            &fc.borrow(),
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION,
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION,
        );
        check_fc(
            s.fc().unwrap(),
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION,
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION,
        );

        // Still no fc update needed.
        assert!(!fc.borrow().frame_needed());
        assert!(!s.fc().unwrap().frame_needed());

        // Receive one more byte that will cause a fc update after it is read.
        s.inbound_stream_frame(false, STREAM_WINDOW / WINDOW_UPDATE_FRACTION, &[0])
            .unwrap();
        check_fc(
            &fc.borrow(),
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION + 1,
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION,
        );
        check_fc(
            s.fc().unwrap(),
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION + 1,
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION,
        );
        // Only consuming data does not cause a fc update to be sent.
        assert!(!fc.borrow().frame_needed());
        assert!(!s.fc().unwrap().frame_needed());

        assert_eq!(s.read(&mut buf).unwrap(), (1, false));
        check_fc(
            &fc.borrow(),
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION + 1,
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION + 1,
        );
        check_fc(
            s.fc().unwrap(),
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION + 1,
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION + 1,
        );
        // Data are retired and the stream fc will send an update.
        assert!(!fc.borrow().frame_needed());
        assert!(s.fc().unwrap().frame_needed());

        // Receive more data to increase fc further.
        s.inbound_stream_frame(
            false,
            STREAM_WINDOW / WINDOW_UPDATE_FRACTION,
            &[0; STREAM_WINDOW_US / WINDOW_UPDATE_FRACTION_US],
        )
        .unwrap();
        assert_eq!(
            s.read(&mut buf).unwrap(),
            (STREAM_WINDOW_US / WINDOW_UPDATE_FRACTION_US - 1, false)
        );
        check_fc(
            &fc.borrow(),
            STREAM_WINDOW * 2 / WINDOW_UPDATE_FRACTION,
            STREAM_WINDOW * 2 / WINDOW_UPDATE_FRACTION,
        );
        check_fc(
            s.fc().unwrap(),
            STREAM_WINDOW * 2 / WINDOW_UPDATE_FRACTION,
            STREAM_WINDOW * 2 / WINDOW_UPDATE_FRACTION,
        );
        assert!(!fc.borrow().frame_needed());
        assert!(s.fc().unwrap().frame_needed());

        // Write the fc update frame
        let mut builder =
            packet::Builder::short(Encoder::default(), false, None::<&[u8]>, packet::LIMIT);
        let mut token = recovery::Tokens::new();
        let mut stats = FrameStats::default();
        fc.borrow_mut().write_frames(
            &mut builder,
            &mut token,
            &mut stats,
            now(),
            Duration::from_millis(100),
        );
        assert_eq!(stats.max_data, 0);
        s.write_frame(
            &mut builder,
            &mut token,
            &mut stats,
            now(),
            Duration::from_millis(100),
        );
        assert_eq!(stats.max_stream_data, 1);

        // Receive 1 byte that will cause a session fc update after it is read.
        s.inbound_stream_frame(false, STREAM_WINDOW * 2 / WINDOW_UPDATE_FRACTION, &[0])
            .unwrap();
        assert_eq!(s.read(&mut buf).unwrap(), (1, false));
        check_fc(
            &fc.borrow(),
            STREAM_WINDOW * 2 / WINDOW_UPDATE_FRACTION + 1,
            STREAM_WINDOW * 2 / WINDOW_UPDATE_FRACTION + 1,
        );
        check_fc(
            s.fc().unwrap(),
            STREAM_WINDOW * 2 / WINDOW_UPDATE_FRACTION + 1,
            STREAM_WINDOW * 2 / WINDOW_UPDATE_FRACTION + 1,
        );
        assert!(fc.borrow().frame_needed());
        assert!(!s.fc().unwrap().frame_needed());
        fc.borrow_mut().write_frames(
            &mut builder,
            &mut token,
            &mut stats,
            now(),
            Duration::from_millis(100),
        );
        assert_eq!(stats.max_data, 1);
        s.write_frame(
            &mut builder,
            &mut token,
            &mut stats,
            now(),
            Duration::from_millis(100),
        );
        assert_eq!(stats.max_stream_data, 1);
    }

    /// Test flow control in `RecvStreamState::SizeKnown`
    #[test]
    fn fc_state_size_known() {
        const SW: u64 = 1024;
        const SW_US: usize = 1024;
        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new((), SW)));

        let mut s = create_stream_with_fc(Rc::clone(&fc), SW);

        check_fc(&fc.borrow(), 0, 0);
        check_fc(s.fc().unwrap(), 0, 0);

        s.inbound_stream_frame(true, SW / 4, &[0; SW_US / 4])
            .unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        // Receiving duplicate frames (already consumed data) will not cause an error or
        // change fc.
        s.inbound_stream_frame(true, SW / 4, &[0; SW_US / 4])
            .unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        // The stream can still receive duplicate data without a fin bit.
        s.inbound_stream_frame(false, SW / 4, &[0; SW_US / 4])
            .unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        // Receiving frame past the final size of a stream will return an error.
        assert_eq!(
            s.inbound_stream_frame(true, SW / 4, &[0; SW_US / 4 + 1]),
            Err(Error::FinalSize)
        );
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        // Add new data to the gap will not change fc.
        s.inbound_stream_frame(false, SW / 8, &[0; SW_US / 8])
            .unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        // Fill the gap
        s.inbound_stream_frame(false, 0, &[0; SW_US / 8]).unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        // Read all data
        let mut buf = [1; SW_US];
        assert_eq!(s.read(&mut buf).unwrap(), (SW_US / 2, true));
        // the stream does not have fc any more. We can only check the session fc.
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        assert!(s.fc().is_none());
    }

    /// Test flow control in `RecvStreamState::DataRecvd`
    #[test]
    fn fc_state_data_recv() {
        const SW: u64 = 1024;
        const SW_US: usize = 1024;
        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new((), SW)));

        let mut s = create_stream_with_fc(Rc::clone(&fc), SW);

        check_fc(&fc.borrow(), 0, 0);
        check_fc(s.fc().unwrap(), 0, 0);

        s.inbound_stream_frame(true, 0, &[0; SW_US / 2]).unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        // Receiving duplicate frames (already consumed data) will not cause an error or
        // change fc.
        s.inbound_stream_frame(true, SW / 4, &[0; SW_US / 4])
            .unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        // The stream can still receive duplicate data without a fin bit.
        s.inbound_stream_frame(false, SW / 4, &[0; SW_US / 4])
            .unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        // Receiving frame past the final size of a stream will return an error.
        assert_eq!(
            s.inbound_stream_frame(true, SW / 4, &[0; SW_US / 4 + 1]),
            Err(Error::FinalSize)
        );
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        // Read all data
        let mut buf = [1; SW_US];
        assert_eq!(s.read(&mut buf).unwrap(), (SW_US / 2, true));
        // the stream does not have fc any more. We can only check the session fc.
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        assert!(s.fc().is_none());
    }

    /// Test flow control in `RecvStreamState::DataRead`
    #[test]
    fn fc_state_data_read() {
        const SW: u64 = 1024;
        const SW_US: usize = 1024;
        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new((), SW)));

        let mut s = create_stream_with_fc(Rc::clone(&fc), SW * 3 / 4);
        check_fc(&fc.borrow(), 0, 0);
        check_fc(s.fc().unwrap(), 0, 0);

        s.inbound_stream_frame(true, 0, &[0; SW_US / 2]).unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        let mut buf = [1; SW_US];
        assert_eq!(s.read(&mut buf).unwrap(), (SW_US / 2, true));
        // the stream does not have fc any more. We can only check the session fc.
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        assert!(s.fc().is_none());

        // Receiving duplicate frames (already consumed data) will not cause an error or
        // change fc.
        s.inbound_stream_frame(true, 0, &[0; SW_US / 2]).unwrap();
        // the stream does not have fc any more. We can only check the session fc.
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        assert!(s.fc().is_none());

        // Receiving frame past the final size of a stream or the stream's fc limit
        // will NOT return an error.
        s.inbound_stream_frame(true, 0, &[0; SW_US / 2 + 1])
            .unwrap();
        s.inbound_stream_frame(true, 0, &[0; SW_US * 3 / 4 + 1])
            .unwrap();
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        assert!(s.fc().is_none());
    }

    /// Test flow control in `RecvStreamState::AbortReading` and final size is known
    #[test]
    fn fc_state_abort_reading_1() {
        const SW: u64 = 1024;
        const SW_US: usize = 1024;
        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new((), SW)));

        let mut s = create_stream_with_fc(Rc::clone(&fc), SW * 3 / 4);
        check_fc(&fc.borrow(), 0, 0);
        check_fc(s.fc().unwrap(), 0, 0);

        s.inbound_stream_frame(true, SW / 4, &[0; SW_US / 4])
            .unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        assert!(!s.stop_sending(Error::None.code()));
        // All data will de retired
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s.fc().unwrap(), SW / 2, SW / 2);

        // Receiving duplicate frames (already consumed data) will not cause an error or
        // change fc.
        s.inbound_stream_frame(true, 0, &[0; SW_US / 2]).unwrap();
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s.fc().unwrap(), SW / 2, SW / 2);

        // The stream can still receive duplicate data without a fin bit.
        s.inbound_stream_frame(false, SW / 4, &[0; SW_US / 4])
            .unwrap();
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s.fc().unwrap(), SW / 2, SW / 2);

        // Receiving frame past the final size of a stream will return an error.
        assert_eq!(
            s.inbound_stream_frame(true, SW / 4, &[0; SW_US / 4 + 1]),
            Err(Error::FinalSize)
        );
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s.fc().unwrap(), SW / 2, SW / 2);
    }

    /// Test flow control in `RecvStreamState::AbortReading` and final size is unknown
    #[test]
    fn fc_state_abort_reading_2() {
        const SW: u64 = 1024;
        const SW_US: usize = 1024;
        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new((), SW)));

        let mut s = create_stream_with_fc(Rc::clone(&fc), SW * 3 / 4);
        check_fc(&fc.borrow(), 0, 0);
        check_fc(s.fc().unwrap(), 0, 0);

        s.inbound_stream_frame(false, 0, &[0; SW_US / 2]).unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        assert!(!s.stop_sending(Error::None.code()));
        // All data will de retired
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s.fc().unwrap(), SW / 2, SW / 2);

        // Receiving duplicate frames (already consumed data) will not cause an error or
        // change fc.
        s.inbound_stream_frame(false, 0, &[0; SW_US / 2]).unwrap();
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s.fc().unwrap(), SW / 2, SW / 2);

        // Receiving data past the flow control limit will cause an error.
        assert_eq!(
            s.inbound_stream_frame(false, 0, &[0; SW_US * 3 / 4 + 1]),
            Err(Error::FlowControl)
        );

        // The stream can still receive duplicate data without a fin bit.
        s.inbound_stream_frame(false, SW / 4, &[0; SW_US / 4])
            .unwrap();
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s.fc().unwrap(), SW / 2, SW / 2);

        // Receiving more data will case the data to be retired.
        // The stream can still receive duplicate data without a fin bit.
        s.inbound_stream_frame(false, SW / 2, &[0; 10]).unwrap();
        check_fc(&fc.borrow(), SW / 2 + 10, SW / 2 + 10);
        check_fc(s.fc().unwrap(), SW / 2 + 10, SW / 2 + 10);

        // We can still receive the final size.
        s.inbound_stream_frame(true, SW / 2, &[0; 20]).unwrap();
        check_fc(&fc.borrow(), SW / 2 + 20, SW / 2 + 20);
        check_fc(s.fc().unwrap(), SW / 2 + 20, SW / 2 + 20);

        // Receiving frame past the final size of a stream will return an error.
        assert_eq!(
            s.inbound_stream_frame(true, SW / 2, &[0; 21]),
            Err(Error::FinalSize)
        );
        check_fc(&fc.borrow(), SW / 2 + 20, SW / 2 + 20);
        check_fc(s.fc().unwrap(), SW / 2 + 20, SW / 2 + 20);
    }

    /// Test flow control in `RecvStreamState::WaitForReset`
    #[test]
    fn fc_state_wait_for_reset() {
        const SW: u64 = 1024;
        const SW_US: usize = 1024;
        let fc = Rc::new(RefCell::new(ReceiverFlowControl::new((), SW)));

        let mut s = create_stream_with_fc(Rc::clone(&fc), SW * 3 / 4);
        check_fc(&fc.borrow(), 0, 0);
        check_fc(s.fc().unwrap(), 0, 0);

        s.inbound_stream_frame(false, 0, &[0; SW_US / 2]).unwrap();
        check_fc(&fc.borrow(), SW / 2, 0);
        check_fc(s.fc().unwrap(), SW / 2, 0);

        assert!(!s.stop_sending(Error::None.code()));
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s.fc().unwrap(), SW / 2, SW / 2);

        assert!(!s.stop_sending_acked());
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s.fc().unwrap(), SW / 2, SW / 2);

        // Receiving duplicate frames (already consumed data) will not cause an error or
        // change fc.
        s.inbound_stream_frame(false, 0, &[0; SW_US / 2]).unwrap();
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s.fc().unwrap(), SW / 2, SW / 2);

        // Receiving data past the flow control limit will cause an error.
        assert_eq!(
            s.inbound_stream_frame(false, 0, &[0; SW_US * 3 / 4 + 1]),
            Err(Error::FlowControl)
        );

        // The stream can still receive duplicate data without a fin bit.
        s.inbound_stream_frame(false, SW / 4, &[0; SW_US / 4])
            .unwrap();
        check_fc(&fc.borrow(), SW / 2, SW / 2);
        check_fc(s.fc().unwrap(), SW / 2, SW / 2);

        // Receiving more data will case the data to be retired.
        // The stream can still receive duplicate data without a fin bit.
        s.inbound_stream_frame(false, SW / 2, &[0; 10]).unwrap();
        check_fc(&fc.borrow(), SW / 2 + 10, SW / 2 + 10);
        check_fc(s.fc().unwrap(), SW / 2 + 10, SW / 2 + 10);
    }
}