Macro for marking future as Send

worker-macros/src/lib.rs

@@ -2,6 +2,8 @@ mod durable_object;
 mod event;
 
 use proc_macro::TokenStream;
+use quote::quote;
+use syn::{parse_macro_input, ItemFn};
 
 #[proc_macro_attribute]
 pub fn durable_object(_attr: TokenStream, item: TokenStream) -> TokenStream {
@@ -21,3 +23,27 @@ pub fn event(attr: TokenStream, item: TokenStream) -> TokenStream {
 pub fn event(attr: TokenStream, item: TokenStream) -> TokenStream {
     event::expand_macro(attr, item, false)
 }
+
+#[proc_macro_attribute]
+pub fn send(_attr: TokenStream, stream: TokenStream) -> TokenStream {
+    let stream_clone = stream.clone();
+    let input = parse_macro_input!(stream_clone as ItemFn);
+
+    let ItemFn {
+        attrs,
+        vis,
+        sig,
+        block,
+    } = input;
+    let stmts = &block.stmts;
+
+    let tokens = quote! {
+        #(#attrs)* #vis #sig {
+            worker::SendFuture::new(async {
+                #(#stmts)*
+            }).await
+        }
+    };
+
+    TokenStream::from(tokens)
+}

worker-sandbox/src/fetch.rs

@@ -6,7 +6,7 @@ use worker::{
     Response, Result, RouteContext,
 };
 
-pub async fn handle_fetch(req: Request, _env: Env, _data: SomeSharedData) -> Result<Response> {
+pub async fn handle_fetch(_req: Request, _env: Env, _data: SomeSharedData) -> Result<Response> {
     let req = Request::new("https://example.com", Method::Post)?;
     let resp = Fetch::Request(req).send().await?;
     let resp2 = Fetch::Url("https://example.com".parse()?).send().await?;
@@ -17,7 +17,12 @@ pub async fn handle_fetch(req: Request, _env: Env, _data: SomeSharedData) -> Res
     ))
 }
 
-pub async fn handle_fetch_json(req: Request, _env: Env, _data: SomeSharedData) -> Result<Response> {
+#[worker::send]
+pub async fn handle_fetch_json(
+    _req: Request,
+    _env: Env,
+    _data: SomeSharedData,
+) -> Result<Response> {
     let data: ApiData = Fetch::Url(
         "https://jsonplaceholder.typicode.com/todos/1"
             .parse()

worker-sandbox/src/router.rs

@@ -5,6 +5,7 @@ use crate::{
 #[cfg(feature = "http")]
 use std::convert::TryInto;
 use std::sync::atomic::Ordering;
+
 use worker::{console_log, Fetch, Headers, Request, Response, Result, RouteContext};
 
 #[cfg(not(feature = "http"))]
@@ -44,6 +45,19 @@ macro_rules! handler (
     }
 );
 
+#[cfg(feature = "http")]
+#[debug_handler]
+#[worker::send]
+async fn test(
+    Extension(env): Extension<Env>,
+    Extension(data): Extension<SomeSharedData>,
+    req: axum::extract::Request,
+) -> &'static str {
+    let foo = env.kv("SOME_NAMESPACE").unwrap();
+    foo.put("test", "test").unwrap().execute().await.unwrap();
+    "hello world"
+}
+
 #[cfg(feature = "http")]
 pub fn make_router(data: SomeSharedData, env: Env) -> axum::Router {
     axum::Router::new()

worker-sandbox/src/user.rs

@@ -1,5 +1,5 @@
 use serde::Serialize;
-use worker::{Date, DateInit, Env, Request, Response, Result, RouteContext};
+use worker::{Date, DateInit, Env, Request, Response, Result};
 
 use crate::SomeSharedData;
 

worker-sandbox/src/ws.rs

@@ -1,8 +1,7 @@
 use super::SomeSharedData;
 use futures_util::StreamExt;
 use worker::{
-    wasm_bindgen_futures, Env, Request, Response, Result, RouteContext, WebSocket, WebSocketPair,
-    WebsocketEvent,
+    wasm_bindgen_futures, Env, Request, Response, Result, WebSocket, WebSocketPair, WebsocketEvent,
 };
 
 pub async fn handle_websocket(_req: Request, env: Env, _data: SomeSharedData) -> Result<Response> {

worker/src/formdata.rs

@@ -186,6 +186,3 @@ impl From<web_sys::File> for File {
         Self(file)
     }
 }
-
-unsafe impl Send for File {}
-unsafe impl Sync for File {}

worker/src/lib.rs

@@ -69,7 +69,7 @@ pub use wasm_bindgen_futures;
 pub use worker_kv as kv;
 
 pub use cf::{Cf, TlsClientAuth};
-pub use worker_macros::{durable_object, event};
+pub use worker_macros::{durable_object, event, send};
 #[doc(hidden)]
 pub use worker_sys;
 pub use worker_sys::{console_debug, console_error, console_log, console_warn};
@@ -100,6 +100,7 @@ pub use crate::request_init::*;
 pub use crate::response::{Response, ResponseBody};
 pub use crate::router::{RouteContext, RouteParams, Router};
 pub use crate::schedule::*;
+pub use crate::send_future::SendFuture;
 pub use crate::socket::*;
 pub use crate::streams::*;
 pub use crate::websocket::*;
@@ -135,6 +136,7 @@ mod request_init;
 mod response;
 mod router;
 mod schedule;
+mod send_future;
 mod socket;
 mod streams;
 mod websocket;

worker/src/send_future.rs

@@ -0,0 +1,28 @@
+use futures_util::future::Future;
+use pin_project::pin_project;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+
+#[pin_project]
+pub struct SendFuture<F> {
+    #[pin]
+    inner: F,
+}
+
+impl<F> SendFuture<F> {
+    pub fn new(inner: F) -> Self {
+        Self { inner }
+    }
+}
+
+unsafe impl<F> Send for SendFuture<F> {}
+
+impl<F: Future> Future for SendFuture<F> {
+    type Output = F::Output;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+        this.inner.poll(cx)
+    }
+}

worker/src/streams.rs

@@ -20,9 +20,6 @@ pub struct ByteStream {
     pub(crate) inner: IntoStream<'static>,
 }
 
-unsafe impl Send for ByteStream {}
-unsafe impl Sync for ByteStream {}
-
 impl Stream for ByteStream {
     type Item = Result<Vec<u8>>;
 

worker/src/websocket.rs

@@ -1,6 +1,6 @@
 use crate::{Error, Method, Request, Result};
 use futures_channel::mpsc::UnboundedReceiver;
-use futures_util::{Future, FutureExt, Stream};
+use futures_util::Stream;
 use serde::Serialize;
 use url::Url;
 use worker_sys::ext::WebSocketExt;
@@ -16,7 +16,7 @@ use std::rc::Rc;
 use std::task::{Context, Poll};
 use wasm_bindgen::convert::FromWasmAbi;
 use wasm_bindgen::prelude::Closure;
-use wasm_bindgen::{JsCast, JsValue};
+use wasm_bindgen::JsCast;
 #[cfg(feature = "http")]
 use wasm_bindgen_futures::JsFuture;