1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
use std::cell::RefCell;
use std::panic::AssertUnwindSafe;
use std::pin::Pin;
use std::rc::Rc;
use futures_util::{Sink, SinkExt};
use js_sys::Promise;
use wasm_bindgen::prelude::*;
use wasm_bindgen_futures::future_to_promise;
#[wasm_bindgen]
pub(crate) struct IntoUnderlyingSink {
inner: Rc<RefCell<Inner>>,
}
impl IntoUnderlyingSink {
pub fn new(sink: Box<dyn Sink<JsValue, Error = JsValue>>) -> Self {
IntoUnderlyingSink {
inner: Rc::new(RefCell::new(Inner::new(sink))),
}
}
}
#[allow(clippy::await_holding_refcell_ref)]
#[wasm_bindgen]
impl IntoUnderlyingSink {
pub fn write(&mut self, chunk: JsValue) -> Promise {
let inner = self.inner.clone();
// SAFETY: We use the take-and-replace pattern in Inner::write() to ensure
// that if a panic occurs, the sink is already taken out of the Option,
// leaving it in a clean None state. This prevents use of corrupted state
// after a panic is caught.
future_to_promise(AssertUnwindSafe(async move {
// This mutable borrow can never panic, since the WritableStream always queues
// each operation on the underlying sink.
let mut inner = inner.try_borrow_mut().unwrap_throw();
inner.write(chunk).await.map(|_| JsValue::undefined())
}))
}
pub fn close(self) -> Promise {
// SAFETY: Inner::close() takes the sink before the fallible operation.
future_to_promise(AssertUnwindSafe(async move {
let mut inner = self.inner.try_borrow_mut().unwrap_throw();
inner.close().await.map(|_| JsValue::undefined())
}))
}
pub fn abort(self, reason: JsValue) -> Promise {
// SAFETY: Inner::abort() just sets sink to None, no fallible operation.
future_to_promise(AssertUnwindSafe(async move {
let mut inner = self.inner.try_borrow_mut().unwrap_throw();
inner.abort(reason).await.map(|_| JsValue::undefined())
}))
}
}
struct Inner {
sink: Option<Pin<Box<dyn Sink<JsValue, Error = JsValue>>>>,
}
impl Inner {
fn new(sink: Box<dyn Sink<JsValue, Error = JsValue>>) -> Self {
Inner {
sink: Some(sink.into()),
}
}
async fn write(&mut self, chunk: JsValue) -> Result<(), JsValue> {
// Take the sink out before the fallible/panickable operation.
// This ensures that if a panic occurs, self.sink is already None,
// so any subsequent call will fail cleanly instead of using corrupted state.
let mut sink = self.sink.take().unwrap_throw();
match sink.send(chunk).await {
Ok(()) => {
// Success: put the sink back for reuse
self.sink = Some(sink);
Ok(())
}
Err(err) => {
// Error: the sink is dropped, self.sink remains None
Err(err)
}
}
// Panic: sink is dropped during unwind, self.sink remains None
}
async fn close(&mut self) -> Result<(), JsValue> {
// Take ownership and close - sink is dropped after close completes
self.sink.take().unwrap_throw().close().await
}
async fn abort(&mut self, _reason: JsValue) -> Result<(), JsValue> {
// Take and drop the sink immediately
self.sink = None;
Ok(())
}
}