fastly / pushpin

A proxy server for adding push to your API, used at the core of Fastly's Fanout service
https://pushpin.org
Apache License 2.0
3.66k stars 153 forks source link

channel: add wait_sendable #48071

Closed jkarneges closed 1 month ago

jkarneges commented 1 month ago

There are some places where instead of asynchronously sending a value to a channel (sender.send(value).await), we asynchronously wait for the ability to send to a channel and then immediately send the value after:

sender.check_send().await;

// immediate send without await
sender.try_send(value);

This API allows the opportunity to defer certain processing until the moment of the actual send. The problem is its usage is a bit fiddly and error-prone. Notably, if the code waiting for the ability to send decides to give up waiting, or if it finishes waiting but then decides not to send anything, it needs to explicitly cancel the operation (sender.cancel()). Failing to cancel can lead to broken coordination when there are multiple senders waiting to send.

This PR introduces a simpler API:

let send: SendOnce = sender.wait_sendable().await;
send.try_send(value);

If the future returned by wait_sendable() is dropped before completion, the operation is automatically canceled. If it completes, it produces a SendOnce struct which can be used to send up to one value via its try_send() method. If the SendOnce is dropped without sending anything, the operation is automatically canceled. Basically it makes the API hard to misuse.

Note that the method is named try_send() and not send() because the channel could still become full after obtaining the SendOnce, if another sender were to send to the channel.

This PR also uses the new API in a couple of places.