compression: no_context_takeover for streaming

Author: ifdario

autobahn/client-test.js

@@ -6,12 +6,9 @@ const pwd = new URL(".", import.meta.url).pathname;
 const AUTOBAHN_TESTSUITE_DOCKER =
   "crossbario/autobahn-testsuite:25.10.1@sha256:519915fb568b04c9383f70a1c405ae3ff44ab9e35835b085239c258b6fac3074";
 
-// Accept optional feature flag from command line (e.g., "zlib")
-const FEATURE_FLAG = Deno.args[0] || "";
-const FEATURE_SUFFIX = FEATURE_FLAG ? `_${FEATURE_FLAG}` : "";
-const CONTAINER_NAME = `fuzzingserver${FEATURE_SUFFIX}`;
-const PORT = FEATURE_FLAG === "zlib" ? 9002 : 9001;
+const CONTAINER_NAME = "fuzzingserver";
 const CLIENT_EXE = "target/release/examples/autobahn_client";
+const WITH_ZLIB = Deno.args[0] === "zlib";
 
 async function containerExists(name) {
   const result =
@@ -26,12 +23,12 @@ async function containerRunning(name) {
 }
 
 async function ensureClientBuilt() {
-  console.log(
-    `Building autobahn_client${FEATURE_FLAG ? ` with feature: ${FEATURE_FLAG}` : ""}...`,
-  );
-  if (FEATURE_FLAG) {
-    await $`cargo build --release --example autobahn_client --features ${FEATURE_FLAG}`;
+  console.log("Building autobahn_client...");
+  if (WITH_ZLIB) {
+    console.log("Building with zlib compression support");
+    await $`cargo build --release --example autobahn_client --features zlib`;
   } else {
+    console.log("Building without zlib compression support");
     await $`cargo build --release --example autobahn_client`;
   }
 }
@@ -47,23 +44,23 @@ if (await containerExists(CONTAINER_NAME)) {
       `Autobahn ${CONTAINER_NAME} docker container exists but is stopped. Starting it.`,
     );
     await $`docker start ${CONTAINER_NAME}`;
+    await sleep(5);
   }
 } else {
-  console.log(
-    `Starting Autobahn ${CONTAINER_NAME} docker container on port ${PORT}...`,
-  );
+  console.log(`Starting Autobahn ${CONTAINER_NAME} docker container...`);
   $`docker run --name ${CONTAINER_NAME} \
     -v ${pwd}/fuzzingserver.json:/fuzzingserver.json:ro \
     -v ${pwd}/reports:/reports \
-    -p ${PORT}:9001 \
+    -p 9001:9001 \
     --rm ${AUTOBAHN_TESTSUITE_DOCKER} \
     wstest -m fuzzingserver -s fuzzingserver.json`.spawn();
+
+  // sleep long because it might take a while to pull the files
+  await sleep(30);
 }
 
 await ensureClientBuilt();
