mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-08-24 00:00:11 +02:00
sync: move broadcast error types into broadcast::error module (#2937)
Refs: #2928
This commit is contained in:
+89
-68
@@ -55,8 +55,8 @@
|
||||
//! [`Sender::subscribe`]: crate::sync::broadcast::Sender::subscribe
|
||||
//! [`Receiver`]: crate::sync::broadcast::Receiver
|
||||
//! [`channel`]: crate::sync::broadcast::channel
|
||||
//! [`RecvError::Lagged`]: crate::sync::broadcast::RecvError::Lagged
|
||||
//! [`RecvError::Closed`]: crate::sync::broadcast::RecvError::Closed
|
||||
//! [`RecvError::Lagged`]: crate::sync::broadcast::error::RecvError::Lagged
|
||||
//! [`RecvError::Closed`]: crate::sync::broadcast::error::RecvError::Closed
|
||||
//! [`recv`]: crate::sync::broadcast::Receiver::recv
|
||||
//!
|
||||
//! # Examples
|
||||
@@ -197,52 +197,96 @@ pub struct Receiver<T> {
|
||||
next: u64,
|
||||
}
|
||||
|
||||
/// Error returned by [`Sender::send`][Sender::send].
|
||||
///
|
||||
/// A **send** operation can only fail if there are no active receivers,
|
||||
/// implying that the message could never be received. The error contains the
|
||||
/// message being sent as a payload so it can be recovered.
|
||||
#[derive(Debug)]
|
||||
pub struct SendError<T>(pub T);
|
||||
pub mod error {
|
||||
//! Broadcast error types
|
||||
|
||||
/// An error returned from the [`recv`] function on a [`Receiver`].
|
||||
///
|
||||
/// [`recv`]: crate::sync::broadcast::Receiver::recv
|
||||
/// [`Receiver`]: crate::sync::broadcast::Receiver
|
||||
#[derive(Debug, PartialEq)]
|
||||
pub enum RecvError {
|
||||
/// There are no more active senders implying no further messages will ever
|
||||
/// be sent.
|
||||
Closed,
|
||||
use std::fmt;
|
||||
|
||||
/// The receiver lagged too far behind. Attempting to receive again will
|
||||
/// return the oldest message still retained by the channel.
|
||||
/// Error returned by from the [`send`] function on a [`Sender`].
|
||||
///
|
||||
/// Includes the number of skipped messages.
|
||||
Lagged(u64),
|
||||
/// A **send** operation can only fail if there are no active receivers,
|
||||
/// implying that the message could never be received. The error contains the
|
||||
/// message being sent as a payload so it can be recovered.
|
||||
///
|
||||
/// [`send`]: crate::sync::broadcast::Sender::send
|
||||
/// [`Sender`]: crate::sync::broadcast::Sender
|
||||
#[derive(Debug)]
|
||||
pub struct SendError<T>(pub T);
|
||||
|
||||
impl<T> fmt::Display for SendError<T> {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "channel closed")
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: fmt::Debug> std::error::Error for SendError<T> {}
|
||||
|
||||
/// An error returned from the [`recv`] function on a [`Receiver`].
|
||||
///
|
||||
/// [`recv`]: crate::sync::broadcast::Receiver::recv
|
||||
/// [`Receiver`]: crate::sync::broadcast::Receiver
|
||||
#[derive(Debug, PartialEq)]
|
||||
pub enum RecvError {
|
||||
/// There are no more active senders implying no further messages will ever
|
||||
/// be sent.
|
||||
Closed,
|
||||
|
||||
/// The receiver lagged too far behind. Attempting to receive again will
|
||||
/// return the oldest message still retained by the channel.
|
||||
///
|
||||
/// Includes the number of skipped messages.
|
||||
Lagged(u64),
|
||||
}
|
||||
|
||||
impl fmt::Display for RecvError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
RecvError::Closed => write!(f, "channel closed"),
|
||||
RecvError::Lagged(amt) => write!(f, "channel lagged by {}", amt),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for RecvError {}
|
||||
|
||||
/// An error returned from the [`try_recv`] function on a [`Receiver`].
|
||||
///
|
||||
/// [`try_recv`]: crate::sync::broadcast::Receiver::try_recv
|
||||
/// [`Receiver`]: crate::sync::broadcast::Receiver
|
||||
#[derive(Debug, PartialEq)]
|
||||
pub enum TryRecvError {
|
||||
/// The channel is currently empty. There are still active
|
||||
/// [`Sender`] handles, so data may yet become available.
|
||||
///
|
||||
/// [`Sender`]: crate::sync::broadcast::Sender
|
||||
Empty,
|
||||
|
||||
/// There are no more active senders implying no further messages will ever
|
||||
/// be sent.
|
||||
Closed,
|
||||
|
||||
/// The receiver lagged too far behind and has been forcibly disconnected.
|
||||
/// Attempting to receive again will return the oldest message still
|
||||
/// retained by the channel.
|
||||
///
|
||||
/// Includes the number of skipped messages.
|
||||
Lagged(u64),
|
||||
}
|
||||
|
||||
impl fmt::Display for TryRecvError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
TryRecvError::Empty => write!(f, "channel empty"),
|
||||
TryRecvError::Closed => write!(f, "channel closed"),
|
||||
TryRecvError::Lagged(amt) => write!(f, "channel lagged by {}", amt),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for TryRecvError {}
|
||||
}
|
||||
|
||||
/// An error returned from the [`try_recv`] function on a [`Receiver`].
|
||||
///
|
||||
/// [`try_recv`]: crate::sync::broadcast::Receiver::try_recv
|
||||
/// [`Receiver`]: crate::sync::broadcast::Receiver
|
||||
#[derive(Debug, PartialEq)]
|
||||
pub enum TryRecvError {
|
||||
/// The channel is currently empty. There are still active
|
||||
/// [`Sender`][Sender] handles, so data may yet become available.
|
||||
Empty,
|
||||
|
||||
/// There are no more active senders implying no further messages will ever
|
||||
/// be sent.
|
||||
Closed,
|
||||
|
||||
/// The receiver lagged too far behind and has been forcibly disconnected.
|
||||
/// Attempting to receive again will return the oldest message still
|
||||
/// retained by the channel.
|
||||
///
|
||||
/// Includes the number of skipped messages.
|
||||
Lagged(u64),
|
||||
}
|
||||
use self::error::*;
|
||||
|
||||
/// Data shared between senders and receivers
|
||||
struct Shared<T> {
|
||||
@@ -371,8 +415,8 @@ const MAX_RECEIVERS: usize = usize::MAX >> 2;
|
||||
/// [`Sender::subscribe`]: crate::sync::broadcast::Sender::subscribe
|
||||
/// [`Receiver`]: crate::sync::broadcast::Receiver
|
||||
/// [`recv`]: crate::sync::broadcast::Receiver::recv
|
||||
/// [`SendError`]: crate::sync::broadcast::SendError
|
||||
/// [`RecvError`]: crate::sync::broadcast::RecvError
|
||||
/// [`SendError`]: crate::sync::broadcast::error::SendError
|
||||
/// [`RecvError`]: crate::sync::broadcast::error::RecvError
|
||||
///
|
||||
/// # Examples
|
||||
///
|
||||
@@ -1112,27 +1156,4 @@ impl<'a, T> Drop for RecvGuard<'a, T> {
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Display for RecvError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
RecvError::Closed => write!(f, "channel closed"),
|
||||
RecvError::Lagged(amt) => write!(f, "channel lagged by {}", amt),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for RecvError {}
|
||||
|
||||
impl fmt::Display for TryRecvError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
TryRecvError::Empty => write!(f, "channel empty"),
|
||||
TryRecvError::Closed => write!(f, "channel closed"),
|
||||
TryRecvError::Lagged(amt) => write!(f, "channel lagged by {}", amt),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for TryRecvError {}
|
||||
|
||||
fn is_unpin<T: Unpin>() {}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
use crate::sync::broadcast;
|
||||
use crate::sync::broadcast::RecvError::{Closed, Lagged};
|
||||
use crate::sync::broadcast::error::RecvError::{Closed, Lagged};
|
||||
|
||||
use loom::future::block_on;
|
||||
use loom::sync::Arc;
|
||||
|
||||
@@ -23,7 +23,7 @@ macro_rules! assert_empty {
|
||||
($e:expr) => {
|
||||
match $e.try_recv() {
|
||||
Ok(value) => panic!("expected empty; got = {:?}", value),
|
||||
Err(broadcast::TryRecvError::Empty) => {}
|
||||
Err(broadcast::error::TryRecvError::Empty) => {}
|
||||
Err(e) => panic!("expected empty; got = {:?}", e),
|
||||
}
|
||||
};
|
||||
@@ -32,7 +32,7 @@ macro_rules! assert_empty {
|
||||
macro_rules! assert_lagged {
|
||||
($e:expr, $n:expr) => {
|
||||
match assert_err!($e) {
|
||||
broadcast::TryRecvError::Lagged(n) => {
|
||||
broadcast::error::TryRecvError::Lagged(n) => {
|
||||
assert_eq!(n, $n);
|
||||
}
|
||||
_ => panic!("did not lag"),
|
||||
@@ -43,7 +43,7 @@ macro_rules! assert_lagged {
|
||||
macro_rules! assert_closed {
|
||||
($e:expr) => {
|
||||
match assert_err!($e) {
|
||||
broadcast::TryRecvError::Closed => {}
|
||||
broadcast::error::TryRecvError::Closed => {}
|
||||
_ => panic!("did not lag"),
|
||||
}
|
||||
};
|
||||
@@ -491,6 +491,6 @@ fn lagging_receiver_recovers_after_wrap_open() {
|
||||
assert_empty!(rx);
|
||||
}
|
||||
|
||||
fn is_closed(err: broadcast::RecvError) -> bool {
|
||||
matches!(err, broadcast::RecvError::Closed)
|
||||
fn is_closed(err: broadcast::error::RecvError) -> bool {
|
||||
matches!(err, broadcast::error::RecvError::Closed)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user