ws verifyciation takes RequestHead; add SendError utility service

actix-framed/src/lib.rs

@@ -2,6 +2,7 @@ mod app;
 mod helpers;
 mod request;
 mod route;
+mod service;
 mod state;
 
 // re-export for convinience
@@ -10,4 +11,5 @@ pub use actix_http::{http, Error, HttpMessage, Response, ResponseError};
 pub use self::app::{FramedApp, FramedAppService};
 pub use self::request::FramedRequest;
 pub use self::route::FramedRoute;
+pub use self::service::{SendError, VerifyWebSockets};
 pub use self::state::State;

actix-framed/src/service.rs

@@ -0,0 +1,112 @@
+use std::marker::PhantomData;
+
+use actix_codec::{AsyncRead, AsyncWrite, Framed};
+use actix_http::error::{Error, ResponseError};
+use actix_http::ws::{verify_handshake, HandshakeError};
+use actix_http::{h1, Request};
+use actix_service::{NewService, Service};
+use futures::future::{ok, Either, FutureResult};
+use futures::{Async, Future, IntoFuture, Poll};
+
+/// Service that verifies incoming request if it is valid websocket
+/// upgrade request. In case of error returns `HandshakeError`
+pub struct VerifyWebSockets<T> {
+    _t: PhantomData<T>,
+}
+
+impl<T> Default for VerifyWebSockets<T> {
+    fn default() -> Self {
+        VerifyWebSockets { _t: PhantomData }
+    }
+}
+
+impl<T> NewService for VerifyWebSockets<T> {
+    type Request = (Request, Framed<T, h1::Codec>);
+    type Response = (Request, Framed<T, h1::Codec>);
+    type Error = (HandshakeError, Framed<T, h1::Codec>);
+    type InitError = ();
+    type Service = VerifyWebSockets<T>;
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self, _: &()) -> Self::Future {
+        ok(VerifyWebSockets { _t: PhantomData })
+    }
+}
+
+impl<T> Service for VerifyWebSockets<T> {
+    type Request = (Request, Framed<T, h1::Codec>);
+    type Response = (Request, Framed<T, h1::Codec>);
+    type Error = (HandshakeError, Framed<T, h1::Codec>);
+    type Future = FutureResult<Self::Response, Self::Error>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, (req, framed): (Request, Framed<T, h1::Codec>)) -> Self::Future {
+        match verify_handshake(req.head()) {
+            Err(e) => Err((e, framed)).into_future(),
+            Ok(_) => Ok((req, framed)).into_future(),
+        }
+    }
+}
+
+/// Send http/1 error response
+pub struct SendError<T, R, E>(PhantomData<(T, R, E)>);
+
+impl<T, R, E> Default for SendError<T, R, E>
+where
+    T: AsyncRead + AsyncWrite,
+    E: ResponseError,
+{
+    fn default() -> Self {
+        SendError(PhantomData)
+    }
+}
+
+impl<T, R, E> NewService for SendError<T, R, E>
+where
+    T: AsyncRead + AsyncWrite + 'static,
+    R: 'static,
+    E: ResponseError + 'static,
+{
+    type Request = Result<R, (E, Framed<T, h1::Codec>)>;
+    type Response = R;
+    type Error = Error;
+    type InitError = ();
+    type Service = SendError<T, R, E>;
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self, _: &()) -> Self::Future {
+        ok(SendError(PhantomData))
+    }
+}
+
+impl<T, R, E> Service for SendError<T, R, E>
+where
+    T: AsyncRead + AsyncWrite + 'static,
+    R: 'static,
+    E: ResponseError + 'static,
+{
+    type Request = Result<R, (E, Framed<T, h1::Codec>)>;
+    type Response = R;
+    type Error = Error;
+    type Future = Either<FutureResult<R, Error>, Box<Future<Item = R, Error = Error>>>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, req: Result<R, (E, Framed<T, h1::Codec>)>) -> Self::Future {
+        match req {
+            Ok(r) => Either::A(ok(r)),
+            Err((e, framed)) => {
+                let res = e.render_response();
+                let e = Error::from(e);
+                Either::B(Box::new(
+                    h1::SendResponse::new(framed, res).then(move |_| Err(e)),
+                ))
+            }
+        }
+    }
+}

actix-framed/tests/test_server.rs

@@ -12,7 +12,7 @@ fn ws_service<T: AsyncRead + AsyncWrite>(
     req: FramedRequest<T>,
 ) -> impl Future<Item = (), Error = Error> {
     let (req, framed, _) = req.into_parts();
-    let res = ws::handshake(&req).unwrap().message_body(());
+    let res = ws::handshake(req.head()).unwrap().message_body(());
 
     framed
         .send((res, body::BodySize::None).into())