actix/actix-web
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-web synced 2025-06-25 06:39:22 +02:00
Code Issues Releases Wiki Activity

use ByteString as container for websocket text message (#1864)

Browse Source
This commit is contained in:
Rob Ede
2021-01-04 11:27:32 +00:00
committed by GitHub
parent 36aee18c64
commit 7d632d0b7b
13 changed files with 131 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
actix-http/CHANGES.md
View File

@ -2,14 +2,11 @@
## Unreleased - 2021-xx-xx
### Changed
* Bumped `rand` to `0.8`.
* Update `actix-*` dependencies to tokio `1.0` based versions. [#1813]
* Bumped `rand` to `0.8`.
* Update `bytes` to `1.0`. [#1813]
* Update `h2` to `0.3`. [#1813]
[#1813]: https://github.com/actix/actix-web/pull/1813
* The `ws::Message::Text` enum variant now contains a `bytestring::ByteString`. [#1864]
### Removed
* Deprecated `on_connect` methods have been removed. Prefer the new
@ -22,6 +19,7 @@
[#1813]: https://github.com/actix/actix-web/pull/1813
[#1857]: https://github.com/actix/actix-web/pull/1857
[#1864]: https://github.com/actix/actix-web/pull/1864
## 2.2.0 - 2020-11-25

3
actix-http/Cargo.toml
View File

@ -51,9 +51,10 @@ actix = { version = "0.11.0-beta.1", optional = true }
base64 = "0.13"
bitflags = "1.2"
bytes = "1"
bytestring = "1"
cookie = { version = "0.14.1", features = ["percent-encode"] }
copyless = "0.1.4"
derive_more = "0.99.2"
derive_more = "0.99.5"
either = "1.5.3"
encoding_rs = "0.8"
futures-channel = { version = "0.3.7", default-features = false }

57
actix-http/src/ws/codec.rs
View File

@ -1,47 +1,60 @@
use actix_codec::{Decoder, Encoder};
use bitflags::bitflags;
use bytes::{Bytes, BytesMut};
use bytestring::ByteString;
use super::frame::Parser;
use super::proto::{CloseReason, OpCode};
use super::ProtocolError;
/// `WebSocket` Message
/// A WebSocket message.
#[derive(Debug, PartialEq)]
pub enum Message {
/// Text message
Text(String),
/// Binary message
/// Text message.
Text(ByteString),
/// Binary message.
Binary(Bytes),
/// Continuation
/// Continuation.
Continuation(Item),
/// Ping message
/// Ping message.
Ping(Bytes),
/// Pong message
/// Pong message.
Pong(Bytes),
/// Close message with optional reason
/// Close message with optional reason.
Close(Option<CloseReason>),
/// No-op. Useful for actix-net services
/// No-op. Useful for low-level services.
Nop,
}
/// `WebSocket` frame
/// A WebSocket frame.
#[derive(Debug, PartialEq)]
pub enum Frame {
/// Text frame, codec does not verify utf8 encoding
/// Text frame. Note that the codec does not validate UTF-8 encoding.
Text(Bytes),
/// Binary frame
/// Binary frame.
Binary(Bytes),
/// Continuation
/// Continuation.
Continuation(Item),
/// Ping message
/// Ping message.
Ping(Bytes),
/// Pong message
/// Pong message.
Pong(Bytes),
/// Close message with optional reason
/// Close message with optional reason.
Close(Option<CloseReason>),
}
/// `WebSocket` continuation item
/// A `WebSocket` continuation item.
#[derive(Debug, PartialEq)]
pub enum Item {
FirstText(Bytes),
@ -51,13 +64,13 @@ pub enum Item {
}
#[derive(Debug, Copy, Clone)]
/// WebSockets protocol codec
/// WebSocket protocol codec.
pub struct Codec {
flags: Flags,
max_size: usize,
}
bitflags::bitflags! {
bitflags! {
struct Flags: u8 {
const SERVER = 0b0000_0001;
const CONTINUATION = 0b0000_0010;
@ -66,7 +79,7 @@ bitflags::bitflags! {
}
impl Codec {
/// Create new websocket frames decoder
/// Create new websocket frames decoder.
pub fn new() -> Codec {
Codec {
max_size: 65_536,
@ -74,9 +87,9 @@ impl Codec {
}
}
/// Set max frame size
/// Set max frame size.
///
/// By default max size is set to 64kb
/// By default max size is set to 64kb.
pub fn max_size(mut self, size: usize) -> Self {
self.max_size = size;
self

99
actix-http/src/ws/mod.rs
View File

@ -1,11 +1,11 @@
//! WebSocket protocol support.
//!
//! To setup a `WebSocket`, first do web socket handshake then on success
//! convert `Payload` into a `WsStream` stream and then use `WsWriter` to
//! communicate with the peer.
//! To setup a WebSocket, first do web socket handshake then on success convert `Payload` into a
//! `WsStream` stream and then use `WsWriter` to communicate with the peer.
use std::io;
use derive_more::{Display, From};
use derive_more::{Display, Error, From};
use http::{header, Method, StatusCode};
use crate::error::ResponseError;
@ -23,86 +23,103 @@ pub use self::dispatcher::Dispatcher;
pub use self::frame::Parser;
pub use self::proto::{hash_key, CloseCode, CloseReason, OpCode};
/// Websocket protocol errors
#[derive(Debug, Display, From)]
/// WebSocket protocol errors.
#[derive(Debug, Display, From, Error)]
pub enum ProtocolError {
/// Received an unmasked frame from client
#[display(fmt = "Received an unmasked frame from client")]
/// Received an unmasked frame from client.
#[display(fmt = "Received an unmasked frame from client.")]
UnmaskedFrame,
/// Received a masked frame from server
#[display(fmt = "Received a masked frame from server")]
/// Received a masked frame from server.
#[display(fmt = "Received a masked frame from server.")]
MaskedFrame,
/// Encountered invalid opcode
#[display(fmt = "Invalid opcode: {}", _0)]
InvalidOpcode(u8),
/// Encountered invalid opcode.
#[display(fmt = "Invalid opcode: {}.", _0)]
InvalidOpcode(#[error(not(source))] u8),
/// Invalid control frame length
#[display(fmt = "Invalid control frame length: {}", _0)]
InvalidLength(usize),
/// Bad web socket op code
#[display(fmt = "Bad web socket op code")]
#[display(fmt = "Invalid control frame length: {}.", _0)]
InvalidLength(#[error(not(source))] usize),
/// Bad opcode.
#[display(fmt = "Bad opcode.")]
BadOpCode,
/// A payload reached size limit.
#[display(fmt = "A payload reached size limit.")]
Overflow,
/// Continuation is not started
/// Continuation is not started.
#[display(fmt = "Continuation is not started.")]
ContinuationNotStarted,
/// Received new continuation but it is already started
#[display(fmt = "Received new continuation but it is already started")]
/// Received new continuation but it is already started.
#[display(fmt = "Received new continuation but it is already started.")]
ContinuationStarted,
/// Unknown continuation fragment
#[display(fmt = "Unknown continuation fragment.")]
ContinuationFragment(OpCode),
/// Io error
#[display(fmt = "io error: {}", _0)]
/// Unknown continuation fragment.
#[display(fmt = "Unknown continuation fragment: {}.", _0)]
ContinuationFragment(#[error(not(source))] OpCode),
/// I/O error.
#[display(fmt = "I/O error: {}", _0)]
Io(io::Error),
}
impl std::error::Error for ProtocolError {}
impl ResponseError for ProtocolError {}
/// Websocket handshake errors
/// WebSocket handshake errors
#[derive(PartialEq, Debug, Display)]
pub enum HandshakeError {
/// Only get method is allowed
#[display(fmt = "Method not allowed")]
/// Only get method is allowed.
#[display(fmt = "Method not allowed.")]
GetMethodRequired,
/// Upgrade header if not set to websocket
#[display(fmt = "Websocket upgrade is expected")]
/// Upgrade header if not set to websocket.
#[display(fmt = "WebSocket upgrade is expected.")]
NoWebsocketUpgrade,
/// Connection header is not set to upgrade
#[display(fmt = "Connection upgrade is expected")]
/// Connection header is not set to upgrade.
#[display(fmt = "Connection upgrade is expected.")]
NoConnectionUpgrade,
/// Websocket version header is not set
#[display(fmt = "Websocket version header is required")]
/// WebSocket version header is not set.
#[display(fmt = "WebSocket version header is required.")]
NoVersionHeader,
/// Unsupported websocket version
#[display(fmt = "Unsupported version")]
/// Unsupported websocket version.
#[display(fmt = "Unsupported version.")]
UnsupportedVersion,
/// Websocket key is not set or wrong
#[display(fmt = "Unknown websocket key")]
/// WebSocket key is not set or wrong.
#[display(fmt = "Unknown websocket key.")]
BadWebsocketKey,
}
impl ResponseError for HandshakeError {
fn error_response(&self) -> Response {
match *self {
match self {
HandshakeError::GetMethodRequired => Response::MethodNotAllowed()
.header(header::ALLOW, "GET")
.finish(),
HandshakeError::NoWebsocketUpgrade => Response::BadRequest()
.reason("No WebSocket UPGRADE header found")
.finish(),
HandshakeError::NoConnectionUpgrade => Response::BadRequest()
.reason("No CONNECTION upgrade")
.finish(),
HandshakeError::NoVersionHeader => Response::BadRequest()
.reason("Websocket version header is required")
.finish(),
HandshakeError::UnsupportedVersion => Response::BadRequest()
.reason("Unsupported version")
.finish(),
HandshakeError::BadWebsocketKey => {
Response::BadRequest().reason("Handshake error").finish()
}

8
actix-http/src/ws/proto.rs
View File

@ -2,21 +2,27 @@ use std::convert::{From, Into};
use std::fmt;
use self::OpCode::*;
/// Operation codes as part of rfc6455.
/// Operation codes as part of RFC6455.
#[derive(Debug, Eq, PartialEq, Clone, Copy)]
pub enum OpCode {
/// Indicates a continuation frame of a fragmented message.
Continue,
/// Indicates a text data frame.
Text,
/// Indicates a binary data frame.
Binary,
/// Indicates a close control frame.
Close,
/// Indicates a ping control frame.
Ping,
/// Indicates a pong control frame.
Pong,
/// Indicates an invalid opcode was received.
Bad,
}

7
actix-http/tests/test_ws.rs
View File

@ -74,7 +74,7 @@ async fn service(msg: ws::Frame) -> Result<ws::Message, Error> {
let msg = match msg {
ws::Frame::Ping(msg) => ws::Message::Pong(msg),
ws::Frame::Text(text) => {
ws::Message::Text(String::from_utf8_lossy(&text).to_string())
ws::Message::Text(String::from_utf8_lossy(&text).into_owned().into())
}
ws::Frame::Binary(bin) => ws::Message::Binary(bin),
ws::Frame::Continuation(item) => ws::Message::Continuation(item),
@ -101,10 +101,7 @@ async fn test_simple() {
// client service
let mut framed = srv.ws().await.unwrap();
framed
.send(ws::Message::Text("text".to_string()))
.await
.unwrap();
framed.send(ws::Message::Text("text".into())).await.unwrap();
let (item, mut framed) = framed.into_future().await;
assert_eq!(
item.unwrap().unwrap(),