Why isn't there a watch::Sender::new in tokio::sync

help
#1

The only way to create a single tokio::sync::watch::Sender without a Receiver seems to be creating a channel and dropping the Receiver

use tokio::sync::watch;

#[allow(unused_variables)]
fn main() {
    let (sender, _) = watch::channel(());
    // or:
    let sender = watch::channel(()).0;
}

(Playground)

By having to use channel instead of being able to create a Sender through soemthing like Sender::new, I needlessly create a Receiver, even if I don't need one (yet). Note that it's always possible to create Receivers later:

use tokio::sync::watch;

fn main() {
    let sender = watch::channel(()).0;
    println!("{:?}", sender.send(()));
    let _receiver = sender.subscribe();
    println!("{:?}", sender.send(()));
}

(Playground)

(Also note Tokio issue #4957 which I filed in that context.)

So I wonder if there is a particular reason why there is no tokio::sync::watch::Sender::new. The same applies to tokio::sync::broadcast::Sender. Maybe it just has been forgotten?

#2

There isn't really any reason other than consistency with mpsc and oneshot channels.

1 Like
#3

Do you think it's okay to file a feature request then?

#4

Yes.

#5

Thank you very much for your feedback. I created a feature request here:

https://github.com/tokio-rs/tokio/issues/4958

#6

I actually ran into a problem when I worked with Senders that have no Receiver. Consider the following example:

use tokio::sync::watch;

#[allow(unused_variables)]
fn main() {
    let (sender, receiver) = watch::channel("initial");
    //drop(receiver); // try to uncomment
    sender.send("updated").ok();
    let receiver2 = sender.subscribe();
    let s = *receiver2.borrow();
    assert_eq!(s, "updated"); // fails if `receiver` is dropped before `send`
}

(Playground)

Apparently the Sender doesn't retain the last value if it was sent when there was no Receiver. Is this intended?

The current behavior is kind-of documented here:

pub fn send(&self, value: T) -> Result<(), SendError>

Sends a new value via the channel, notifying all receivers.

This method fails if the channel has been closed, which happens when every receiver has been dropped.

It's not 100% clear what "fails" means in this context (i.e. returning Err(_) or failing to update the value). Either way, I think it would be helpful if the sent value was retained for a future subscriber.

Due to the current behavior, I have to keep a "dummy" Receiver in my code:

pub struct Receiver<T> {
    watch: watch::Sender<Option<broadcast::Sender<Message<T>>>>,
    _dummy: watch::Receiver<Option<broadcast::Sender<Message<T>>>>,
}

impl<T> Receiver<T>
where
    T: Clone,
{
    pub fn new() -> Self {
        let (sender, receiver) = watch::channel(None);
        Self {
            watch: sender,
            _dummy: receiver,
        }
    }
}

Update: Using Sender::send_replace is a simpler solution than keeping a dummy receiver around.

More trouble with tokio::sync::watch
#7

Well, yes, the failure behavior is quite unfortunate. It was designed that way because you couldn't reopen the watch channel after closing it in the past, but now that you can reopen it, it would be preferable for send to not fail.

#8

If you want, I can open an issue on it. I did make a comment to the previous two issues though already.

Not sure if this can/should be fixed without breaking compatibility. The documentation isn't very clear (yet), and I would assume that the chances of someone relying on the current behavior is pretty low, but not sure.

#9

I would always be happy to see additional documentation!

The behavior cannot be fixed. The value is returned in the error, so there isn't a way to change the behavior without changing the error type.

1 Like
#10

Ah, I missed that.