linch 0.6.3

//! # Linch - High-Performance Async/Sync Channel
//!
//! Linch is a high-performance channel implementation that supports both synchronous
//! and asynchronous operations. It's designed for scenarios where you need maximum
//! throughput and low latency with mixed sync/async workloads.
//!
//! The name "linch" is short for "**lined channel**" - like a concrete-lined channel
//! that provides a smooth, reliable pathway for water flow. Similarly, Linch provides
//! a smooth, reliable pathway for data flow between different parts of your application.
//!
//! ## Features
//!
//! - **Mixed Sync/Async**: Send and receive both synchronously and asynchronously
//! - **High Performance**: Optimized for throughput and low latency
//! - **Select Operations**: Support for selecting over multiple channels
//! - **Multiple Implementations**: Choose between the main channel and schannel variants
//!
//! ## Basic Usage
//!
//! ```rust
//! use linch::bounded;
//!
//! // Create a bounded channel with capacity 10
//! let (sender, receiver) = bounded(10);
//!
//! // Send synchronously
//! sender.send(42).unwrap();
//!
//! // Receive synchronously
//! let value = receiver.recv().unwrap();
//! assert_eq!(value, 42);
//! ```
//!
//! ## Async Usage
//!
//! ```rust
//! use linch::bounded;
//!
//! # tokio_test::block_on(async {
//! let (sender, receiver) = bounded(10);
//!
//! // Send asynchronously
//! sender.send_async(42).await.unwrap();
//!
//! // Receive asynchronously
//! let value = receiver.recv_async().await.unwrap();
//! assert_eq!(value, 42);
//! # });
//! ```
//!
//! ## Select Operations
//!
//! ```rust
//! use linch::{bounded, Select};
//!
//! # tokio_test::block_on(async {
//! let (tx1, rx1) = bounded(1);
//! let (tx2, rx2) = bounded(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();
//!         println!("Received {} from first channel", value);
//!     },
//!     i if i == idx2 => {
//!         let value = op.recv(&rx2).unwrap();
//!         println!("Received {} from second channel", value);
//!     },
//!     _ => unreachable!(),
//! }
//! # });
//! ```

mod error;
mod inner;
mod mutex;
mod receiver;
pub mod schannel;
mod select;
mod sender;
mod util;

use std::sync::Arc;

pub use error::{
    ReadyTimeoutError, RecvError, RecvTimeoutError, SelectTimeoutError, SendError,
    SendTimeoutError, TryRecvError, TrySelectError, TrySendError,
};
pub use receiver::{Receiver, RecvFut, RecvStream};
pub use select::{Select, SelectedOperation};
pub use sender::{SendFut, Sender};

// Re-export schannel types for convenience
pub use schannel::{Select as SchannelSelect, SelectedOperation as SchannelSelectedOperation};

use crate::inner::Inner;

/// Creates a bounded channel with the specified capacity.
///
/// This function creates a pair of [`Sender`] and [`Receiver`] that can be used
/// to send and receive values both synchronously and asynchronously.
///
/// # Arguments
///
/// * `cap` - The capacity of the channel buffer. Must be greater than 0.
///
/// # Returns
///
/// A tuple containing a [`Sender`] and [`Receiver`] pair.
///
/// # Panics
///
/// Panics if `cap` is 0.
///
/// # Examples
///
/// ```rust
/// use linch::bounded;
///
/// // Create a channel with capacity 5
/// let (sender, receiver) = bounded(5);
///
/// // Send a value
/// sender.send("Hello").unwrap();
///
/// // Receive the value
/// let msg = receiver.recv().unwrap();
/// assert_eq!(msg, "Hello");
/// ```
///
/// # Async Example
///
/// ```rust
/// use linch::bounded;
///
/// # tokio_test::block_on(async {
/// let (sender, receiver) = bounded(5);
///
/// // Send asynchronously
/// sender.send_async("Hello").await.unwrap();
///
/// // Receive asynchronously
/// let msg = receiver.recv_async().await.unwrap();
/// assert_eq!(msg, "Hello");
/// # });
/// ```
pub fn bounded<T>(cap: usize) -> (Sender<T>, Receiver<T>) {
    assert!(cap > 0);
    let (tx, rx) = crossbeam_channel::bounded(cap);
    let inner = Arc::new(Inner::default());
    (Sender::new(tx, inner.clone()), Receiver::new(rx, inner))
}

/// Creates an unbounded channel.
///
/// This function creates a pair of [`Sender`] and [`Receiver`] that can be used
/// to send and receive values both synchronously and asynchronously.
///
/// 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::unbounded;
///
/// // Create an unbounded channel
/// let (sender, receiver) = unbounded();
///
/// // Send a value - never blocks
/// sender.send("Hello").unwrap();
///
/// // Receive the value
/// let msg = receiver.recv().unwrap();
/// assert_eq!(msg, "Hello");
/// ```
///
/// # Async Example
///
/// ```rust
/// use linch::unbounded;
///
/// # tokio_test::block_on(async {
/// let (sender, receiver) = unbounded();
///
/// // Send asynchronously - never blocks
/// sender.send_async("Hello").await.unwrap();
///
/// // Receive asynchronously
/// let msg = receiver.recv_async().await.unwrap();
/// assert_eq!(msg, "Hello");
/// # });
/// ```
pub fn unbounded<T>() -> (Sender<T>, Receiver<T>) {
    let (tx, rx) = crossbeam_channel::unbounded();
    let inner = Arc::new(Inner::default());
    (Sender::new(tx, inner.clone()), Receiver::new(rx, inner))
}