-await $`${CLIENT_EXE}`
-  .env("RUST_BACKTRACE", "full")
-  .env("AUTOBAHN_PORT", PORT.toString());
+await $`${CLIENT_EXE}`.env("RUST_BACKTRACE", "full");
 
 const { yawc } = JSON.parse(
   Deno.readTextFileSync("./autobahn/reports/client/index.json"),
@@ -72,7 +69,12 @@ const { yawc } = JSON.parse(
 const result = Object.values(yawc);
 
 function failed(name) {
-  return name !== "OK" && name !== "INFORMATIONAL" && name !== "NON-STRICT";
+  return (
+    name !== "OK" &&
+    name !== "INFORMATIONAL" &&
+    name !== "NON-STRICT" &&
+    name !== "UNIMPLEMENTED"
+  );
 }
 
 const failedtests = result.filter((outcome) => failed(outcome.behavior));

autobahn/fuzzingclient.json

@@ -3,7 +3,7 @@
   "servers": [
     {
       "agent": "yawc",
-      "url": "ws://host.docker.internal:9004"
+      "url": "ws://host.docker.internal:9002"
     }
   ],
   "cases": [

autobahn/server-test.js

@@ -10,7 +10,6 @@ const pwd = new URL(".", import.meta.url).pathname;
 const FEATURE_FLAG = Deno.args[0] || "";
 const FEATURE_SUFFIX = FEATURE_FLAG ? `_${FEATURE_FLAG}` : "";
 const CONTAINER_NAME = `fuzzingclient${FEATURE_SUFFIX}`;
-const PORT = FEATURE_FLAG === "zlib" ? 9004 : 9003;
 const ECHO_SERVER_EXE = "target/release/examples/autobahn_server";
 
 const isMac = Deno.build.os === "darwin";
@@ -20,7 +19,7 @@ const networkArgs = isMac ? "" : "--net=host";
 async function containerExists(name) {
   const r =
     await $`docker ps -a --filter name=^/${name}$ --format "{{.Names}}"`.quiet();
-  return r.stdout.trim().length > 0;
+  return r.stdout.trim().length > 9002;
 }
 
 async function containerRunning(name) {
@@ -41,7 +40,7 @@ async function ensureEchoServerBuilt() {
 
 const configPath = `${pwd}/fuzzingclient.json`;
 const config = JSON.parse(Deno.readTextFileSync(configPath));
-config.servers[0].url = `ws://${dockerHost}:${PORT}`;
+config.servers[0].url = `ws://${dockerHost}:9002`;
 Deno.writeTextFileSync(configPath, JSON.stringify(config, null, 2));
 
 await ensureEchoServerBuilt();
@@ -51,7 +50,6 @@ const server = new Deno.Command(ECHO_SERVER_EXE, {
   signal: controller.signal,
   env: {
     ...Deno.env.toObject(),
-    AUTOBAHN_PORT: PORT.toString(),
   },
 }).spawn();
 
@@ -71,14 +69,14 @@ if (await containerExists(CONTAINER_NAME)) {
   }
 } else {
   console.log(
-    `Starting Autobahn ${CONTAINER_NAME} fuzzing client container on port ${PORT}...`,
+    `Starting Autobahn ${CONTAINER_NAME} fuzzing client container...`,
   );
   const cmd = [
     "docker run",
     `--name ${CONTAINER_NAME}`,
     `-v ${pwd}/fuzzingclient.json:/fuzzingclient.json:ro`,
     `-v ${pwd}/reports:/reports`,
-    `-p ${PORT}:${PORT}`,
+    `-p 9002:9002`,
     networkArgs,
     "--rm",
     AUTOBAHN_TESTSUITE_DOCKER,
@@ -98,7 +96,9 @@ const indexJson = JSON.parse(
 const testResults = Object.values(indexJson.yawc);
 
 function isFailure(behavior) {
-  return !["OK", "INFORMATIONAL", "NON-STRICT"].includes(behavior);
+  return !["OK", "INFORMATIONAL", "NON-STRICT", "UNIMPLEMENTED"].includes(
+    behavior,
+  );
 }
 
 const failedTests = testResults.filter((o) => isFailure(o.behavior));

examples/autobahn_client.rs

@@ -1,26 +1,14 @@
 use anyhow::Result;
 use futures::{SinkExt, StreamExt};
 use serde_json::Value;
-use std::{
-    collections::HashMap,
-    time::{Duration, Instant},
-};
-use tokio::time::sleep;
+use std::collections::HashMap;
 use yawc::{close::CloseCode, frame::OpCode, Frame, Options, TcpWebSocket, WebSocket};
 
-fn get_port() -> u16 {
-    std::env::var("AUTOBAHN_PORT")
-        .ok()
-        .and_then(|p| p.parse().ok())
-        .unwrap_or(9001)
-}
-
 async fn get_case_info(case: u32) -> String {
-    let port = get_port();
     for attempt in 1..=3 {
         match async {
             let mut ws = WebSocket::connect(
-                format!("ws://localhost:{port}/getCaseInfo?case={case}")
+                format!("ws://localhost:9001/getCaseInfo?case={case}")
                     .parse()
                     .unwrap(),
             )
@@ -82,8 +70,7 @@ fn get_options_for_case_id(case_id: &str) -> Options {
                 return base_options.with_client_max_window_bits(15);
             } else if case_id.starts_with("13.5.") {
                 return base_options
-                    // .server_no_context_takeover()
-                    // .client_no_context_takeover()
+                    .client_no_context_takeover()
                     .with_client_max_window_bits(9);
             } else if case_id.starts_with("13.6.") {
                 return base_options
@@ -106,32 +93,18 @@ fn get_options_for_case_id(case_id: &str) -> Options {
 
 async fn connect(path: &str, case_id: Option<&str>) -> Result<TcpWebSocket> {
     let options = case_id.map(get_options_for_case_id).unwrap_or_default();
-    let port = get_port();
 
-    let client = WebSocket::connect(format!("ws://localhost:{port}/{path}").parse().unwrap())
+    let client = WebSocket::connect(format!("ws://localhost:9001/{path}").parse().unwrap())
         .with_options(options)
         .await?;
     Ok(client)
 }
 
 async fn get_case_count() -> Result<u32> {
-    let started_at = Instant::now();
-    loop {
-        let mut ws = match connect("getCaseCount", None).await {
-            Ok(ok) => ok,
-            Err(_) => {
-                if started_at.elapsed() > Duration::from_secs(300) {
-                    panic!("unable to connect")
-                }
-                // ...
-                sleep(Duration::from_millis(250)).await;
-                continue;
-            }
-        };
-        let msg = ws.next().await.ok_or_else(|| anyhow::Error::msg("idk"))?;
-        ws.send(Frame::close(CloseCode::Normal, [])).await?;
-        break Ok(std::str::from_utf8(msg.payload())?.parse()?);
-    }
+    let mut ws = connect("getCaseCount", None).await?;
+    let msg = ws.next().await.ok_or_else(|| anyhow::Error::msg("idk"))?;
+    ws.send(Frame::close(CloseCode::Normal, [])).await?;
+    Ok(std::str::from_utf8(msg.payload())?.parse()?)
 }
 
 #[tokio::main]

examples/autobahn_server.rs

@@ -2,7 +2,7 @@ use std::net::SocketAddr;
 
 use axum::{response::IntoResponse, routing::get, Router};
 use tokio::net::TcpListener;
-use yawc::{frame::OpCode, CompressionLevel, HttpStream, IncomingUpgrade, Options, WebSocket};
+use yawc::{frame::OpCode, HttpStream, IncomingUpgrade, Options, WebSocket};
 
 async fn ws_handler(ws: IncomingUpgrade) -> impl IntoResponse {
     // Configure options based on what Autobahn tests need
@@ -25,7 +25,6 @@ fn get_server_options() -> Options {
         .with_max_payload_read(100 * 1024 * 1024)
         .with_max_read_buffer(200 * 1024 * 1024)
         .with_low_latency_compression()
-        .with_compression_level(CompressionLevel::none())
 }
 
 async fn handle_socket(mut ws: WebSocket<HttpStream>) {
@@ -56,12 +55,7 @@ async fn main() {
 
     let app = Router::new().route("/", get(ws_handler));
 
-    let port = std::env::var("AUTOBAHN_PORT")
-        .ok()
-        .and_then(|p| p.parse().ok())
-        .unwrap_or(9002);
-
-    let addr = SocketAddr::from(([127, 0, 0, 1], port));
+    let addr = SocketAddr::from(([127, 0, 0, 1], 9002));
     log::info!("Autobahn test server listening on {}", addr);
 
     let listener = TcpListener::bind(addr).await.unwrap();