Implemented hash_receiver method on mpsc

futures-channel/src/mpsc/mod.rs

@@ -552,6 +552,13 @@ impl<T> SenderInner<T> {
         Arc::ptr_eq(&self.inner, &other.inner)
     }
 
+    /// Returns pointer to the Arc containing sender
+    ///
+    /// The returned pointer is not referenced and should be only used for hashing!
+    fn ptr(&self) -> *const Inner<T> {
+        &*self.inner
+    }
+
     /// Returns whether this channel is closed without needing a context.
     fn is_closed(&self) -> bool {
         !decode_state(self.inner.state.load(SeqCst)).is_open
@@ -666,6 +673,14 @@ impl<T> Sender<T> {
             _ => false,
         }
     }
+
+    /// Hashes the receiver into the provided hasher
+    pub fn hash_receiver<H>(&self, hasher: &mut H) where H: std::hash::Hasher {
+        use std::hash::Hash;
+
+        let ptr = self.0.as_ref().map(|inner| inner.ptr());
+        ptr.hash(hasher);
+    }
 }
 
 impl<T> UnboundedSender<T> {
@@ -739,6 +754,14 @@ impl<T> UnboundedSender<T> {
             _ => false,
         }
     }
+
+    /// Hashes the receiver into the provided hasher
+    pub fn hash_receiver<H>(&self, hasher: &mut H) where H: std::hash::Hasher {
+        use std::hash::Hash;
+
+        let ptr = self.0.as_ref().map(|inner| inner.ptr());
+        ptr.hash(hasher);
+    }
 }
 
 impl<T> Clone for Sender<T> {

futures-channel/tests/mpsc.rs

@@ -529,6 +529,55 @@ fn same_receiver() {
     assert!(txb1.same_receiver(&txb2));
 }
 
+#[test]
+fn hash_receiver() {
+    use std::hash::Hasher;
+    use std::collections::hash_map::DefaultHasher;
+
+    let mut hasher_a1 = DefaultHasher::new();
+    let mut hasher_a2 = DefaultHasher::new();
+    let mut hasher_b1 = DefaultHasher::new();
+    let mut hasher_b2 = DefaultHasher::new();
+    let (mut txa1, _) = mpsc::channel::<i32>(1);
+    let txa2 = txa1.clone();
+
+    let (mut txb1, _) = mpsc::channel::<i32>(1);
+    let txb2 = txb1.clone();
+
+    txa1.hash_receiver(&mut hasher_a1);
+    let hash_a1 = hasher_a1.finish();
+    txa2.hash_receiver(&mut hasher_a2);
+    let hash_a2 = hasher_a2.finish();
+    txb1.hash_receiver(&mut hasher_b1);
+    let hash_b1 = hasher_b1.finish();
+    txb2.hash_receiver(&mut hasher_b2);
+    let hash_b2 = hasher_b2.finish();
+
+    assert_eq!(hash_a1, hash_a2);
+    assert_eq!(hash_b1, hash_b2);
+    assert!(hash_a1 != hash_b1);
+
+    txa1.disconnect();
+    txb1.close_channel();
+
+    let mut hasher_a1 = DefaultHasher::new();
+    let mut hasher_a2 = DefaultHasher::new();
+    let mut hasher_b1 = DefaultHasher::new();
+    let mut hasher_b2 = DefaultHasher::new();
+
+    txa1.hash_receiver(&mut hasher_a1);
+    let hash_a1 = hasher_a1.finish();
+    txa2.hash_receiver(&mut hasher_a2);
+    let hash_a2 = hasher_a2.finish();
+    txb1.hash_receiver(&mut hasher_b1);
+    let hash_b1 = hasher_b1.finish();
+    txb2.hash_receiver(&mut hasher_b2);
+    let hash_b2 = hasher_b2.finish();
+
+    assert!(hash_a1 != hash_a2);
+    assert_eq!(hash_b1, hash_b2);
+}
+
 #[test]
 fn send_backpressure() {
     let (waker, counter) = new_count_waker();