Test

Author: paberr

README.md

@@ -182,28 +182,23 @@ pub async fn process_with_progress(data: Vec<u8>, channel: Channel) -> ProcessRe
 Then use the `webworker_channel!` macro and `run_channel` method:
 
 ```rust,ignore
-use wasmworker::{webworker_channel, Channel, WebWorker};
+use wasmworker::{webworker_channel, WebWorker};
 
 let worker = WebWorker::new(None).await?;
 
-// Create a channel for bidirectional communication
-let (main_channel, worker_port) = Channel::new()?;
-
-// Start the async task
-let task = worker.run_channel(
-   webworker_channel!(process_with_progress),
-   &data,
-   worker_port,
-);
+// Start the async task — returns a ChannelTask for communication + result
+let task = worker
+   .run_channel(webworker_channel!(process_with_progress), &data)
+   .await;
 
 // Receive progress from worker
-let progress: Progress = main_channel.recv().await.unwrap();
+let progress: Progress = task.recv().await.unwrap();
 
 // Send response back to worker
-main_channel.send(&Continue { should_continue: true });
+task.send(&Continue { should_continue: true });
 
 // Wait for task completion
-let result = task.await;
+let result = task.result().await;
 ```
 
 ### Bundler support (Vite)

src/channel_task.rs

@@ -0,0 +1,78 @@
+use std::marker::PhantomData;
+
+use serde::{de::DeserializeOwned, Serialize};
+use tokio::sync::oneshot;
+
+use crate::{channel::Channel, convert::from_bytes};
+
+/// A handle to a running channel task on a WebWorker.
+///
+/// `ChannelTask` combines a bidirectional [`Channel`] for sending and receiving
+/// messages with the worker, and a future that resolves to the task's final result.
+///
+/// This type is returned by [`crate::WebWorker::run_channel`] and
+/// [`crate::pool::WebWorkerPool::run_channel`]. It allows you to exchange messages
+/// with the worker (e.g., for progress reporting) and then consume the final result.
+///
+/// # Example
+///
+/// ```ignore
+/// let task = worker
+///     .run_channel(webworker_channel!(process_with_progress), &data)
+///     .await;
+///
+/// let progress: Progress = task.recv().await.expect("progress");
+/// task.send(&Continue { should_continue: true });
+///
+/// let result: ProcessResult = task.result().await;
+/// ```
+pub struct ChannelTask<R> {
+    channel: Channel,
+    result_rx: oneshot::Receiver<Vec<u8>>,
+    _phantom: PhantomData<R>,
+}
+
+impl<R: DeserializeOwned> ChannelTask<R> {
+    /// Create a new `ChannelTask` from a channel and a result receiver.
+    pub(crate) fn new(channel: Channel, result_rx: oneshot::Receiver<Vec<u8>>) -> Self {
+        Self {
+            channel,
+            result_rx,
+            _phantom: PhantomData,
+        }
+    }
+
+    /// Receive the next deserialized message from the worker.
+    ///
+    /// Returns `None` if the channel's sender side has been dropped
+    /// (i.e., the worker has finished and closed the channel).
+    pub async fn recv<T: DeserializeOwned>(&self) -> Option<T> {
+        self.channel.recv().await
+    }
+
+    /// Receive raw bytes from the worker.
+    ///
+    /// Returns `None` if the channel's sender side has been dropped.
+    pub async fn recv_bytes(&self) -> Option<Box<[u8]>> {
+        self.channel.recv_bytes().await
+    }
+
+    /// Send a serialized message to the worker.
+    pub fn send<T: Serialize>(&self, msg: &T) {
+        self.channel.send(msg);
+    }
+
+    /// Send raw bytes to the worker.
+    pub fn send_bytes(&self, bytes: &[u8]) {
+        self.channel.send_bytes(bytes);
+    }
+
+    /// Await the task's final result, consuming the `ChannelTask`.
+    pub async fn result(self) -> R {
+        let bytes = self
+            .result_rx
+            .await
+            .expect("WebWorker result sender dropped");
+        from_bytes(&bytes)
+    }
+}

src/lib.rs

@@ -1,17 +1,21 @@
 #![doc = include_str!("../README.md")]
 #![allow(clippy::borrowed_box)]
 pub use channel::Channel;
+pub use channel_task::ChannelTask;
 pub use global::{
     has_worker_pool, init_optimized_worker_pool, init_worker_pool, worker_pool, AlreadyInitialized,
 };
 pub use pool::WorkerPoolOptions;
