.github/dependabot.yml

@@ -0,0 +1,9 @@
+version: 2
+updates:
+  - package-ecosystem: cargo
+    directory: /
+    schedule:
+      interval: weekly
+    commit-message:
+      prefix: ''
+    labels: []

.github/workflows/ci.yml

@@ -1,16 +1,29 @@
 name: CI
 
+permissions:
+  contents: read
+
 on:
   pull_request:
   push:
     branches:
       - master
   schedule:
-    - cron: '0 2 * * *'
+    - cron: '0 2 * * 0'
 
 env:
-  RUSTFLAGS: -D warnings
+  CARGO_INCREMENTAL: 0
+  CARGO_NET_GIT_FETCH_WITH_CLI: true
+  CARGO_NET_RETRY: 10
+  CARGO_TERM_COLOR: always
   RUST_BACKTRACE: 1
+  RUSTFLAGS: -D warnings
+  RUSTDOCFLAGS: -D warnings
+  RUSTUP_MAX_RETRIES: 10
+
+defaults:
+  run:
+    shell: bash
 
 jobs:
   test:
@@ -36,7 +49,7 @@ jobs:
       matrix:
         # When updating this, the reminder to update the minimum supported
         # Rust version in Cargo.toml.
-        rust: ['1.38']
+        rust: ['1.45']
     steps:
       - uses: actions/checkout@v3
       - name: Install Rust
@@ -60,9 +73,14 @@ jobs:
       - run: cargo fmt --all --check
 
   security_audit:
+    permissions:
+      checks: write
+      contents: read
+      issues: write
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
-      - uses: actions-rs/audit-check@v1
+      # https://github.com/rustsec/audit-check/issues/2
+      - uses: rustsec/audit-check@master
         with:
           token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/release.yml

@@ -1,5 +1,8 @@
 name: Release
 
+permissions:
+  contents: write
+
 on:
   push:
     tags:

CHANGELOG.md

@@ -1,3 +1,8 @@
+# Version 1.9.0
+
+- Fix a bug where `WeakSender/WeakReceiver` could incorrectly return `Some` even if the channel is already closed (#60)
+- Remove the unnecessary `T: Clone` bound from `WeakSender/WeakReceiver`'s `Clone` implementation (#62)
+
 # Version 1.8.0
 
 - Prevent deadlock if sender/receiver is forgotten (#49)

Cargo.toml

@@ -3,10 +3,10 @@ name = "async-channel"
 # When publishing a new version:
 # - Update CHANGELOG.md
 # - Create "v1.x.y" git tag
-version = "1.8.0"
+version = "1.9.0"
 authors = ["Stjepan Glavina <stjepang@gmail.com>"]
 edition = "2018"
-rust-version = "1.38"
+rust-version = "1.45"
 description = "Async multi-producer multi-consumer channel"
 license = "Apache-2.0 OR MIT"
 repository = "https://github.com/smol-rs/async-channel"

src/lib.rs

@@ -828,7 +828,6 @@ impl<T> futures_core::stream::FusedStream for Receiver<T> {
 ///
 /// This is created through the [`Sender::downgrade`] method. In order to use it, it needs
 /// to be upgraded into a [`Sender`] through the `upgrade` method.
-#[derive(Clone)]
 pub struct WeakSender<T> {
     channel: Arc<Channel<T>>,
 }
@@ -839,23 +838,32 @@ impl<T> WeakSender<T> {
         if self.channel.queue.is_closed() {
             None
         } else {
-            let old_count = self.channel.sender_count.fetch_add(1, Ordering::Relaxed);
-            if old_count == 0 {
-                // Channel was closed while we were incrementing the count.
-                self.channel.sender_count.store(0, Ordering::Release);
-                None
-            } else if old_count > usize::MAX / 2 {
-                // Make sure the count never overflows, even if lots of sender clones are leaked.
-                process::abort();
-            } else {
-                Some(Sender {
+            match self.channel.sender_count.fetch_update(
+                Ordering::Relaxed,
+                Ordering::Relaxed,
+                |count| if count == 0 { None } else { Some(count + 1) },
+            ) {
+                Err(_) => None,
+                Ok(new_value) if new_value > usize::MAX / 2 => {
+                    // Make sure the count never overflows, even if lots of sender clones are leaked.
+                    process::abort();
+                }
+                Ok(_) => Some(Sender {
                     channel: self.channel.clone(),
-                })
+                }),
             }
         }
     }
 }
 
+impl<T> Clone for WeakSender<T> {
+    fn clone(&self) -> Self {
+        WeakSender {
+            channel: self.channel.clone(),
+        }
+    }
+}
+
 impl<T> fmt::Debug for WeakSender<T> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "WeakSender {{ .. }}")
@@ -866,7 +874,6 @@ impl<T> fmt::Debug for WeakSender<T> {
 ///
 /// This is created through the [`Receiver::downgrade`] method. In order to use it, it needs
 /// to be upgraded into a [`Receiver`] through the `upgrade` method.
-#[derive(Clone)]
 pub struct WeakReceiver<T> {
     channel: Arc<Channel<T>>,
 }
@@ -877,24 +884,33 @@ impl<T> WeakReceiver<T> {
         if self.channel.queue.is_closed() {
             None
         } else {
-            let old_count = self.channel.receiver_count.fetch_add(1, Ordering::Relaxed);
-            if old_count == 0 {
-                // Channel was closed while we were incrementing the count.
-                self.channel.receiver_count.store(0, Ordering::Release);
-                None
-            } else if old_count > usize::MAX / 2 {
-                // Make sure the count never overflows, even if lots of receiver clones are leaked.
-                process::abort();
-            } else {
-                Some(Receiver {
+            match self.channel.receiver_count.fetch_update(
+                Ordering::Relaxed,
+                Ordering::Relaxed,
+                |count| if count == 0 { None } else { Some(count + 1) },
+            ) {
+                Err(_) => None,
+                Ok(new_value) if new_value > usize::MAX / 2 => {
+                    // Make sure the count never overflows, even if lots of receiver clones are leaked.
+                    process::abort();
+                }
+                Ok(_) => Some(Receiver {
                     channel: self.channel.clone(),
                     listener: None,
-                })
+                }),
             }
         }
     }
 }
 
+impl<T> Clone for WeakReceiver<T> {
+    fn clone(&self) -> Self {
+        WeakReceiver {
+            channel: self.channel.clone(),
+        }
+    }
+}
+
 impl<T> fmt::Debug for WeakReceiver<T> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "WeakReceiver {{ .. }}")