feat(socketio/svc): impl ws creation svc

crates/engineioxide/src/service/mod.rs

@@ -44,6 +44,7 @@ use tower_service::Service as TowerSvc;
 
 use crate::{
     body::ResponseBody, config::EngineIoConfig, engine::EngineIo, handler::EngineIoHandler,
+    service::parser::RequestInfo,
 };
 
 mod futures;
@@ -136,6 +137,24 @@ where
     }
 }
 
+impl<St, H, Svc> TowerSvc<(St, http::Request<()>)> for EngineIoService<H, Svc>
+where
+    H: EngineIoHandler,
+    St: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
+{
+    type Response = ();
+    type Error = parser::ParseError;
+    type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, (conn, req): (St, http::Request<()>)) -> Self::Future {
+        std::future::ready(ws_conn_inner(self.engine.clone(), conn, req))
+    }
+}
+
 /// Hyper 1.0 Service implementation.
 impl<H, ReqBody, ResBody, S> HyperSvc<Request<ReqBody>> for EngineIoService<H, S>
 where
@@ -160,6 +179,38 @@ where
     }
 }
 
+impl<St, H, Svc> HyperSvc<(St, http::Request<()>)> for EngineIoService<H, Svc>
+where
+    H: EngineIoHandler,
+    St: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
+{
+    type Response = ();
+    type Error = parser::ParseError;
+    type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
+
+    fn call(&self, (conn, req): (St, http::Request<()>)) -> Self::Future {
+        std::future::ready(ws_conn_inner(self.engine.clone(), conn, req))
+    }
+}
+
+fn ws_conn_inner<St, H>(
+    engine: Arc<EngineIo<H>>,
+    conn: St,
+    req: http::Request<()>,
+) -> Result<(), parser::ParseError>
+where
+    H: EngineIoHandler,
+    St: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
+{
+    let RequestInfo { protocol, sid, .. } = RequestInfo::parse(&req, &engine.config)?;
+
+    let (parts, _) = req.into_parts();
+    let fut = crate::transport::ws::on_init(engine, conn, protocol, sid, parts);
+    tokio::spawn(fut);
+
+    Ok(())
+}
+
 #[cfg(feature = "__test_harness")]
 #[doc(hidden)]
 impl<H, Svc> EngineIoService<H, Svc>

crates/engineioxide/src/service/parser.rs

@@ -127,7 +127,7 @@ pub struct RequestInfo {
 
 impl RequestInfo {
     /// Parse the request URI to extract the [`TransportType`](crate::service::TransportType) and the socket id.
-    fn parse<B>(req: &Request<B>, config: &EngineIoConfig) -> Result<Self, ParseError> {
+    pub(crate) fn parse<B>(req: &Request<B>, config: &EngineIoConfig) -> Result<Self, ParseError> {
         use ParseError::*;
         let query = req.uri().query().ok_or(UnknownTransport)?;
 

crates/socketioxide/src/service.rs

@@ -54,13 +54,34 @@ where
     type Error = <EngineIoService<Client<A>, S> as TowerSvc<Request<ReqBody>>>::Error;
     type Future = <EngineIoService<Client<A>, S> as TowerSvc<Request<ReqBody>>>::Future;
 
-    #[inline(always)]
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.engine_svc.poll_ready(cx)
+        TowerSvc::<Request<ReqBody>>::poll_ready(&mut self.engine_svc, cx)
     }
-    #[inline(always)]
+
     fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
-        self.engine_svc.call(req)
+        TowerSvc::<Request<ReqBody>>::call(&mut self.engine_svc, req)
+    }
+}
+
+type WsReq<St> = (St, http::Request<()>);
+
+/// Service to create a websocket connection
+impl<St, A, S> TowerSvc<(St, http::Request<()>)> for SocketIoService<S, A>
+where
+    A: Adapter,
+    S: Clone,
+    St: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
+{
+    type Response = <EngineIoService<Client<A>, S> as HyperSvc<WsReq<St>>>::Response;
+    type Error = <EngineIoService<Client<A>, S> as HyperSvc<WsReq<St>>>::Error;
+    type Future = <EngineIoService<Client<A>, S> as HyperSvc<WsReq<St>>>::Future;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        TowerSvc::<WsReq<St>>::poll_ready(&mut self.engine_svc, cx)
+    }
+
+    fn call(&mut self, req: WsReq<St>) -> Self::Future {
+        TowerSvc::<WsReq<St>>::call(&mut self.engine_svc, req)
     }
 }
 
@@ -80,7 +101,23 @@ where
 
     #[inline(always)]
     fn call(&self, req: Request<ReqBody>) -> Self::Future {
-        self.engine_svc.call(req)
+        HyperSvc::<Request<ReqBody>>::call(&self.engine_svc, req)
+    }
+}
+
+/// Service to create a websocket connection
+impl<St, A, S> HyperSvc<WsReq<St>> for SocketIoService<S, A>
+where
+    S: Clone,
+    A: Adapter,
+    St: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
+{
+    type Response = <EngineIoService<Client<A>, S> as HyperSvc<WsReq<St>>>::Response;
+    type Error = <EngineIoService<Client<A>, S> as HyperSvc<WsReq<St>>>::Error;
+    type Future = <EngineIoService<Client<A>, S> as HyperSvc<WsReq<St>>>::Future;
+
+    fn call(&self, req: WsReq<St>) -> Self::Future {
+        HyperSvc::<WsReq<St>>::call(&self.engine_svc, req)
     }
 }