#[cfg(test)]
mod tests {
    use crate::bounded;
    use futures::{StreamExt, TryStreamExt};
    use std::thread;
    use std::time::Duration;
    use tokio::time::timeout;

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

        // Test basic sync send/receive
        sender.send(42).unwrap();
        let received = receiver.recv().unwrap();
        assert_eq!(received, 42);
    }

    #[test]
    fn test_send_receive_sync_threaded() {
        let (sender, receiver) = bounded::<String>(1);

        let sender_handle = thread::spawn(move || {
            for i in 0..5 {
                sender.send(format!("message_{}", i)).unwrap();
                thread::sleep(Duration::from_millis(10));
            }
        });

        let receiver_handle = thread::spawn(move || {
            let mut messages = Vec::new();
            for _ in 0..5 {
                messages.push(receiver.recv().unwrap());
            }
            messages
        });

        sender_handle.join().unwrap();
        let messages = receiver_handle.join().unwrap();

        assert_eq!(messages.len(), 5);
        for (i, msg) in messages.iter().enumerate() {
            assert_eq!(msg, &format!("message_{}", i));
        }
    }

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

        // Test basic async send/receive
        sender.send_async(42).await.unwrap();
        let received = receiver.recv_async().await.unwrap();
        assert_eq!(received, 42);
    }

    #[tokio::test]
    async fn test_send_receive_async_concurrent() {
        let (sender, receiver) = bounded::<String>(1);

        let sender_task = tokio::spawn(async move {
            for i in 0..5 {
                sender
                    .send_async(format!("async_message_{}", i))
                    .await
                    .unwrap();
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        });

        let receiver_task = tokio::spawn(async move {
            let mut messages = Vec::new();
            for _ in 0..5 {
                messages.push(receiver.recv_async().await.unwrap());
            }
            messages
        });

        sender_task.await.unwrap();
        let messages = receiver_task.await.unwrap();

        assert_eq!(messages.len(), 5);
        for (i, msg) in messages.iter().enumerate() {
            assert_eq!(msg, &format!("async_message_{}", i));
        }
    }

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

        // Send synchronously in a blocking task
        let _ = std::thread::spawn(move || {
            for i in 0..5 {
                sender.send(i).unwrap();
            }
        });

        // Receive asynchronously
        let receiver_task = tokio::spawn(async move {
            let mut values = Vec::new();
            for _ in 0..5 {
                values.push(receiver.recv_async().await.unwrap());
            }
            values
        });

        let values = receiver_task.await.unwrap();

        assert_eq!(values, vec![0, 1, 2, 3, 4]);
    }

    #[tokio::test]
    async fn test_send_async_receive_sync() {
        let (sender, receiver) = bounded::<String>(1);

        // Send asynchronously
        let sender_task = tokio::spawn(async move {
            for i in 0..5 {
                sender.send_async(format!("mixed_{}", i)).await.unwrap();
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        });

        // Receive synchronously in a blocking task
        let receiver_task = tokio::task::spawn_blocking(move || {
            let mut messages = Vec::new();
            for _ in 0..5 {
                messages.push(receiver.recv().unwrap());
            }
            messages
        });

        sender_task.await.unwrap();
        let messages = receiver_task.await.unwrap();

        assert_eq!(messages.len(), 5);
        for (i, msg) in messages.iter().enumerate() {
            assert_eq!(msg, &format!("mixed_{}", i));
        }
    }

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

        // Send a message
        sender.send(42).unwrap();

        // Drop the sender
        drop(sender);

        // Should still be able to receive the message
        assert_eq!(receiver.recv().unwrap(), 42);

        // Further receives should fail
        assert!(receiver.recv().is_err());
    }

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

        // Send a message asynchronously
        sender.send_async(42).await.unwrap();

        // Drop the sender
        drop(sender);

        // Should still be able to receive the message
        assert_eq!(receiver.recv_async().await.unwrap(), 42);

        // Further receives should fail
        assert!(receiver.recv_async().await.is_err());
    }

    #[tokio::test]
    async fn test_stream_basic_functionality() {
        let (sender, receiver) = bounded::<i32>(5);

        // Send some values
        sender.send(1).unwrap();
        sender.send(2).unwrap();
        sender.send(3).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, 3]);
    }

    #[tokio::test]
    async fn test_stream_empty_channel() {
        let (sender, receiver) = bounded::<i32>(1);
        drop(sender); // Close the channel immediately

        let mut stream = receiver.into_stream();
        let value = stream.next().await;

        assert!(value.is_none());
    }

    #[tokio::test]
    async fn test_stream_with_collect() {
        let (sender, receiver) = bounded::<String>(10);

        // Send multiple values
        for i in 0..5 {
            sender.send(format!("message_{}", i)).unwrap();
        }
        drop(sender);

        let stream = receiver.into_stream();
        let values: Vec<String> = stream.collect().await;

        assert_eq!(values.len(), 5);
        for (i, value) in values.iter().enumerate() {
            assert_eq!(value, &format!("message_{}", i));
        }
    }

    #[tokio::test]
    async fn test_stream_concurrent_senders() {
        let (sender, receiver) = bounded::<i32>(100);

        // Spawn multiple sender tasks
        let mut handles = Vec::new();
        for i in 0..3 {
            let sender_clone = sender.clone();
            let handle = tokio::spawn(async move {
                for j in 0..10 {
                    sender_clone.send_async(i * 10 + j).await.unwrap();
                }
            });
            handles.push(handle);
        }

        // Wait for all senders to complete
        for handle in handles {
            handle.await.unwrap();
        }
        drop(sender);

        let stream = receiver.into_stream();
        let values: Vec<i32> = stream.collect().await;

        assert_eq!(values.len(), 30);
        // Values should be from 0-29, but order is not guaranteed
        let mut sorted_values = values;
        sorted_values.sort();
        let expected: Vec<i32> = (0..30).collect();
        assert_eq!(sorted_values, expected);
    }

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

        // Send one value and keep sender alive
        sender.send(42).unwrap();

        let mut stream = receiver.into_stream();

        // First value should be received immediately
        let result = timeout(Duration::from_millis(100), stream.next()).await;
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), Some(42));

        // Second value should timeout since no more values are sent
        let result = timeout(Duration::from_millis(50), stream.next()).await;
        assert!(result.is_err()); // Timeout error
    }

    #[tokio::test]
    async fn test_stream_partial_consumption() {
        let (sender, receiver) = bounded::<i32>(10);

        // Send multiple values
        for i in 0..10 {
            sender.send(i).unwrap();
        }
        drop(sender);

        let mut stream = receiver.into_stream();

        // Only consume first 3 values
        let mut consumed = Vec::new();
        for _ in 0..3 {
            if let Some(value) = stream.next().await {
                consumed.push(value);
            }
        }

        assert_eq!(consumed, vec![0, 1, 2]);

        // Stream should still have remaining values
        let remaining: Vec<i32> = stream.collect().await;
        assert_eq!(remaining, vec![3, 4, 5, 6, 7, 8, 9]);
    }

    #[tokio::test]
    async fn test_stream_async_senders() {
        let (sender, receiver) = bounded::<i32>(5);

        // Send values asynchronously first
        for i in 0..5 {
            sender.send_async(i).await.unwrap();
        }
        drop(sender); // Close the channel

        // Then collect from stream
        let stream = receiver.into_stream();
        let values: Vec<i32> = stream.collect().await;

        assert_eq!(values, vec![0, 1, 2, 3, 4]);
    }

    #[tokio::test]
    async fn test_stream_mixed_send_modes() {
        let (sender, receiver) = bounded::<String>(10);

        // Send some values synchronously
        sender.send("sync_1".to_string()).unwrap();
        sender.send("sync_2".to_string()).unwrap();

        // Send some values asynchronously
        let sender_clone = sender.clone();
        let async_task = tokio::spawn(async move {
            sender_clone
                .send_async("async_1".to_string())
                .await
                .unwrap();
            sender_clone
                .send_async("async_2".to_string())
                .await
                .unwrap();
        });

        async_task.await.unwrap();
        drop(sender);

        let stream = receiver.into_stream();
        let values: Vec<String> = stream.collect().await;

        assert_eq!(values.len(), 4);
        assert!(values.contains(&"sync_1".to_string()));
        assert!(values.contains(&"sync_2".to_string()));
        assert!(values.contains(&"async_1".to_string()));
        assert!(values.contains(&"async_2".to_string()));
    }

    #[tokio::test]
    async fn test_stream_with_filter_map() {
        let (sender, receiver) = bounded::<i32>(10);

        // Send values 0-9
        for i in 0..10 {
            sender.send(i).unwrap();
        }
        drop(sender);

        let stream = receiver.into_stream();

        // Filter even numbers and double them
        let filtered: Vec<i32> = stream
            .filter_map(|x| async move {
                if x % 2 == 0 {
                    Some(x * 2)
                } else {
                    None
                }
            })
            .collect()
            .await;

        assert_eq!(filtered, vec![0, 4, 8, 12, 16]);
    }

    #[tokio::test]
    async fn test_stream_take() {
        let (sender, receiver) = bounded::<i32>(10);

        // Send more values than we'll take
        for i in 0..10 {
            sender.send(i).unwrap();
        }
        drop(sender);

        let stream = receiver.into_stream();
        let taken: Vec<i32> = stream.take(3).collect().await;

        assert_eq!(taken, vec![0, 1, 2]);
    }

    #[tokio::test]
    async fn test_stream_fold() {
        let (sender, receiver) = bounded::<i32>(5);

        // Send values 1-5
        for i in 1..=5 {
            sender.send(i).unwrap();
        }
        drop(sender);

        let stream = receiver.into_stream();
        let sum = stream.fold(0, |acc, x| async move { acc + x }).await;

        assert_eq!(sum, 15); // 1+2+3+4+5 = 15
    }

    #[tokio::test]
    async fn test_stream_multiple_receivers() {
        let (sender, receiver) = bounded::<i32>(10);

        // Clone the receiver to create multiple stream consumers
        let receiver2 = receiver.clone();

        // Send values
        for i in 0..6 {
            sender.send(i).unwrap();
        }
        drop(sender);

        // Both streams should be able to receive values (they compete for messages)
        let task1 = tokio::spawn(async move { receiver.into_stream().collect::<Vec<i32>>().await });

        let task2 =
            tokio::spawn(async move { receiver2.into_stream().collect::<Vec<i32>>().await });

        let values1 = task1.await.unwrap();
        let values2 = task2.await.unwrap();

        // Total values received should equal what was sent
        assert_eq!(values1.len() + values2.len(), 6);

        // Combine and sort to verify all values were received
        let mut all_values = values1;
        all_values.extend(values2);
        all_values.sort();

        assert_eq!(all_values, vec![0, 1, 2, 3, 4, 5]);
    }

    #[tokio::test]
    async fn test_stream_error_handling() {
        let (sender, receiver) = bounded::<Result<i32, String>>(5);

        // Send mix of Ok and Err values
        sender.send(Ok(1)).unwrap();
        sender.send(Err("error".to_string())).unwrap();
        sender.send(Ok(2)).unwrap();
        drop(sender);

        let stream = receiver.into_stream();
        let results: Vec<Result<i32, String>> = stream.collect().await;

        assert_eq!(results.len(), 3);
        assert_eq!(results[0], Ok(1));
        assert_eq!(results[1], Err("error".to_string()));
        assert_eq!(results[2], Ok(2));

        // Test with try_collect (should stop at first error)
        let (sender2, receiver2) = bounded::<Result<i32, String>>(5);
        sender2.send(Ok(1)).unwrap();
        sender2.send(Err("error".to_string())).unwrap();
        sender2.send(Ok(2)).unwrap();
        drop(sender2);

        let stream2 = receiver2.into_stream();
        let result: Result<Vec<i32>, String> = stream2.try_collect().await;

        assert_eq!(result, Err("error".to_string()));
    }

    // ===== COMPREHENSIVE STREAMING TESTS =====
    // These tests specifically focus on waker behavior and timing to ensure
    // the stream implementation doesn't hang under normal workloads

    #[tokio::test]
    async fn test_stream_waker_behavior_immediate_values() {
        let (sender, receiver) = bounded::<i32>(10);

        // Send values immediately before creating stream
        for i in 0..5 {
            sender.send(i).unwrap();
        }
        drop(sender);

        let start = std::time::Instant::now();
        let stream = receiver.into_stream();
        let values: Vec<i32> = stream.collect().await;
        let elapsed = start.elapsed();

        assert_eq!(values, vec![0, 1, 2, 3, 4]);
        // Should be very fast since values are already available
        assert!(elapsed < Duration::from_millis(10));
    }

    #[tokio::test]
    async fn test_stream_waker_behavior_delayed_values() {
        let (sender, receiver) = bounded::<i32>(10);

        // Start collecting in a task that owns the receiver
        let collect_task = tokio::spawn(async move {
            let mut stream = receiver.into_stream();
            let mut values = Vec::new();
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 5 {
                    break;
                }
            }
            values
        });

        // Send values with delays to test waker behavior
        for i in 0..5 {
            tokio::time::sleep(Duration::from_millis(50)).await;
            sender.send(i).unwrap();
        }
        drop(sender);

        let values = collect_task.await.unwrap();
        assert_eq!(values, vec![0, 1, 2, 3, 4]);
    }

    #[tokio::test]
    async fn test_stream_waker_behavior_mixed_timing() {
        let (sender, receiver) = bounded::<String>(10);

        // Send some values immediately
        sender.send("immediate_1".to_string()).unwrap();
        sender.send("immediate_2".to_string()).unwrap();

        let collect_task = tokio::spawn(async move {
            let mut stream = receiver.into_stream();
            let mut values = Vec::new();
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 4 {
                    break;
                }
            }
            values
        });

        // Send more values with delays
        tokio::time::sleep(Duration::from_millis(30)).await;
        sender.send("delayed_1".to_string()).unwrap();

        tokio::time::sleep(Duration::from_millis(30)).await;
        sender.send("delayed_2".to_string()).unwrap();

        drop(sender);

        let values = collect_task.await.unwrap();
        assert_eq!(values.len(), 4);
        assert!(values.contains(&"immediate_1".to_string()));
        assert!(values.contains(&"immediate_2".to_string()));
        assert!(values.contains(&"delayed_1".to_string()));
        assert!(values.contains(&"delayed_2".to_string()));
    }

    #[tokio::test]
    async fn test_stream_concurrent_producers_single_consumer() {
        let (sender, receiver) = bounded::<i32>(100);

        let collect_task = tokio::spawn(async move {
            let mut stream = receiver.into_stream();
            let mut values = Vec::new();
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 30 {
                    break;
                }
            }
            values
        });

        // Spawn multiple producer tasks
        let mut producer_handles = Vec::new();
        for producer_id in 0..3 {
            let sender_clone = sender.clone();
            let handle = tokio::spawn(async move {
                for i in 0..10 {
                    tokio::time::sleep(Duration::from_millis(10)).await;
                    sender_clone.send_async(producer_id * 10 + i).await.unwrap();
                }
            });
            producer_handles.push(handle);
        }

        // Wait for all producers to complete
        for handle in producer_handles {
            handle.await.unwrap();
        }
        drop(sender);

        let values = collect_task.await.unwrap();
        assert_eq!(values.len(), 30);

        // Verify all expected values are present
        let mut sorted_values = values;
        sorted_values.sort();
        let expected: Vec<i32> = (0..30).collect();
        assert_eq!(sorted_values, expected);
    }

    #[tokio::test]
    async fn test_stream_multiple_consumers_competing() {
        let (sender, receiver) = bounded::<i32>(50);

        // Create multiple streams from the same receiver
        let receiver2 = receiver.clone();
        let receiver3 = receiver.clone();

        // Spawn consumer tasks
        let consumer1 = tokio::spawn(async move {
            let mut stream = receiver.into_stream();
            let mut values = Vec::new();
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 10 {
                    break;
                }
            }
            values
        });

        let consumer2 = tokio::spawn(async move {
            let mut stream = receiver2.into_stream();
            let mut values = Vec::new();
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 10 {
                    break;
                }
            }
            values
        });

        let consumer3 = tokio::spawn(async move {
            let mut stream = receiver3.into_stream();
            let mut values = Vec::new();
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 10 {
                    break;
                }
            }
            values
        });

        // Send values with timing
        for i in 0..30 {
            tokio::time::sleep(Duration::from_millis(5)).await;
            sender.send_async(i).await.unwrap();
        }
        drop(sender);

        let values1 = consumer1.await.unwrap();
        let values2 = consumer2.await.unwrap();
        let values3 = consumer3.await.unwrap();

        // Each consumer should get some values
        assert_eq!(values1.len(), 10);
        assert_eq!(values2.len(), 10);
        assert_eq!(values3.len(), 10);

        // All values should be unique across consumers
        let mut all_values = values1;
        all_values.extend(values2);
        all_values.extend(values3);
        all_values.sort();
        all_values.dedup();
        assert_eq!(all_values.len(), 30);
    }

    #[tokio::test]
    async fn test_stream_backpressure_behavior() {
        let (sender, receiver) = bounded::<i32>(2); // Small buffer

        let collect_task = tokio::spawn(async move {
            let mut stream = receiver.into_stream();
            let mut values = Vec::new();
            // Consume slowly to create backpressure
            while let Some(value) = stream.next().await {
                values.push(value);
                tokio::time::sleep(Duration::from_millis(100)).await;
                if values.len() == 5 {
                    break;
                }
            }
            values
        });

        // Send values faster than consumption
        for i in 0..5 {
            sender.send_async(i).await.unwrap();
        }
        drop(sender);

        let values = collect_task.await.unwrap();
        assert_eq!(values, vec![0, 1, 2, 3, 4]);
    }

    #[tokio::test]
    async fn test_stream_rapid_send_receive_cycles() {
        let (sender, receiver) = bounded::<i32>(10);

        let collect_task = tokio::spawn(async move {
            let mut stream = receiver.into_stream();
            let mut values = Vec::new();
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 20 {
                    break;
                }
            }
            values
        });

        // Rapid send-receive cycles
        for batch in 0..4 {
            for i in 0..5 {
                sender.send_async(batch * 5 + i).await.unwrap();
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        drop(sender);

        let values = collect_task.await.unwrap();
        assert_eq!(values.len(), 20);
        let mut sorted_values = values;
        sorted_values.sort();
        let expected: Vec<i32> = (0..20).collect();
        assert_eq!(sorted_values, expected);
    }

    #[tokio::test]
    async fn test_stream_intermittent_sending() {
        let (sender, receiver) = bounded::<String>(10);

        let collect_task = tokio::spawn(async move {
            let mut stream = receiver.into_stream();
            let mut values = Vec::new();
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 6 {
                    break;
                }
            }
            values
        });

        // Intermittent sending pattern
        sender.send("burst_1".to_string()).unwrap();
        sender.send("burst_2".to_string()).unwrap();
        sender.send("burst_3".to_string()).unwrap();

        tokio::time::sleep(Duration::from_millis(100)).await;

        sender.send("pause_1".to_string()).unwrap();

        tokio::time::sleep(Duration::from_millis(100)).await;

        sender.send("pause_2".to_string()).unwrap();
        sender.send("pause_3".to_string()).unwrap();

        drop(sender);

        let values = collect_task.await.unwrap();
        assert_eq!(values.len(), 6);
        assert!(values.contains(&"burst_1".to_string()));
        assert!(values.contains(&"burst_2".to_string()));
        assert!(values.contains(&"burst_3".to_string()));
        assert!(values.contains(&"pause_1".to_string()));
        assert!(values.contains(&"pause_2".to_string()));
        assert!(values.contains(&"pause_3".to_string()));
    }

    #[tokio::test]
    async fn test_stream_early_termination() {
        let (sender, receiver) = bounded::<i32>(10);

        let collect_task = tokio::spawn(async move {
            let mut stream = receiver.into_stream();
            let mut values = Vec::new();
            // Only consume 3 values then stop
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 3 {
                    break;
                }
            }
            values
        });

        // Send more values than will be consumed
        for i in 0..10 {
            sender.send_async(i).await.unwrap();
        }
        drop(sender);

        let values = collect_task.await.unwrap();
        assert_eq!(values.len(), 3);
        // Should be the first 3 values
        assert_eq!(values, vec![0, 1, 2]);
    }

    #[tokio::test]
    async fn test_stream_with_select_macro() {
        let (sender1, receiver1) = bounded::<i32>(5);
        let (sender2, receiver2) = bounded::<String>(5);

        let collect_task = tokio::spawn(async move {
            let mut stream1 = receiver1.into_stream();
            let mut stream2 = receiver2.into_stream();
            let mut int_values = Vec::new();
            let mut string_values = Vec::new();

            // Use select! to handle both streams
            loop {
                tokio::select! {
                    Some(value) = stream1.next() => {
                        int_values.push(value);
                        if int_values.len() == 3 && string_values.len() == 2 {
                            break;
                        }
                    }
                    Some(value) = stream2.next() => {
                        string_values.push(value);
                        if int_values.len() == 3 && string_values.len() == 2 {
                            break;
                        }
                    }
                    else => {
                        // Both streams are done
                        break;
                    }
                }
            }
            (int_values, string_values)
        });

        // Send values to both channels
        sender1.send_async(1).await.unwrap();
        sender2.send_async("hello".to_string()).await.unwrap();
        sender1.send_async(2).await.unwrap();
        sender2.send_async("world".to_string()).await.unwrap();
        sender1.send_async(3).await.unwrap();

        drop(sender1);
        drop(sender2);

        let (int_values, string_values) = collect_task.await.unwrap();
        assert_eq!(int_values.len(), 3);
        assert_eq!(string_values.len(), 2);
        assert!(int_values.contains(&1));
        assert!(int_values.contains(&2));
        assert!(int_values.contains(&3));
        assert!(string_values.contains(&"hello".to_string()));
        assert!(string_values.contains(&"world".to_string()));
    }

    #[tokio::test]
    async fn test_stream_stress_high_throughput() {
        let (sender, receiver) = bounded::<i32>(1000);

        let collect_task = tokio::spawn(async move {
            let mut stream = receiver.into_stream();
            let mut values = Vec::new();
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 1000 {
                    break;
                }
            }
            values
        });

        // High throughput sending
        let sender_task = tokio::spawn(async move {
            for i in 0..1000 {
                sender.send_async(i).await.unwrap();
            }
            drop(sender);
        });

        sender_task.await.unwrap();

        let values = collect_task.await.unwrap();
        assert_eq!(values.len(), 1000);

        let mut sorted_values = values;
        sorted_values.sort();
        let expected: Vec<i32> = (0..1000).collect();
        assert_eq!(sorted_values, expected);
    }

    #[tokio::test]
    async fn test_stream_timeout_behavior() {
        let (sender, receiver) = bounded::<i32>(5);

        // Test that stream properly handles timeouts
        let collect_task = tokio::spawn(async move {
            let mut stream = receiver.into_stream();
            let mut values = Vec::new();
            loop {
                match tokio::time::timeout(Duration::from_millis(100), stream.next()).await {
                    Ok(Some(value)) => {
                        values.push(value);
                        if values.len() == 3 {
                            break;
                        }
                    }
                    Ok(None) => break, // Channel closed
                    Err(_) => {
                        // Timeout - this is expected behavior
                        if values.len() == 3 {
                            break;
                        }
                    }
                }
            }
            values
        });

        // Send values with delays
        tokio::time::sleep(Duration::from_millis(50)).await;
        sender.send_async(1).await.unwrap();

        tokio::time::sleep(Duration::from_millis(50)).await;
        sender.send_async(2).await.unwrap();

        tokio::time::sleep(Duration::from_millis(50)).await;
        sender.send_async(3).await.unwrap();

        drop(sender);

        let values = collect_task.await.unwrap();
        assert_eq!(values.len(), 3);
        assert_eq!(values, vec![1, 2, 3]);
    }

    #[tokio::test]
    async fn test_stream_drop_behavior_cleanup() {
        let (sender, receiver) = bounded::<i32>(10);

        // Send some values
        for i in 0..5 {
            sender.send_async(i).await.unwrap();
        }

        let stream = receiver.into_stream();
        let mut stream = stream;

        // Consume a few values then drop the stream
        let first_value = stream.next().await.unwrap();
        let second_value = stream.next().await.unwrap();

        assert_eq!(first_value, 0);
        assert_eq!(second_value, 1);

        // Drop the stream - this should clean up properly
        drop(stream);
        drop(sender);

        // The remaining values should be lost, but no panic should occur
    }

    #[tokio::test]
    async fn test_stream_with_chain_operations() {
        let (sender, receiver) = bounded::<i32>(10);

        // Send values
        for i in 0..10 {
            sender.send_async(i).await.unwrap();
        }
        drop(sender);

        let stream = receiver.into_stream();

        // Chain multiple operations
        let result: Vec<i32> = stream
            .filter_map(|x| async move {
                if x % 2 == 0 {
                    Some(x * 2)
                } else {
                    None
                }
            }) // Keep only even numbers and double them
            .take(3) // Take first 3
            .collect()
            .await;

        assert_eq!(result, vec![0, 4, 8]); // 0*2, 2*2, 4*2
    }

    // ===== COMPREHENSIVE DROP BEHAVIOR TESTS =====

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

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

        // Start a thread that will block trying to send
        let sender_clone = sender.clone();
        let send_handle = thread::spawn(move || {
            // This should block until receiver consumes the value
            let result = sender_clone.send(43);
            // Should get disconnected error since receiver was dropped
            assert!(result.is_err());
        });

        // Give the sender time to block
        thread::sleep(Duration::from_millis(10));

        // Drop the receiver - this should wake up the blocked sender
        drop(receiver);

        // The blocked sender should now get a disconnected error
        send_handle.join().unwrap();
    }

    #[test]
    fn test_multiple_sender_drop_behavior() {
        let (sender, receiver) = bounded::<i32>(10);

        // Create multiple senders
        let sender1 = sender.clone();
        let sender2 = sender.clone();
        let sender3 = sender.clone();

        // Send from different senders
        sender1.send(1).unwrap();
        sender2.send(2).unwrap();
        sender3.send(3).unwrap();

        // Drop some senders
        drop(sender1);
        drop(sender2);

        // Should still be able to send from remaining senders
        sender3.send(4).unwrap();
        sender.send(5).unwrap();

        // Receive all values
        let mut values = Vec::new();
        for _ in 0..5 {
            values.push(receiver.recv().unwrap());
        }
        values.sort();
        assert_eq!(values, vec![1, 2, 3, 4, 5]);

        // Drop all remaining senders
        drop(sender3);
        drop(sender);

        // Further receives should fail
        assert!(receiver.recv().is_err());
    }

    #[test]
    fn test_multiple_receiver_drop_behavior() {
        let (sender, receiver) = bounded::<i32>(10);

        // Create multiple receivers
        let receiver1 = receiver.clone();
        let receiver2 = receiver.clone();
        let receiver3 = receiver.clone();

        // Send some values
        sender.send(1).unwrap();
        sender.send(2).unwrap();
        sender.send(3).unwrap();

        // Drop some receivers
        drop(receiver1);
        drop(receiver2);

        // Should still be able to receive from remaining receivers
        assert_eq!(receiver3.recv().unwrap(), 1);
        assert_eq!(receiver.recv().unwrap(), 2);
        assert_eq!(receiver.recv().unwrap(), 3);

        // Drop all remaining receivers
        drop(receiver3);
        drop(receiver);

        // Further sends should fail
        assert!(sender.send(4).is_err());
    }

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

        // Fill the channel to capacity
        sender.send_async(42).await.unwrap();

        // Start a task that will block trying to receive
        let receiver_clone = receiver.clone();
        let recv_task = tokio::spawn(async move {
            // This should block until a value is available
            receiver_clone.recv_async().await.unwrap()
        });

        // Give the receiver time to block
        tokio::time::sleep(Duration::from_millis(10)).await;

        // Drop the original sender - this should wake up the blocked receiver
        drop(sender);

        // The blocked receiver should now get a disconnected error
        let result = recv_task.await.unwrap();
        // The receiver should get the value that was already in the channel
        assert_eq!(result, 42);

        // Further receives should fail
        assert!(receiver.recv_async().await.is_err());
    }

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

        // Fill the channel
        sender.send_async(42).await.unwrap();

        // Start a task that will block trying to send
        let sender_clone = sender.clone();
        let send_task = tokio::spawn(async move {
            // This should block until receiver consumes the value
            let result = sender_clone.send_async(43).await;
            // Should get disconnected error since receiver was dropped
            assert!(result.is_err());
        });

        // Give the sender time to block
        tokio::time::sleep(Duration::from_millis(10)).await;

        // Drop the receiver - this should wake up the blocked sender
        drop(receiver);

        // The blocked sender should now get a disconnected error
        send_task.await.unwrap();
    }

    #[tokio::test]
    async fn test_send_fut_drop_cleanup() {
        let (sender, receiver) = bounded::<i32>(2);

        // Fill the channel
        sender.send_async(42).await.unwrap();

        // Create a send future that will block
        let send_fut = sender.send_async(43);

        // Drop the future before it completes - this should clean up properly
        drop(send_fut);

        // Should still be able to send normally
        sender.send_async(44).await.unwrap();

        // Receive values
        assert_eq!(receiver.recv_async().await.unwrap(), 42);
        assert_eq!(receiver.recv_async().await.unwrap(), 44);
    }

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

        // Create a receive future that will block
        let recv_fut = receiver.recv_async();

        // Drop the future before it completes - this should clean up properly
        drop(recv_fut);

        // Should still be able to receive normally
        sender.send_async(42).await.unwrap();
        assert_eq!(receiver.recv_async().await.unwrap(), 42);
    }

    #[tokio::test]
    async fn test_stream_drop_cleanup() {
        let (sender, receiver) = bounded::<i32>(10);

        // Send some values
        for i in 0..5 {
            sender.send_async(i).await.unwrap();
        }

        // Create a stream (this takes ownership of the receiver)
        let stream = receiver.into_stream();

        // Drop the stream before consuming all values - this should clean up properly
        drop(stream);

        // Since the receiver is gone, sending should fail
        assert!(sender.send_async(99).await.is_err());

        // Verify no panics occurred during stream drop
    }

    #[tokio::test]
    async fn test_concurrent_drop_behavior() {
        let (sender, receiver) = bounded::<i32>(10);

        // Create multiple senders and receivers
        let senders: Vec<_> = (0..5).map(|_| sender.clone()).collect();
        let receivers: Vec<_> = (0..5).map(|_| receiver.clone()).collect();

        // Send values from different senders
        for (i, sender) in senders.iter().enumerate() {
            sender.send_async(i as i32).await.unwrap();
        }

        // Receive values from different receivers
        let mut received_values = Vec::new();
        for receiver in &receivers {
            received_values.push(receiver.recv_async().await.unwrap());
        }
        received_values.sort();
        assert_eq!(received_values, vec![0, 1, 2, 3, 4]);

        // Drop all senders and receivers concurrently
        drop(senders);
        drop(receivers);
        drop(sender);
        drop(receiver);

        // No panics should occur
    }

    #[tokio::test]
    async fn test_complex_drop_scenarios() {
        let (sender, receiver) = bounded::<i32>(10);

        // Test dropping futures while they're pending
        let send_fut1 = sender.send_async(1);
        let send_fut2 = sender.send_async(2);
        let recv_fut1 = receiver.recv_async();
        let recv_fut2 = receiver.recv_async();

        // Drop some futures before they complete
        drop(send_fut1);
        drop(recv_fut1);

        // The remaining futures should still work
        send_fut2.await.unwrap();
        assert_eq!(recv_fut2.await.unwrap(), 2);

        // Test dropping streams mid-consumption
        for i in 3..8 {
            sender.send_async(i).await.unwrap();
        }

        let stream = receiver.into_stream();
        let mut stream = stream;

        // Consume a few values
        assert_eq!(stream.next().await.unwrap(), 3);
        assert_eq!(stream.next().await.unwrap(), 4);

        // Drop the stream mid-consumption
        drop(stream);

        // Since the receiver is gone, sending should fail
        assert!(sender.send_async(99).await.is_err());

        // Verify no panics occurred during stream drop
    }

    #[tokio::test]
    async fn test_stream_concurrent_stream_creation_final() {
        let (sender, receiver) = bounded::<i32>(20);

        // Create multiple streams concurrently
        let receiver2 = receiver.clone();
        let receiver3 = receiver.clone();

        let task1 = tokio::spawn(async move {
            let mut stream = receiver.into_stream();
            let mut values = Vec::new();
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 5 {
                    break;
                }
            }
            values
        });

        let task2 = tokio::spawn(async move {
            let mut stream = receiver2.into_stream();
            let mut values = Vec::new();
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 5 {
                    break;
                }
            }
            values
        });

        let task3 = tokio::spawn(async move {
            let mut stream = receiver3.into_stream();
            let mut values = Vec::new();
            while let Some(value) = stream.next().await {
                values.push(value);
                if values.len() == 5 {
                    break;
                }
            }
            values
        });

        // Send values
        for i in 0..15 {
            sender.send_async(i).await.unwrap();
        }
        drop(sender);

        let values1 = task1.await.unwrap();
        let values2 = task2.await.unwrap();
        let values3 = task3.await.unwrap();

        assert_eq!(values1.len(), 5);
        assert_eq!(values2.len(), 5);
        assert_eq!(values3.len(), 5);

        // All values should be unique
        let mut all_values = values1;
        all_values.extend(values2);
        all_values.extend(values3);
        all_values.sort();
        all_values.dedup();
        assert_eq!(all_values.len(), 15);
    }
}