//! 多生产者多消费者通道。

// 该模块目前未公开，但用作 `sync::mpsc` 中通道的实现。实现来自 crossbeam - 通道 crate:
//
// 版权所有 (c) 2019 The Crossbeam Project Developers
//
// 特此免费授予获得本软件和相关文档文件 ("Software") 副本的任何人不受限制地处理本软件的权限，包括但不限于使用、复制、修改、合并、发布的权利、分发、再许可或者出售软件的副本，并允许获得软件的人这样做，但须满足以下条件:
//
//
// 上述版权声明和本许可声明应包含在本软件的所有副本或重要部分中。
//
// 本软件提供 "AS IS"，不提供任何明示或暗示的保证，包括但不限于对适销性、特定用途的适用性和非侵权性的保证。
// 在任何情况下，作者或版权持有人均不对任何索赔、损害或其他责任负责，无论是在合同诉讼、侵权行为还是其他方面，由软件或软件的使用或其他交易引起、由软件引起或与之相关软件。
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//
//

mod array;
mod context;
mod counter;
mod error;
mod list;
mod select;
mod utils;
mod waker;
mod zero;

use crate::fmt;
use crate::panic::{RefUnwindSafe, UnwindSafe};
use crate::time::{Duration, Instant};
pub use error::*;

/// 创建无限容量的通道。
///
/// 这个通道有一个可增长的缓冲区，可以一次容纳任意数量的消息。
pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
    let (s, r) = counter::new(list::Channel::new());
    let s = Sender { flavor: SenderFlavor::List(s) };
    let r = Receiver { flavor: ReceiverFlavor::List(r) };
    (s, r)
}

/// 创建容量有限的通道。
///
/// 这个通道有一个缓冲区，一次最多可以容纳 `cap` 条消息。
///
/// 一种特殊情况是零容量通道，它不能容纳任何消息。
/// 相反，发送和接收操作必须同时出现，以便配对并传递消息。
pub fn sync_channel<T>(cap: usize) -> (Sender<T>, Receiver<T>) {
    if cap == 0 {
        let (s, r) = counter::new(zero::Channel::new());
        let s = Sender { flavor: SenderFlavor::Zero(s) };
        let r = Receiver { flavor: ReceiverFlavor::Zero(r) };
        (s, r)
    } else {
        let (s, r) = counter::new(array::Channel::with_capacity(cap));
        let s = Sender { flavor: SenderFlavor::Array(s) };
        let r = Receiver { flavor: ReceiverFlavor::Array(r) };
        (s, r)
    }
}

/// 一个通道的发送方。
pub struct Sender<T> {
    flavor: SenderFlavor<T>,
}

/// 发送者口味。
enum SenderFlavor<T> {
    /// 基于预分配数组的有界通道。
    Array(counter::Sender<array::Channel<T>>),

    /// 无界通道实现为链表。
    List(counter::Sender<list::Channel<T>>),

    /// 零容量通道。
    Zero(counter::Sender<zero::Channel<T>>),
}

unsafe impl<T: Send> Send for Sender<T> {}
unsafe impl<T: Send> Sync for Sender<T> {}

impl<T> UnwindSafe for Sender<T> {}
impl<T> RefUnwindSafe for Sender<T> {}

impl<T> Sender<T> {
    /// 尝试在不阻塞的情况下将消息发送到通道。
    ///
    /// 此方法将立即向通道发送消息，或者在通道已满或断开时返回错误。
    /// 返回的错误包含原始消息。
    ///
    /// 如果在零容量通道上调用，只有当通道的另一端恰好同时有接收操作时，此方法才会发送消息。
    ///
    pub fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> {
        match &self.flavor {
            SenderFlavor::Array(chan) => chan.try_send(msg),
            SenderFlavor::List(chan) => chan.try_send(msg),
            SenderFlavor::Zero(chan) => chan.try_send(msg),
        }
    }

