Native WebSocket support

[net]

I think Rocket would benefit heavily from native WebSocket support. Even better would be something similar to Phoenix's channels.

It's great to see such a comprehensive web framework for Rust coming together!

[jsen-]

@net did you manage to combine rocket with eg. websocket on the same host:port?
I'm trying to upgrade connection to ws, but no luck so far :(

[SergioBenitez]

This is indeed planned.

[wagenet]

@SergioBenitez got any suggestions for how one could manually add web socket support?

[SergioBenitez]

@wagenet Do you mean to Rocket or to your web application? The former is subtle is it requires a lot of design work, but the latter should be straightforward. You should be able to use the websocket or ws-rs crates as they are intended. You won't be able to use the same port for both servers, of course. And, you'll need to spawn one of the two servers in a separate thread so that one server doesn't block the other one from starting. Should you choose to do this, your main function would look something like:

fn main() {
    thread::spawn(|| {
        listen("127.0.0.1:3012", |out| {
            move |msg| {
                out.send(msg)
           }
        })
    })

    rocket.ignite()..more()..launch()
}

[wagenet]

@SergioBenitez thanks :) I ended up discovering this sort of approach myself.

[jhpratt]

@SergioBenitez is this waiting on a future version of Rocket that migrates to Actix or some other async framework?

[sparky8251]

Is there no way to add web socket support on the same port as HTTP?

If not, this is quite problematic for me.

[jhpratt]

@sparky8251 Not currently, at least easily.

[sparky8251]

Well, that sucks. Liked the readability of Rocket but without the ability to have websockets on the same port Rocket is useless to me.

Actix just doesnt have the same readability after all...

If you happen to know a way to do websockets on the same port, even if its not clean, I'd love to hear it. If I end up porting to Actix I won't be returning to Rocket but if I can make the websocket stuff work, hopefully I can stick around until native support lands.

[jebrosen]

@sparky8251 nginx ("Since version 1.3.13") should be able to forward to a second server on a per-location basis. The second server could even be another listening port connected to the same binary that's serving the rocket HTTP server (e.g. via #90 (comment)).

I've made an example (completely untested so far, sorry) based on the nginx example for WebSocket:

