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 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189
// Copyright (c) 2019 Parity Technologies (UK) Ltd.
// Copyright (c) 2016 twist developers
//
// Licensed under the Apache License, Version 2.0
// <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0> or the MIT
// license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
// option. All files in the project carrying such notice may not be copied,
// modified, or distributed except according to those terms.
//! An implementation of the [RFC 6455][rfc6455] websocket protocol.
//!
//! To begin a websocket connection one first needs to perform a [handshake],
//! either as [client] or [server], in order to upgrade from HTTP.
//! Once successful, the client or server can transition to a connection,
//! i.e. a [Sender]/[Receiver] pair and send and receive textual or
//! binary data.
//!
//! **Note**: While it is possible to only receive websocket messages it is
//! not possible to only send websocket messages. Receiving data is required
//! in order to react to control frames such as PING or CLOSE. While those will be
//! answered transparently they have to be received in the first place, so
//! calling [`connection::Receiver::receive`] is imperative.
//!
//! **Note**: None of the `async` methods are safe to cancel so their `Future`s
//! must not be dropped unless they return `Poll::Ready`.
//!
//! # Client example
//!
//! ```no_run
//! # use tokio_util::compat::TokioAsyncReadCompatExt;
//! # async fn doc() -> Result<(), soketto::BoxedError> {
//! use soketto::handshake::{Client, ServerResponse};
//!
//! // First, we need to establish a TCP connection.
//! let socket = tokio::net::TcpStream::connect("...").await?;
//!
//! // Then we configure the client handshake.
//! let mut client = Client::new(socket.compat(), "...", "/");
//!
//! // And finally we perform the handshake and handle the result.
//! let (mut sender, mut receiver) = match client.handshake().await? {
//! ServerResponse::Accepted { .. } => client.into_builder().finish(),
//! ServerResponse::Redirect { status_code, location } => unimplemented!("follow location URL"),
//! ServerResponse::Rejected { status_code } => unimplemented!("handle failure")
//! };
//!
//! // Over the established websocket connection we can send
//! sender.send_text("some text").await?;
//! sender.send_text("some more text").await?;
//! sender.flush().await?;
//!
//! // ... and receive data.
//! let mut data = Vec::new();
//! receiver.receive_data(&mut data).await?;
//!
//! # Ok(())
//! # }
//!
//! ```
//!
//! # Server example
//!
//! ```no_run
//! # use tokio_util::compat::TokioAsyncReadCompatExt;
//! # use tokio_stream::{wrappers::TcpListenerStream, StreamExt};
//! # async fn doc() -> Result<(), soketto::BoxedError> {
//! use soketto::{handshake::{Server, ClientRequest, server::Response}};
//!
//! // First, we listen for incoming connections.
//! let listener = tokio::net::TcpListener::bind("...").await?;
//! let mut incoming = TcpListenerStream::new(listener);
//!
//! while let Some(socket) = incoming.next().await {
//! // For each incoming connection we perform a handshake.
//! let mut server = Server::new(socket?.compat());
//!
//! let websocket_key = {
//! let req = server.receive_request().await?;
//! req.key()
//! };
//!
//! // Here we accept the client unconditionally.
//! let accept = Response::Accept { key: websocket_key, protocol: None };
//! server.send_response(&accept).await?;
//!
//! // And we can finally transition to a websocket connection.
//! let (mut sender, mut receiver) = server.into_builder().finish();
//!
//! let mut data = Vec::new();
//! let data_type = receiver.receive_data(&mut data).await?;
//!
//! if data_type.is_text() {
//! sender.send_text(std::str::from_utf8(&data)?).await?
//! } else {
//! sender.send_binary(&data).await?
//! }
//!
//! sender.close().await?
//! }
//!
//! # Ok(())
//! # }
//!
//! ```
//!
//! See `examples/hyper_server.rs` from this crate's repository for an example of
//! starting up a WebSocket server alongside an Hyper HTTP server.
//!
//! [client]: handshake::Client
//! [server]: handshake::Server
//! [Sender]: connection::Sender
//! [Receiver]: connection::Receiver
//! [rfc6455]: https://tools.ietf.org/html/rfc6455
//! [handshake]: https://tools.ietf.org/html/rfc6455#section-4
#![forbid(unsafe_code)]
pub mod base;
pub mod connection;
pub mod data;
pub mod extension;
pub mod handshake;
use bytes::BytesMut;
use futures::io::{AsyncRead, AsyncReadExt};
use std::io;
pub use connection::{Mode, Receiver, Sender};
pub use data::{Data, Incoming};
pub type BoxedError = Box<dyn std::error::Error + Send + Sync>;
/// A parsing result.
#[derive(Debug, Clone)]
pub enum Parsing<T, N = ()> {
/// Parsing completed.
Done {
/// The parsed value.
value: T,
/// The offset into the byte slice that has been consumed.
offset: usize,
},
/// Parsing is incomplete and needs more data.
NeedMore(N),
}
/// A buffer type used for implementing `Extension`s.
#[derive(Debug)]
pub enum Storage<'a> {
/// A read-only shared byte slice.
Shared(&'a [u8]),
/// A mutable byte slice.
Unique(&'a mut [u8]),
/// An owned byte buffer.
Owned(Vec<u8>),
}
impl AsRef<[u8]> for Storage<'_> {
fn as_ref(&self) -> &[u8] {
match self {
Storage::Shared(d) => d,
Storage::Unique(d) => d,
Storage::Owned(b) => b.as_ref(),
}
}
}
/// Helper function to allow casts from `usize` to `u64` only on platforms
/// where the sizes are guaranteed to fit.
#[cfg(any(target_pointer_width = "32", target_pointer_width = "64"))]
const fn as_u64(a: usize) -> u64 {
a as u64
}
/// Fill the buffer from the given `AsyncRead` impl with up to `max` bytes.
async fn read<R>(reader: &mut R, dest: &mut BytesMut, max: usize) -> io::Result<()>
where
R: AsyncRead + Unpin,
{
let i = dest.len();
dest.resize(i + max, 0u8);
let n = reader.read(&mut dest[i..]).await?;
dest.truncate(i + n);
if n == 0 {
return Err(io::ErrorKind::UnexpectedEof.into());
}
log::trace!("read {} bytes", n);
Ok(())
}