Add channel for blocking context

(Currently exactly like crossbeam)

Author: frostyplanet

src/mpmc.rs

@@ -5,6 +5,19 @@ use crate::blocking_tx::*;
 /// Multi producers, multi consumers
 use crate::channel::*;
 
+/// Initiate a unbounded channel for blocking context.
+///
+/// Sender will never block, so we use the same TxBlocking for threads
+pub fn unbounded_blocking<T: Unpin>() -> (MTx<T>, MRx<T>) {
+    let (tx, rx) = crossbeam::channel::unbounded();
+    let send_wakers = SendWakersBlocking::new();
+    let recv_wakers = RecvWakersBlocking::new();
+    let shared = ChannelShared::new(send_wakers, recv_wakers);
+    let tx = MTx::new(tx, shared.clone());
+    let rx = MRx::new(rx, shared);
+    (tx, rx)
+}
+
 /// Initiate a unbounded channel.
 ///
 /// Although sender type is MTx, will never block.
@@ -19,6 +32,23 @@ pub fn unbounded_async<T: Unpin>() -> (MTx<T>, MAsyncRx<T>) {
     (tx, rx)
 }
 
+/// Initiate a bounded channel for blocking context
+///
+/// Special case: 0 size is not supported yet, threat it as 1 size for now.
+pub fn bounded_blocking<T: Unpin>(mut size: usize) -> (MTx<T>, MRx<T>) {
+    if size == 0 {
+        size = 1;
+    }
+    let (tx, rx) = crossbeam::channel::bounded(size);
+    let send_wakers = SendWakersBlocking::new();
+    let recv_wakers = RecvWakersBlocking::new();
+    let shared = ChannelShared::new(send_wakers, recv_wakers);
+
+    let tx = MTx::new(tx, shared.clone());
+    let rx = MRx::new(rx, shared);
+    (tx, rx)
+}
+
 /// Initiate a bounded channel that sender and receiver are async.
 ///
 /// Special case: 0 size is not supported yet, threat it as 1 size for now.

src/mpsc.rs

@@ -5,6 +5,19 @@ use crate::blocking_tx::*;
 /// Multi producers, single consumer
 use crate::channel::*;
 
+/// Initiate a unbounded channel for blocking context.
+///
+/// Sender will never block, so we use the same TxBlocking for threads
+pub fn unbounded_blocking<T: Unpin>() -> (MTx<T>, Rx<T>) {
+    let (tx, rx) = crossbeam::channel::unbounded();
+    let send_wakers = SendWakersBlocking::new();
+    let recv_wakers = RecvWakersBlocking::new();
+    let shared = ChannelShared::new(send_wakers, recv_wakers);
+    let tx = MTx::new(tx, shared.clone());
+    let rx = Rx::new(rx, shared);
+    (tx, rx)
+}
+
 /// Initiate a unbounded channel.
 ///
 /// Although sender type is MTx, will never block.
@@ -18,6 +31,23 @@ pub fn unbounded_async<T: Unpin>() -> (MTx<T>, AsyncRx<T>) {
     (tx, rx)
 }
 
+/// Initiate a bounded channel for blocking context
+///
+/// Special case: 0 size is not supported yet, threat it as 1 size for now.
+pub fn bounded_blocking<T: Unpin>(mut size: usize) -> (MTx<T>, Rx<T>) {
+    if size == 0 {
+        size = 1;
+    }
+    let (tx, rx) = crossbeam::channel::bounded(size);
+    let send_wakers = SendWakersBlocking::new();
+    let recv_wakers = RecvWakersBlocking::new();
+    let shared = ChannelShared::new(send_wakers, recv_wakers);
+
+    let tx = MTx::new(tx, shared.clone());
+    let rx = Rx::new(rx, shared);
+    (tx, rx)
+}
+
 /// Initiate a bounded channel that sender and receiver is async.
 ///
 /// Special case: 0 size is not supported yet, threat it as 1 size for now.

src/spsc.rs

@@ -5,7 +5,21 @@ use crate::blocking_tx::*;
 /// Single producer, single consumer
 use crate::channel::*;
 
+/// Initiate a unbounded channel for blocking context.
+///
+/// Sender will never block, so we use the same TxBlocking for threads
+pub fn unbounded_blocking<T: Unpin>() -> (Tx<T>, Rx<T>) {
+    let (tx, rx) = crossbeam::channel::unbounded();
+    let send_wakers = SendWakersBlocking::new();
+    let recv_wakers = RecvWakersBlocking::new();
+    let shared = ChannelShared::new(send_wakers, recv_wakers);
+    let tx = Tx::new(tx, shared.clone());
+    let rx = Rx::new(rx, shared);
+    (tx, rx)
+}
+
 /// Initiate a unbounded channel.
+///
 /// Sender will never block, so we use the same TxBlocking for threads
 pub fn unbounded_async<T: Unpin>() -> (Tx<T>, AsyncRx<T>) {
     let (tx, rx) = crossbeam::channel::unbounded();
@@ -17,6 +31,23 @@ pub fn unbounded_async<T: Unpin>() -> (Tx<T>, AsyncRx<T>) {
     (tx, rx)
 }
 
+/// Initiate a bounded channel for blocking context
+///
+/// Special case: 0 size is not supported yet, threat it as 1 size for now.
+pub fn bounded_blocking<T: Unpin>(mut size: usize) -> (Tx<T>, Rx<T>) {
+    if size == 0 {
+        size = 1;
+    }
+    let (tx, rx) = crossbeam::channel::bounded(size);
+    let send_wakers = SendWakersBlocking::new();
+    let recv_wakers = RecvWakersBlocking::new();
+    let shared = ChannelShared::new(send_wakers, recv_wakers);
+
+    let tx = Tx::new(tx, shared.clone());
+    let rx = Rx::new(rx, shared);
+    (tx, rx)
+}
+
 /// Initiate a bounded channel that sender and receiver are async
 ///
 /// Special case: 0 size is not supported yet, threat it as 1 size for now.