fastwebsockets 0.10.0

A fast RFC6455 WebSocket server implementation
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

// https://github.com/de-vri-es/hyper-tungstenite-rs/tree/main/tests

use http_body_util::Empty;
use hyper::body::Bytes;
use hyper::body::Incoming;
use hyper::header::CONNECTION;
use hyper::header::UPGRADE;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::Request;
use hyper::Response;
use hyper_util::rt::TokioIo;
use std::future::Future;
use std::net::Ipv6Addr;
use tokio::net::TcpStream;

use assert2::assert;
use assert2::let_assert;

struct TestExecutor;

impl<Fut> hyper::rt::Executor<Fut> for TestExecutor
where
  Fut: Future + Send + 'static,
  Fut::Output: Send + 'static,
{
  fn execute(&self, fut: Fut) {
    tokio::spawn(fut);
  }
}

#[tokio::test]
async fn hyper() {
  // Bind a TCP listener to an ephemeral port.
  let_assert!(
    Ok(listener) =
      tokio::net::TcpListener::bind((Ipv6Addr::LOCALHOST, 0u16)).await
  );
  let_assert!(Ok(bind_addr) = listener.local_addr());

  // Spawn the server in a task.
  tokio::spawn(async move {
    loop {
      let (stream, _) = listener.accept().await.unwrap();
      let io = TokioIo::new(stream);

      tokio::spawn(async move {
        if let Err(err) = http1::Builder::new()
          .serve_connection(io, service_fn(upgrade_websocket))
          .with_upgrades()
          .await
        {
          println!("Error serving connection: {:?}", err);
        }
      });
    }
  });

  // Try to create a websocket connection with the server.
  let_assert!(Ok(stream) = TcpStream::connect(bind_addr).await);
  let_assert!(
    Ok(req) = Request::builder()
      .method("GET")
      .uri("ws://localhost/foo")
      .header("Host", "localhost")
      .header(UPGRADE, "websocket")
      .header(CONNECTION, "upgrade")
      .header(
        "Sec-WebSocket-Key",
        fastwebsockets::handshake::generate_key(),
      )
      .header("Sec-WebSocket-Version", "13")
      .body(Empty::<Bytes>::new())
  );
  let_assert!(Ok((mut stream, _response)) = fastwebsockets::handshake::client(&TestExecutor, req, stream).await);

  let_assert!(Ok(message) = stream.read_frame().await);
  assert!(message.opcode == fastwebsockets::OpCode::Text);
  assert!(message.payload == b"Hello!");

  let_assert!(
    Ok(()) = stream
      .write_frame(fastwebsockets::Frame::text(b"Goodbye!".to_vec().into()))
      .await
  );
  let_assert!(Ok(close_frame) = stream.read_frame().await);
  assert!(close_frame.opcode == fastwebsockets::OpCode::Close);
}

async fn upgrade_websocket(
  mut request: Request<Incoming>,
) -> Result<Response<Empty<Bytes>>, fastwebsockets::WebSocketError> {
  assert!(fastwebsockets::upgrade::is_upgrade_request(&request) == true);

  let (response, stream) = fastwebsockets::upgrade::upgrade(&mut request)?;
  tokio::spawn(async move {
    let_assert!(Ok(mut stream) = stream.await);
    assert!(let Ok(()) = stream.write_frame(fastwebsockets::Frame::text(b"Hello!".to_vec().into())).await);
    let_assert!(Ok(reply) = stream.read_frame().await);
    assert!(reply.opcode == fastwebsockets::OpCode::Text);
    assert!(reply.payload == b"Goodbye!");

    assert!(let Ok(()) = stream.write_frame(fastwebsockets::Frame::close_raw(vec![].into())).await);
  });

  Ok(response)
}