    /// 阻塞当前线程，直到发送消息或断开通信。
    ///
    /// 如果通道已满且未断开连接，则此调用将阻塞，直到发送操作可以继续进行。
    /// 如果通道断开连接，这个调用将被唤醒并返回一个错误。
    /// 返回的错误包含原始消息。
    ///
    /// 如果在零容量通道上调用，此方法将等待接收操作出现在通道的另一侧。
    ///
    pub fn send(&self, msg: T) -> Result<(), SendError<T>> {
        match &self.flavor {
            SenderFlavor::Array(chan) => chan.send(msg, None),
            SenderFlavor::List(chan) => chan.send(msg, None),
            SenderFlavor::Zero(chan) => chan.send(msg, None),
        }
        .map_err(|err| match err {
            SendTimeoutError::Disconnected(msg) => SendError(msg),
            SendTimeoutError::Timeout(_) => unreachable!(),
        })
    }
}

// `sync::mpsc` 未使用以下方法，但很有用，我们可能最终会公开它们
//
//
#[allow(unused)]
impl<T> Sender<T> {
    /// 等待消息发送到通道，但时间有限。
    ///
    /// 如果通道已满且未断开，则此调用将阻塞，直到发送操作可以继续或操作超时。
    /// 如果通道断开连接，这个调用将被唤醒并返回一个错误。
    /// 返回的错误包含原始消息。
    ///
    /// 如果在零容量通道上调用，此方法将等待接收操作出现在通道的另一侧。
    ///
    pub fn send_timeout(&self, msg: T, timeout: Duration) -> Result<(), SendTimeoutError<T>> {
        match Instant::now().checked_add(timeout) {
            Some(deadline) => self.send_deadline(msg, deadline),
            // 到目前为止，在 future 中，它实际上与无限期等待相同。
            None => self.send(msg).map_err(SendTimeoutError::from),
        }
    }

    /// 等待一条消息被发送到通道，但只在给定的截止日期之前。
    ///
    /// 如果通道已满且未断开，则此调用将阻塞，直到发送操作可以继续或操作超时。
    /// 如果通道断开连接，这个调用将被唤醒并返回一个错误。
    /// 返回的错误包含原始消息。
    ///
    /// 如果在零容量通道上调用，此方法将等待接收操作出现在通道的另一侧。
    ///
    pub fn send_deadline(&self, msg: T, deadline: Instant) -> Result<(), SendTimeoutError<T>> {
        match &self.flavor {
            SenderFlavor::Array(chan) => chan.send(msg, Some(deadline)),
            SenderFlavor::List(chan) => chan.send(msg, Some(deadline)),
            SenderFlavor::Zero(chan) => chan.send(msg, Some(deadline)),
        }
    }

    /// 如果通道为空，则返回 `true`。
    ///
    /// Note: 零容量通道永远是空的。
    pub fn is_empty(&self) -> bool {
        match &self.flavor {
            SenderFlavor::Array(chan) => chan.is_empty(),
            SenderFlavor::List(chan) => chan.is_empty(),
            SenderFlavor::Zero(chan) => chan.is_empty(),
        }
    }

    /// 如果通道已满，则返回 `true`。
    ///
    /// Note: 零容量通道总是满的。
    pub fn is_full(&self) -> bool {
        match &self.flavor {
            SenderFlavor::Array(chan) => chan.is_full(),
            SenderFlavor::List(chan) => chan.is_full(),
            SenderFlavor::Zero(chan) => chan.is_full(),
        }
    }

    /// 返回通道中的消息数。
    pub fn len(&self) -> usize {
        match &self.flavor {
            SenderFlavor::Array(chan) => chan.len(),
            SenderFlavor::List(chan) => chan.len(),
            SenderFlavor::Zero(chan) => chan.len(),
        }
    }

    /// 如果通道有界，则返回其容量。
    pub fn capacity(&self) -> Option<usize> {
        match &self.flavor {
            SenderFlavor::Array(chan) => chan.capacity(),
            SenderFlavor::List(chan) => chan.capacity(),
            SenderFlavor::Zero(chan) => chan.capacity(),
        }
    }

