fix: improved API for unsync actions that doesn't require SendWrapper on input

examples/server_fns_axum/src/app.rs

@@ -1,6 +1,7 @@
 use futures::StreamExt;
 use http::Method;
 use leptos::{html::Input, prelude::*, spawn::spawn_local};
+use send_wrapper::SendWrapper;
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use server_fn::{
     client::{browser::BrowserClient, Client},
@@ -361,8 +362,8 @@ pub fn FileUpload() -> impl IntoView {
         Ok(count)
     }
 
-    let upload_action = Action::new_unsync(|data: &FormData| {
-        let data = data.to_owned();
+    let upload_action = Action::new(|data: &SendWrapper<FormData>| {
+        let data = (**data).clone();
         // `MultipartData` implements `From<FormData>`
         file_length(data.into())
     });
@@ -374,7 +375,7 @@ pub fn FileUpload() -> impl IntoView {
             ev.prevent_default();
             let target = ev.target().unwrap().unchecked_into::<HtmlFormElement>();
             let form_data = FormData::new_with_form(&target).unwrap();
-            upload_action.dispatch_unsync(form_data);
+            upload_action.dispatch(SendWrapper::new(form_data));
         }>
             <input type="file" name="file_to_upload"/>
             <input type="submit"/>
@@ -554,8 +555,7 @@ pub fn FileUploadWithProgress() -> impl IntoView {
         </form>
         {move || filename.get().map(|filename| view! { <p>Uploading {filename}</p> })}
         {move || {
-            max
-                .get()
+            max.get()
                 .map(|max| {
                     view! {
                         <progress
@@ -632,6 +632,7 @@ pub fn FileWatcher() -> impl IntoView {
                     })
                     .collect::<Vec<_>>()
             }}
+
         </ul>
         <p>
             <em>

reactive_graph/src/actions/action.rs

@@ -260,9 +260,59 @@ where
             });
         }
     }
+
+    #[track_caller]
+    pub fn dispatch_local(&self, input: I) {
+        if !is_suppressing_resource_load() {
+            let mut fut = (self.action_fn)(&input).fuse();
+
+            // abort this task if the owner is cleaned up
+            let (abort_tx, mut abort_rx) = oneshot::channel();
+            Owner::on_cleanup(move || {
+                abort_tx.send(()).expect(
+                    "tried to cancel a future in ArcAction::dispatch(), but \
+                     the channel has already closed",
+                );
+            });
+
+            // Update the state before loading
+            self.in_flight.update(|n| *n += 1);
+            let current_version =
+                self.version.try_get_untracked().unwrap_or_default();
+            self.input.try_update(|inp| *inp = Some(input));
+
+            // Spawn the task
+            Executor::spawn_local({
+                let input = self.input.clone();
+                let version = self.version.clone();
+                let value = self.value.clone();
+                let in_flight = self.in_flight.clone();
+                async move {
+                    select! {
+                        // if the abort message has been sent, bail and do nothing
+                        _ = abort_rx => {
+                            in_flight.update(|n| *n = n.saturating_sub(1));
+                        },
+                        // otherwise, update the value
+                        result = fut => {
+                            in_flight.update(|n| *n = n.saturating_sub(1));
+                            let is_latest = version.get_untracked() <= current_version;
+                            if is_latest {
+                                version.update(|n| *n += 1);
+                                value.update(|n| *n = Some(result));
+                            }
+                            if in_flight.get_untracked() == 0 {
+                                input.update(|inp| *inp = None);
+                            }
+                        }
+                    }
+                }
+            });
+        }
+    }
 }
 
-impl<I, O> ArcAction<SendWrapper<I>, O>
+impl<I, O> ArcAction<I, O>
 where
     I: 'static,
     O: Send + Sync + 'static,
@@ -302,11 +352,6 @@ where
             defined_at: Location::caller(),
         }
     }
-
-    #[track_caller]
-    pub fn dispatch_unsync(&self, input: I) {
-        self.dispatch(SendWrapper::new(input));
-    }
 }
 
 impl<I, O> ArcAction<I, O> {
@@ -764,9 +809,9 @@ where
     }
 }
 
-impl<I, O> Action<SendWrapper<I>, O>
+impl<I, O> Action<I, O>
 where
-    I: 'static,
+    I: Send + Sync + 'static,
     O: Send + Sync + 'static,
 {
     /// Creates a new action, which does not require its input to be `Send`.
@@ -803,11 +848,6 @@ where
             defined_at: Location::caller(),
         }
     }
-
-    #[track_caller]
-    pub fn dispatch_unsync(&self, input: I) {
-        self.dispatch(SendWrapper::new(input));
-    }
 }
 
 impl<I, O> DefinedAt for Action<I, O>