linch/

schannel.rs

use std::{
    future::Future,
    pin::Pin,
    task::{Context, Poll},
    time::{Duration, Instant},
};

use crate::error::{
    ReadyTimeoutError, RecvError, RecvTimeoutError, SelectTimeoutError, SendError,
    SendTimeoutError, TryRecvError, TrySelectError, TrySendError,
};
use crossbeam_channel::{Receiver as CrossbeamReceiver, Sender as CrossbeamSender};
use futures::Stream;
use pin_project_lite::pin_project;

/// Creates a bounded channel with the specified capacity using the schannel implementation.
///
/// This channel is a wrapper around the crossbeam channel that provides a more
/// ergonomic API for sending and receiving items with both synchronous and asynchronous operations.
///
/// # Performance Characteristics
///
/// The asynchronous send/receive operations in this implementation are optimized for high throughput
/// scenarios. They use active polling which makes them suitable for situations where:
///
/// * Active async polling is acceptable and desired
/// * Throughput and latency are more important than CPU efficiency
/// * You want maximum performance for benchmarking against other channel implementations
///
/// This implementation does not expect channels to remain empty for extended periods and
/// will continuously poll when operations are pending.
///
/// # Arguments
///
/// * `capacity` - The capacity of the channel buffer. Must be greater than 0.
///
/// # Returns
///
/// A tuple containing a [`Sender`] and [`Receiver`] pair.
///
/// # Panics
///
/// Panics if `capacity` is 0.
///
/// # Examples
///
/// ```rust
/// use linch::schannel;
///
/// // Create a channel with capacity 10
/// let (sender, receiver) = schannel::bounded(10);
///
/// // Send synchronously
/// sender.send(42).unwrap();
///
/// // Receive synchronously
/// let value = receiver.recv().unwrap();
/// assert_eq!(value, 42);
/// ```
///
/// # Async Example
///
/// ```rust
/// use linch::schannel;
///
/// # tokio_test::block_on(async {
/// let (sender, receiver) = schannel::bounded(10);
///
/// // Send asynchronously
/// sender.send_async(42).await.unwrap();
///
/// // Receive asynchronously  
/// let value = receiver.recv_async().await.unwrap();
/// assert_eq!(value, 42);
/// # });
/// ```
pub fn bounded<T>(capacity: usize) -> (Sender<T>, Receiver<T>) {
    assert!(capacity > 0);
    let (tx, rx) = crossbeam_channel::bounded(capacity);
    (Sender::new(tx), Receiver::new(rx))
}

/// Creates an unbounded channel using the schannel implementation.
///
/// This channel is a wrapper around the crossbeam unbounded channel that provides a more
/// ergonomic API for sending and receiving items with both synchronous and asynchronous operations.
///
/// # Performance Characteristics
///
/// The asynchronous send/receive operations in this implementation are optimized for high throughput
/// scenarios. They use active polling which makes them suitable for situations where:
///
/// * Active async polling is acceptable and desired
/// * Throughput and latency are more important than CPU efficiency
/// * You want maximum performance for benchmarking against other channel implementations
///
/// This implementation does not expect channels to remain empty for extended periods and
/// will continuously poll when operations are pending.
///
/// Unlike bounded channels, unbounded channels have no capacity limit, so send operations
/// will never block due to a full buffer. However, this can lead to unbounded memory growth
/// if the receiver cannot keep up with the sender.
///
/// # Returns
///
/// A tuple containing a [`Sender`] and [`Receiver`] pair.
///
/// # Examples
///
/// ```rust
/// use linch::schannel;
///
/// // Create an unbounded channel
/// let (sender, receiver) = schannel::unbounded();
///
/// // Send synchronously - never blocks
/// sender.send(42).unwrap();
///
/// // Receive synchronously
/// let value = receiver.recv().unwrap();
/// assert_eq!(value, 42);
/// ```
///
/// # Async Example
///
/// ```rust
/// use linch::schannel;
///
/// # tokio_test::block_on(async {
/// let (sender, receiver) = schannel::unbounded();
///
/// // Send asynchronously - never blocks
/// sender.send_async(42).await.unwrap();
///
/// // Receive asynchronously  
/// let value = receiver.recv_async().await.unwrap();
/// assert_eq!(value, 42);
/// # });
/// ```
pub fn unbounded<T>() -> (Sender<T>, Receiver<T>) {
    let (tx, rx) = crossbeam_channel::unbounded();
    (Sender::new(tx), Receiver::new(rx))
}

/// The sending half of an schannel channel.
///
/// This struct allows sending values both synchronously and asynchronously using
/// an optimized implementation designed for high throughput scenarios.
///
/// Senders can be cloned to create multiple sending endpoints for the same channel.
/// All synchronous crossbeam channel operations are available through [`Deref`].
///
/// # Examples
///
/// ```rust
/// use linch::schannel;
///
/// let (sender, receiver) = schannel::bounded(5);
///
/// // Send synchronously
/// sender.send(42).unwrap();
///
/// // Clone the sender
/// let sender2 = sender.clone();
/// sender2.send(43).unwrap();
/// ```
#[derive(Debug)]
pub struct Sender<T> {
    tx: CrossbeamSender<T>,
}

impl<T> Clone for Sender<T> {
    fn clone(&self) -> Self {
        Self {
            tx: self.tx.clone(),
        }
    }
}

impl<T> Sender<T> {
    /// Creates a new sender from the underlying crossbeam sender.
    ///
    /// This is typically not called directly by users, but rather through
    /// [`bounded`].
    pub fn new(tx: CrossbeamSender<T>) -> Self {
        Self { tx }
    }

    /// Sends a value synchronously.
    ///
    /// This method blocks until there is space in the channel buffer or all receivers have been dropped.
    ///
    /// # Arguments
    ///
    /// * `value` - The value to send
    ///
    /// # Returns
    ///
    /// * `Ok(())` if the value was sent successfully
    /// * `Err(SendError(value))` if all receivers have been dropped
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let (sender, receiver) = schannel::bounded(1);
    /// sender.send(42).unwrap();
    /// assert_eq!(receiver.recv().unwrap(), 42);
    /// ```
    pub fn send(&self, value: T) -> Result<(), SendError<T>> {
        Ok(self.tx.send(value)?)
    }

    /// Sends a value synchronously with a timeout.
    ///
    /// This method blocks until there is space in the channel buffer, the timeout
    /// expires, or all receivers have been dropped.
    ///
    /// # Arguments
    ///
    /// * `value` - The value to send
    /// * `timeout` - The maximum duration to wait
    ///
    /// # Returns
    ///
    /// * `Ok(())` if the value was sent successfully
    /// * `Err(SendTimeoutError::Timeout(value))` if the timeout expired
    /// * `Err(SendTimeoutError::Disconnected(value))` if all receivers have been dropped
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    /// use std::time::Duration;
    ///
    /// let (sender, _receiver) = schannel::bounded(1);
    /// sender.send(1).unwrap(); // Fill the buffer
    ///
    /// // This will timeout since the buffer is full
    /// let result = sender.send_timeout(2, Duration::from_millis(10));
    /// assert!(result.is_err());
    /// ```
    pub fn send_timeout(&self, value: T, timeout: Duration) -> Result<(), SendTimeoutError<T>> {
        Ok(self.tx.send_timeout(value, timeout)?)
    }