    /// 如果发送者属于同一个通道，则返回 `true`。
    pub fn same_channel(&self, other: &Sender<T>) -> bool {
        match (&self.flavor, &other.flavor) {
            (SenderFlavor::Array(ref a), SenderFlavor::Array(ref b)) => a == b,
            (SenderFlavor::List(ref a), SenderFlavor::List(ref b)) => a == b,
            (SenderFlavor::Zero(ref a), SenderFlavor::Zero(ref b)) => a == b,
            _ => false,
        }
    }
}

impl<T> Drop for Sender<T> {
    fn drop(&mut self) {
        unsafe {
            match &self.flavor {
                SenderFlavor::Array(chan) => chan.release(|c| c.disconnect_senders()),
                SenderFlavor::List(chan) => chan.release(|c| c.disconnect_senders()),
                SenderFlavor::Zero(chan) => chan.release(|c| c.disconnect()),
            }
        }
    }
}

impl<T> Clone for Sender<T> {
    fn clone(&self) -> Self {
        let flavor = match &self.flavor {
            SenderFlavor::Array(chan) => SenderFlavor::Array(chan.acquire()),
            SenderFlavor::List(chan) => SenderFlavor::List(chan.acquire()),
            SenderFlavor::Zero(chan) => SenderFlavor::Zero(chan.acquire()),
        };

        Sender { flavor }
    }
}

impl<T> fmt::Debug for Sender<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.pad("Sender { .. }")
    }
}

/// 一个通道的接收端。
pub struct Receiver<T> {
    flavor: ReceiverFlavor<T>,
}

/// 接收者口味。
enum ReceiverFlavor<T> {
    /// 基于预分配数组的有界通道。
    Array(counter::Receiver<array::Channel<T>>),

    /// 无界通道实现为链表。
    List(counter::Receiver<list::Channel<T>>),

    /// 零容量通道。
    Zero(counter::Receiver<zero::Channel<T>>),
}

unsafe impl<T: Send> Send for Receiver<T> {}
unsafe impl<T: Send> Sync for Receiver<T> {}

impl<T> UnwindSafe for Receiver<T> {}
impl<T> RefUnwindSafe for Receiver<T> {}

impl<T> Receiver<T> {
    /// 尝试在不阻塞的情况下接收来自通道的消息。
    ///
    /// 此方法将立即从通道接收消息，或者如果通道为空则返回错误。
    ///
    /// 如果在零容量的通道上调用，只有当通道的另一端恰好同时有发送操作时，此方法才会收到消息。
    ///
    ///
    pub fn try_recv(&self) -> Result<T, TryRecvError> {
        match &self.flavor {
            ReceiverFlavor::Array(chan) => chan.try_recv(),
            ReceiverFlavor::List(chan) => chan.try_recv(),
            ReceiverFlavor::Zero(chan) => chan.try_recv(),
        }
    }

    /// 阻塞当前线程，直到收到消息或通道为空并断开连接。
    ///
    /// 如果通道为空且未断开连接，则此调用将阻塞，直到接收操作可以继续进行。
    /// 如果通道为空并断开连接，则此调用将唤醒并返回错误。
    ///
    /// 如果在零容量通道上调用，此方法将等待发送操作出现在通道的另一侧。
    ///
    ///
    ///
    pub fn recv(&self) -> Result<T, RecvError> {
        match &self.flavor {
            ReceiverFlavor::Array(chan) => chan.recv(None),
            ReceiverFlavor::List(chan) => chan.recv(None),
            ReceiverFlavor::Zero(chan) => chan.recv(None),
        }
        .map_err(|_| RecvError)
    }

    /// 等待从通道收到消息，但时间有限。
    ///
    /// 如果通道为空且未断开连接，则此调用将阻塞，直到接收操作可以继续或操作超时。
    /// 如果通道为空并断开连接，则此调用将唤醒并返回错误。
    ///
    /// 如果在零容量通道上调用，此方法将等待发送操作出现在通道的另一侧。
    ///
    ///
    pub fn recv_timeout(&self, timeout: Duration) -> Result<T, RecvTimeoutError> {
        match Instant::now().checked_add(timeout) {
            Some(deadline) => self.recv_deadline(deadline),
            // 到目前为止，在 future 中，它实际上与无限期等待相同。
            None => self.recv().map_err(RecvTimeoutError::from),
        }
    }

