mirror of
https://github.com/pjh456/OxideJS.git
synced 2026-10-24 02:31:13 +08:00
319 lines
10 KiB
Rust
319 lines
10 KiB
Rust
//! 通用消息队列：`std::sync::mpsc` 的薄封装，提供命名发送端/接收端类型与统一入口。
|
|
//!
|
|
//! 关键约定：
|
|
//! - `Sender<T>` 可克隆、可跨线程共享（要求 `T: Send`）；`Receiver<T>` 单消费者、非 Send。
|
|
//! - FIFO 序：`send` 的顺序与 `recv` 的顺序一致。
|
|
//! - 断开检测：全部发送端（含克隆）drop 后，`recv` / `try_recv` 返回断开，
|
|
//! `is_disconnected` 返回 true（与 std 语义一致，不取决于队列是否还有待取值）。
|
|
//! - 语义与 `std::sync::mpsc` 完全一致；本模块只提供命名类型与统一入口，零新依赖。
|
|
|
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
|
use std::sync::{mpsc, Arc};
|
|
|
|
/// 创建一对消息队列发送端与接收端。
|
|
///
|
|
/// 发送端可克隆、可跨线程共享；接收端单消费者。全部发送端 drop 后队列断开，
|
|
/// 接收端此后的所有接收都返回断开。
|
|
pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
|
|
let (tx, rx) = mpsc::channel();
|
|
// 存活发送端计数：std 的 `mpsc::Receiver::is_disconnected` 尚不稳定，
|
|
// 用自维护计数复现 std 断开语义（全部发送端 drop 即断开）。
|
|
let live = Arc::new(AtomicUsize::new(1));
|
|
(Sender { inner: tx, live: live.clone() }, Receiver { inner: rx, live })
|
|
}
|
|
|
|
/// 发送端：消息队列的可克隆、可跨线程共享端。
|
|
///
|
|
/// 多个 `Sender` 克隆（含移入其他线程的）写入同一队列；全部发送端 drop 后
|
|
/// 队列断开。
|
|
pub struct Sender<T> {
|
|
inner: mpsc::Sender<T>,
|
|
live: Arc<AtomicUsize>,
|
|
}
|
|
|
|
impl<T> Sender<T> {
|
|
/// 发送一个值进队列。
|
|
///
|
|
/// # 边界与前提
|
|
/// - 接收端已全部 drop 时返回 `SendError`，值归还调用方。
|
|
///
|
|
/// # 注意事项
|
|
/// - `Sender<T>: Send` 要求 `T: Send`，值随发送跨线程转移。
|
|
pub fn send(&self, value: T) -> Result<(), SendError<T>> {
|
|
self.inner.send(value).map_err(|e| SendError(e.0))
|
|
}
|
|
}
|
|
|
|
// std 的 `mpsc::Sender` 对任意 `T` 都可克隆，手动实现避免 derive 引入多余的 `T: Clone` 约束。
|
|
impl<T> Clone for Sender<T> {
|
|
fn clone(&self) -> Self {
|
|
self.live.fetch_add(1, Ordering::Relaxed);
|
|
Self {
|
|
inner: self.inner.clone(),
|
|
live: self.live.clone(),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<T> Drop for Sender<T> {
|
|
fn drop(&mut self) {
|
|
self.live.fetch_sub(1, Ordering::Release);
|
|
}
|
|
}
|
|
|
|
/// 接收端：消息队列的单消费者端，非 Send。
|
|
///
|
|
/// 接收端只在持有线程内使用；阻塞 `recv` 与非阻塞 `try_recv` 均可用。
|
|
pub struct Receiver<T> {
|
|
inner: mpsc::Receiver<T>,
|
|
live: Arc<AtomicUsize>,
|
|
}
|
|
|
|
impl<T> Receiver<T> {
|
|
/// 阻塞接收：等待直到有值到达或队列断开。
|
|
///
|
|
/// # 边界与前提
|
|
/// - 队列已断开（全部发送端 drop 且无值可取）时返回 `RecvError`。
|
|
///
|
|
/// # 注意事项
|
|
/// - 阻塞调用，调用方须保证所在线程可被阻塞。
|
|
pub fn recv(&self) -> Result<T, RecvError> {
|
|
self.inner.recv().map_err(|_| RecvError)
|
|
}
|
|
|
|
/// 非阻塞接收：立即返回。
|
|
///
|
|
/// # 边界与前提
|
|
/// - 队列为空时返回 `TryRecvError::Empty`（发送端仍存活，其后可有值）。
|
|
/// - 队列已断开时返回 `TryRecvError::Disconnected`（其后无值）。
|
|
pub fn try_recv(&self) -> Result<T, TryRecvError> {
|
|
self.inner.try_recv().map_err(|e| match e {
|
|
mpsc::TryRecvError::Empty => TryRecvError::Empty,
|
|
mpsc::TryRecvError::Disconnected => TryRecvError::Disconnected,
|
|
})
|
|
}
|
|
|
|
/// 带超时接收：在 `duration` 内返回一个值，否则返回超时错误。
|
|
///
|
|
/// # 边界与前提
|
|
/// - `duration` 内无值到达且队列未断开时返回 `Err(Timeout)`。
|
|
/// - 队列已断开（全部发送端 drop）且无值可取时同样返回 `Err(Timeout)`
|
|
/// （与 std `recv_timeout` 语义一致，不区分超时与断开；断开判定经
|
|
/// [`Receiver::is_disconnected`] 单独查询）。
|
|
///
|
|
/// # 注意事项
|
|
/// - 透传 std `mpsc::Receiver::recv_timeout` 语义。
|
|
pub fn recv_timeout(&self, duration: std::time::Duration) -> Result<T, Timeout> {
|
|
self.inner.recv_timeout(duration).map_err(|_| Timeout)
|
|
}
|
|
|
|
/// 队列是否已断开（全部发送端已 drop）。
|
|
///
|
|
/// 与 std 断开语义一致：不取决于队列是否还有待取值。断开是单向状态：
|
|
/// 一旦为 true 不再回到 false。
|
|
pub fn is_disconnected(&self) -> bool {
|
|
self.live.load(Ordering::Acquire) == 0
|
|
}
|
|
}
|
|
|
|
/// 发送失败：接收端已 drop，值归还调用方。
|
|
pub struct SendError<T>(pub T);
|
|
|
|
/// 接收失败：队列已断开（全部发送端已 drop）且无值可取。
|
|
pub struct RecvError;
|
|
|
|
/// 非阻塞接收失败：队列为空或队列已断开。
|
|
pub enum TryRecvError {
|
|
/// 队列为空（发送端仍存活，其后可有值）。
|
|
Empty,
|
|
/// 队列已断开（全部发送端已 drop，其后无值）。
|
|
Disconnected,
|
|
}
|
|
|
|
/// 带超时接收失败：指定时间内无值到达且队列未断开（或已断开且无值可取）。
|
|
pub struct Timeout;
|
|
|
|
impl<T: std::fmt::Debug> std::fmt::Debug for SendError<T> {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_tuple("SendError").field(&self.0).finish()
|
|
}
|
|
}
|
|
|
|
impl std::fmt::Debug for RecvError {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
write!(f, "RecvError")
|
|
}
|
|
}
|
|
|
|
impl std::fmt::Debug for TryRecvError {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
match self {
|
|
TryRecvError::Empty => write!(f, "TryRecvError::Empty"),
|
|
TryRecvError::Disconnected => write!(f, "TryRecvError::Disconnected"),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl std::fmt::Debug for Timeout {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
write!(f, "Timeout")
|
|
}
|
|
}
|
|
|
|
impl<T> std::fmt::Display for SendError<T> {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
write!(f, "发送失败：接收端已 drop")
|
|
}
|
|
}
|
|
|
|
impl std::fmt::Display for RecvError {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
write!(f, "接收失败：消息队列已断开")
|
|
}
|
|
}
|
|
|
|
impl std::fmt::Display for TryRecvError {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
match self {
|
|
TryRecvError::Empty => write!(f, "队列为空"),
|
|
TryRecvError::Disconnected => write!(f, "消息队列已断开"),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl std::fmt::Display for Timeout {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
write!(f, "接收超时")
|
|
}
|
|
}
|
|
|
|
impl<T> std::error::Error for SendError<T> where T: std::fmt::Debug {}
|
|
impl std::error::Error for RecvError {}
|
|
impl std::error::Error for TryRecvError {}
|
|
impl std::error::Error for Timeout {}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn send_recv_roundtrip() {
|
|
let (tx, rx) = channel();
|
|
tx.send(42).unwrap();
|
|
assert_eq!(rx.recv().unwrap(), 42);
|
|
}
|
|
|
|
#[test]
|
|
fn fifo_order() {
|
|
let (tx, rx) = channel();
|
|
for i in 0..3 {
|
|
tx.send(i).unwrap();
|
|
}
|
|
for i in 0..3 {
|
|
assert_eq!(rx.recv().unwrap(), i);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn multi_producer() {
|
|
let (tx, rx) = channel();
|
|
let mut handles = Vec::new();
|
|
for t in 0..4 {
|
|
let tx = tx.clone();
|
|
handles.push(std::thread::spawn(move || {
|
|
for i in 0..25 {
|
|
tx.send(t * 100 + i).unwrap();
|
|
}
|
|
}));
|
|
}
|
|
drop(tx);
|
|
let mut received = Vec::new();
|
|
for _ in 0..100 {
|
|
received.push(rx.recv().unwrap());
|
|
}
|
|
received.sort_unstable();
|
|
let expected: Vec<i32> = (0..4).flat_map(|t| (0..25).map(move |i| t * 100 + i)).collect();
|
|
assert_eq!(received, expected);
|
|
for h in handles {
|
|
h.join().unwrap();
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn try_recv_empty() {
|
|
let (_tx, rx) = channel::<i32>();
|
|
assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
|
|
}
|
|
|
|
#[test]
|
|
fn try_recv_disconnected() {
|
|
let (tx, rx) = channel::<i32>();
|
|
drop(tx);
|
|
assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected)));
|
|
}
|
|
|
|
#[test]
|
|
fn recv_disconnected() {
|
|
let (tx, rx) = channel::<i32>();
|
|
drop(tx);
|
|
assert!(rx.recv().is_err());
|
|
}
|
|
|
|
#[test]
|
|
fn is_disconnected_flag() {
|
|
let (tx, rx) = channel::<i32>();
|
|
assert!(!rx.is_disconnected());
|
|
drop(tx);
|
|
assert!(rx.is_disconnected());
|
|
}
|
|
|
|
#[test]
|
|
fn is_disconnected_with_pending_values() {
|
|
// 断开语义不取决于队列是否还有待取值：发送端 drop 即断开。
|
|
let (tx, rx) = channel::<i32>();
|
|
tx.send(1).unwrap();
|
|
drop(tx);
|
|
assert!(rx.is_disconnected());
|
|
assert_eq!(rx.recv().unwrap(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn send_error_returns_value() {
|
|
let (tx, rx) = channel::<i32>();
|
|
drop(rx);
|
|
let err = tx.send(7).unwrap_err();
|
|
assert_eq!(err.0, 7);
|
|
}
|
|
|
|
#[test]
|
|
fn recv_timeout_returns_value_in_time() {
|
|
let (tx, rx) = channel::<i32>();
|
|
tx.send(42).unwrap();
|
|
assert_eq!(rx.recv_timeout(std::time::Duration::from_millis(50)).unwrap(), 42);
|
|
}
|
|
|
|
#[test]
|
|
fn recv_timeout_empty_returns_timeout() {
|
|
let (_tx, rx) = channel::<i32>();
|
|
assert!(rx.recv_timeout(std::time::Duration::from_millis(50)).is_err());
|
|
}
|
|
|
|
#[test]
|
|
fn recv_timeout_disconnected_returns_timeout() {
|
|
let (tx, rx) = channel::<i32>();
|
|
drop(tx);
|
|
assert!(rx.recv_timeout(std::time::Duration::from_millis(50)).is_err());
|
|
assert!(rx.is_disconnected());
|
|
}
|
|
|
|
#[test]
|
|
fn sender_is_send_across_threads() {
|
|
let (tx, rx) = channel::<String>();
|
|
let handle = std::thread::spawn(move || {
|
|
tx.send("from worker".to_string()).unwrap();
|
|
});
|
|
handle.join().unwrap();
|
|
assert_eq!(rx.recv().unwrap(), "from worker");
|
|
}
|
|
}
|