-pub use web_sys::MessagePort;
 pub use webworker::WebWorker;
 
+#[doc(hidden)]
+pub use web_sys::MessagePort;
+
 // Re-export WebWorkerPool from pool module
 pub use pool::WebWorkerPool;
 
 mod channel;
+mod channel_task;
 pub mod convert;
 pub mod error;
 pub mod func;

src/pool/mod.rs

@@ -7,9 +7,10 @@ pub use scheduler::Strategy;
 use serde::{Deserialize, Serialize};
 
 use wasm_bindgen_futures::JsFuture;
-use web_sys::{window, MessagePort};
+use web_sys::window;
 
 use crate::{
+    channel_task::ChannelTask,
     error::InitError,
     func::{WebWorkerChannelFn, WebWorkerFn},
     WebWorker,
@@ -211,25 +212,29 @@ impl WebWorkerPool {
 
     /// Run an async function with bidirectional channel support on this [`WebWorkerPool`].
     ///
+    /// Returns a [`ChannelTask`] that provides both the communication channel and the
+    /// task result. The `MessageChannel` is created internally.
+    ///
     /// The `func`: [`WebWorkerChannelFn`] argument should normally be instantiated using the
     /// [`crate::webworker_channel!`] macro. This ensures type safety and that the function
     /// is correctly exposed to the worker.
     ///
     /// Example:
     /// ```ignore
-    /// worker_pool().await.run_channel(webworker_channel!(process_with_progress), &my_data, port).await
+    /// let task = worker_pool().await
+    ///     .run_channel(webworker_channel!(process_with_progress), &data)
+    ///     .await;
+    ///
+    /// let progress: Progress = task.recv().await.expect("progress");
+    /// task.send(&Continue { should_continue: true });
+    /// let result: ProcessResult = task.result().await;
     /// ```
-    pub async fn run_channel<T, R>(
-        &self,
-        func: WebWorkerChannelFn<T, R>,
-        arg: &T,
-        port: MessagePort,
-    ) -> R
+    pub async fn run_channel<T, R>(&self, func: WebWorkerChannelFn<T, R>, arg: &T) -> ChannelTask<R>
     where
         T: Serialize + for<'de> Deserialize<'de>,
         R: Serialize + for<'de> Deserialize<'de>,
     {
-        self.run_channel_internal(func, arg, port).await
+        self.run_channel_internal(func, arg).await
     }
 
     /// This function can outsource a task on a [`WebWorkerPool`] which has `Box<[u8]>` both as input and output.
@@ -271,15 +276,14 @@ impl WebWorkerPool {
         &self,
         func: WebWorkerChannelFn<T, R>,
         arg: &T,
-        port: MessagePort,
-    ) -> R
+    ) -> ChannelTask<R>
     where
         T: Serialize + for<'de> Deserialize<'de>,
         R: Serialize + for<'de> Deserialize<'de>,
     {
         let worker_id = self.scheduler.schedule(self);
         self.workers[worker_id]
-            .run_channel_internal(func, arg, port)
+            .run_channel_internal(func, arg)
             .await
     }
 

src/webworker/worker.rs

@@ -12,10 +12,13 @@ use serde::{Deserialize, Serialize};
 use tokio::sync::{oneshot, Semaphore};
 use wasm_bindgen::{prelude::Closure, JsCast, JsValue, UnwrapThrowExt};
 use web_sys::{
-    Blob, BlobPropertyBag, MessageEvent, MessagePort, Url, Worker, WorkerOptions, WorkerType,
+    Blob, BlobPropertyBag, MessageChannel, MessageEvent, MessagePort, Url, Worker, WorkerOptions,
+    WorkerType,
 };
 
 use crate::{
+    channel::Channel,
+    channel_task::ChannelTask,
     convert::{from_bytes, to_bytes},
     error::{Full, InitError},
     func::{WebWorkerChannelFn, WebWorkerFn},
@@ -234,6 +237,10 @@ impl WebWorker {
 
     /// Run an async function with bidirectional channel support on this [`WebWorker`].
     ///
+    /// Returns a [`ChannelTask`] that provides both the communication channel and the
+    /// task result. The `MessageChannel` is created internally — callers interact only
+    /// through the returned `ChannelTask`.
+    ///
     /// The `func`: [`WebWorkerChannelFn`] argument should normally be instantiated using the
     /// [`crate::webworker_channel!`] macro. This ensures type safety and that the function
     /// is correctly exposed to the worker.
@@ -242,19 +249,20 @@ impl WebWorker {
     ///
     /// Example:
     /// ```ignore
-    /// worker.run_channel(webworker_channel!(process_with_progress), &my_data, port).await
+    /// let task = worker
+    ///     .run_channel(webworker_channel!(process_with_progress), &data)
+    ///     .await;
+    ///
+    /// let progress: Progress = task.recv().await.expect("progress");
+    /// task.send(&Continue { should_continue: true });
+    /// let result: ProcessResult = task.result().await;
     /// ```
-    pub async fn run_channel<T, R>(
-        &self,
-        func: WebWorkerChannelFn<T, R>,
-        arg: &T,
-        port: MessagePort,
-    ) -> R
+    pub async fn run_channel<T, R>(&self, func: WebWorkerChannelFn<T, R>, arg: &T) -> ChannelTask<R>
     where
         T: Serialize + for<'de> Deserialize<'de>,
         R: Serialize + for<'de> Deserialize<'de>,
     {
-        self.run_channel_internal(func, arg, port).await
+        self.run_channel_internal(func, arg).await
     }
 
     /// This function differs from [`WebWorker::run`] by returning early if the given task limit is reached.
@@ -364,12 +372,13 @@ impl WebWorker {
     }
 
     /// Internal function to schedule a channel task to the worker.
+    /// Creates a `MessageChannel` internally, sends one port to the worker,
+    /// and returns a `ChannelTask` wrapping the other port and the result future.
     pub(crate) async fn run_channel_internal<T, R>(
         &self,
         func: WebWorkerChannelFn<T, R>,
         arg: &T,
-        port: MessagePort,
-    ) -> R
+    ) -> ChannelTask<R>
     where
         T: Serialize + for<'de> Deserialize<'de>,
         R: Serialize + for<'de> Deserialize<'de>,
@@ -381,8 +390,15 @@ impl WebWorker {
             None
         };
 
-        // Convert arg and result.
-        self.force_run(func.name, arg, true, Some(port)).await
+        // Create the MessageChannel internally.
+        let msg_channel = MessageChannel::new().expect_throw("Could not create MessageChannel");
+        let channel = Channel::from(msg_channel.port1());
+        let worker_port = msg_channel.port2();
+
+        // Send the request and get a receiver for the result bytes.
+        let result_rx = self.send_channel_request(func.name, arg, worker_port);
+
+        ChannelTask::new(channel, result_rx)
     }
 
     /// This function handles the communication with the worker
@@ -447,6 +463,50 @@ impl WebWorker {
             .expect_throw("Could not find function")
     }
 
+    /// Sends a channel request to the worker and returns a receiver for the result bytes.
+    /// Unlike `send_request`, this does not await the result — it returns immediately
+    /// so the caller can interact with the channel before consuming the result.
+    fn send_channel_request<T>(
+        &self,
+        func_name: &'static str,
+        arg: &T,
+        port: MessagePort,
+    ) -> oneshot::Receiver<Vec<u8>>
+    where
+        T: Serialize + for<'de> Deserialize<'de>,
+    {
+        let id = self.current_task.fetch_add(1, Ordering::Relaxed);
+        let request = Request {
+            id,
+            func_name,
+            is_channel: true,
+            arg: to_bytes(arg),
+        };
+
+        let (sender, receiver) = oneshot::channel();
+        self.open_tasks.borrow_mut().insert(id, sender);
+
+        let transfer = Array::new();
+        transfer.push(&port);
+
+        self.worker
+            .post_message_with_transfer(
+                &serde_wasm_bindgen::to_value(&request).expect_throw("Could not serialize request"),
+                &transfer,
+            )
+            .expect_throw("WebWorker gone");
+
+        // Map the receiver to extract just the response bytes.
+        let (byte_sender, byte_receiver) = oneshot::channel();
+        wasm_bindgen_futures::spawn_local(async move {
+            if let Ok(response) = receiver.await {
+                let _ = byte_sender.send(response.response.expect("Could not find function"));
+            }
+        });
+
+        byte_receiver
+    }
+
     /// Return the current capacity for new tasks.
     pub fn capacity(&self) -> Option<usize> {
         self.task_limit.as_ref().map(|s| s.available_permits())