support non-async {stream,future}.cancel-{read,write} (#11625)

* support non-async `{stream,future}.cancel-{read,write}`

During my earlier stream API refactoring, I had forgotten to support or test
synchronous cancellation; this commit does both.  In the process, I realized the
future API ought to be updated to support blocking cancellation just like the
stream API, so I made that change as well.

This also adds `{Source,Destination}::reborrow` functions, allowing instances of
those types to be reborrowed, such that they may be passed as parameters but
also used again.

Note that I had to move some functions from `impl ConcurrentState` to `impl
Instance` in order to access the store and suspend the current fiber when
synchronously cancelling.

Signed-off-by: Joel Dice <[email protected]>

* reduce code duplication

Signed-off-by: Joel Dice <[email protected]>

---------

Signed-off-by: Joel Dice <[email protected]>

Author: dicej

crates/misc/component-async-tests/src/util.rs

@@ -9,7 +9,7 @@ use std::{
 use wasmtime::{
     StoreContextMut,
     component::{
-        Accessor, Destination, FutureConsumer, FutureProducer, Lift, Lower, Source, StreamConsumer,
+        Destination, FutureConsumer, FutureProducer, Lift, Lower, Source, StreamConsumer,
         StreamProducer, StreamResult,
     },
 };
@@ -139,24 +139,41 @@ impl<T> OneshotProducer<T> {
 impl<D, T: Send + 'static> FutureProducer<D> for OneshotProducer<T> {
     type Item = T;
 
-    async fn produce(self, _: &Accessor<D>) -> Result<T> {
-        Ok(self.0.await?)
+    fn poll_produce(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        _: StoreContextMut<D>,
+        finish: bool,
+    ) -> Poll<Result<Option<T>>> {
+        match Pin::new(&mut self.get_mut().0).poll(cx) {
+            Poll::Pending if finish => Poll::Ready(Ok(None)),
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(result) => Poll::Ready(Ok(Some(result?))),
+        }
     }
 }
 
-pub struct OneshotConsumer<T>(oneshot::Sender<T>);
+pub struct OneshotConsumer<T>(Option<oneshot::Sender<T>>);
 
 impl<T> OneshotConsumer<T> {
     pub fn new(tx: oneshot::Sender<T>) -> Self {
-        Self(tx)
+        Self(Some(tx))
     }
 }
 
-impl<D, T: Send + 'static> FutureConsumer<D> for OneshotConsumer<T> {
+impl<D, T: Lift + Send + 'static> FutureConsumer<D> for OneshotConsumer<T> {
     type Item = T;
 
-    async fn consume(self, _: &Accessor<D>, value: T) -> Result<()> {
-        _ = self.0.send(value);
-        Ok(())
+    fn poll_consume(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        store: StoreContextMut<D>,
+        mut source: Source<'_, T>,
+        _: bool,
+    ) -> Poll<Result<()>> {
+        let value = &mut None;
+        source.read(store, value)?;
+        _ = self.get_mut().0.take().unwrap().send(value.take().unwrap());
+        Poll::Ready(Ok(()))
     }
 }