Adds support for outbound message chunking

pull/144/head
SirCipher 5 years ago
parent 31b50bf6b5
commit 3cf0b83949
  1. 123
      src/protocol/mod.rs

@ -19,6 +19,7 @@ use crate::error::{Error, Result};
use crate::extensions::uncompressed::UncompressedExt; use crate::extensions::uncompressed::UncompressedExt;
use crate::extensions::WebSocketExtension; use crate::extensions::WebSocketExtension;
use crate::util::NonBlockingResult; use crate::util::NonBlockingResult;
use crate::protocol::frame::coding::Data;
pub(crate) const MAX_MESSAGE_SIZE: usize = 64 << 20; pub(crate) const MAX_MESSAGE_SIZE: usize = 64 << 20;
@ -34,8 +35,8 @@ pub enum Role {
/// The configuration for WebSocket connection. /// The configuration for WebSocket connection.
#[derive(Debug, Copy, Clone)] #[derive(Debug, Copy, Clone)]
pub struct WebSocketConfig<E = UncompressedExt> pub struct WebSocketConfig<E = UncompressedExt>
where where
E: WebSocketExtension, E: WebSocketExtension,
{ {
/// The size of the send queue. You can use it to turn on/off the backpressure features. `None` /// The size of the send queue. You can use it to turn on/off the backpressure features. `None`
/// means here that the size of the queue is unlimited. The default value is the unlimited /// means here that the size of the queue is unlimited. The default value is the unlimited
@ -51,8 +52,8 @@ where
} }
impl<E> Default for WebSocketConfig<E> impl<E> Default for WebSocketConfig<E>
where where
E: WebSocketExtension, E: WebSocketExtension,
{ {
fn default() -> Self { fn default() -> Self {
WebSocketConfig { WebSocketConfig {
@ -64,8 +65,8 @@ where
} }
impl<E> WebSocketConfig<E> impl<E> WebSocketConfig<E>
where where
E: WebSocketExtension, E: WebSocketExtension,
{ {
/// Creates a `WebSocketConfig` instance using the default configuration and the provided /// Creates a `WebSocketConfig` instance using the default configuration and the provided
/// encoder for new connections. /// encoder for new connections.
@ -84,8 +85,8 @@ where
/// It may be created by calling `connect`, `accept` or `client` functions. /// It may be created by calling `connect`, `accept` or `client` functions.
#[derive(Debug)] #[derive(Debug)]
pub struct WebSocket<Stream, Ext> pub struct WebSocket<Stream, Ext>
where where
Ext: WebSocketExtension, Ext: WebSocketExtension,
{ {
/// The underlying socket. /// The underlying socket.
socket: Stream, socket: Stream,
@ -94,8 +95,8 @@ where
} }
impl<Stream, Ext> WebSocket<Stream, Ext> impl<Stream, Ext> WebSocket<Stream, Ext>
where where
Ext: WebSocketExtension, Ext: WebSocketExtension,
{ {
/// Convert a raw socket into a WebSocket without performing a handshake. /// Convert a raw socket into a WebSocket without performing a handshake.
/// ///
@ -167,9 +168,9 @@ where
} }
impl<Stream, Ext> WebSocket<Stream, Ext> impl<Stream, Ext> WebSocket<Stream, Ext>
where where
Stream: Read + Write, Stream: Read + Write,
Ext: WebSocketExtension, Ext: WebSocketExtension,
{ {
/// Read a message from stream, if possible. /// Read a message from stream, if possible.
/// ///
@ -254,8 +255,8 @@ where
/// A context for managing WebSocket stream. /// A context for managing WebSocket stream.
#[derive(Debug)] #[derive(Debug)]
pub struct WebSocketContext<Ext = UncompressedExt> pub struct WebSocketContext<Ext = UncompressedExt>
where where
Ext: WebSocketExtension, Ext: WebSocketExtension,
{ {
/// Server or client? /// Server or client?
role: Role, role: Role,
@ -274,8 +275,8 @@ where
} }
impl<Ext> WebSocketContext<Ext> impl<Ext> WebSocketContext<Ext>
where where
Ext: WebSocketExtension, Ext: WebSocketExtension,
{ {
/// Create a WebSocket context that manages a post-handshake stream. /// Create a WebSocket context that manages a post-handshake stream.
pub fn new(role: Role, config: Option<WebSocketConfig<Ext>>) -> Self { pub fn new(role: Role, config: Option<WebSocketConfig<Ext>>) -> Self {
@ -334,8 +335,8 @@ where
/// This function sends pong and close responses automatically. /// This function sends pong and close responses automatically.
/// However, it never blocks on write. /// However, it never blocks on write.
pub fn read_message<Stream>(&mut self, stream: &mut Stream) -> Result<Message> pub fn read_message<Stream>(&mut self, stream: &mut Stream) -> Result<Message>
where where
Stream: Read + Write, Stream: Read + Write,
{ {
// Do not read from already closed connections. // Do not read from already closed connections.
self.state.check_active()?; self.state.check_active()?;
@ -362,8 +363,8 @@ where
/// Note that only the last pong frame is stored to be sent, and only the /// Note that only the last pong frame is stored to be sent, and only the
/// most recent pong frame is sent if multiple pong frames are queued. /// most recent pong frame is sent if multiple pong frames are queued.
pub fn write_message<Stream>(&mut self, stream: &mut Stream, message: Message) -> Result<()> pub fn write_message<Stream>(&mut self, stream: &mut Stream, message: Message) -> Result<()>
where where
Stream: Read + Write, Stream: Read + Write,
{ {
// When terminated, return AlreadyClosed. // When terminated, return AlreadyClosed.
self.state.check_active()?; self.state.check_active()?;
@ -405,8 +406,8 @@ where
/// Flush the pending send queue. /// Flush the pending send queue.
pub fn write_pending<Stream>(&mut self, stream: &mut Stream) -> Result<()> pub fn write_pending<Stream>(&mut self, stream: &mut Stream) -> Result<()>
where where
Stream: Read + Write, Stream: Read + Write,
{ {
// First, make sure we have no pending frame sending. // First, make sure we have no pending frame sending.
self.frame.write_pending(stream)?; self.frame.write_pending(stream)?;
@ -448,8 +449,8 @@ where
/// There is no need to call it again. Calling this function is /// There is no need to call it again. Calling this function is
/// the same as calling `write(Message::Close(..))`. /// the same as calling `write(Message::Close(..))`.
pub fn close<Stream>(&mut self, stream: &mut Stream, code: Option<CloseFrame>) -> Result<()> pub fn close<Stream>(&mut self, stream: &mut Stream, code: Option<CloseFrame>) -> Result<()>
where where
Stream: Read + Write, Stream: Read + Write,
{ {
if let WebSocketState::Active = self.state { if let WebSocketState::Active = self.state {
self.state = WebSocketState::ClosedByUs; self.state = WebSocketState::ClosedByUs;
@ -463,8 +464,8 @@ where
/// Try to decode one message frame. May return None. /// Try to decode one message frame. May return None.
fn read_message_frame<Stream>(&mut self, stream: &mut Stream) -> Result<Option<Message>> fn read_message_frame<Stream>(&mut self, stream: &mut Stream) -> Result<Option<Message>>
where where
Stream: Read + Write, Stream: Read + Write,
{ {
if let Some(mut frame) = self if let Some(mut frame) = self
.frame .frame
@ -589,8 +590,8 @@ where
/// Send a single pending frame. /// Send a single pending frame.
fn send_one_frame<Stream>(&mut self, stream: &mut Stream, mut frame: Frame) -> Result<()> fn send_one_frame<Stream>(&mut self, stream: &mut Stream, mut frame: Frame) -> Result<()>
where where
Stream: Read + Write, Stream: Read + Write,
{ {
match self.role { match self.role {
Role::Server => {} Role::Server => {}
@ -608,10 +609,27 @@ where
}; };
} }
trace!("Sending frame: {:?}", frame); let max_frame_size = self.config.max_frame_size.unwrap_or_else(usize::max_value);
self.frame if frame.payload().len() > max_frame_size {
.write_frame(stream, frame) let mut chunks = frame.payload().chunks(self.config.max_frame_size.unwrap_or_else(usize::max_value)).peekable();
.check_connection_reset(self.state) let data_frame = Frame::message(Vec::from(chunks.next().unwrap()), frame.header().opcode, false);
self.frame.write_frame(stream, data_frame).check_connection_reset(self.state)?;
while let Some(chunk) = chunks.next() {
let frame = Frame::message(Vec::from(chunk), OpCode::Data(Data::Continue), chunks.peek().is_none());
trace!("Sending frame: {:?}", frame);
self.frame.write_frame(stream, frame).check_connection_reset(self.state)?;
}
Ok(())
} else {
trace!("Sending frame: {:?}", frame);
self.frame
.write_frame(stream, frame)
.check_connection_reset(self.state)
}
} }
} }
@ -685,6 +703,8 @@ mod tests {
use crate::extensions::uncompressed::UncompressedExt; use crate::extensions::uncompressed::UncompressedExt;
use std::io; use std::io;
use std::io::Cursor; use std::io::Cursor;
use crate::protocol::frame::Frame;
use crate::protocol::frame::coding::{OpCode, Data};
struct WriteMoc<Stream>(Stream); struct WriteMoc<Stream>(Stream);
@ -756,4 +776,41 @@ mod tests {
"Space limit exceeded: Message too big: 0 + 3 > 2" "Space limit exceeded: Message too big: 0 + 3 > 2"
); );
} }
#[test]
fn fragmented_tx() {
let max_message_size = 2;
let input_str = "hello unit test";
let limit = WebSocketConfig {
max_send_queue: None,
max_frame_size: Some(2),
encoder: UncompressedExt::new(Some(max_message_size)),
};
let mut socket = WebSocket::from_raw_socket(Cursor::new(Vec::new()), Role::Client, Some(limit));
socket.write_message(Message::text(input_str)).unwrap();
socket.socket.set_position(0);
let WebSocket { mut socket, mut context } = socket;
let vec = input_str.chars().collect::<Vec<_>>();
let mut iter = vec.chunks(max_message_size).map(|c| c.iter().collect::<String>())
.into_iter().peekable();
let frame_eq = |expected: Frame, actual: Frame| {
assert_eq!(expected.payload(), actual.payload());
assert_eq!(expected.header().opcode, actual.header().opcode);
assert_eq!(expected.header().rsv1, actual.header().rsv1);
};
let expected = Frame::message(iter.next().unwrap().into(), OpCode::Data(Data::Text), false);
frame_eq(expected, context.frame.read_frame(&mut socket, Some(max_message_size)).unwrap().unwrap());
while let Some(chars) = iter.next() {
let expected = Frame::message(chars.into(), OpCode::Data(Data::Continue), iter.peek().is_none());
frame_eq(expected, context.frame.read_frame(&mut socket, Some(max_message_size)).unwrap().unwrap());
}
}
} }

Loading…
Cancel
Save