cancel_safe_futures/sink/
permit.rs

1use core::{marker::PhantomData, pin::Pin};
2use futures_sink::Sink;
3use futures_util::{sink::Flush, SinkExt};
4
5/// A permit to send an item to a sink.
6///
7/// Permits are issued by the [`reserve`] and [`flush_reserve`] adapters, and indicate that
8/// [`Sink::poll_ready`] has completed and that the sink is now ready to accept an item.
9///
10/// [`reserve`]: crate::sink::SinkExt::reserve
11/// [`flush_reserve`]: crate::sink::SinkExt::flush_reserve
12#[derive(Debug)]
13#[must_use]
14pub struct Permit<'a, Si: ?Sized, Item> {
15    sink: &'a mut Si,
16    _phantom: PhantomData<fn(Item)>,
17}
18
19// By default, Unpin would be implemented for Permit even if Si isn't Unpin. But we explicitly only
20// support Unpin sinks.
21impl<Si: Unpin + ?Sized, Item> Unpin for Permit<'_, Si, Item> {}
22
23impl<'a, Item, Si: Sink<Item> + Unpin + ?Sized> Permit<'a, Si, Item> {
24    pub(super) fn new(sink: &'a mut Si) -> Self {
25        Self {
26            sink,
27            _phantom: PhantomData,
28        }
29    }
30
31    /// Sends an item to the sink, akin to the [`SinkExt::feed`] adapter.
32    ///
33    /// Unlike [`SinkExt::feed`], `Permit::feed` is a synchronous method. This is because a `Permit`
34    /// indicates that [`Sink::poll_ready`] has been called already, so the sink is immediately
35    /// ready to accept an item.
36    pub fn feed(self, item: Item) -> Result<(), Si::Error> {
37        Pin::new(self.sink).start_send(item)
38    }
39
40    /// Sends an item to the sink and then flushes it, akin to the [`SinkExt::send`] adapter.
41    ///
42    /// Unlike [`SinkExt::send`], `Permit::send` has two parts:
43    ///
44    /// 1. A synchronous part, which sends the item to the sink. This part is identical to
45    ///    [`Self::feed`].
46    /// 2. An asynchronous part, which flushes the sink via the [`SinkExt::flush`] adapter.
47    ///
48    /// This structure means that users get immediate feedback about, and can then await the
49    /// resulting [`Flush`] future, or cancel it if necessary.
50    ///
51    /// # Cancel safety
52    ///
53    /// The returned [`Flush`] future is cancel-safe. If it is dropped, the sink will no longer be
54    /// flushed. It is recommended that `flush()` be called explicitly, either by itself or via
55    /// the [`flush_reserve`](crate::SinkExt::flush_reserve) adapter.
56    pub fn send(mut self, item: Item) -> Result<Flush<'a, Si, Item>, Si::Error> {
57        Pin::new(&mut self.sink).start_send(item)?;
58        Ok(self.sink.flush())
59    }
60}
cancel_safe_futures/sink/permit.rs

cancel_safe_futures/sink/
permit.rs