    /// Attempts to send a value without blocking.
    ///
    /// This method returns immediately without blocking.
    ///
    /// # Arguments
    ///
    /// * `value` - The value to send
    ///
    /// # Returns
    ///
    /// * `Ok(())` if the value was sent successfully
    /// * `Err(TrySendError::Full(value))` if the channel is full
    /// * `Err(TrySendError::Disconnected(value))` if all receivers have been dropped
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let (sender, receiver) = schannel::bounded(1);
    /// sender.send(1).unwrap(); // Fill the buffer
    ///
    /// // This will fail since the buffer is full
    /// let result = sender.try_send(2);
    /// assert!(result.is_err());
    /// ```
    pub fn try_send(&self, value: T) -> Result<(), TrySendError<T>> {
        Ok(self.tx.try_send(value)?)
    }

    /// Sends a value asynchronously with a timeout.
    ///
    /// The timeout is calculated from the time this method is called, not when
    /// the future is first polled.
    ///
    /// # Arguments
    ///
    /// * `item` - The value to send
    /// * `timeout` - The maximum duration to wait
    ///
    /// # Returns
    ///
    /// A [`SendTimeoutFut`] that resolves to:
    /// * `Ok(())` if the value was sent successfully
    /// * `Err(SendTimeoutError::Timeout(item))` if the timeout expired
    /// * `Err(SendTimeoutError::Disconnected(item))` if all receivers have been dropped
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    /// use std::time::Duration;
    ///
    /// # tokio_test::block_on(async {
    /// let (sender, _receiver) = schannel::bounded(1);
    /// sender.send(1).unwrap(); // Fill the buffer
    ///
    /// // This will timeout since the buffer is full
    /// let result = sender.send_timeout_async(2, Duration::from_millis(10)).await;
    /// assert!(result.is_err());
    /// # });
    /// ```
    pub fn send_timeout_async(&self, item: T, timeout: Duration) -> SendTimeoutFut<'_, T> {
        SendTimeoutFut {
            tx: self,
            item: Some(item),
            deadline: Some(Instant::now() + timeout),
        }
    }

    /// Sends a value asynchronously.
    ///
    /// This method returns a future that will complete when there is space in the
    /// channel buffer or when all receivers have been dropped.
    ///
    /// # Arguments
    ///
    /// * `item` - The value to send
    ///
    /// # Returns
    ///
    /// A [`SendFut`] that resolves to:
    /// * `Ok(())` if the value was sent successfully
    /// * `Err(SendError(item))` if all receivers have been dropped
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// # tokio_test::block_on(async {
    /// let (sender, receiver) = schannel::bounded(1);
    /// sender.send_async(42).await.unwrap();
    /// assert_eq!(receiver.recv().unwrap(), 42);
    /// # });
    /// ```
    pub fn send_async(&self, item: T) -> SendFut<'_, T> {
        SendFut {
            fut: SendTimeoutFut {
                tx: self,
                item: Some(item),
                deadline: None,
            },
        }
    }
}

impl<T> From<CrossbeamSender<T>> for Sender<T> {
    fn from(tx: CrossbeamSender<T>) -> Self {
        Self { tx }
    }
}

/// A future representing an asynchronous send operation.
///
/// This future is created by the [`send_async`](Sender::send_async) method and will
/// complete when the value has been sent or when all receivers have been dropped.
///
/// # Examples
///
/// ```rust
/// use linch::schannel;
///
/// # tokio_test::block_on(async {
/// let (sender, receiver) = schannel::bounded(1);
/// let send_fut = sender.send_async(42);
/// send_fut.await.unwrap();
/// assert_eq!(receiver.recv().unwrap(), 42);
/// # });
/// ```
pub struct SendFut<'a, T> {
    fut: SendTimeoutFut<'a, T>,
}

impl<'a, T> SendFut<'a, T> {
    pub fn new(tx: &'a Sender<T>, item: T, timeout: Option<Duration>) -> Self {
        Self {
            fut: SendTimeoutFut {
                tx,
                item: Some(item),
                deadline: timeout.map(|d| Instant::now() + d),
            },
        }
    }
}

impl<'a, T> Future for SendFut<'a, T> {
    type Output = Result<(), SendError<T>>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        match Pin::new(&mut self.fut).poll(cx) {
            Poll::Ready(Ok(_)) => Poll::Ready(Ok(())),
            Poll::Ready(Err(SendTimeoutError::Timeout(_))) => {
                unreachable!("SendTimeoutError::Timeout should not be returned by SendFut");
            }
            Poll::Ready(Err(SendTimeoutError::Disconnected(item))) => {
                Poll::Ready(Err(SendError(item)))
            }
            Poll::Pending => Poll::Pending,
        }
    }
}

pin_project! {
    /// A future representing an asynchronous send operation with a timeout.
    ///
    /// This future is created by the [`send_timeout_async`](Sender::send_timeout_async) method and will
    /// complete when the value has been sent, the timeout expires, or when all receivers have been dropped.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    /// use std::time::Duration;
    ///
    /// # tokio_test::block_on(async {
    /// let (sender, receiver) = schannel::bounded(1);
    /// let timeout_fut = sender.send_timeout_async(42, Duration::from_secs(1));
    /// timeout_fut.await.unwrap();
    /// assert_eq!(receiver.recv().unwrap(), 42);
    /// # });
    /// ```
    pub struct SendTimeoutFut<'a, T> {
        tx: &'a Sender<T>,
        item: Option<T>,
        deadline: Option<Instant>,
    }
}

// this is not very efficient. we should not always take/replace the item.
// but in order to avoid that we would need to pin the item, send a pointer to the item,
// and then have the receiver take the item.
// this is not worth the effort right now.
impl<'a, T> Future for SendTimeoutFut<'a, T> {
    type Output = Result<(), SendTimeoutError<T>>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();

        if let Some(deadline) = this.deadline {
            if deadline <= &mut Instant::now() {
                return Poll::Ready(Err(SendTimeoutError::Timeout(this.item.take().unwrap())));
            }
        }

        match this.tx.try_send(this.item.take().unwrap()) {
            Ok(_) => Poll::Ready(Ok(())),
            Err(e) => match e {
                TrySendError::Disconnected(item) => {
                    Poll::Ready(Err(SendTimeoutError::Disconnected(item)))
                }
                TrySendError::Full(item) => {
                    this.item.replace(item);
                    cx.waker().wake_by_ref();
                    Poll::Pending
                }
            },
        }
    }
}

/// The receiving half of an schannel channel.
///
/// This struct allows receiving values both synchronously and asynchronously using
/// an optimized implementation designed for high throughput scenarios.
///
/// Receivers can be cloned to create multiple receiving endpoints for the same channel.
/// All synchronous crossbeam channel operations are available through [`Deref`].
/// The receiver can also be used as a [`Stream`] for async iteration.
///
/// # Examples
///
/// ```rust
/// use linch::schannel;
///
/// let (sender, receiver) = schannel::bounded(5);
/// sender.send(42).unwrap();
///
/// // Receive synchronously
/// let value = receiver.recv().unwrap();
/// assert_eq!(value, 42);
///
/// // Clone the receiver
/// let receiver2 = receiver.clone();
/// ```
#[derive(Debug)]
pub struct Receiver<T> {
    rx: CrossbeamReceiver<T>,
}

impl<T> Clone for Receiver<T> {
    fn clone(&self) -> Self {
        Self {
            rx: self.rx.clone(),
        }
    }
}

