Repository: launchbadge/sqlx
File: sqlx-postgres/src/listener.rs (Rust)

Code before:
                }
            };

            match message.format {
                // We've received an async notification, return it.
                BackendMessageFormat::NotificationResponse => {
                    return Ok(Some(PgNotification(message.decode()?)));
                }

                // Mark the connection as ready for another query
                BackendMessageFormat::ReadyForQuery => {
                    self.connection().await?.inner.pending_ready_for_query_count -= 1;
                }

                // Ignore unexpected messages
                _ => {}
            }
        }
    }

    /// Receives the next notification that already exists in the connection buffer, if any.
    ///
    /// This is similar to `try_recv`, except it will not wait if the connection has not yet received a notification.
    ///
    /// This is helpful if you want to retrieve all buffered notifications and process them in batches.
    pub fn try_recv_buffered(&mut self) -> Option<PgNotification> {
        if let Ok(Some(notification)) = self.buffer_rx.try_next() {
            Some(PgNotification(notification))
        } else {
            None
        }
    }

    /// Consume this listener, returning a `Stream` of notifications.
    ///
    /// The backing connection will be automatically reconnected should it be lost.
    ///
    /// This has the same potential drawbacks as [`recv`](PgListener::recv).
    ///
    pub fn into_stream(mut self) -> impl Stream<Item = Result<PgNotification, Error>> + Unpin {
        Box::pin(try_stream! {
            loop {
                r#yield!(self.recv().await?);
            }
        })
    }
}

impl Drop for PgListener {
    fn drop(&mut self) {
        if let Some(mut conn) = self.connection.take() {

Review comment (suggestion):
The API is fine but the naming is a bit clunky. It's a shame we used `try_recv` for something else here. I think for that reason we should avoid `recv` for something that doesn't do I/O.

How about `.pop_buffered()`? Or `.next_buffered()`? I'd even be okay with `.pop()` though it may not be immediately obvious from context exactly what it does.

Code after:
                }
            };

            match message.format {
                // We've received an async notification, return it.
                BackendMessageFormat::NotificationResponse => {
                    return Ok(Some(PgNotification(message.decode()?)));
                }

                // Mark the connection as ready for another query
                BackendMessageFormat::ReadyForQuery => {
                    self.connection().await?.inner.pending_ready_for_query_count -= 1;
                }

                // Ignore unexpected messages
                _ => {}
            }
        }
    }

    /// Receives the next notification that already exists in the connection buffer, if any.
    ///
    /// This is similar to `try_recv`, except it will not wait if the connection has not yet received a notification.
    ///
    /// This is helpful if you want to retrieve all buffered notifications and process them in batches.
    pub fn next_buffered(&mut self) -> Option<PgNotification> {
        if let Ok(Some(notification)) = self.buffer_rx.try_next() {
            Some(PgNotification(notification))
        } else {
            None
        }
    }

    /// Consume this listener, returning a `Stream` of notifications.
    ///
    /// The backing connection will be automatically reconnected should it be lost.
    ///
    /// This has the same potential drawbacks as [`recv`](PgListener::recv).
    ///
    pub fn into_stream(mut self) -> impl Stream<Item = Result<PgNotification, Error>> + Unpin {
        Box::pin(try_stream! {
            loop {
                r#yield!(self.recv().await?);
            }
        })
    }
}

impl Drop for PgListener {
    fn drop(&mut self) {
        if let Some(mut conn) = self.connection.take() {