wasm_streams/writable/
into_async_write.rs

1use std::pin::Pin;
2use std::task::{Context, Poll};
3
4use futures_util::io::AsyncWrite;
5use futures_util::ready;
6use futures_util::sink::SinkExt;
7use js_sys::Uint8Array;
8use wasm_bindgen::JsValue;
9
10use crate::util::js_to_io_error;
11
12use super::IntoSink;
13
14/// An [`AsyncWrite`] for the [`into_async_write`](super::WritableStream::into_async_write) method.
15///
16/// This `AsyncWrite` holds a writer, and therefore locks the [`WritableStream`](super::WritableStream).
17/// When this `AsyncWrite` is dropped, it also drops its writer which in turn
18/// [releases its lock](https://streams.spec.whatwg.org/#release-a-lock).
19///
20/// [`AsyncWrite`]: https://docs.rs/futures/0.3.18/futures/io/trait.AsyncWrite.html
21#[must_use = "writers do nothing unless polled"]
22#[derive(Debug)]
23pub struct IntoAsyncWrite<'writer> {
24    sink: IntoSink<'writer>,
25}
26
27impl<'writer> IntoAsyncWrite<'writer> {
28    #[inline]
29    pub(super) fn new(sink: IntoSink<'writer>) -> Self {
30        Self { sink }
31    }
32
33    /// [Aborts](https://streams.spec.whatwg.org/#abort-a-writable-stream) the stream,
34    /// signaling that the producer can no longer successfully write to the stream.
35    pub async fn abort(self) -> Result<(), JsValue> {
36        self.sink.abort().await
37    }
38
39    /// [Aborts](https://streams.spec.whatwg.org/#abort-a-writable-stream) the stream,
40    /// signaling that the producer can no longer successfully write to the stream.
41    pub async fn abort_with_reason(self, reason: &JsValue) -> Result<(), JsValue> {
42        self.sink.abort_with_reason(reason).await
43    }
44}
45
46impl<'writer> AsyncWrite for IntoAsyncWrite<'writer> {
47    fn poll_write(
48        mut self: Pin<&mut Self>,
49        cx: &mut Context<'_>,
50        buf: &[u8],
51    ) -> Poll<std::io::Result<usize>> {
52        ready!(self
53            .as_mut()
54            .sink
55            .poll_ready_unpin(cx)
56            .map_err(js_to_io_error))?;
57        self.as_mut()
58            .sink
59            .start_send_unpin(Uint8Array::from(buf).into())
60            .map_err(js_to_io_error)?;
61        Poll::Ready(Ok(buf.len()))
62    }
63
64    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
65        self.as_mut()
66            .sink
67            .poll_flush_unpin(cx)
68            .map_err(js_to_io_error)
69    }
70
71    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
72        self.as_mut()
73            .sink
74            .poll_close_unpin(cx)
75            .map_err(js_to_io_error)
76    }
77}
wasm_streams/writable/into_async_write.rs

wasm_streams/writable/
into_async_write.rs