http {
    map $http_upgrade $connection_upgrade {
        default upgrade;
        ''      close;
    }

    server {
        listen 80 default server;
        server_name example.com;

        location /websocket/ {
            proxy_pass http://localhost:8080;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection $connection_upgrade;
        }

        location / {
            proxy_pass http://localhost:8000;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection $connection_upgrade;
        }
    }

I presume something similar can be done with Apache as well.

[sparky8251]

Unfortunately, using nginx is a non-answer for my use case. I'd like to not need a 3rd party program to make this work.

[VictorKoenders]

Hyper 0.12 supports websockets: https://github.com/hyperium/hyper/blob/master/examples/upgrades.rs

Rocket is still on hyper 0.10 though, so we'll need to upgrade that dependency first

Edit: which seems blocked on 0.5.0: #17 (comment)

[fenhl]

With #1008 merged, would it make sense to start implementing websocket support on the async branch?

[jebrosen]

I think we would need to see some progress on #1066 first - which reminds me, I need to write a bit more about that. And WebSocket support is also big enough that I'd like to see a hypothetical API design before going too far into implementation.

[jhpratt]

WebSockets will certainly be one of the larger issues. I'm actually looking into the two main websocket crates (ws and websocket). ws is, frankly, a bit of a mess, and will be extremely difficult to upgrade to asynchronous. websocket already has support, though it's running on tokio 0.1 and hyper 0.10. I've just asked on an issue there to see what sort of changes will be accepted.

While this wouldn't directly add websocket support to Rocket, it would allow for an asynchronous server that could be spawned on the same runtime.

[jebrosen]

Actually, I don't think #1066 is necessary for this after all - it would help with SSE, but not websockets. Instead, websockets would likely have to be handled at the hyper-service or connection level with an API such as https://docs.rs/websocket/0.23.0/websocket/server/upgrade/async/trait.IntoWs.html.

[schrieveslaach]

@jebrosen, based on your async branch I created a branch with rudimentary websocket support and I would like to contribute this new feature to rocket.

The first draft of the API looks likes this:

let (tx, rx) = channel();

std::thread::spawn(move || {
    let duration = std::time::Duration::from_secs(1);
    loop {
        println!("Sending message");
        tx.unbounded_send(Message{}).unwrap();
        std::thread::sleep(duration);
    }
});

let _ = rocket::ignite().receivers(vec![rx]).mount("/", routes![hello]).launch();

@SergioBenitez, @jebrosen, what do you think about this first implementation.

BTW: here is the branch, implementing the behavior.

[jebrosen]

My vague idea for a websocket API was to be able to do something like the following:

#[websocket("/chat/<room>")]
async fn chat(room: String, conn: WebSocket) {
    // let message = conn.recv().await;
    // conn.send(message).await;
}

fn main() {
    rocket::ignite().websocket(websocket![chat]);
}

That is, have it work similarly and in parallel to catchers and routes. It should also have a lower-level API like Handler. I'm not sure if we should support only Websocket or expose more flexibility in terms of other protocols via Upgrade.

As far as I can tell your current implementation takes any number of receivers, and every message that comes on any receiver is sent to all connected clients - and that pattern is too limiting.

[schrieveslaach]

Okay, that looks much nicer, but I have one question. If I understand your example correctly, wouldn't this require that Rocket run the chat function every x milliseconds?

As far as I can tell your current implementation takes any number of receivers, and every message that comes on any receiver is sent to all connected clients - and that pattern is too limiting.

This is not intended to be the final result of a future PR. It was just a very basic implementation to get it running.

Currently, I'm wondering if we could combine our API designs:

#[websocket("/chat/<room>")]
async fn chat(room: String, msg: Message) -> Json<Value> {
    // Will be invoced when rx emits a message and here you could 
   // transform the message into a certain paylod.
}

fn main() {
    let (tx, rx) = channel();

    std::thread::spawn(move || {
        let duration = std::time::Duration::from_secs(1);
        loop {
            println!("Sending message");
            // Message could be a trait and here you could create a 
            // text message (similar to websockets crate API)
            tx.unbounded_send(Message{}).unwrap();
            std::thread::sleep(duration);
        }
    });

    rocket::ignite().websocket(vec![websocket!(chat, rx)]);
}

[SergioBenitez]

For those watching: design discussions are taking place on Matrix. Here is my proposed API for websockets, by example:

//! A multiroom chat application with a special `echo` room which echoes its
//! messages to `lobby`.

// `&Channel`, `&Broker` are channel guards, provided by Rocket.
// `&Broker` is additionally a request guard for sending messages from anywhere.
//
// A `Channel` is:
//   1) A bidrectional channel between two endpoints: client and server.
//   2) Linked to a topic URI, for broadcasting to that topic.
//   3) A cache storer, like `&Request`. The cache is dropped on `leave`.
//
// A `Broker` knows about all channels.
//
// &Username is an application channel guard that asserts that the connection
// sending a message has been granted the reflected username.

type UserList = State<ConcurrentHashSet<String>>;

enum Event<'r> {
    /// A join with a requested `username` in `.0`.
    Join(&'r str),
    /// Message `.1` sent by user with username `.0`.
    Message(&'r Username, &'r str),
    /// User with username `.0` has left.
    Leave(&'r Username),
}

/// A new client has connected to a room. Inform the room of this, say hello to
/// the client with a "join" message.
#[join("/<_>", data = "<username>")]
fn join(channel: &Channel, username: &str, users: &UserList) {
    if users.contains_or_add(username) {
        channel.send(Event::Leave("username taken")).await;
        return channel.close();
    }

    channel.local_cache(|| username.to_string());
    channel.send(Event::Join("hello!")).await;
    channel.broadcast(Event::Join(username)).await;
}

/// A message sent to a room that's not `echo`.
#[message("/<_>", data = "<msg>")]
fn room(channel: &Channel, user: &Username, msg: &str) {
    channel.broadcast(Event::Message(user, msg)).await;
}

/// A message sent to `/echo`: send it to `/echo` and `/lobby` concurrently.
#[message("/echo", data = "<msg>")]
fn echo(broker: &Broker, channel: &Channel, user: &Username, msg: &str) {
    let lobby_msg = channel!(broker, room("/lobby")).broadcast(Event::Message(user, msg));
    let room_msg = channel.broadcast(Event::Message(user, msg));
    join!(lobby_msg, room_msg).await;
}

/// A user has left a room. Let the room know.
#[leave("/<_>")]
fn leave(channel: &Channel, user: &Username, users: &UserList) {
    users.remove(user);
    channel.broadcast(Event::Leave(user)).await;
}

[entropylost]

So I guess there isn't any working prototype for this and I should just use the preexisting websocket thing mentioned above?

[entropylost]

Also maybe add tungstenite to the possible options?

[ponyatov]

Can SocketIO.js on the client side be used as a fallback for Rocket in longpoll mode?
Maybe it can be a more light variant requires much lesser implementation efforts, or using already existing infrastructure.

[jebrosen]

Maybe it can be a more light variant requires much lesser implementation efforts, or using already existing infrastructure.

No; in fact this would probably require even more implementation efforts: Socket.IO uses its own protocol on top of long-polling and/or WebSockets, and the few rust implementations of socket.io servers I found appear to be ready or suitable for "plugging in" to Rocket at the moment.

[the10thWiz]

So I guess there isn't any working prototype for this and I should just use the preexisting websocket thing mentioned above?

If you are interested, there is a pull request I've been working on to integrate websocket support directly into Rocket itself. However, I would caution against using it in production yet, since it's not quite ready yet.

There are several option questions related to websocket support, and some serious testing need to happen. Since it sounds like you are implementing a service that uses websockets, how do you handle user authentication? I've implemented a solution that takes advantage of temporary URLs, but I'm curious to hear what your solution is.

[entropylost]

Well for emulating WS via EventStream, I just send out a unique token and use that as the data field and use a custom eventstream impl on the client that supports payloads.

[Kiiyya]

Looks like WebSockets are currently planned for the 0.8.0 milestone. Sounds it'll be a few years still, then?

[hf29h8sh321]

This proposed API is the best IMO, since it offers the greatest flexibility. A channel-based API may be too limiting for certain use cases.

[TimBoettcher]

Did the design discussions get anywhere? I have a feeling this is a highly anticipated feature, and I myself would appreciate it for my app, too, but it looks like it's not feasible to implement for the time being?

Another question: Does rocket somehow expose internal state? Because if I run a separate server for WS handling (eg. with tokio-tungstenite), I'd need to somehow get the internal state and possibly DB pool of the rocket app, I think.

[toxeus]

@TimBoettcher I recommend reading up the discussion on the PR to get an idea about alternatives and why rocket cannot be used with tokio-tungstenite.

[tvallotton]

What I really like about this proposal is that is the one that is most consistent with Rockets current API. Macros and guards generally deal with request from the user, while the response is dealt by the return type . Here Websocket would be just an ordinary type that implements Handler.

[hf29h8sh321]

I've seen this crate which might provide a somewhat easy way to provide an implementation.

[leap0x7b]

This one is definitely better than most proposals. The WebSocket request API isn't too complex I could easily write it. It's also very flexible than what most other proposal offers.

[Mai-Lapyst]

Worked the last few weeks on a solution similar to the one presented in this comment, by adding an API to expose the hyper upgrade to rocket and implementing websockets with it. Have submitted an combined PR (#2466) for you guys to try it out.

There are some things missing for a complete websocket implementation such as subprotocols and extensions; but for now simple sending/recieving should suffice.

Example for the upgrade API:

struct TestSocket {}

#[crate::async_trait]
impl Upgrade<'static> for TestSocket {
  async fn start(&self, upgraded: crate::http::hyper::upgrade::Upgraded) {
    // can fully use the hyper::upgrade::Upgraded struct
  }
}

impl<'r, F> Responder<'r, 'r> for TestSocket {
  fn respond_to(self, req: &'r Request<'_>) -> response::Result<'r> {
    Response::build()
      .status(Status::SwitchingProtocols)
      .raw_header("Connection", "upgrade")
      .raw_header("Upgrade", "testsocket")
      .upgrade(Some(Box::new(self)))
      .ok()
  }
}

