crossfire 2.1.11

channels for async and threads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

//! Multiple producers, single consumer.
//!
//! The optimization assumes a single consumer. The waker registration of the receiver is lossless compared to `mpmc`.
//!
//! **NOTE**: For the SC (single consumer) version, [AsyncRx] and [Rx] are not `Clone` and do not implement `Sync`.
//! Although they can be moved to other threads, they are not allowed to be used with `send`/`recv` while in an `Arc`.
//!
//! The following code is OK:
//!
//! ``` rust
//! use crossfire::*;
//! async fn foo() {
//!     let (tx, rx) = mpsc::bounded_async::<usize>(100);
//!     tokio::spawn(async move {
//!         let _ = rx.recv().await;
//!     });
//!     drop(tx);
//! }
//! ```
//!
//! Because the `AsyncRx` does not have the `Sync` marker, using `Arc<AsyncRx>` will lose the `Send` marker.
//!
//! For your safety, the following code **should not compile**:
//!
//! ``` compile_fail
//! use crossfire::*;
//! use std::sync::Arc;
//! async fn foo() {
//!     let (tx, rx) = mpsc::bounded_async::<usize>(100);
//!     let rx = Arc::new(rx);
//!     tokio::spawn(async move {
//!         let _ = rx.recv().await;
//!     });
//!     drop(tx);
//! }
//! ```

use crate::async_rx::*;
use crate::async_tx::*;
use crate::blocking_rx::*;
use crate::blocking_tx::*;
use crate::shared::*;

macro_rules! init_share {
    ($flavor: expr) => {{
        let send_wakers = $flavor.new_reg_sender::<true>();
        let recv_wakers = $flavor.new_reg_recv::<false>();
        ChannelShared::new($flavor.to_flavor(), send_wakers, recv_wakers)
    }};
}

macro_rules! init_array {
    ($bound: expr) => {{
        Array::<T>::new($bound)
    }};
}

/// Creates an unbounded channel for use in a blocking context.
///
/// The sender will never block, so we use the same `Tx` for all threads.
pub fn unbounded_blocking<T: Unpin>() -> (MTx<T>, Rx<T>) {
    let share = init_share!(List::<T>::new());
    let tx = MTx::new(share.clone());
    let rx = Rx::new(share);
    (tx, rx)
}

/// Creates an unbounded channel for use in an async context.
///
/// Although the sender type is `MTx`, it will never block.
pub fn unbounded_async<T: Unpin>() -> (MTx<T>, AsyncRx<T>) {
    let share = init_share!(List::<T>::new());
    let tx = MTx::new(share.clone());
    let rx = AsyncRx::new(share);
    (tx, rx)
}

/// Creates a bounded channel for use in a blocking context.
///
/// As a special case, a channel size of 0 is not supported and will be treated as a channel of size 1.
pub fn bounded_blocking<T: Unpin>(size: usize) -> (MTx<T>, Rx<T>) {
    let share = init_share!(init_array!(size));
    let tx = MTx::new(share.clone());
    let rx = Rx::new(share);
    (tx, rx)
}

/// Creates a bounded channel where both the sender and receiver are async.
///
/// As a special case, a channel size of 0 is not supported and will be treated as a channel of size 1.
pub fn bounded_async<T: Unpin>(size: usize) -> (MAsyncTx<T>, AsyncRx<T>) {
    let share = init_share!(init_array!(size));
    let tx = MAsyncTx::new(share.clone());
    let rx = AsyncRx::new(share);
    (tx, rx)
}

/// Creates a bounded channel where the sender is async and the receiver is blocking.
///
/// As a special case, a channel size of 0 is not supported and will be treated as a channel of size 1.
pub fn bounded_tx_async_rx_blocking<T: Unpin>(size: usize) -> (MAsyncTx<T>, Rx<T>) {
    let share = init_share!(init_array!(size));
    let tx = MAsyncTx::new(share.clone());
    let rx = Rx::new(share);
    (tx, rx)
}

/// Creates a bounded channel where the sender is blocking and the receiver is async.
///
/// As a special case, a channel size of 0 is not supported and will be treated as a channel of size 1.
pub fn bounded_tx_blocking_rx_async<T>(size: usize) -> (MTx<T>, AsyncRx<T>) {
    let share = init_share!(init_array!(size));
    let tx = MTx::new(share.clone());
    let rx = AsyncRx::new(share);
    (tx, rx)
}