Add Receiver::new_sender to channels

crossbeam-channel/src/channel.rs

@@ -3,6 +3,7 @@
 use std::fmt;
 use std::iter::FusedIterator;
 use std::mem;
+use std::ops::Deref;
 use std::panic::{RefUnwindSafe, UnwindSafe};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
@@ -1506,3 +1507,63 @@ pub(crate) unsafe fn read<T>(r: &Receiver<T>, token: &mut Token) -> Result<T, ()
  ReceiverFlavor::Never(chan) => chan.read(token),
  }
 }
+
+/// Channel flavors that are reconnectable, ie, may survive
+/// without senders.
+///
+pub mod reconnectable {
+ // note: we need separate types because the At/Tick/Never
+ // channel types do not support this, but they all use the
+ // same exact Receiver type as array and list flavours.
+
+ // The intended usage of this module is just to replace any `use crossbeam_channel::...`
+ // with `use crossbeam_channel::reconnectable::...` and have everything work out of the box.
+
+ use super::*;
+ pub use crate::*;
+
+ use super::Receiver as NormalReceiver;
+
+ /// An [unbounded](super::unbounded) channel whose receiver
+ /// also is reconnectable.
+ pub fn unbounded<T>() -> (Sender<T>, Receiver<T>) {
+ let (s, r) = super::unbounded();
+ (s, Receiver(r))
+ }
+
+ /// A [receiver](super::Receiver) that can survive periods
+ /// of time where no [Sender]s are alive. New senders may
+ /// be created from the receiver directly ([Self::new_sender]).
+ /// The channel is only deallocated when the last receiver dies.
+ /// If there are live senders at that point, they start producing
+ /// [SendError]s as usual.
+ #[derive(Debug)]
+ pub struct Receiver<T>(NormalReceiver<T>);
+
+ impl<T> Receiver<T> {
+ /// Returns a new sender that communicates with this
+ /// receiver.
+ pub fn new_sender(&self) -> Sender<T> {
+ match &self.0.flavor {
+ ReceiverFlavor::Array(chan) => Sender {
+ flavor: SenderFlavor::Array(chan.new_sender(|_| unreachable!())),
+ },
+ ReceiverFlavor::List(chan) => Sender {
+ flavor: SenderFlavor::List(chan.new_sender(|c| c.reconnect_senders())),
+ },
+ ReceiverFlavor::Zero(chan) => Sender {
+ flavor: SenderFlavor::Zero(chan.new_sender(|_| unreachable!())),
+ },
+ _ => unreachable!("This type cannot be built with at/never/tick"),
+ }
+ }
+ }
+
+ impl<T> Deref for self::Receiver<T> {
+ type Target = NormalReceiver<T>;
+
+ fn deref(&self) -> &Self::Target {
+ &self.0
+ }
+ }
+}

crossbeam-channel/src/counter.rs

@@ -127,6 +127,18 @@ impl<C> Receiver<C> {
  }
  }
  }
+
+ pub(crate) fn new_sender(&self, reconnect: impl FnOnce(&C) -> bool) -> Sender<C> {
+ if 0 == self.counter().senders.fetch_add(1, Ordering::SeqCst) {
+ // we're the first sender to be created
+ reconnect(&self.counter().chan);
+ self.counter().destroy.store(false, Ordering::Relaxed);
+ }
+
+ Sender {
+ counter: self.counter,
+ }
+ }
 }
 
 impl<C> ops::Deref for Receiver<C> {

crossbeam-channel/src/flavors/list.rs

@@ -544,6 +544,17 @@ impl<T> Channel<T> {
  }
  }
 
+ /// Reconnects senders. There is no need to notify threads
+ /// as nobody is currently blocking, since we were in an
+ /// idle phase.
+ ///
+ /// Returns `true` if this call reconnected the channel.
+ pub(crate) fn reconnect_senders(&self) -> bool {
+ let tail = self.tail.index.fetch_and(!MARK_BIT, Ordering::SeqCst);
+
+ tail & MARK_BIT == 0
+ }
+
  /// Disconnects receivers.
  ///
  /// Returns `true` if this call disconnected the channel.

crossbeam-channel/src/lib.rs

@@ -361,6 +361,7 @@ cfg_if! {
  pub use crate::channel::{bounded, unbounded};
  pub use crate::channel::{IntoIter, Iter, TryIter};
  pub use crate::channel::{Receiver, Sender};
+ pub use crate::channel::reconnectable;
 
  pub use crate::select::{Select, SelectedOperation};
 

crossbeam-channel/tests/list.rs

@@ -6,7 +6,7 @@ use std::sync::atomic::Ordering;
 use std::thread;
 use std::time::Duration;
 
-use crossbeam_channel::{select, unbounded, Receiver};
+use crossbeam_channel::{reconnectable, select, unbounded, Receiver};
 use crossbeam_channel::{RecvError, RecvTimeoutError, TryRecvError};
 use crossbeam_channel::{SendError, SendTimeoutError, TrySendError};
 use crossbeam_utils::thread::scope;
@@ -200,6 +200,23 @@ fn recv_after_disconnect() {
  assert_eq!(r.recv(), Err(RecvError));
 }
 
+#[test]
+fn zero_receiver_revival() {
+ let (s, r) = reconnectable::unbounded();
+
+ s.send(1).unwrap();
+
+ drop(s);
+
+ assert_eq!(r.recv(), Ok(1));
+ assert_eq!(r.recv(), Err(RecvError));
+
+ let s = r.new_sender();
+
+ s.send(1).unwrap();
+ assert_eq!(r.recv(), Ok(1));
+}
+
 #[test]
 fn len() {
  let (s, r) = unbounded();