tokio-websockets 0.3.1

High performance, strict, tokio-util based websockets 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

//! Helper for creating [`http::Request`] objects for HTTP/1.1 Upgrade requests
//! with websocket servers.
use bytes::Bytes;
use http::{
    header::{CONNECTION, SEC_WEBSOCKET_KEY, SEC_WEBSOCKET_VERSION, UPGRADE},
    HeaderValue, Request, Uri,
};
use http_body::Empty;

use crate::client::make_key;

/// Create a HTTP upgrade request for use with HTTP libraries.
///
/// This can be sent with a client and then waiting for the upgrade to complete.
///
/// For example, using [hyper](https://docs.rs/hyper/latest/hyper/):
///
/// ```rust
/// use hyper::{self, client::HttpConnector, Client, StatusCode, Uri};
/// use tokio_websockets::{upgrade_request, ClientBuilder};
///
/// # use futures_util::SinkExt;
/// # use tokio_websockets::ServerBuilder;
/// # use tokio::net::TcpListener;
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// # let listener = TcpListener::bind("127.0.0.1:3333").await?;
/// # tokio::spawn(async move {
/// #     let (stream, _) = listener.accept().await?;
/// #     let mut ws_stream = ServerBuilder::new()
/// #         .accept(stream)
/// #         .await.unwrap();
/// #
/// #     while let Some(msg) = ws_stream.next().await {
/// #         let msg = msg?;
/// #
/// #         if msg.is_text() || msg.is_binary() {
/// #             ws_stream.send(msg).await?;
/// #         }
/// #     }
/// #
/// #     Ok::<_, tokio_websockets::Error>(())
/// # });
/// #
/// // Create a new hyper client, ideally you'd reuse an existing one
/// // The HTTP Connector does not allow ws:// per default
/// let mut connector = HttpConnector::new();
/// connector.enforce_http(false);
/// let client = Client::builder().build(connector);
///
/// let uri = Uri::from_static("ws://localhost:3333");
/// let response = client.request(upgrade_request(uri)?).await?;
///
/// assert!(
///     response.status() == StatusCode::SWITCHING_PROTOCOLS,
///     "Our server didn't upgrade: {}",
///     response.status()
/// );
///
/// let stream = match hyper::upgrade::on(response).await {
///     Ok(upgraded) => ClientBuilder::new().take_over(upgraded),
///     Err(e) => panic!("upgrade error: {}", e),
/// };
///
/// // stream is a websocket stream like any other one
///
/// # Ok(()) }
/// ```
///
/// # Errors
///
/// This method returns a [`http::Error`] if assembling the request failed,
/// which should never happen.
pub fn upgrade_request<T>(uri: T) -> Result<Request<Empty<Bytes>>, http::Error>
where
    Uri: TryFrom<T>,
    <Uri as TryFrom<T>>::Error: Into<http::Error>,
{
    let key_bytes = make_key();

    Request::builder()
        .uri(uri)
        .header(CONNECTION, "Upgrade")
        .header(UPGRADE, "websocket")
        .header(SEC_WEBSOCKET_VERSION, "13")
        .header(SEC_WEBSOCKET_KEY, HeaderValue::from_bytes(&key_bytes)?)
        .body(Empty::new())
}