impl<T> Receiver<T> {
    /// Creates a new receiver from the underlying crossbeam receiver.
    ///
    /// This is typically not called directly by users, but rather through
    /// [`bounded`].
    pub fn new(rx: CrossbeamReceiver<T>) -> Self {
        Self { rx }
    }

    /// Receives a value synchronously.
    ///
    /// This method blocks until a value is available in the channel or all senders have been dropped.
    ///
    /// # Returns
    ///
    /// * `Ok(value)` if a value was received successfully
    /// * `Err(RecvError)` if all senders have been dropped and the channel is empty
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let (sender, receiver) = schannel::bounded(1);
    /// sender.send(42).unwrap();
    /// assert_eq!(receiver.recv().unwrap(), 42);
    /// ```
    pub fn recv(&self) -> Result<T, RecvError> {
        Ok(self.rx.recv()?)
    }

    /// Receives a value synchronously with a timeout.
    ///
    /// This method blocks until a value is available in the channel, the timeout
    /// expires, or all senders have been dropped.
    ///
    /// # Arguments
    ///
    /// * `timeout` - The maximum duration to wait
    ///
    /// # Returns
    ///
    /// * `Ok(value)` if a value was received successfully
    /// * `Err(RecvTimeoutError::Timeout)` if the timeout expired
    /// * `Err(RecvTimeoutError::Disconnected)` if all senders have been dropped
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    /// use std::time::Duration;
    ///
    /// let (_sender, receiver) = schannel::bounded::<i32>(1);
    ///
    /// // This will timeout since no values are sent
    /// let result = receiver.recv_timeout(Duration::from_millis(10));
    /// assert!(result.is_err());
    /// ```
    pub fn recv_timeout(&self, timeout: Duration) -> Result<T, RecvTimeoutError> {
        Ok(self.rx.recv_timeout(timeout)?)
    }

    /// Attempts to receive a value without blocking.
    ///
    /// This method returns immediately without blocking.
    ///
    /// # Returns
    ///
    /// * `Ok(value)` if a value was received successfully
    /// * `Err(TryRecvError::Empty)` if the channel is empty but not closed
    /// * `Err(TryRecvError::Disconnected)` if all senders have been dropped and the channel is empty
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let (sender, receiver) = schannel::bounded(1);
    ///
    /// // This will fail since no values are available
    /// let result = receiver.try_recv();
    /// assert!(result.is_err());
    ///
    /// sender.send(42).unwrap();
    /// assert_eq!(receiver.try_recv().unwrap(), 42);
    /// ```
    pub fn try_recv(&self) -> Result<T, TryRecvError> {
        Ok(self.rx.try_recv()?)
    }

    /// Receives a value asynchronously with a timeout.
    ///
    /// The timeout is calculated from the time this method is called, not when
    /// the future is first polled.
    ///
    /// # Arguments
    ///
    /// * `timeout` - The maximum duration to wait
    ///
    /// # Returns
    ///
    /// A [`RecvTimeoutFut`] that resolves to:
    /// * `Ok(value)` if a value was received successfully
    /// * `Err(RecvTimeoutError::Timeout)` if the timeout expired
    /// * `Err(RecvTimeoutError::Disconnected)` if all senders have been dropped
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    /// use std::time::Duration;
    ///
    /// # tokio_test::block_on(async {
    /// let (sender, receiver) = schannel::bounded(1);
    /// sender.send(42).unwrap();
    /// let timeout_fut = receiver.recv_timeout_async(Duration::from_secs(1));
    /// assert_eq!(timeout_fut.await.unwrap(), 42);
    /// # });
    /// ```
    pub fn recv_timeout_async(&self, timeout: Duration) -> RecvTimeoutFut<'_, T> {
        RecvTimeoutFut {
            rx: self,
            deadline: Some(Instant::now() + timeout),
        }
    }

    /// Receives a value asynchronously.
    ///
    /// This method returns a future that will complete when a value is available
    /// in the channel or when all senders have been dropped.
    ///
    /// # Returns
    ///
    /// A [`RecvFut`] that resolves to:
    /// * `Ok(value)` if a value was received successfully
    /// * `Err(RecvError)` if all senders have been dropped and the channel is empty
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// # tokio_test::block_on(async {
    /// let (sender, receiver) = schannel::bounded(1);
    /// sender.send(42).unwrap();
    /// assert_eq!(receiver.recv_async().await.unwrap(), 42);
    /// # });
    /// ```
    pub fn recv_async(&self) -> RecvFut<'_, T> {
        RecvFut {
            fut: RecvTimeoutFut {
                rx: self,
                deadline: None,
            },
        }
    }

    /// Converts the receiver into a stream.
    ///
    /// This allows using the receiver with async stream utilities and for iteration.
    ///
    /// # Returns
    ///
    /// A [`RecvStream`] that yields values from the channel
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    /// use futures::StreamExt;
    ///
    /// # tokio_test::block_on(async {
    /// let (sender, receiver) = schannel::bounded(3);
    /// sender.send(1).unwrap();
    /// sender.send(2).unwrap();
    /// sender.send(3).unwrap();
    /// drop(sender); // Close the channel
    ///
    /// let mut stream = receiver.into_stream();
    /// let values: Vec<_> = stream.collect().await;
    /// assert_eq!(values, vec![1, 2, 3]);
    /// # });
    /// ```
    pub fn into_stream(self) -> RecvStream<T> {
        RecvStream { rx: self }
    }
}

impl<T> From<CrossbeamReceiver<T>> for Receiver<T> {
    fn from(rx: CrossbeamReceiver<T>) -> Self {
        Self { rx }
    }
}

pin_project! {
    /// A future representing an asynchronous receive operation with a timeout.
    ///
    /// This future is created by the [`recv_timeout_async`](Receiver::recv_timeout_async) method and will
    /// complete when a value is available, the timeout expires, or when all senders have been dropped.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    /// use std::time::Duration;
    ///
    /// # tokio_test::block_on(async {
    /// let (sender, receiver) = schannel::bounded(1);
    /// sender.send(42).unwrap();
    /// let timeout_fut = receiver.recv_timeout_async(Duration::from_secs(1));
    /// assert_eq!(timeout_fut.await.unwrap(), 42);
    /// # });
    /// ```
    pub struct RecvTimeoutFut<'a, T> {
        rx: &'a Receiver<T>,
        deadline: Option<Instant>,
    }
}

impl<'a, T> Future for RecvTimeoutFut<'a, T> {
    type Output = Result<T, RecvTimeoutError>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();

        if let Some(deadline) = this.deadline {
            if deadline <= &mut Instant::now() {
                return Poll::Ready(Err(RecvTimeoutError::Timeout));
            }
        }

        match this.rx.try_recv() {
            Ok(item) => Poll::Ready(Ok(item)),
            Err(e) => match e {
                TryRecvError::Disconnected => Poll::Ready(Err(RecvTimeoutError::Disconnected)),
                TryRecvError::Empty => {
                    cx.waker().wake_by_ref();
                    Poll::Pending
                }
            },
        }
    }
}

/// A future representing an asynchronous receive operation.
///
/// This future is created by the [`recv_async`](Receiver::recv_async) method and will
/// complete when a value is available or when all senders have been dropped.
///
/// # Examples
///
/// ```rust
/// use linch::schannel;
///
/// # tokio_test::block_on(async {
/// let (sender, receiver) = schannel::bounded(1);
/// sender.send(42).unwrap();
/// let recv_fut = receiver.recv_async();
/// assert_eq!(recv_fut.await.unwrap(), 42);
/// # });
/// ```
pub struct RecvFut<'a, T> {
    fut: RecvTimeoutFut<'a, T>,
}