and one for the websockets:

use rocket::response::websocket::{WebsocketMessage, WebsocketChannel, CreateWebsocket, Websocket};

#[get("/ws")]
fn websock() -> Websocket![] {
  CreateWebsocket! {
    while let Some(msg) = ch.next().await {
      println!("handler {msg:?}");
      ch.send(msg);
    }
  }
}

[SergioBenitez]

Connection upgrading support just landed via #2488 and d97c83d! Thanks to @Mai-Lapyst for the push and idea. This means we can implement support for WebSockets outside of Rocket. The new upgrade example does exactly this! Here's the echo server:

#[get("/echo")]
fn echo(ws: ws::WebSocket) -> ws::Channel {
    ws.channel(|mut stream| Box::pin(async move {
        while let Some(message) = stream.next().await {
            let _ = stream.send(message?).await;
        }

        Ok(())
    }))
}

I still believe we want an API like the one in #90 (comment), but this brings us to parity with the echo system ecosystem. Soon, I'll modify the example to make use of Rocket's async stream support and macros, which will make the example closer to:

#[get("/echo")]
fn echo(ws: ws::WebSocket) -> ws::Channel {
    ws::channel! { ws =>
        for await message in ws {
            yield message;
        }
    }
}

[SergioBenitez]

The stream API is now implemented in the example!

