Remove some redundancies from wasm32 rust sdk

Author: thlsrms

Cargo.lock

@@ -37,14 +37,14 @@ getrandom = { version = "0.3.2", features = ["wasm_js"]}
 gloo-console = "0.3.0"
 gloo-storage = "0.3.0"
 rustls-pki-types = { version = "1.11.0", features = ["web"] }
-tokio-tungstenite-wasm = { git = "https://github.com/thlsrms/tokio-tungstenite-wasm", rev = "c788b7cfc30f576c" }
 tokio = { version = "1.37", default-features = false, features = [
-    "rt", "macros", "sync"
+    "rt", "macros", "sync", "io-util"
 ] }
+tokio-tungstenite-wasm = "0.6.0"
 tungstenite = { version = "0.26.2", features = ["rustls"] }
 wasm-bindgen = "0.2.100"
 wasm-bindgen-futures = "0.4.45"
-web-sys = { version = "0.3.77", features = [ "Document", "HtmlDocument", "Window", "Storage" ] }
+web-sys = { version = "0.3.77", features = [ "Document", "HtmlDocument", "Window" ] }
 
 [dev-dependencies]
 # for quickstart-chat and cursive-chat examples

crates/sdk/Cargo.toml

@@ -67,7 +67,6 @@ pub struct DbContextImpl<M: SpacetimeModule> {
     /// which are pre-parsed in the background by [`parse_loop`].
     #[cfg(not(target_arch = "wasm32"))]
     recv: Arc<TokioMutex<mpsc::UnboundedReceiver<ParsedMessage<M>>>>,
-
     #[cfg(target_arch = "wasm32")]
     recv: SharedCell<mpsc::UnboundedReceiver<ParsedMessage<M>>>,
 
@@ -81,7 +80,6 @@ pub struct DbContextImpl<M: SpacetimeModule> {
     /// from which [Self::apply_pending_mutations] and friends read mutations.
     #[cfg(not(target_arch = "wasm32"))]
     pending_mutations_recv: Arc<TokioMutex<mpsc::UnboundedReceiver<PendingMutation<M>>>>,
-
     #[cfg(target_arch = "wasm32")]
     pending_mutations_recv: SharedCell<mpsc::UnboundedReceiver<PendingMutation<M>>>,
 
@@ -473,7 +471,6 @@ impl<M: SpacetimeModule> DbContextImpl<M> {
     /// If no WebSocket messages are in the queue, immediately return `false`.
     ///
     /// Called by the autogenerated `DbConnection` method of the same name.
-    #[cfg(not(target_arch = "wasm32"))]
     pub fn advance_one_message(&self) -> crate::Result<bool> {
         // Apply any pending mutations before processing a WS message,
         // so that pending callbacks don't get skipped.
@@ -490,42 +487,27 @@ impl<M: SpacetimeModule> DbContextImpl<M> {
         // returns `Err(_)`. Similar behavior as `Iterator::next` and
         // `Stream::poll_next`. No comment on whether this is a good mental
         // model or not.
-        let res = match self.recv.blocking_lock().try_next() {
-            Ok(None) => {
-                let disconnect_ctx = self.make_event_ctx(None);
-                self.invoke_disconnected(&disconnect_ctx);
-                Err(crate::Error::Disconnected)
-            }
-            Err(_) => Ok(false),
-            Ok(Some(msg)) => self.process_message(msg).map(|_| true),
-        };
 
-        // Also apply any new pending messages afterwards,
-        // so that outgoing WS messages get sent as soon as possible.
-        self.apply_pending_mutations()?;
+        let res = {
+            #[cfg(not(target_arch = "wasm32"))]
+            let mut recv = self.recv.blocking_lock();
 
-        res
-    }
+            #[cfg(target_arch = "wasm32")]
+            let mut recv = self.recv.lock().unwrap();
 
-    #[cfg(target_arch = "wasm32")]
-    pub fn advance_one_message(&self) -> crate::Result<bool> {
-        self.apply_pending_mutations()?;
-        // Synchronously try to pull one server message
-        let res = {
-            let mut chan = self.recv.lock().unwrap();
-            match chan.try_next() {
+            match recv.try_next() {
                 Ok(None) => {
-                    // Shouldn’t happen on unbounded, treat as disconnect
-                    let ctx = self.make_event_ctx(None);
-                    self.invoke_disconnected(&ctx);
+                    let disconnect_ctx = self.make_event_ctx(None);
+                    self.invoke_disconnected(&disconnect_ctx);
                     Err(crate::Error::Disconnected)
                 }
                 Err(_) => Ok(false),
                 Ok(Some(msg)) => self.process_message(msg).map(|_| true),
             }
         };
 
-        // send any pending outgoing mutations now that we've done a read
+        // Also apply any new pending messages afterwards,
+        // so that outgoing WS messages get sent as soon as possible.
         self.apply_pending_mutations()?;
 
         res
@@ -971,7 +953,7 @@ but you must call one of them, or else the connection will never progress.
         })?;
 
         let (raw_msg_recv, raw_msg_send) = ws_connection.spawn_message_loop();
-        let parsed_recv_chan = spawn_parse_loop::<M>(raw_msg_recv, &handle);
+        let parsed_recv_chan = spawn_parse_loop::<M>(raw_msg_recv);
 
         let inner = Arc::new(StdMutex::new(DbContextImplInner {
             runtime,
@@ -997,10 +979,8 @@ but you must call one of them, or else the connection will never progress.
             inner,
             send_chan,
             cache,
-            #[cfg(target_arch = "wasm32")]
             recv: Arc::new(StdMutex::new(parsed_recv_chan)),
             pending_mutations_send,
-            #[cfg(target_arch = "wasm32")]
             pending_mutations_recv: Arc::new(StdMutex::new(pending_mutations_recv)),
             identity: Arc::new(StdMutex::new(None)),
         };
@@ -1125,14 +1105,12 @@ fn enter_or_create_runtime() -> crate::Result<(Option<Runtime>, runtime::Handle)
     match runtime::Handle::try_current() {
         Err(e) if e.is_missing_context() => {
             #[cfg(not(target_arch = "wasm32"))]
-            let rt = tokio::runtime::Builder::new_multi_thread()
-                .enable_all()
-                .worker_threads(1)
-                .thread_name("spacetimedb-background-connection")
-                .build()
-                .map_err(|source| InternalError::new("Failed to create Tokio runtime").with_cause(source))?;
+            let mut rt = tokio::runtime::Builder::new_multi_thread();
             #[cfg(target_arch = "wasm32")]
-            let rt = tokio::runtime::Builder::new_current_thread()
+            let mut rt = tokio::runtime::Builder::new_current_thread();
+
+            let rt = rt
+                .enable_all()
                 .worker_threads(1)
                 .thread_name("spacetimedb-background-connection")
                 .build()
@@ -1173,7 +1151,6 @@ fn spawn_parse_loop<M: SpacetimeModule>(
 #[cfg(target_arch = "wasm32")]
 fn spawn_parse_loop<M: SpacetimeModule>(
     raw_message_recv: mpsc::UnboundedReceiver<ws::ServerMessage<BsatnFormat>>,
-    _handle: &runtime::Handle,
 ) -> mpsc::UnboundedReceiver<ParsedMessage<M>> {
     let (parsed_message_send, parsed_message_recv) = mpsc::unbounded();
     wasm_bindgen_futures::spawn_local(parse_loop(raw_message_recv, parsed_message_send));

crates/sdk/src/db_connection.rs

@@ -525,11 +525,6 @@ impl WsConnection {
                             }
                         },
 
-                        Some(Ok(WebSocketMessage::Ping(payload)))
-                        | Some(Ok(WebSocketMessage::Pong(payload))) => {
-                            record_metrics(payload.len());
-                        },
-
                         Some(Ok(WebSocketMessage::Close(r))) => {
                             let reason: String = if let Some(r) = r {
                                 format!("{}:{:?}", r, r.code)
@@ -557,12 +552,14 @@ impl WsConnection {
                     outbound = outgoing.next() => if let Some(client_msg) = outbound {
                         let raw = Self::encode_message(client_msg);
                         if let Err(e) = ws_writer.send(raw).await {
-                            gloo_console::warn!("WS Send error: ", format!("{:?}",e));
+                            gloo_console::warn!("Error sending outgoing message:", format!("{:?}",e));
                             break;
                         }
                     } else {
                         // channel closed, so we're done  sending
-                        let _ = ws_writer.close().await;
+                        if let Err(e) = ws_writer.close().await {
+                            gloo_console::warn!("Error sending close frame:", format!("{:?}", e));
+                        }
                         break;
                     },
                 }