impl<'a, T> Future for RecvFut<'a, T> {
    type Output = Result<T, RecvError>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        match Pin::new(&mut self.fut).poll(cx) {
            Poll::Ready(Ok(item)) => Poll::Ready(Ok(item)),
            Poll::Ready(Err(RecvTimeoutError::Timeout)) => {
                unreachable!("RecvTimeoutError::Timeout should not be returned by RecvTimeoutFut");
            }
            Poll::Ready(Err(RecvTimeoutError::Disconnected)) => Poll::Ready(Err(RecvError)),
            Poll::Pending => Poll::Pending,
        }
    }
}

/// A stream that yields values from a channel receiver.
///
/// This stream is created by the [`into_stream`](Receiver::into_stream) method and implements
/// the [`Stream`] trait for async iteration over channel values.
///
/// The stream will yield values until all senders are dropped and the channel is empty.
///
/// # Examples
///
/// ```rust
/// use linch::schannel;
/// use futures::StreamExt;
///
/// # tokio_test::block_on(async {
/// let (sender, receiver) = schannel::bounded(3);
/// sender.send(1).unwrap();
/// sender.send(2).unwrap();
/// drop(sender); // Close the channel
///
/// let mut stream = receiver.into_stream();
/// let mut values = Vec::new();
/// while let Some(value) = stream.next().await {
///     values.push(value);
/// }
/// assert_eq!(values, vec![1, 2]);
/// # });
/// ```
pub struct RecvStream<T> {
    rx: Receiver<T>,
}

impl<T> Stream for RecvStream<T> {
    type Item = T;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let mut fut = self.rx.recv_async();

        match Pin::new(&mut fut).poll(cx) {
            Poll::Ready(Ok(item)) => Poll::Ready(Some(item)),
            Poll::Ready(Err(_e)) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}

/// A handle for selecting over multiple schannel operations.
///
/// This allows you to wait on multiple send or receive operations simultaneously
/// and get notified when any one of them becomes ready.
///
/// # Examples
///
/// ```rust
/// use linch::schannel;
///
/// let (tx1, rx1) = schannel::bounded(1);
/// let (tx2, rx2) = schannel::bounded(1);
///
/// tx1.send(1).unwrap();
/// tx2.send(2).unwrap();
///
/// let mut sel = schannel::Select::new();
/// let idx1 = sel.recv(&rx1);
/// let idx2 = sel.recv(&rx2);
///
/// let op = sel.select();
/// match op.index() {
///     i if i == idx1 => {
///         let value = op.recv(&rx1).unwrap();
///         println!("Received {} from first channel", value);
///     },
///     i if i == idx2 => {
///         let value = op.recv(&rx2).unwrap();
///         println!("Received {} from second channel", value);
///     },
///     _ => unreachable!(),
/// }
/// ```
pub struct Select<'a> {
    select: crossbeam_channel::Select<'a>,
}

impl<'a> Default for Select<'a> {
    fn default() -> Self {
        Self::new()
    }
}

impl<'a> Select<'a> {
    /// Creates a new select handle.
    ///
    /// Operations will be selected in a pseudo-random order to ensure fairness.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let mut sel = schannel::Select::new();
    /// ```
    pub fn new() -> Self {
        let select = crossbeam_channel::Select::new();
        Self { select }
    }

    /// Creates a new biased select handle.
    ///
    /// Operations will be selected in the order they were added, giving
    /// priority to earlier operations.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let mut sel = schannel::Select::new_biased();
    /// ```
    pub fn new_biased() -> Self {
        let select = crossbeam_channel::Select::new_biased();
        Self { select }
    }

    /// Adds a send operation to the select.
    ///
    /// Returns the index of this operation, which can be used to identify
    /// which operation was selected.
    ///
    /// # Arguments
    ///
    /// * `sender` - The sender to add to the select
    ///
    /// # Returns
    ///
    /// The index of this send operation
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let (tx, _rx) = schannel::bounded::<i32>(1);
    /// let mut sel = schannel::Select::new();
    /// let send_idx = sel.send(&tx);
    /// ```
    pub fn send<T>(&mut self, sender: &'a Sender<T>) -> usize {
        self.select.send(&sender.tx)
    }

    /// Adds a receive operation to the select.
    ///
    /// Returns the index of this operation, which can be used to identify
    /// which operation was selected.
    ///
    /// # Arguments
    ///
    /// * `receiver` - The receiver to add to the select
    ///
    /// # Returns
    ///
    /// The index of this receive operation
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let (_tx, rx) = schannel::bounded::<i32>(1);
    /// let mut sel = schannel::Select::new();
    /// let recv_idx = sel.recv(&rx);
    /// ```
    pub fn recv<T>(&mut self, receiver: &'a Receiver<T>) -> usize {
        self.select.recv(&receiver.rx)
    }

    /// Blocks until one of the operations becomes ready and returns it.
    ///
    /// # Returns
    ///
    /// A [`SelectedOperation`] that can be used to complete the operation
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let (tx, rx) = schannel::bounded(1);
    /// tx.send(42).unwrap();
    ///
    /// let mut sel = schannel::Select::new();
    /// let recv_idx = sel.recv(&rx);
    ///
    /// let op = sel.select();
    /// if op.index() == recv_idx {
    ///     let value = op.recv(&rx).unwrap();
    ///     assert_eq!(value, 42);
    /// }
    /// ```
    pub fn select(&mut self) -> SelectedOperation<'a> {
        self.select.select().into()
    }

    /// Attempts to select a ready operation without blocking.
    ///
    /// # Returns
    ///
    /// * `Ok(SelectedOperation)` if an operation was ready
    /// * `Err(TrySelectError)` if no operations were ready
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let (tx, rx) = schannel::bounded::<i32>(1);
    /// let mut sel = schannel::Select::new();
    /// let recv_idx = sel.recv(&rx);
    ///
    /// // This will return an error since no values are available
    /// assert!(sel.try_select().is_err());
    ///
    /// tx.send(42).unwrap();
    /// // Now it will succeed
    /// let op = sel.try_select().unwrap();
    /// if op.index() == recv_idx {
    ///     let value = op.recv(&rx).unwrap();
    ///     assert_eq!(value, 42);
    /// }
    /// ```
    pub fn try_select(&mut self) -> Result<SelectedOperation<'a>, TrySelectError> {
        Ok(SelectedOperation::from(self.select.try_select()?))
    }

    /// Blocks until one of the operations becomes ready or the timeout expires.
    ///
    /// # Arguments
    ///
    /// * `timeout` - The maximum duration to wait
    ///
    /// # Returns
    ///
    /// * `Ok(SelectedOperation)` if an operation became ready
    /// * `Err(SelectTimeoutError)` if the timeout expired
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    /// use std::time::Duration;
    ///
    /// let (_tx, rx) = schannel::bounded::<i32>(1);
    /// let mut sel = schannel::Select::new();
    /// sel.recv(&rx);
    ///
    /// // This will timeout since no values are available
    /// let result = sel.select_timeout(Duration::from_millis(10));
    /// assert!(result.is_err());
    /// ```
    pub fn select_timeout(
        &mut self,
        timeout: Duration,
    ) -> Result<SelectedOperation<'a>, SelectTimeoutError> {
        Ok(SelectedOperation::from(
            self.select.select_timeout(timeout)?,
        ))
    }

    /// Returns the index of a ready operation, if any.
    ///
    /// This is a non-blocking operation that returns immediately.
    ///
    /// # Returns
    ///
    /// The index of a ready operation, or a value indicating no operations are ready
    pub fn ready(&mut self) -> usize {
        self.select.ready()
    }

    /// Returns the index of a ready operation, waiting up to the timeout.
    ///
    /// # Arguments
    ///
    /// * `timeout` - The maximum duration to wait
    ///
    /// # Returns
    ///
    /// * `Ok(index)` if an operation became ready
    /// * `Err(ReadyTimeoutError)` if the timeout expired
    pub fn ready_timeout(&mut self, timeout: Duration) -> Result<usize, ReadyTimeoutError> {
        Ok(self.select.ready_timeout(timeout)?)
    }

    /// Removes an operation from the select.
    ///
    /// # Arguments
    ///
    /// * `index` - The index of the operation to remove
    pub fn remove(&mut self, index: usize) {
        self.select.remove(index);
    }
}