https://github.com/SergioBenitez/Rocket/blob/2abddd923e36b49b01821f09f4dc82ee19d69dcc/examples/upgrade/src/main.rs#L19-L26

[SergioBenitez]

WebSocket support is now in the contrib ws crate. Full docs are available at api.rocket.rs. Note that the crate is not yet released, so it cannot be used off of crates.io and must be used from the git repository until the next release. If you wish to test this, you'll also need to depend on rocket from GitHub (as well as rocket_ws from GitHub, of course).

As WebSocket support is now available from an officially supported crate, I believe we can close this issue. I would still like Rocket to eventually support a channels/group-based API as in #90 (comment), but we can leave that for another issue.

Thank you all, for your patience. ❤️

[Krysztal112233]

Congratulations! This feature is really important and I'm really excited to see WebSocket merged into Rocket!

[HarikrishnanBalagopal]

WebSocket support is now in the contrib ws crate. Full docs are available at api.rocket.rs. Note that the crate is not yet released, so it cannot be used off of crates.io and must be used from the git repository until the next release. If you wish to test this, you'll also need to depend on rocket from GitHub (as well as rocket_ws from GitHub, of course).

As WebSocket support is now available from an officially supported crate, I believe we can close this issue. I would still like Rocket to eventually support a channels/group-based API as in #90 (comment), but we can leave that for another issue.

Thank you all, for your patience. ❤️

@SergioBenitez The docs seem to be wrong

The docs say to do

[dependencies]
ws = { package = "rocket_ws", version ="=0.1.0-rc.3" }

but the crate is not yet in crates.io so you need to do
https://github.com/dani-garcia/vaultwarden/blob/bc26bfa589c007da9b9be37e1172060f38a948b9/Cargo.toml#L60

rocket_ws = { git = 'https://github.com/SergioBenitez/Rocket', rev = "ce441b5f46fdf5cd99cb32b8b8638835e4c2a5fa" } # v0.5 branch

[SergioBenitez]

WebSocket support is now in the contrib ws crate. Full docs are available at api.rocket.rs. Note that the crate is not yet released, so it cannot be used off of crates.io and must be used from the git repository until the next release. If you wish to test this, you'll also need to depend on rocket from GitHub (as well as rocket_ws from GitHub, of course).

As WebSocket support is now available from an officially supported crate, I believe we can close this issue. I would still like Rocket to eventually support a channels/group-based API as in #90 (comment), but we can leave that for another issue.

Thank you all, for your patience. ❤️

@SergioBenitez The docs seem to be wrong

The docs say to do

[dependencies]

ws = { package = "rocket_ws", version ="=0.1.0-rc.3" }

but the crate is not yet in crates.io so you need to do

https://github.com/dani-garcia/vaultwarden/blob/bc26bfa589c007da9b9be37e1172060f38a948b9/Cargo.toml#L60

rocket_ws = { git = 'https://github.com/SergioBenitez/Rocket', rev = "ce441b5f46fdf5cd99cb32b8b8638835e4c2a5fa" } # v0.5 branch

We always write docs on master as if they're part of the release master last targeted. Otherwise we'd need to update the docs with a commit rev every time we push them, which is simply untenable. The docs are there as a reference for those using master, so the assumption is they understand how to depend on Rocket and its dependencies from master properly.