    /// 等待从通道收到消息，但时间有限。
    ///
    /// 如果通道为空且未断开连接，则此调用将阻塞，直到接收操作可以继续或操作超时。
    /// 如果通道为空并断开连接，则此调用将唤醒并返回错误。
    ///
    /// 如果在零容量通道上调用，此方法将等待发送操作出现在通道的另一侧。
    ///
    ///
    pub fn recv_deadline(&self, deadline: Instant) -> Result<T, RecvTimeoutError> {
        match &self.flavor {
            ReceiverFlavor::Array(chan) => chan.recv(Some(deadline)),
            ReceiverFlavor::List(chan) => chan.recv(Some(deadline)),
            ReceiverFlavor::Zero(chan) => chan.recv(Some(deadline)),
        }
    }
}

// `sync::mpsc` 未使用以下方法，但很有用，我们可能最终会公开它们
//
//
#[allow(unused)]
impl<T> Receiver<T> {
    /// 如果通道为空，则返回 `true`。
    ///
    /// Note: 零容量通道永远是空的。
    pub fn is_empty(&self) -> bool {
        match &self.flavor {
            ReceiverFlavor::Array(chan) => chan.is_empty(),
            ReceiverFlavor::List(chan) => chan.is_empty(),
            ReceiverFlavor::Zero(chan) => chan.is_empty(),
        }
    }

    /// 如果通道已满，则返回 `true`。
    ///
    /// Note: 零容量通道总是满的。
    pub fn is_full(&self) -> bool {
        match &self.flavor {
            ReceiverFlavor::Array(chan) => chan.is_full(),
            ReceiverFlavor::List(chan) => chan.is_full(),
            ReceiverFlavor::Zero(chan) => chan.is_full(),
        }
    }

    /// 返回通道中的消息数。
    pub fn len(&self) -> usize {
        match &self.flavor {
            ReceiverFlavor::Array(chan) => chan.len(),
            ReceiverFlavor::List(chan) => chan.len(),
            ReceiverFlavor::Zero(chan) => chan.len(),
        }
    }

    /// 如果通道有界，则返回其容量。
    pub fn capacity(&self) -> Option<usize> {
        match &self.flavor {
            ReceiverFlavor::Array(chan) => chan.capacity(),
            ReceiverFlavor::List(chan) => chan.capacity(),
            ReceiverFlavor::Zero(chan) => chan.capacity(),
        }
    }

    /// 如果接收者属于同一个通道，则返回 `true`。
    pub fn same_channel(&self, other: &Receiver<T>) -> bool {
        match (&self.flavor, &other.flavor) {
            (ReceiverFlavor::Array(a), ReceiverFlavor::Array(b)) => a == b,
            (ReceiverFlavor::List(a), ReceiverFlavor::List(b)) => a == b,
            (ReceiverFlavor::Zero(a), ReceiverFlavor::Zero(b)) => a == b,
            _ => false,
        }
    }
}

impl<T> Drop for Receiver<T> {
    fn drop(&mut self) {
        unsafe {
            match &self.flavor {
                ReceiverFlavor::Array(chan) => chan.release(|c| c.disconnect_receivers()),
                ReceiverFlavor::List(chan) => chan.release(|c| c.disconnect_receivers()),
                ReceiverFlavor::Zero(chan) => chan.release(|c| c.disconnect()),
            }
        }
    }
}

impl<T> Clone for Receiver<T> {
    fn clone(&self) -> Self {
        let flavor = match &self.flavor {
            ReceiverFlavor::Array(chan) => ReceiverFlavor::Array(chan.acquire()),
            ReceiverFlavor::List(chan) => ReceiverFlavor::List(chan.acquire()),
            ReceiverFlavor::Zero(chan) => ReceiverFlavor::Zero(chan.acquire()),
        };

        Receiver { flavor }
    }
}

impl<T> fmt::Debug for Receiver<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.pad("Receiver { .. }")
    }
}