/// A selected operation that is ready to complete.
///
/// This struct represents an operation that was selected by a [`Select`] and is
/// ready to be executed. You can use the [`index`](SelectedOperation::index) method
/// to determine which operation was selected, and then call the appropriate
/// [`send`](SelectedOperation::send) or [`recv`](SelectedOperation::recv) method.
///
/// # Examples
///
/// ```rust
/// use linch::schannel;
///
/// let (tx, rx) = schannel::bounded(1);
/// tx.send(42).unwrap();
///
/// let mut sel = schannel::Select::new();
/// let recv_idx = sel.recv(&rx);
///
/// let op = sel.select();
/// if op.index() == recv_idx {
///     let value = op.recv(&rx).unwrap();
///     assert_eq!(value, 42);
/// }
/// ```
pub struct SelectedOperation<'a>(crossbeam_channel::SelectedOperation<'a>);

impl<'a> From<crossbeam_channel::SelectedOperation<'a>> for SelectedOperation<'a> {
    fn from(value: crossbeam_channel::SelectedOperation<'a>) -> Self {
        Self(value)
    }
}

impl<'a> SelectedOperation<'a> {
    /// Returns the index of the selected operation.
    ///
    /// This index corresponds to the value returned by the [`send`](Select::send)
    /// or [`recv`](Select::recv) method that added this operation to the select.
    ///
    /// # Returns
    ///
    /// The index of the selected operation
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let (tx, rx) = schannel::bounded::<i32>(1);
    /// tx.send(42).unwrap();
    ///
    /// let mut sel = schannel::Select::new();
    /// let recv_idx = sel.recv(&rx);
    ///
    /// let op = sel.select();
    /// assert_eq!(op.index(), recv_idx);
    /// let value = op.recv(&rx).unwrap();
    /// assert_eq!(value, 42);
    /// ```
    pub fn index(&self) -> usize {
        self.0.index()
    }

    /// Completes the selected send operation.
    ///
    /// This method should only be called if the selected operation was a send operation.
    ///
    /// # Arguments
    ///
    /// * `sender` - The sender that was selected
    /// * `msg` - The message to send
    ///
    /// # Returns
    ///
    /// * `Ok(())` if the message was sent successfully
    /// * `Err(SendError(msg))` if all receivers have been dropped
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let (tx, rx) = schannel::bounded(1);
    /// let mut sel = schannel::Select::new();
    /// let send_idx = sel.send(&tx);
    ///
    /// let op = sel.select();
    /// if op.index() == send_idx {
    ///     op.send(&tx, 42).unwrap();
    ///     assert_eq!(rx.recv().unwrap(), 42);
    /// }
    /// ```
    pub fn send<T>(self, sender: &'a Sender<T>, msg: T) -> Result<(), SendError<T>> {
        Ok(self.0.send(&sender.tx, msg)?)
    }

