Expose the max limit for payload sizes in Websocket Actors.

actix-web-actors/src/ws.rs

@@ -27,7 +27,7 @@ use futures::{Async, Future, Poll, Stream};
 
 /// Do websocket handshake and start ws actor.
 pub fn start<A, T>(actor: A, req: &HttpRequest, stream: T) -> Result<HttpResponse, Error>
-where
+    where
         A: Actor<Context = WebsocketContext<A>> + StreamHandler<Message, ProtocolError>,
         T: Stream<Item = Bytes, Error = PayloadError> + 'static,
 {
@@ -35,6 +35,16 @@ where
     Ok(res.streaming(WebsocketContext::create(actor, stream)))
 }
 
+/// Do websocket handshake and start ws actor with provided limit.
+pub fn start_with_limit<A, T>(actor: A, req: &HttpRequest, stream: T, limit: usize) -> Result<HttpResponse, Error>
+    where
+        A: Actor<Context = WebsocketContext<A>> + StreamHandler<Message, ProtocolError>,
+        T: Stream<Item = Bytes, Error = PayloadError> + 'static,
+{
+    let mut res = handshake(req)?;
+    Ok(res.streaming(WebsocketContext::create_with_limit(actor, stream, limit)))
+}
+
 /// Prepare `WebSocket` handshake response.
 ///
 /// This function returns handshake `HttpResponse`, ready to send to peer.
@@ -101,7 +111,7 @@ pub fn handshake(req: &HttpRequest) -> Result<HttpResponseBuilder, HandshakeErro
 
 /// Execution context for `WebSockets` actors
 pub struct WebsocketContext<A>
-where
+    where
         A: Actor<Context = WebsocketContext<A>>,
 {
     inner: ContextParts<A>,
@@ -109,7 +119,7 @@ where
 }
 
 impl<A> ActorContext for WebsocketContext<A>
-where
+    where
         A: Actor<Context = Self>,
 {
     fn stop(&mut self) {
@@ -126,7 +136,7 @@ where
 }
 
 impl<A> AsyncContext<A> for WebsocketContext<A>
-where
+    where
         A: Actor<Context = Self>,
 {
     fn spawn<F>(&mut self, fut: F) -> SpawnHandle
@@ -162,7 +172,7 @@ where
 }
 
 impl<A> WebsocketContext<A>
-where
+    where
         A: Actor<Context = Self>,
 {
     #[inline]
@@ -171,15 +181,25 @@ where
         where
             A: StreamHandler<Message, ProtocolError>,
             S: Stream<Item = Bytes, Error = PayloadError> + 'static,
+    {
+        WebsocketContext::create_with_limit(actor, stream, 65_536)
+    }
+
+    #[inline]
+    /// Create a new Websocket context from a request, an actor, and a limit
+    pub fn create_with_limit<S>(actor: A, stream: S, limit: usize) -> impl Stream<Item = Bytes, Error = Error>
+        where
+            A: StreamHandler<Message, ProtocolError>,
+            S: Stream<Item = Bytes, Error = PayloadError> + 'static,
     {
         let mb = Mailbox::default();
         let mut ctx = WebsocketContext {
             inner: ContextParts::new(mb.sender_producer()),
             messages: VecDeque::new(),
         };
-        ctx.add_stream(WsStream::new(stream));
+        ctx.add_stream(WsStream::new(stream, limit));
 
-        WebsocketContextFut::new(ctx, actor, mb)
+        WebsocketContextFut::new(ctx, actor, mb, limit)
     }
 
     /// Create a new Websocket context
@@ -191,22 +211,36 @@ where
             F: FnOnce(&mut Self) -> A + 'static,
             A: StreamHandler<Message, ProtocolError>,
             S: Stream<Item = Bytes, Error = PayloadError> + 'static,
+    {
+        WebsocketContext::with_factory_codec(stream, f, 65_536)
+    }
+
+    /// Create a new Websocket context with a limit
+    pub fn with_factory_codec<S, F>(
+        stream: S,
+        f: F,
+        limit: usize,
+    ) -> impl Stream<Item = Bytes, Error = Error>
+        where
+            F: FnOnce(&mut Self) -> A + 'static,
+            A: StreamHandler<Message, ProtocolError>,
+            S: Stream<Item = Bytes, Error = PayloadError> + 'static,
     {
         let mb = Mailbox::default();
         let mut ctx = WebsocketContext {
             inner: ContextParts::new(mb.sender_producer()),
             messages: VecDeque::new(),
         };
-        ctx.add_stream(WsStream::new(stream));
+        ctx.add_stream(WsStream::new(stream, limit));
 
         let act = f(&mut ctx);
 
-        WebsocketContextFut::new(ctx, act, mb)
+        WebsocketContextFut::new(ctx, act, mb, limit)
     }
 }
 
 impl<A> WebsocketContext<A>
-where
+    where
         A: Actor<Context = Self>,
 {
     /// Write payload
@@ -266,7 +300,7 @@ where
 }
 
 impl<A> AsyncContextParts<A> for WebsocketContext<A>
-where
+    where
         A: Actor<Context = Self>,
 {
     fn parts(&mut self) -> &mut ContextParts<A> {
@@ -275,7 +309,7 @@ where
 }
 
 struct WebsocketContextFut<A>
-where
+    where
         A: Actor<Context = WebsocketContext<A>>,
 {
     fut: ContextFut<A, WebsocketContext<A>>,
@@ -285,14 +319,14 @@ where
 }
 
 impl<A> WebsocketContextFut<A>
-where
+    where
         A: Actor<Context = WebsocketContext<A>>,
 {
-    fn new(ctx: WebsocketContext<A>, act: A, mailbox: Mailbox<A>) -> Self {
+    fn new(ctx: WebsocketContext<A>, act: A, mailbox: Mailbox<A>, limit: usize) -> Self {
         let fut = ContextFut::new(ctx, act, mailbox);
         WebsocketContextFut {
             fut,
-            encoder: Codec::new(),
+            encoder: Codec::new().max_size(limit),
             buf: BytesMut::new(),
             closed: false,
         }
@@ -300,7 +334,7 @@ where
 }
 
 impl<A> Stream for WebsocketContextFut<A>
-where
+    where
         A: Actor<Context = WebsocketContext<A>>,
 {
     type Item = Bytes;
@@ -332,7 +366,7 @@ where
 }
 
 impl<A, M> ToEnvelope<A, M> for WebsocketContext<A>
-where
+    where
         A: Actor<Context = WebsocketContext<A>> + Handler<M>,
         M: ActixMessage + Send + 'static,
         M::Result: Send,
@@ -350,13 +384,13 @@ struct WsStream<S> {
 }
 
 impl<S> WsStream<S>
-where
+    where
         S: Stream<Item = Bytes, Error = PayloadError>,
 {
-    fn new(stream: S) -> Self {
+    fn new(stream: S, limit: usize) -> Self {
         Self {
             stream,
-            decoder: Codec::new(),
+            decoder: Codec::new().max_size(limit),
             buf: BytesMut::new(),
             closed: false,
         }
@@ -364,7 +398,7 @@ where
 }
 
 impl<S> Stream for WsStream<S>
-where
+    where
         S: Stream<Item = Bytes, Error = PayloadError>,
 {
     type Item = Message;