Support wasm transport.

Author: jjeffcaii

examples/Cargo.toml

@@ -10,7 +10,7 @@ rsocket_rust_transport_tcp = { path = "../rsocket-transport-tcp" }
 rsocket_rust_transport_websocket = { path = "../rsocket-transport-websocket" }
 log = "0.4.8"
 env_logger = "0.7.1"
-futures = "0.3.3"
+futures = "0.3.4"
 
 [dev-dependencies.tokio]
 version = "0.2.11"

rsocket-test/Cargo.toml

@@ -6,13 +6,13 @@ edition = "2018"
 
 [dev-dependencies]
 log = "0.4"
-futures = "0.3.3"
+futures = "0.3.4"
 env_logger = "0.7.1"
 rsocket_rust = { path = "../rsocket", features = ["frame"] }
 rsocket_rust_transport_tcp = { path = "../rsocket-transport-tcp" }
 rsocket_rust_transport_websocket = { path = "../rsocket-transport-websocket" }
 bytes = "0.5.4"
-hex = "0.4.0"
+hex = "0.4.2"
 rand = "0.7.3"
 
 [dev-dependencies.tokio]

rsocket-test/tests/test_clients.rs

@@ -16,6 +16,16 @@ fn init() {
         .try_init();
 }
 
+#[tokio::main]
+#[test]
+async fn test_connect_must_failed() {
+    let result = RSocketFactory::connect()
+        .transport(TcpClientTransport::from("tcp://127.0.0.1:6789"))
+        .start()
+        .await;
+    assert_eq!(false, result.is_ok());
+}
+
 #[test]
 fn test_websocket() {
     init();

rsocket-test/tests/test_composite_metadata.rs

@@ -1,4 +1,4 @@
-use bytes::{Buf, BufMut, Bytes, BytesMut};
+use bytes::BytesMut;
 use rsocket_rust::extension::{CompositeMetadata, Metadata};
 use rsocket_rust::utils::Writeable;
 

rsocket-transport-tcp/Cargo.toml

@@ -6,7 +6,7 @@ edition = "2018"
 
 [dependencies]
 log = "0.4.8"
-futures = "0.3.3"
+futures = "0.3.4"
 bytes = "0.5.4"
 rsocket_rust = { path = "../rsocket", features = ["frame"] }
 

rsocket-transport-tcp/src/client.rs

@@ -1,9 +1,13 @@
 use super::codec::LengthBasedFrameCodec;
 use futures::{SinkExt, StreamExt};
+use rsocket_rust::error::RSocketError;
 use rsocket_rust::frame::Frame;
-use rsocket_rust::transport::{BoxResult, ClientTransport, Rx, SafeFuture, Tx};
+use rsocket_rust::runtime::{DefaultSpawner, Spawner};
+use rsocket_rust::transport::{ClientTransport, Rx, Tx, TxOnce};
+use std::future::Future;
 use std::net::SocketAddr;
 use std::net::TcpStream as StdTcpStream;
+use std::pin::Pin;
 use tokio::net::TcpStream;
 use tokio_util::codec::Framed;
 
@@ -23,37 +27,53 @@ impl TcpClientTransport {
     }
 
     #[inline]
-    async fn connect(self) -> BoxResult<TcpStream> {
+    async fn connect(self) -> Result<TcpStream, RSocketError> {
         match self.connector {
             Connector::Direct(stream) => Ok(stream),
             Connector::Lazy(addr) => match StdTcpStream::connect(&addr) {
                 Ok(raw) => match TcpStream::from_std(raw) {
                     Ok(stream) => Ok(stream),
-                    Err(e) => Err(Box::new(e)),
+                    Err(e) => Err(RSocketError::from(e)),
                 },
-                Err(e) => Err(Box::new(e)),
+                Err(e) => Err(RSocketError::from(e)),
             },
         }
     }
 }
 
 impl ClientTransport for TcpClientTransport {
-    fn attach(self, incoming: Tx<Frame>, mut sending: Rx<Frame>) -> SafeFuture<BoxResult<()>> {
-        Box::pin(async move {
-            let socket = self.connect().await?;
-            let (mut writer, mut reader) = Framed::new(socket, LengthBasedFrameCodec).split();
-            tokio::spawn(async move {
-                while let Some(it) = reader.next().await {
-                    incoming.unbounded_send(it.unwrap()).unwrap();
+    fn attach(
+        self,
+        incoming: Tx<Frame>,
+        mut sending: Rx<Frame>,
+        connected: Option<TxOnce<Result<(), RSocketError>>>,
+    ) {
+        DefaultSpawner.spawn(async move {
+            match self.connect().await {
+                Ok(socket) => {
+                    if let Some(sender) = connected {
+                        sender.send(Ok(())).unwrap();
+                    }
+                    let (mut writer, mut reader) =
+                        Framed::new(socket, LengthBasedFrameCodec).split();
+                    DefaultSpawner.spawn(async move {
+                        while let Some(it) = reader.next().await {
+                            incoming.unbounded_send(it.unwrap()).unwrap();
+                        }
+                    });
+                    // loop write
+                    while let Some(it) = sending.next().await {
+                        debug!("===> SND: {:?}", &it);
+                        writer.send(it).await.unwrap()
+                    }
+                }
+                Err(e) => {
+                    if let Some(sender) = connected {
+                        sender.send(Err(e)).unwrap();
+                    }
                 }
-            });
-            // loop write
-            while let Some(it) = sending.next().await {
-                debug!("===> SND: {:?}", &it);
-                writer.send(it).await.unwrap()
             }
-            Ok(())
-        })
+        });
     }
 }
 
@@ -65,7 +85,12 @@ impl From<SocketAddr> for TcpClientTransport {
 
 impl From<&str> for TcpClientTransport {
     fn from(addr: &str) -> TcpClientTransport {
-        let socket_addr: SocketAddr = addr.parse().unwrap();
+        let socket_addr: SocketAddr = if addr.starts_with("tcp://") {
+            let ss: String = addr.chars().skip(6).collect();
+            ss.parse().unwrap()
+        } else {
+            addr.parse().unwrap()
+        };
         TcpClientTransport::new(Connector::Lazy(socket_addr))
     }
 }

rsocket-transport-tcp/src/server.rs

@@ -1,6 +1,9 @@
 use super::client::TcpClientTransport;
-use rsocket_rust::transport::{BoxResult, ClientTransport, SafeFuture, ServerTransport};
+use rsocket_rust::transport::{ClientTransport, ServerTransport};
+use std::error::Error;
+use std::future::Future;
 use std::net::SocketAddr;
+use std::pin::Pin;
 use tokio::net::TcpListener;
 
 pub struct TcpServerTransport {
@@ -20,7 +23,7 @@ impl ServerTransport for TcpServerTransport {
         self,
         starter: Option<fn()>,
         acceptor: impl Fn(Self::Item) + Send + Sync + 'static,
-    ) -> SafeFuture<BoxResult<()>>
+    ) -> Pin<Box<dyn Send + Future<Output = Result<(), Box<dyn Send + Sync + Error>>>>>
     where
         Self::Item: ClientTransport + Sized,
     {

rsocket-transport-wasm/Cargo.toml

@@ -6,11 +6,18 @@ edition = "2018"
 
 [dependencies]
 bytes = "0.5.4"
-wasm-bindgen = "0.2.58"
 wasm-bindgen-futures = "0.4.8"
+futures-channel = "0.3.4"
+futures-util = "0.3.4"
 js-sys = "0.3.35"
+serde = "1.0.104"
+serde_derive = "1.0.104"
 rsocket_rust = { path = "../rsocket", features = ["frame"] }
 
+[dependencies.wasm-bindgen]
+version = "0.2.58"
+features = ["serde-serialize"]
+
 [dependencies.web-sys]
 version = "0.3.35"
 features = [
@@ -20,4 +27,5 @@ features = [
   "ErrorEvent",
   "MessageEvent",
   "WebSocket",
+  "Event",
 ]

rsocket-transport-wasm/src/client.rs

@@ -1,13 +1,19 @@
 use bytes::BytesMut;
+use futures_channel::{mpsc, oneshot};
+use futures_util::{SinkExt, StreamExt, TryStreamExt};
 use js_sys::{ArrayBuffer, Uint8Array};
+use rsocket_rust::error::RSocketError;
 use rsocket_rust::frame::Frame;
-use rsocket_rust::transport::{BoxResult, ClientTransport, Rx, SafeFuture, Tx};
+use rsocket_rust::transport::{ClientTransport, Rx, Tx, TxOnce};
 use rsocket_rust::utils::Writeable;
 use std::cell::RefCell;
+use std::future::Future;
+use std::pin::Pin;
 use std::rc::Rc;
 use wasm_bindgen::prelude::*;
 use wasm_bindgen::JsCast;
-use web_sys::{ErrorEvent, FileReader, MessageEvent, ProgressEvent, WebSocket};
+use wasm_bindgen_futures::spawn_local;
+use web_sys::{ErrorEvent, Event, FileReader, MessageEvent, ProgressEvent, WebSocket};
 
 macro_rules! console_log {
     ($($t:tt)*) => (log(&format_args!($($t)*).to_string()))
@@ -23,46 +29,83 @@ pub struct WebsocketClientTransport {
     url: String,
 }
 
+impl WebsocketClientTransport {
+    #[inline]
+    fn wait_for_open(ws: &WebSocket) -> impl Future<Output = ()> {
+        let (sender, receiver) = oneshot::channel();
+        // The Closure is only called once, so we can use Closure::once
+        let on_open = Closure::once(move |_e: Event| {
+            // We don't need to send a value, so we just send ()
+            sender.send(()).unwrap();
+        });
+        ws.set_onopen(Some(on_open.as_ref().unchecked_ref()));
+        async move {
+            // Wait for it to open
+            receiver.await.unwrap();
+            // Clean up the Closure so we don't leak any memory
+            drop(on_open);
+        }
+    }
+}
+
 impl ClientTransport for WebsocketClientTransport {
-    fn attach(self, incoming: Tx<Frame>, mut sending: Rx<Frame>) -> SafeFuture<BoxResult<()>> {
-        Box::pin(async move {
-            let sending = RefCell::new(sending);
-            // Connect to an echo server
-            let ws = WebSocket::new(&self.url).unwrap();
-
-            // on message
-            let on_message = Closure::wrap(Box::new(move |e: MessageEvent| {
-                let data: JsValue = e.data();
-                read_binary(data, incoming.clone());
-            }) as Box<dyn FnMut(MessageEvent)>);
-            ws.set_onmessage(Some(on_message.as_ref().unchecked_ref()));
-            on_message.forget();
-
-            // on error
-            let on_error = Closure::wrap(Box::new(move |_e: ErrorEvent| {
-                // TODO: handle error
-            }) as Box<dyn FnMut(ErrorEvent)>);
-            ws.set_onerror(Some(on_error.as_ref().unchecked_ref()));
-            on_error.forget();
-
-            // on open
-            let cloned_ws = ws.clone();
-            let on_open = Closure::wrap(Box::new(move |_| {
-                let mut sending = sending.borrow_mut();
-                while let Ok(Some(f)) = sending.try_next() {
-                    let mut bf = BytesMut::new();
-                    f.write_to(&mut bf);
-                    let mut raw = bf.to_vec();
-                    cloned_ws.send_with_u8_array(&mut raw[..]).unwrap();
+    fn attach(
+        self,
+        incoming: Tx<Frame>,
+        mut sending: Rx<Frame>,
+        connected: Option<TxOnce<Result<(), RSocketError>>>,
+    ) {
+        spawn_local(async move {
+            match WebSocket::new(&self.url) {
+                Ok(ws) => {
+                    // on message
+                    let on_message = Closure::wrap(Box::new(move |e: MessageEvent| {
+                        let data: JsValue = e.data();
+                        read_binary(data, incoming.clone());
+                    })
+                        as Box<dyn FnMut(MessageEvent)>);
+                    ws.set_onmessage(Some(on_message.as_ref().unchecked_ref()));
+                    on_message.forget();
+
+                    // on error
+                    let on_error = Closure::wrap(Box::new(move |e: ErrorEvent| {
+                        console_log!("websocket error: {}", e.message());
+                    })
+                        as Box<dyn FnMut(ErrorEvent)>);
+                    ws.set_onerror(Some(on_error.as_ref().unchecked_ref()));
+                    on_error.forget();
+
+                    // on_close
+                    let on_close = Closure::once(Box::new(move |_e: Event| {
+                        console_log!("websocket closed");
+                    }) as Box<dyn FnMut(Event)>);
+                    ws.set_onclose(Some(on_close.as_ref().unchecked_ref()));
+                    on_close.forget();
+
+                    Self::wait_for_open(&ws).await;
+
+                    if let Some(sender) = connected {
+                        sender.send(Ok(())).unwrap();
+                    }
+
+                    while let Some(v) = sending.next().await {
+                        let mut bf = BytesMut::new();
+                        v.write_to(&mut bf);
+                        let mut raw = bf.to_vec();
+                        ws.send_with_u8_array(&mut raw[..])
+                            .expect("write data into websocket failed.");
+                    }
+                    console_log!("***** attch end *****");
                 }
-            }) as Box<dyn FnMut(JsValue)>);
-            ws.set_onopen(Some(on_open.as_ref().unchecked_ref()));
-            on_open.forget();
-
-            console_log!("***** attch end *****");
-
-            Ok(())
-        })
+                Err(e) => {
+                    if let Some(sender) = connected {
+                        sender
+                            .send(Err(RSocketError::from(e.as_string().unwrap())))
+                            .unwrap();
+                    }
+                }
+            }
+        });
     }
 }
 

rsocket-transport-wasm/src/lib.rs

@@ -1,7 +1,10 @@
+#[macro_use]
+extern crate serde_derive;
+
 mod client;
 mod misc;
 mod runtime;
 
 pub use client::WebsocketClientTransport;
-pub use misc::{JsClient, JsPayload};
+pub use misc::{connect, new_payload, JsClient, JsPayload};
 pub use runtime::WASMSpawner;