    /// Completes the selected receive operation.
    ///
    /// This method should only be called if the selected operation was a receive operation.
    ///
    /// # Arguments
    ///
    /// * `receiver` - The receiver that was selected
    ///
    /// # Returns
    ///
    /// * `Ok(value)` if a value was received successfully
    /// * `Err(RecvError)` if all senders have been dropped and the channel is empty
    ///
    /// # Examples
    ///
    /// ```rust
    /// use linch::schannel;
    ///
    /// let (tx, rx) = schannel::bounded(1);
    /// tx.send(42).unwrap();
    ///
    /// let mut sel = schannel::Select::new();
    /// let recv_idx = sel.recv(&rx);
    ///
    /// let op = sel.select();
    /// if op.index() == recv_idx {
    ///     let value = op.recv(&rx).unwrap();
    ///     assert_eq!(value, 42);
    /// }
    /// ```
    pub fn recv<T>(self, receiver: &'a Receiver<T>) -> Result<T, RecvError> {
        Ok(self.0.recv(&receiver.rx)?)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crossbeam_channel;
    use std::{thread, time::Duration};
    use tokio::time::timeout;

    fn create_bounded_channel<T>(capacity: usize) -> (Sender<T>, Receiver<T>) {
        let (tx, rx) = crossbeam_channel::bounded(capacity);
        (Sender::new(tx), Receiver::new(rx))
    }

    // Test 1: Synchronous send, synchronous receive
    #[test]
    fn test_sync_send_sync_recv_success() {
        let (sender, receiver) = create_bounded_channel(1);

        // Send synchronously
        sender.send(42).unwrap();

        // Receive synchronously
        let received = receiver.recv().unwrap();
        assert_eq!(received, 42);
    }

    #[test]
    fn test_sync_send_sync_recv_with_timeout_success() {
        let (sender, receiver) = create_bounded_channel(1);

        // Send synchronously
        sender.send(42).unwrap();

        // Receive synchronously with timeout
        let received = receiver.recv_timeout(Duration::from_millis(100)).unwrap();
        assert_eq!(received, 42);
    }

    #[test]
    fn test_sync_send_sync_recv_timeout() {
        let (_sender, receiver) = create_bounded_channel::<i32>(1);

        // Try to receive with timeout - should timeout since nothing was sent
        let result = receiver.recv_timeout(Duration::from_millis(10));
        assert!(matches!(result, Err(RecvTimeoutError::Timeout)));
    }

    #[test]
    fn test_sync_send_sync_recv_disconnected() {
        let (sender, receiver) = create_bounded_channel::<i32>(1);

        // Drop sender to simulate disconnection
        drop(sender);

        // Try to receive - should return disconnected error
        let result = receiver.recv();
        assert!(matches!(result, Err(RecvError)));
    }

    // Test 2: Synchronous send, asynchronous receive
    #[tokio::test]
    async fn test_sync_send_async_recv_success() {
        let (sender, receiver) = create_bounded_channel(1);

        // Send synchronously in a separate thread
        let sender_clone = sender.clone();
        thread::spawn(move || {
            thread::sleep(Duration::from_millis(10));
            sender_clone.send(42).unwrap();
        });

        // Receive asynchronously
        let received = receiver.recv_async().await.unwrap();
        assert_eq!(received, 42);
    }

    #[tokio::test]
    async fn test_sync_send_async_recv_with_timeout_success() {
        let (sender, receiver) = create_bounded_channel(1);

        // Send synchronously in a separate thread
        let sender_clone = sender.clone();
        thread::spawn(move || {
            thread::sleep(Duration::from_millis(10));
            sender_clone.send(42).unwrap();
        });

        // Receive asynchronously with timeout
        let received = receiver
            .recv_timeout_async(Duration::from_millis(100))
            .await
            .unwrap();
        assert_eq!(received, 42);
    }

    #[tokio::test]
    async fn test_sync_send_async_recv_timeout() {
        let (_sender, receiver) = create_bounded_channel::<i32>(1);

        // Try to receive with timeout - should timeout since nothing was sent
        let result = receiver.recv_timeout_async(Duration::from_millis(10)).await;
        assert!(matches!(result, Err(RecvTimeoutError::Timeout)));
    }

    #[tokio::test]
    async fn test_sync_send_async_recv_disconnected() {
        let (sender, receiver) = create_bounded_channel::<i32>(1);

        // Drop sender to simulate disconnection
        drop(sender);

        // Try to receive asynchronously - should return disconnected error
        let result = receiver.recv_async().await;
        assert!(matches!(result, Err(RecvError)));
    }

    // Test 3: Asynchronous send, synchronous receive
    #[tokio::test]
    async fn test_async_send_sync_recv_success() {
        let (sender, receiver) = create_bounded_channel(1);

        // Send asynchronously and receive synchronously in separate tasks
        let send_task = tokio::spawn(async move {
            sender.send_async(42).await.unwrap();
            // rx may have already dropped, so we need to ignore the error
            let _ = sender.send_async(43).await;
        });

        let recv_task = tokio::spawn(async move {
            // Small delay to ensure send happens first
            tokio::time::sleep(Duration::from_millis(10)).await;
            receiver.recv().unwrap()
        });

        let (_, received) = tokio::try_join!(send_task, recv_task).unwrap();
        assert_eq!(received, 42);
    }

    #[tokio::test]
    async fn test_async_send_sync_recv_with_timeout_success() {
        let (sender, receiver) = create_bounded_channel(1);

        // Send asynchronously
        let send_task = tokio::spawn(async move {
            sender.send_async(42).await.unwrap();
            // rx may have already dropped, so we need to ignore the error
            let _ = sender.send_async(43).await;
        });

        let recv_task = tokio::spawn(async move {
            // Small delay to ensure send happens first
            tokio::time::sleep(Duration::from_millis(10)).await;
            receiver.recv_timeout(Duration::from_millis(100)).unwrap()
        });

        let (_, received) = tokio::try_join!(send_task, recv_task).unwrap();
        assert_eq!(received, 42);
    }

    #[tokio::test]
    async fn test_async_send_sync_recv_timeout() {
        let (_sender, receiver) = create_bounded_channel::<i32>(1);

        // Try to receive with timeout - should timeout since nothing was sent
        let result =
            tokio::task::spawn_blocking(move || receiver.recv_timeout(Duration::from_millis(10)))
                .await
                .unwrap();

        assert!(matches!(result, Err(RecvTimeoutError::Timeout)));
    }

    #[tokio::test]
    async fn test_async_send_with_timeout_success() {
        let (sender, receiver) = create_bounded_channel(1);

        // Send with timeout should succeed when channel has capacity
        let result = sender
            .send_timeout_async(42, Duration::from_millis(100))
            .await;
        assert!(result.is_ok());

        // Verify the message was sent
        let received = receiver.recv().unwrap();
        assert_eq!(received, 42);
    }

    #[tokio::test]
    async fn test_async_send_timeout() {
        let (sender, _receiver) = create_bounded_channel(1);

        // Fill the channel
        sender.send(1).unwrap();

        // Try to send with a short timeout - should timeout
        let result = sender
            .send_timeout_async(2, Duration::from_millis(10))
            .await;
        assert!(matches!(result, Err(SendTimeoutError::Timeout(_))));
    }

    #[tokio::test]
    async fn test_async_send_disconnected() {
        let (sender, receiver) = create_bounded_channel::<i32>(1);

        // Drop receiver to simulate disconnection
        drop(receiver);

        // Try to send asynchronously - should return disconnected error
        let result = sender.send_async(42).await;
        assert!(matches!(result, Err(SendError(_))));
    }

    // Test 4: Asynchronous send, asynchronous receive
    #[tokio::test]
    async fn test_async_send_async_recv_success() {
        let (sender, receiver) = create_bounded_channel(1);

        // Send and receive asynchronously
        let send_task = tokio::spawn(async move {
            sender.send_async(42).await.unwrap();
        });

        let recv_task = tokio::spawn(async move { receiver.recv_async().await.unwrap() });

        let (_, received) = tokio::try_join!(send_task, recv_task).unwrap();
        assert_eq!(received, 42);
    }

    #[tokio::test]
    async fn test_async_send_async_recv_with_timeout_success() {
        let (sender, receiver) = create_bounded_channel(1);

        // Send asynchronously
        let send_task = tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(10)).await;
            sender.send_async(42).await.unwrap();
        });

        let recv_task = tokio::spawn(async move {
            receiver
                .recv_timeout_async(Duration::from_millis(100))
                .await
                .unwrap()
        });

        let (_, received) = tokio::try_join!(send_task, recv_task).unwrap();
        assert_eq!(received, 42);
    }

    #[tokio::test]
    async fn test_async_send_async_recv_timeout() {
        let (_sender, receiver) = create_bounded_channel::<i32>(1);

        // Try to receive with timeout - should timeout since nothing was sent
        let result = receiver.recv_timeout_async(Duration::from_millis(10)).await;
        assert!(matches!(result, Err(RecvTimeoutError::Timeout)));
    }

    #[tokio::test]
    async fn test_async_send_async_recv_both_timeout() {
        let (sender, _receiver) = create_bounded_channel(1);

        // Fill the channel
        sender.send(1).unwrap();

        // Try to send with timeout (should timeout)
        let send_result = sender
            .send_timeout_async(2, Duration::from_millis(10))
            .await;
        assert!(matches!(send_result, Err(SendTimeoutError::Timeout(_))));

        // Try to receive from empty channel with timeout (should timeout)
        let (_sender2, receiver2) = create_bounded_channel::<i32>(1);
        let recv_result = receiver2
            .recv_timeout_async(Duration::from_millis(10))
            .await;
        assert!(matches!(recv_result, Err(RecvTimeoutError::Timeout)));
    }

    #[tokio::test]
    async fn test_async_send_async_recv_disconnected() {
        let (sender, receiver) = create_bounded_channel::<i32>(1);

        // Test send disconnection
        drop(receiver);
        let send_result = sender.send_async(42).await;
        assert!(matches!(send_result, Err(SendError(_))));

        // Test receive disconnection
        let (sender2, receiver2) = create_bounded_channel::<i32>(1);
        drop(sender2);
        let recv_result = receiver2.recv_async().await;
        assert!(matches!(recv_result, Err(RecvError)));
    }

    // Additional edge case tests
    #[tokio::test]
    async fn test_multiple_senders_single_receiver() {
        let (sender, receiver) = create_bounded_channel(1);

        let sender1 = sender.clone();
        let sender2 = sender.clone();

        // Send from multiple senders
        let send_task1 = tokio::spawn(async move {
            sender1.send_async(1).await.unwrap();
        });

        let send_task2 = tokio::spawn(async move {
            sender2.send_async(2).await.unwrap();
        });

        // Receive both messages
        let recv_task = tokio::spawn(async move {
            let mut received = vec![];
            received.push(receiver.recv_async().await.unwrap());
            received.push(receiver.recv_async().await.unwrap());
            received.sort(); // Sort since order is not guaranteed
            received
        });

        let (_, _, received) = tokio::try_join!(send_task1, send_task2, recv_task).unwrap();
        assert_eq!(received, vec![1, 2]);
    }

    #[tokio::test]
    async fn test_bounded_channel_backpressure() {
        let (sender, receiver) = create_bounded_channel(2);

        // Fill the channel to capacity
        sender.send(1).unwrap();
        sender.send(2).unwrap();

        // This send should block/timeout since channel is full
        let send_result = timeout(Duration::from_millis(10), sender.send_async(3)).await;

        assert!(send_result.is_err()); // Should timeout

        // Receive one item to make space
        let received = receiver.recv().unwrap();
        assert_eq!(received, 1);

        // Now the send should succeed
        let send_result = sender.send_async(3).await;
        assert!(send_result.is_ok());
    }

    #[test]
    fn test_sender_receiver_debug_clone() {
        let (sender, receiver) = create_bounded_channel::<i32>(1);

        // Test Debug implementation
        let debug_str = format!("{:?}", sender);
        assert!(debug_str.contains("Sender"));

        let debug_str = format!("{:?}", receiver);
        assert!(debug_str.contains("Receiver"));

        // Test Clone implementation for Sender
        let _sender_clone = sender.clone();

        // Test Clone implementation for Receiver
        let _receiver_clone = receiver.clone();
    }

    #[test]
    fn test_deref_implementations() {
        let (sender, receiver) = create_bounded_channel(1);

        // Test that we can use CrossbeamSender methods directly
        sender.send(42).unwrap();

        // Test that we can use CrossbeamReceiver methods directly
        let received = receiver.recv().unwrap();
        assert_eq!(received, 42);

        // Test try_send and try_recv
        let result = sender.try_send(43);
        assert!(result.is_ok());

        let result = receiver.try_recv();
        assert_eq!(result.unwrap(), 43);
    }

    // ===== SELECT TESTS =====

    #[test]
    fn test_select_basic_recv() {
        let (tx1, rx1) = create_bounded_channel(1);
        let (tx2, rx2) = create_bounded_channel(1);

        tx1.send(1).unwrap();
        tx2.send(2).unwrap();

        let mut sel = Select::new();
        let idx1 = sel.recv(&rx1);
        let idx2 = sel.recv(&rx2);

        let op = sel.select();
        match op.index() {
            i if i == idx1 => {
                let value = op.recv(&rx1).unwrap();
                assert_eq!(value, 1);
            }
            i if i == idx2 => {
                let value = op.recv(&rx2).unwrap();
                assert_eq!(value, 2);
            }
            _ => panic!("Unexpected index"),
        }
    }

    #[test]
    fn test_select_basic_send() {
        let (tx1, rx1) = create_bounded_channel(1);
        let (tx2, rx2) = create_bounded_channel(1);

        let mut sel = Select::new();
        let idx1 = sel.send(&tx1);
        let idx2 = sel.send(&tx2);

        let op = sel.select();
        match op.index() {
            i if i == idx1 => {
                op.send(&tx1, 1).unwrap();
                assert_eq!(rx1.recv().unwrap(), 1);
            }
            i if i == idx2 => {
                op.send(&tx2, 2).unwrap();
                assert_eq!(rx2.recv().unwrap(), 2);
            }
            _ => panic!("Unexpected index"),
        }
    }

    #[test]
    fn test_select_mixed_operations() {
        let (tx1, rx1) = create_bounded_channel(1);
        let (tx2, rx2) = create_bounded_channel(1);

        // Send to first channel
        tx1.send(1).unwrap();

        let mut sel = Select::new();
        let recv_idx = sel.recv(&rx1);
        let send_idx = sel.send(&tx2);

        let op = sel.select();
        match op.index() {
            i if i == recv_idx => {
                let value = op.recv(&rx1).unwrap();
                assert_eq!(value, 1);
            }
            i if i == send_idx => {
                op.send(&tx2, 2).unwrap();
                assert_eq!(rx2.recv().unwrap(), 2);
            }
            _ => panic!("Unexpected index"),
        }
    }

    #[test]
    fn test_select_try_select() {
        let (tx, rx) = create_bounded_channel::<i32>(1);

        let mut sel = Select::new();
        let recv_idx = sel.recv(&rx);

        // Should fail since no values are available
        assert!(sel.try_select().is_err());

        // Send a value
        tx.send(42).unwrap();

        // Now it should succeed
        let op = sel.try_select().unwrap();
        assert_eq!(op.index(), recv_idx);
        let value = op.recv(&rx).unwrap();
        assert_eq!(value, 42);
    }

    #[test]
    fn test_select_timeout() {
        let (_tx, rx) = create_bounded_channel::<i32>(1);

        let mut sel = Select::new();
        sel.recv(&rx);

        // Should timeout since no values are available
        let result = sel.select_timeout(Duration::from_millis(10));
        assert!(result.is_err());
    }

    #[test]
    fn test_select_timeout_success() {
        let (tx, rx) = create_bounded_channel(1);

        // Send a value in a separate thread
        let tx_clone = tx.clone();
        thread::spawn(move || {
            thread::sleep(Duration::from_millis(50));
            tx_clone.send(42).unwrap();
        });

        let mut sel = Select::new();
        let recv_idx = sel.recv(&rx);

        // Should succeed within timeout
        let result = sel.select_timeout(Duration::from_millis(100));
        assert!(result.is_ok());

        let op = result.unwrap();
        assert_eq!(op.index(), recv_idx);
        let value = op.recv(&rx).unwrap();
        assert_eq!(value, 42);
    }

    #[test]
    fn test_select_ready() {
        let (tx, rx) = create_bounded_channel(1);

        let mut sel = Select::new();
        let recv_idx = sel.recv(&rx);

        // Send a value first
        tx.send(42).unwrap();

        // Now the receive operation should be ready
        assert_eq!(sel.ready(), recv_idx);
    }

    #[test]
    fn test_select_ready_timeout() {
        let (_tx, rx) = create_bounded_channel::<i32>(1);

        let mut sel = Select::new();
        sel.recv(&rx);

        // Should timeout since no values are available
        let result = sel.ready_timeout(Duration::from_millis(10));
        assert!(result.is_err());
    }

    #[test]
    fn test_select_ready_timeout_success() {
        let (tx, rx) = create_bounded_channel(1);

        // Send a value in a separate thread
        let tx_clone = tx.clone();
        thread::spawn(move || {
            thread::sleep(Duration::from_millis(50));
            tx_clone.send(42).unwrap();
        });

        let mut sel = Select::new();
        let recv_idx = sel.recv(&rx);

        // Should succeed within timeout
        let result = sel.ready_timeout(Duration::from_millis(100));
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), recv_idx);
    }

    #[test]
    fn test_select_remove() {
        let (tx1, rx1) = create_bounded_channel(1);
        let (tx2, rx2) = create_bounded_channel(1);

        let mut sel = Select::new();
        let idx1 = sel.recv(&rx1);
        let idx2 = sel.recv(&rx2);

        // Remove the first operation
        sel.remove(idx1);

        // Send to both channels
        tx1.send(1).unwrap();
        tx2.send(2).unwrap();

        // Only the second operation should be selected
        let op = sel.select();
        assert_eq!(op.index(), idx2);
        let value = op.recv(&rx2).unwrap();
        assert_eq!(value, 2);
    }

    #[test]
    fn test_select_biased() {
        let (tx1, rx1) = create_bounded_channel(1);
        let (tx2, rx2) = create_bounded_channel(1);

        tx1.send(1).unwrap();
        tx2.send(2).unwrap();

        let mut sel = Select::new_biased();
        let idx1 = sel.recv(&rx1);
        let _idx2 = sel.recv(&rx2);

        // With biased selection, the first operation should be selected
        let op = sel.select();
        assert_eq!(op.index(), idx1);
        let value = op.recv(&rx1).unwrap();
        assert_eq!(value, 1);
    }

    #[test]
    fn test_select_multiple_channels() {
        let (tx1, rx1) = create_bounded_channel(1);
        let (tx2, rx2) = create_bounded_channel(1);
        let (tx3, rx3) = create_bounded_channel(1);

        tx1.send(1).unwrap();
        tx2.send(2).unwrap();
        tx3.send(3).unwrap();

        let mut sel = Select::new();
        let idx1 = sel.recv(&rx1);
        let idx2 = sel.recv(&rx2);
        let idx3 = sel.recv(&rx3);

        let mut received = Vec::new();

        // Select all three operations
        for _ in 0..3 {
            let op = sel.select();
            match op.index() {
                i if i == idx1 => {
                    let value = op.recv(&rx1).unwrap();
                    received.push(value);
                }
                i if i == idx2 => {
                    let value = op.recv(&rx2).unwrap();
                    received.push(value);
                }
                i if i == idx3 => {
                    let value = op.recv(&rx3).unwrap();
                    received.push(value);
                }
                _ => panic!("Unexpected index"),
            }
        }

        received.sort();
        assert_eq!(received, vec![1, 2, 3]);
    }

    #[test]
    fn test_select_send_blocking() {
        let (tx1, rx1) = create_bounded_channel(1);
        let (tx2, rx2) = create_bounded_channel(1);

        // Fill both channels
        tx1.send(1).unwrap();
        tx2.send(2).unwrap();

        let mut sel = Select::new();
        let _send_idx1 = sel.send(&tx1);
        let _send_idx2 = sel.send(&tx2);

        // Both sends should block, but we can still select
        // This will block until one of the receivers consumes a value
        let tx1_clone = tx1.clone();
        let tx2_clone = tx2.clone();
        let rx1_clone = rx1.clone();
        let _rx2_clone = rx2.clone();

        let sender_handle = thread::spawn(move || {
            let mut sel = Select::new();
            let send_idx1 = sel.send(&tx1_clone);
            let send_idx2 = sel.send(&tx2_clone);

            let op = sel.select();
            match op.index() {
                i if i == send_idx1 => {
                    op.send(&tx1_clone, 3).unwrap();
                }
                i if i == send_idx2 => {
                    op.send(&tx2_clone, 4).unwrap();
                }
                _ => panic!("Unexpected index"),
            }
        });

        // Consume a value to unblock one of the sends
        thread::sleep(Duration::from_millis(10));
        let _ = rx1_clone.recv().unwrap();

        sender_handle.join().unwrap();

        // Verify the new value was sent
        let value = rx1.recv().unwrap();
        assert_eq!(value, 3);
    }

    #[test]
    fn test_select_receive_blocking() {
        let (tx1, rx1) = create_bounded_channel::<i32>(1);
        let (tx2, rx2) = create_bounded_channel::<i32>(1);

        let mut sel = Select::new();
        let _recv_idx1 = sel.recv(&rx1);
        let _recv_idx2 = sel.recv(&rx2);

        // Both receives should block, but we can still select
        // This will block until one of the senders sends a value
        let _tx1_clone = tx1.clone();
        let _tx2_clone = tx2.clone();
        let rx1_clone = rx1.clone();
        let rx2_clone = rx2.clone();

        let receiver_handle = thread::spawn(move || {
            let mut sel = Select::new();
            let recv_idx1 = sel.recv(&rx1_clone);
            let recv_idx2 = sel.recv(&rx2_clone);

            let op = sel.select();
            match op.index() {
                i if i == recv_idx1 => {
                    let value = op.recv(&rx1_clone).unwrap();
                    value
                }
                i if i == recv_idx2 => {
                    let value = op.recv(&rx2_clone).unwrap();
                    value
                }
                _ => panic!("Unexpected index"),
            }
        });

        // Send a value to unblock one of the receives
        thread::sleep(Duration::from_millis(10));
        tx1.send(42).unwrap();

        let received_value = receiver_handle.join().unwrap();
        assert_eq!(received_value, 42);
    }

    #[test]
    fn test_select_disconnected_send() {
        let (tx, rx) = create_bounded_channel::<i32>(1);

        // Drop the receiver
        drop(rx);

        let mut sel = Select::new();
        let send_idx = sel.send(&tx);

        let op = sel.select();
        assert_eq!(op.index(), send_idx);

        // Send should fail with disconnected error
        let result = op.send(&tx, 42);
        assert!(result.is_err());
    }

    #[test]
    fn test_select_disconnected_recv() {
        let (tx, rx) = create_bounded_channel::<i32>(1);

        // Drop the sender
        drop(tx);

        let mut sel = Select::new();
        let recv_idx = sel.recv(&rx);

        let op = sel.select();
        assert_eq!(op.index(), recv_idx);

        // Receive should fail with disconnected error
        let result = op.recv(&rx);
        assert!(result.is_err());
    }

    #[test]
    fn test_select_default() {
        let (tx, rx) = create_bounded_channel(1);
        tx.send(42).unwrap();

        let mut sel = Select::default(); // Test Default implementation
        let recv_idx = sel.recv(&rx);

        let op = sel.select();
        assert_eq!(op.index(), recv_idx);
        let value = op.recv(&rx).unwrap();
        assert_eq!(value, 42);
    }

    #[test]
    fn test_select_complex_scenario() {
        let (tx1, rx1) = create_bounded_channel(1);
        let (tx2, rx2) = create_bounded_channel(1);
        let (tx3, rx3) = create_bounded_channel(1);

        // Send to first two channels
        tx1.send(1).unwrap();
        tx2.send(2).unwrap();

        let mut sel = Select::new();
        let recv_idx1 = sel.recv(&rx1);
        let recv_idx2 = sel.recv(&rx2);
        let send_idx3 = sel.send(&tx3);

        let mut received = Vec::new();

        // First selection should be one of the receives
        let op = sel.select();
        match op.index() {
            i if i == recv_idx1 => {
                let value = op.recv(&rx1).unwrap();
                received.push(value);
            }
            i if i == recv_idx2 => {
                let value = op.recv(&rx2).unwrap();
                received.push(value);
            }
            _ => panic!("Unexpected index for first selection"),
        }

        // Second selection should be the other receive
        let op = sel.select();
        match op.index() {
            i if i == recv_idx1 => {
                let value = op.recv(&rx1).unwrap();
                received.push(value);
            }
            i if i == recv_idx2 => {
                let value = op.recv(&rx2).unwrap();
                received.push(value);
            }
            _ => panic!("Unexpected index for second selection"),
        }

        // Third selection should be the send
        let op = sel.select();
        assert_eq!(op.index(), send_idx3);
        op.send(&tx3, 3).unwrap();

        // Verify all operations completed correctly
        received.sort();
        assert_eq!(received, vec![1, 2]);
        assert_eq!(rx3.recv().unwrap(), 3);
    }
}