Awaiting on WatchStream channel gets output twice

[reza-ebrahimi]

Version

[dependencies]
tokio = { version = "1.1.0", features = ["full"] }
tokio-stream = { version = "0.1.5", features = ["sync"] }

Platform

$ uname -a
Linux dev 5.4.0-67-generic #75~18.04.1-Ubuntu SMP Tue Feb 23 19:17:50 UTC 2021 x86_64 x86_64 x86_64 GNU/Linux

Description
Awaiting on WatchStream in a loop, will get the last value twice, While were sending once.

Testcode:

#[tokio::main]
async fn main() {
    let (tx, rx) = watch::channel("hello");

    let mut stream = WatchStream::new(rx).map(|payload| {
        println!("{}", payload);
    });

    let task = tokio::spawn(async move {
        loop {
          stream.next().await;
        }
    });

    // Send goodbye just once
    tx.send("goodbye").unwrap();

    task.await.ok();
}

Output:

goodbye
goodbye

[Darksonn]

I don't think this can be fixed without exposing new methods on the watch receiver in Tokio. E.g. we might expose a version method. This would involve changing send to increment version before the write lock is released, as it is otherwise not possible to reliably get the current version of the value returned by borrow(). We could also add a method similar to changed() that returns a reference to the actual value.

Thoughts @carllerche ?