faye-websocket client race condition issue

Cargo.toml

[package]
name = "helloworld"
version = "0.1.0"
authors = ["Ryan Brewster <ryanpbrewster@gmail.com>"]
edition = "2018"

[dependencies]
actix = "0.8.3"
actix-web = "1.0.0"
actix-web-actors = "1.0.0"

main.rs

use actix::*;
use actix_web::*;
use actix_web_actors::ws;
use std::time::Duration;

fn main() -> Result<(), std::io::Error> {
    HttpServer::new(|| {
        App::new().service(web::resource("/").route(
            web::get().to(|req: HttpRequest, stream: web::Payload| {
                ws::start(Ws, &req, stream)
            }),
        ))
    })
    .bind("127.0.0.1:8080")?
    .run()
}

struct Ws;

impl Actor for Ws {
    type Context = ws::WebsocketContext<Self>;
    fn started(&mut self, ctx: &mut Self::Context) {
        ctx.run_later(Duration::from_millis(0), |_, c| {
            c.text(format!("hello"));
        });
    }
}

impl StreamHandler<ws::Message, ws::ProtocolError> for Ws {
    fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
        println!("received {:?}", msg);
        ctx.close(None);
    }
}

output.log

> node ws.js ws
connect: 12.882ms ok
connect: 2.403ms ok
connect: 2.038ms ok
connect: 0.979ms ok
connect: 1.908ms ok
connect: 2.008ms ok
connect: 2.256ms ok
connect: 3.504ms ok
connect: 2.916ms ok
connect: 2.741ms ok
connect: 2.617ms ok



> node ws.js faye
connect: 3.512ms ok
connect: 4.956ms ok
connect: 4.756ms ok
connect: 4.125ms ok
connect: 3.536ms ok
connect: 2.863ms ok
connect: 3.763ms ok
connect: 3.523ms ok
connect: 3.749ms ok
connect: 500.985ms timeout
connect: 6.456ms ok
connect: 5.415ms ok
connect: 5.455ms ok
connect: 4.845ms ok

package.json

{
  "devDependencies": {
    "faye-websocket": "^0.11.1",
    "ws": "^7.0.0"
  }
}

test.js

const WebSocket = require('ws');
const faye = require('faye-websocket');

const timeout = 500;

async function runFaye() {
  return new Promise((resolve, reject) => {
    const ws = new faye.Client('ws://localhost:8080/');
    ws.on("close", resolve);
    ws.on("message", (evt) => ws.close());
    ws.on("error", reject);
    setTimeout(reject, timeout);
  });
}

async function runWs() {
  return new Promise((resolve, reject) => {
    const ws = new WebSocket('ws://localhost:8080/');
    ws.on("close", resolve);
    ws.on("message", (evt) => ws.terminate());
    ws.on("error", reject);
    setTimeout(reject, timeout);
  });
}

async function main() {
  const f = process.argv.pop() == "faye" ? runFaye : runWs;
  while (true) {
    console.time("connect");
    const msg = await f().then(() => "ok").catch(err => "timeout");
    console.timeLog("connect", msg);
  }
}

main();