get it actually working

client/demo.js

@@ -30,20 +30,24 @@ onmessage = async (msg) => {
 		postMessage(JSON.stringify(str, null, 4));
 	}
 
-    const { EpoxyClient } = await epoxy();
+	const { EpoxyClient, EpoxyClientOptions, EpoxyHandlers } = await epoxy();
 
 	console.log("certs:", CERTS);
 
+	let epoxy_client_options = new EpoxyClientOptions();
+	epoxy_client_options.user_agent = navigator.userAgent;
+
+	let epoxy_client = new EpoxyClient("ws://localhost:4000", CERTS, epoxy_client_options);
+
 	const tconn0 = performance.now();
-    // args: websocket url, user agent, redirect limit, certs
+	await epoxy_client.replace_stream_provider();
-    let epoxy_client = await new EpoxyClient("ws://localhost:4000", navigator.userAgent, 10, CERTS);
 	const tconn1 = performance.now();
 	log(`conn establish took ${tconn1 - tconn0} ms or ${(tconn1 - tconn0) / 1000} s`);
 
 	// epoxy classes are inspectable
 	console.log(epoxy_client);
 	// you can change the user agent and redirect limit in JS
-    epoxy_client.redirectLimit = 15;
+	epoxy_client.redirect_limit = 15;
 
 	const test_mux = async (url) => {
 		const t0 = performance.now();
@@ -170,27 +174,32 @@ onmessage = async (msg) => {
 		log(`avg native (${num_tests}) took ${total_native} ms or ${total_native / 1000} s`);
 		log(`avg mux - avg native (${num_tests}): ${total_mux - total_native} ms or ${(total_mux - total_native) / 1000} s`);
 	} else if (should_ws_test) {
-        let ws = await epoxy_client.connect_ws(
+		let handlers = new EpoxyHandlers(
 			() => log("opened"),
 			() => log("closed"),
 			err => console.error(err),
-            msg => log(msg),
+			msg => log(`got "${msg}"`)
+		);
+		let ws = await epoxy_client.connect_websocket(
+			handlers,
 			"wss://echo.websocket.events",
 			[],
-            "localhost"
 		);
 		while (true) {
 			log("sending `data`");
-            await ws.send_text("data");
+			await ws.send("data");
-            await (new Promise((res, _) => setTimeout(res, 50)));
+			await (new Promise((res, _) => setTimeout(res, 10)));
 		}
 	} else if (should_tls_test) {
 		let decoder = new TextDecoder();
-        let ws = await epoxy_client.connect_tls(
+		let handlers = new EpoxyHandlers(
 			() => log("opened"),
 			() => log("closed"),
 			err => console.error(err),
-            msg => { console.log(msg); log(decoder.decode(msg)) },
+			msg => { console.log(msg); console.log(decoder.decode(msg).split("\r\n\r\n")[1].length); log(decoder.decode(msg)) },
+		);
+		let ws = await epoxy_client.connect_tls(
+			handlers,
 			"google.com:443",
 		);
 		await ws.send("GET / HTTP 1.1\r\nHost: google.com\r\nConnection: close\r\n\r\n");
@@ -198,18 +207,21 @@ onmessage = async (msg) => {
 		await ws.close();
 	} else if (should_udp_test) {
 		let decoder = new TextDecoder();
-        // tokio example: `cargo r --example echo-udp -- 127.0.0.1:5000`
+		let handlers = new EpoxyHandlers(
-        let ws = await epoxy_client.connect_udp(
 			() => log("opened"),
 			() => log("closed"),
 			err => console.error(err),
 			msg => { console.log(msg); log(decoder.decode(msg)) },
+		);
+		// tokio example: `cargo r --example echo-udp -- 127.0.0.1:5000`
+		let ws = await epoxy_client.connect_udp(
+			handlers,
 			"127.0.0.1:5000",
 		);
 		while (true) {
 			log("sending `data`");
 			await ws.send("data");
-            await (new Promise((res, _) => setTimeout(res, 50)));
+			await (new Promise((res, _) => setTimeout(res, 10)));
 		}
 	} else if (should_reconnect_test) {
 		while (true) {

client/src/io_stream.rs

@@ -1,4 +1,4 @@
-use bytes::{BufMut, BytesMut};
+use bytes::{buf::UninitSlice, BufMut, BytesMut};
 use futures_util::{
     io::WriteHalf, lock::Mutex, stream::SplitSink, AsyncReadExt, AsyncWriteExt, SinkExt, StreamExt,
 };
@@ -18,6 +18,7 @@ pub struct EpoxyIoStream {
     onerror: Function,
 }
 
+#[wasm_bindgen]
 impl EpoxyIoStream {
     pub(crate) fn connect(stream: ProviderAsyncRW, handlers: EpoxyHandlers) -> Self {
         let (mut rx, tx) = stream.split();
@@ -32,17 +33,24 @@ impl EpoxyIoStream {
 
         let onerror_cloned = onerror.clone();
 
-        // similar to tokio::io::ReaderStream
+        // similar to tokio_util::io::ReaderStream
         spawn_local(async move {
             let mut buf = BytesMut::with_capacity(4096);
             loop {
-                match rx.read(buf.as_mut()).await {
+                match rx
+                    .read(unsafe {
+                        std::mem::transmute::<&mut UninitSlice, &mut [u8]>(buf.chunk_mut())
+                    })
+                    .await
+                {
                     Ok(cnt) => {
+                        if cnt > 0 {
                             unsafe { buf.advance_mut(cnt) };
 
                             let _ = onmessage
                                 .call1(&JsValue::null(), &Uint8Array::from(buf.split().as_ref()));
                         }
+                    }
                     Err(err) => {
                         let _ = onerror.call1(&JsValue::null(), &JsError::from(err).into());
                         break;
@@ -101,6 +109,7 @@ pub struct EpoxyUdpStream {
     onerror: Function,
 }
 
+#[wasm_bindgen]
 impl EpoxyUdpStream {
     pub(crate) fn connect(stream: ProviderUnencryptedStream, handlers: EpoxyHandlers) -> Self {
         let (tx, mut rx) = stream.split();

client/src/lib.rs

@@ -37,23 +37,23 @@ type HttpBody = http_body_util::Full<Bytes>;
 
 #[derive(Debug, Error)]
 pub enum EpoxyError {
-    #[error(transparent)]
+    #[error("Invalid DNS name: {0:?}")]
     InvalidDnsName(#[from] futures_rustls::rustls::pki_types::InvalidDnsNameError),
-    #[error(transparent)]
+    #[error("Wisp: {0:?}")]
     Wisp(#[from] wisp_mux::WispError),
-    #[error(transparent)]
+    #[error("IO: {0:?}")]
     Io(#[from] std::io::Error),
-    #[error(transparent)]
+    #[error("HTTP: {0:?}")]
     Http(#[from] http::Error),
-    #[error(transparent)]
+    #[error("Hyper client: {0:?}")]
     HyperClient(#[from] hyper_util_wasm::client::legacy::Error),
-    #[error(transparent)]
+    #[error("Hyper: {0:?}")]
     Hyper(#[from] hyper::Error),
-    #[error(transparent)]
+    #[error("HTTP ToStr: {0:?}")]
     ToStr(#[from] http::header::ToStrError),
-    #[error(transparent)]
+    #[error("Getrandom: {0:?}")]
     GetRandom(#[from] getrandom::Error),
-    #[error(transparent)]
+    #[error("Fastwebsockets: {0:?}")]
     FastWebSockets(#[from] fastwebsockets::WebSocketError),
 
     #[error("Invalid URL scheme")]
@@ -196,13 +196,14 @@ impl EpoxyHandlers {
     }
 }
 
-#[wasm_bindgen]
+#[wasm_bindgen(inspectable)]
 pub struct EpoxyClient {
     stream_provider: Arc<StreamProvider>,
     client: Client<StreamProviderService, HttpBody>,
 
-    redirect_limit: usize,
+    pub redirect_limit: usize,
-    user_agent: String,
+    #[wasm_bindgen(getter_with_clone)]
+    pub user_agent: String,
 }
 
 #[wasm_bindgen]
@@ -235,6 +236,10 @@ impl EpoxyClient {
         })
     }
 
+    pub async fn replace_stream_provider(&self) -> Result<(), EpoxyError> {
+        self.stream_provider.replace_client().await
+    }
+
     pub async fn connect_websocket(
         &self,
         handlers: EpoxyHandlers,

client/src/stream_provider.rs

@@ -4,7 +4,11 @@ use futures_rustls::{
     rustls::{ClientConfig, RootCertStore},
     TlsConnector, TlsStream,
 };
-use futures_util::{future::Either, lock::Mutex, AsyncRead, AsyncWrite, Future};
+use futures_util::{
+    future::Either,
+    lock::{Mutex, MutexGuard},
+    AsyncRead, AsyncWrite, Future,
+};
 use hyper_util_wasm::client::legacy::connect::{Connected, Connection};
 use js_sys::{Array, Reflect, Uint8Array};
 use pin_project_lite::pin_project;
@@ -14,7 +18,7 @@ use wasm_bindgen::{JsCast, JsValue};
 use wasm_bindgen_futures::spawn_local;
 use wisp_mux::{
     extensions::{udp::UdpProtocolExtensionBuilder, ProtocolExtensionBuilder},
-    ClientMux, IoStream, MuxStreamIo, StreamType, WispError,
+    ClientMux, IoStream, MuxStreamIo, StreamType,
 };
 
 use crate::{ws_wrapper::WebSocketWrapper, EpoxyClientOptions, EpoxyError};
@@ -75,7 +79,10 @@ impl StreamProvider {
         })
     }
 
-    async fn create_client(&self) -> Result<(), EpoxyError> {
+    async fn create_client(
+        &self,
+        mut locked: MutexGuard<'_, Option<ClientMux>>,
+    ) -> Result<(), EpoxyError> {
         let extensions_vec: Vec<Box<dyn ProtocolExtensionBuilder + Send + Sync>> =
             vec![Box::new(UdpProtocolExtensionBuilder())];
         let extensions = if self.wisp_v2 {
@@ -93,7 +100,7 @@ impl StreamProvider {
         } else {
             client.with_no_required_extensions()
         };
-        self.current_client.lock().await.replace(mux);
+        locked.replace(mux);
         let current_client = self.current_client.clone();
         spawn_local(async move {
             fut.await;
@@ -102,6 +109,10 @@ impl StreamProvider {
         Ok(())
     }
 
+    pub async fn replace_client(&self) -> Result<(), EpoxyError> {
+        self.create_client(self.current_client.lock().await).await
+    }
+
     pub async fn get_stream(
         &self,
         stream_type: StreamType,
@@ -109,13 +120,14 @@ impl StreamProvider {
         port: u16,
     ) -> Result<ProviderUnencryptedStream, EpoxyError> {
         Box::pin(async {
-            if let Some(mux) = self.current_client.lock().await.as_ref() {
+            let locked = self.current_client.lock().await;
+            if let Some(mux) = locked.as_ref() {
                 Ok(mux
                     .client_new_stream(stream_type, host, port)
                     .await?
                     .into_io())
             } else {
-                self.create_client().await?;
+                self.create_client(locked).await?;
                 self.get_stream(stream_type, host, port).await
             }
         })
@@ -231,12 +243,16 @@ impl Service<hyper::Uri> for StreamProviderService {
         let provider = self.0.clone();
         Box::pin(async move {
             let scheme = req.scheme_str().ok_or(EpoxyError::InvalidUrlScheme)?;
-            let host = req.host().ok_or(WispError::UriHasNoHost)?.to_string();
+            let host = req.host().ok_or(EpoxyError::NoUrlHost)?.to_string();
-            let port = req.port_u16().ok_or(WispError::UriHasNoPort)?;
+            let port = req.port_u16().map(Ok).unwrap_or_else(|| match scheme {
+                "https" | "wss" => Ok(443),
+                "http" | "ws" => Ok(80),
+                _ => Err(EpoxyError::NoUrlPort),
+            })?;
             Ok(HyperIo {
                 inner: match scheme {
-                    "https" => Either::Left(provider.get_tls_stream(host, port).await?),
+                    "https" | "wss" => Either::Left(provider.get_tls_stream(host, port).await?),
-                    "http" => {
+                    "http" | "ws" => {
                         Either::Right(provider.get_asyncread(StreamType::Tcp, host, port).await?)
                     }
                     _ => return Err(EpoxyError::InvalidUrlScheme),

client/src/websocket.rs

@@ -30,6 +30,7 @@ pub struct EpoxyWebSocket {
     onerror: Function,
 }
 
+#[wasm_bindgen]
 impl EpoxyWebSocket {
     pub(crate) async fn connect(
         client: &EpoxyClient,