bump version with backpressure

Cargo.toml

@@ -124,6 +124,7 @@ ring = "=0.16.20"
 rusqlite = { version = "=0.29.0", features = ["unlock_notify", "bundled"] }
 rustls = "0.21.0"
 rustls-pemfile = "1.0.0"
+rustls-tokio-stream = "0.2.1"
 rustls-webpki = "0.101.4"
 rustls-native-certs = "0.6.2"
 webpki-roots = "0.25.2"

cli/tests/testdata/run/websocket_test.ts

@@ -163,7 +163,7 @@ Deno.test("websocket error", async () => {
     // Error message got changed because we don't use warp in test_util
     assertEquals(
       err.message,
-      "InvalidData: received corrupt message of type InvalidContentType",
+      "InvalidData: invalid data",
     );
     promise1.resolve();
   };

cli/tests/unit/websocket_test.ts

@@ -21,21 +21,77 @@ Deno.test(async function websocketConstructorTakeURLObjectAsParameter() {
   const promise = deferred();
   const ws = new WebSocket(new URL("ws://localhost:4242/"));
   assertEquals(ws.url, "ws://localhost:4242/");
-  ws.onerror = () => fail();
+  ws.onerror = (e) => promise.reject(e);
   ws.onopen = () => ws.close();
   ws.onclose = () => {
     promise.resolve();
   };
   await promise;
 });
 
+// Ignored until split websocket
+Deno.test({ ignore: true }, async function websocketSendLargePacket() {
+  const promise = deferred();
+  const ws = new WebSocket(new URL("wss://localhost:4243/"));
+  assertEquals(ws.url, "wss://localhost:4243/");
+  ws.onerror = (e) => promise.reject(e);
+  ws.onopen = () => {
+    ws.send("a".repeat(65000));
+  };
+  ws.onmessage = () => {
+    ws.close();
+  };
+  ws.onclose = () => {
+    promise.resolve();
+  };
+  await promise;
+});
+
+// Ignored until split websocket
+Deno.test({ ignore: true }, async function websocketSendLargeBinaryPacket() {
+  const promise = deferred();
+  const ws = new WebSocket(new URL("wss://localhost:4243/"));
+  assertEquals(ws.url, "wss://localhost:4243/");
+  ws.onerror = (e) => promise.reject(e);
+  ws.onopen = () => {
+    ws.send(new Uint8Array(65000));
+  };
+  ws.onmessage = (msg) => {
+    console.log(msg);
+    ws.close();
+  };
+  ws.onclose = () => {
+    promise.resolve();
+  };
+  await promise;
+});
+
+// Ignored until split websocket
+Deno.test({ ignore: true }, async function websocketSendLargeBlobPacket() {
+  const promise = deferred();
+  const ws = new WebSocket(new URL("wss://localhost:4243/"));
+  assertEquals(ws.url, "wss://localhost:4243/");
+  ws.onerror = (e) => promise.reject(e);
+  ws.onopen = () => {
+    ws.send(new Blob(["a".repeat(10)]));
+  };
+  ws.onmessage = (msg) => {
+    console.log(msg);
+    ws.close();
+  };
+  ws.onclose = () => {
+    promise.resolve();
+  };
+  await promise;
+});
+
 // https://github.com/denoland/deno/pull/17762
 // https://github.com/denoland/deno/issues/17761
 Deno.test(async function websocketPingPong() {
   const promise = deferred();
   const ws = new WebSocket("ws://localhost:4245/");
   assertEquals(ws.url, "ws://localhost:4245/");
-  ws.onerror = () => fail();
+  ws.onerror = (e) => promise.reject(e);
   ws.onmessage = (e) => {
     ws.send(e.data);
   };

ext/websocket/Cargo.toml

@@ -22,6 +22,7 @@ fastwebsockets = { workspace = true, features = ["upgrade"] }
 http.workspace = true
 hyper = { workspace = true, features = ["backports"] }
 once_cell.workspace = true
+rustls.workspace = true
+rustls-tokio-stream.workspace = true
 serde.workspace = true
 tokio.workspace = true
-tokio-rustls.workspace = true

ext/websocket/lib.rs

@@ -30,22 +30,23 @@ use http::Request;
 use http::Uri;
 use hyper::Body;
 use once_cell::sync::Lazy;
+use rustls::RootCertStore;
+use rustls::ServerName;
+use rustls_tokio_stream::TlsStream;
 use serde::Serialize;
 use std::borrow::Cow;
 use std::cell::Cell;
 use std::cell::RefCell;
 use std::convert::TryFrom;
 use std::fmt;
 use std::future::Future;
+use std::num::NonZeroUsize;
 use std::path::PathBuf;
 use std::rc::Rc;
 use std::sync::Arc;
 use tokio::io::AsyncRead;
 use tokio::io::AsyncWrite;
 use tokio::net::TcpStream;
-use tokio_rustls::rustls::RootCertStore;
-use tokio_rustls::rustls::ServerName;
-use tokio_rustls::TlsConnector;
 
 use fastwebsockets::CloseCode;
 use fastwebsockets::FragmentCollector;
@@ -281,11 +282,16 @@ where
         unsafely_ignore_certificate_errors,
         None,
       )?;
-      let tls_connector = TlsConnector::from(Arc::new(tls_config));
       let dnsname = ServerName::try_from(domain.as_str())
         .map_err(|_| invalid_hostname(domain))?;
-      let tls_socket = tls_connector.connect(dnsname, tcp_socket).await?;
-      handshake(cancel_resource, request, tls_socket).await?
+      let mut tls_connector = TlsStream::new_client_side(
+        tcp_socket,
+        tls_config.into(),
+        dnsname,
+        NonZeroUsize::new(65536),
+      );
+      let _hs = tls_connector.handshake().await?;
+      handshake(cancel_resource, request, tls_connector).await?
     }
     _ => unreachable!(),
   };
@@ -630,7 +636,8 @@ pub async fn op_ws_next_event(
 
   let mut ws = RcRef::map(&resource, |r| &r.ws).borrow_mut().await;
   loop {
-    let val = match ws.read_frame().await {
+    let val = ws.read_frame().await;
+    let val = match val {
       Ok(val) => val,
       Err(err) => {
         // No message was received, socket closed while we waited.

test_util/src/lib.rs

@@ -338,6 +338,7 @@ async fn echo_websocket_handler(
     match frame.opcode {
       fastwebsockets::OpCode::Close => break,
       fastwebsockets::OpCode::Text | fastwebsockets::OpCode::Binary => {
+        println!("got frame! {}", frame.payload.len());
         ws.write_frame(frame).await.unwrap();
       }
       _ => {}

tools/wpt.ts

@@ -597,7 +597,7 @@ function reportVariation(result: TestResult, expectation: boolean | string[]) {
 
     const expectFail = expectation === false;
     const failReason = result.status !== 0
-      ? "runner failed during test"
+      ? `runner failed during test: >>>>>>>\n${result.stderr}\n<<<<<<<\n`
       : "the event loop run out of tasks during the test";
     console.log(
       `\nfile result: ${