Repository: tokio-rs/tokio
File: tokio-util/src/time/delay_queue.rs (Rust)

Code before:
///     fn get(&self, key: &CacheKey) -> Option<&Value> {
///         self.entries.get(key)
///             .map(|&(ref v, _)| v)
///     }
///
///     fn remove(&mut self, key: &CacheKey) {
///         if let Some((_, cache_key)) = self.entries.remove(key) {
///             self.expirations.remove(&cache_key);
///         }
///     }
///
///     fn poll_purge(&mut self, cx: &mut Context<'_>) -> Poll<()> {
///         while let Some(entry) = ready!(self.expirations.poll_expired(cx)) {
///             self.entries.remove(entry.get_ref());
///         }
///
///         Poll::Ready(())
///     }
/// }
/// ```
///
/// [`insert`]: method@Self::insert
/// [`insert_at`]: method@Self::insert_at
/// [`Key`]: struct@Key
/// [`Stream`]: https://docs.rs/futures/0.3.31/futures/stream/trait.Stream.html
/// [`StreamExt::next`]: https://docs.rs/futures/0.3.31/futures/stream/trait.StreamExt.html#method.next
/// [`poll_expired`]: method@Self::poll_expired
/// [`Stream::poll_expired`]: method@Self::poll_expired
/// [`DelayQueue`]: struct@DelayQueue
/// [`sleep`]: fn@tokio::time::sleep
/// [`slab`]: slab
/// [`capacity`]: method@Self::capacity
/// [`reserve`]: method@Self::reserve
#[derive(Debug)]
pub struct DelayQueue<T> {
    /// Stores data associated with entries
    slab: SlabStorage<T>,

    /// Lookup structure tracking all delays in the queue
    wheel: Wheel<Stack<T>>,

    /// Delays that were inserted when already expired. These cannot be stored
    /// in the wheel
    expired: Stack<T>,

    /// Delay expiring when the *first* item in the queue expires
    delay: Option<Pin<Box<Sleep>>>,

    /// Wheel polling state
    wheel_now: u64,


Review comment (suggestion):
Since [`tokio_stream::StreamExt::next`] documents its own cancellation safety, and `tokio::select!` [explicitly references](https://github.com/tokio-rs/tokio/blob/0284d1b5c8ea5aff5b30c254200fb0a46c21d67c/tokio/src/macros/select.rs#L111) it, it’s more appropriate to link directly to [tokio_stream::StreamExt::next].

[`tokio_stream::StreamExt::next`]: https://docs.rs/tokio-stream/latest/tokio_stream/trait.StreamExt.html#method.next

<img width="1578" height="1004" src="https://github.com/user-attachments/assets/7406f019-3a5a-4c32-97d5-a971dd6b3ca7" />


Code after:
///     fn get(&self, key: &CacheKey) -> Option<&Value> {
///         self.entries.get(key)
///             .map(|&(ref v, _)| v)
///     }
///
///     fn remove(&mut self, key: &CacheKey) {
///         if let Some((_, cache_key)) = self.entries.remove(key) {
///             self.expirations.remove(&cache_key);
///         }
///     }
///
///     fn poll_purge(&mut self, cx: &mut Context<'_>) -> Poll<()> {
///         while let Some(entry) = ready!(self.expirations.poll_expired(cx)) {
///             self.entries.remove(entry.get_ref());
///         }
///
///         Poll::Ready(())
///     }
/// }
/// ```
///
/// [`insert`]: method@Self::insert
/// [`insert_at`]: method@Self::insert_at
/// [`Key`]: struct@Key
/// [`Stream`]: https://docs.rs/futures/0.3.31/futures/stream/trait.Stream.html
/// [`StreamExt::next`]: https://docs.rs/tokio-stream/0.1.17/tokio_stream/trait.StreamExt.html#method.next
/// [`poll_expired`]: method@Self::poll_expired
/// [`Stream::poll_expired`]: method@Self::poll_expired
/// [`DelayQueue`]: struct@DelayQueue
/// [`sleep`]: fn@tokio::time::sleep
/// [`slab`]: slab
/// [`capacity`]: method@Self::capacity
/// [`reserve`]: method@Self::reserve
#[derive(Debug)]
pub struct DelayQueue<T> {
    /// Stores data associated with entries
    slab: SlabStorage<T>,

    /// Lookup structure tracking all delays in the queue
    wheel: Wheel<Stack<T>>,

    /// Delays that were inserted when already expired. These cannot be stored
    /// in the wheel
    expired: Stack<T>,

    /// Delay expiring when the *first* item in the queue expires
    delay: Option<Pin<Box<Sleep>>>,

    /// Wheel polling state
    wheel_now: u64,