Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -838,6 +838,19 @@ impl<T> Sender<T> {
pub fn same_channel(&self, other: &Sender<T>) -> bool {
Arc::ptr_eq(&self.shared, &other.shared)
}

/// Create a new [`Receiver`] for the same channel as this sender.
///
/// This is handy when only one half of the channel is stored somewhere (such as inside a
/// data structure) and the other half needs to be recovered without threading it through
/// separately. If every receiver has already been dropped, the returned receiver will
/// observe the channel as disconnected, just as the sender does.
pub fn make_receiver(&self) -> Receiver<T> {
self.shared.receiver_count.fetch_add(1, Ordering::Relaxed);
Receiver {
shared: self.shared.clone(),
}
}
}

impl<T> Clone for Sender<T> {
Expand Down Expand Up @@ -1040,6 +1053,19 @@ impl<T> Receiver<T> {
pub fn same_channel(&self, other: &Receiver<T>) -> bool {
Arc::ptr_eq(&self.shared, &other.shared)
}

/// Create a new [`Sender`] for the same channel as this receiver.
///
/// This is handy when only one half of the channel is stored somewhere (such as inside a
/// data structure) and the other half needs to be recovered without threading it through
/// separately. If every sender has already been dropped, the returned sender will observe
/// the channel as disconnected, just as the receiver does.
pub fn make_sender(&self) -> Sender<T> {
self.shared.sender_count.fetch_add(1, Ordering::Relaxed);
Sender {
shared: self.shared.clone(),
}
}
}

impl<T> Clone for Receiver<T> {
Expand Down
37 changes: 37 additions & 0 deletions tests/basic.rs
Original file line number Diff line number Diff line change
Expand Up @@ -454,3 +454,40 @@ fn weak_upgrade() {
tx2.send(()).unwrap();
assert!(rx.try_recv().is_ok());
}

#[test]
fn make_receiver() {
let (tx, rx) = unbounded();
let rx2 = tx.make_receiver();
assert!(rx.same_channel(&rx2));
assert_eq!(tx.receiver_count(), 2);

tx.send(42).unwrap();
// A message is only received by a single receiver, so exactly one of the two gets it.
let a = rx.try_recv().ok();
let b = rx2.try_recv().ok();
assert_eq!(a.or(b), Some(42));

// The channel stays open as long as the recovered receiver lives.
drop(rx);
assert!(!tx.is_disconnected());
drop(rx2);
assert!(tx.is_disconnected());
}

#[test]
fn make_sender() {
let (tx, rx) = unbounded();
let tx2 = rx.make_sender();
assert!(tx.same_channel(&tx2));
assert_eq!(rx.sender_count(), 2);

tx2.send(1).unwrap();
assert_eq!(rx.try_recv().unwrap(), 1);

// The channel stays open as long as the recovered sender lives.
drop(tx);
assert!(!rx.is_disconnected());
drop(tx2);
assert!(rx.is_disconnected());
}