feat(client): implement an engineio client based on Stream + Sink

crates/engineioxide-client/Cargo.toml

@@ -0,0 +1,44 @@
+[package]
+name = "engineioxide-client"
+description = "Engine IO client implementation in rust"
+version = "0.17.0"
+edition.workspace = true
+rust-version.workspace = true
+authors.workspace = true
+repository.workspace = true
+homepage.workspace = true
+keywords.workspace = true
+categories.workspace = true
+license.workspace = true
+readme = "README.md"
+
+[dependencies]
+engineioxide-core = { path = "../engineioxide-core", version = "0.2" }
+bytes.workspace = true
+futures-core.workspace = true
+futures-util.workspace = true
+http.workspace = true
+http-body.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+thiserror.workspace = true
+tokio = { workspace = true, features = ["rt", "time"] }
+hyper = { workspace = true, features = ["client", "http1"] }
+tokio-tungstenite.workspace = true
+http-body-util.workspace = true
+pin-project-lite.workspace = true
+smallvec.workspace = true
+hyper-util = { workspace = true, features = ["tokio"] }
+
+# Tracing
+tracing = { workspace = true } # TODO: make optional
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["macros", "parking_lot"] }
+tracing-subscriber = { workspace = true, features = ["env-filter"] }
+engineioxide = { path = "../engineioxide", features = ["tracing", "v3"] }
+
+[features]
+v3 = ["engineioxide-core/v3"]
+# tracing = ["dep:tracing", "engineioxide-core/tracing"]
+__test_harness = []

crates/engineioxide-client/src/client.rs

@@ -0,0 +1,161 @@
+use std::{
+    fmt,
+    pin::Pin,
+    task::{Context, Poll, ready},
+};
+
+use engineioxide_core::{Packet, PacketParseError, Sid};
+use futures_core::Stream;
+use futures_util::{
+    Sink, StreamExt,
+    stream::{SplitSink, SplitStream},
+};
+
+use crate::{
+    HttpClient,
+    transport::{Transport, polling::PollingSvc},
+};
+
+#[derive(Debug, thiserror::Error)]
+pub enum ClientError<S: PollingSvc> {
+    #[error("packet parse error")]
+    PacketParse(#[from] PacketParseError),
+    #[error("transport error")]
+    Transport(S::Error),
+}
+
+pin_project_lite::pin_project! {
+    pub struct Client<S: PollingSvc> {
+        #[pin]
+        pub transport_rx: SplitStream<Transport<S>>,
+        #[pin]
+        pub transport_tx: SplitSink<Transport<S>, Packet>,
+
+        should_send_pong: bool,
+        should_flush: bool,
+        pub sid: Sid,
+    }
+}
+
+impl<S: PollingSvc> Client<S>
+where
+    S::Error: fmt::Debug,
+    <S::Body as http_body::Body>::Error: fmt::Debug,
+{
+    pub async fn connect(svc: S) -> Result<Self, PacketParseError> {
+        let (inner, open) = HttpClient::connect(svc).await?;
+        let transport = Transport::Polling { inner };
+        let (transport_tx, transport_rx) = transport.split();
+        let client = Client {
+            transport_tx,
+            transport_rx,
+            sid: open.sid,
+
+            should_flush: false,
+            should_send_pong: false,
+        };
+
+        Ok(client)
+    }
+}
+impl<S: PollingSvc> Client<S>
+where
+    S::Error: fmt::Debug,
+    <S::Body as http_body::Body>::Error: fmt::Debug,
+{
+    #[tracing::instrument(skip(cx))]
+    fn heartbeat(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), S::Error>> {
+        let mut this = self.project();
+        if let Err(e) = ready!(this.transport_tx.as_mut().poll_ready(cx)) {
+            return Poll::Ready(Err(e));
+        }
+        if let Err(e) = this.transport_tx.as_mut().start_send(Packet::Pong) {
+            return Poll::Ready(Err(e));
+        }
+
+        *this.should_send_pong = false;
+        *this.should_flush = true;
+        Poll::Ready(Ok(()))
+    }
+
+    #[tracing::instrument(skip(cx))]
+    fn flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), S::Error>> {
+        let mut this = self.project();
+        if let Err(e) = ready!(this.transport_tx.as_mut().poll_flush(cx)) {
+            return Poll::Ready(Err(e));
+        }
+        *this.should_flush = false;
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl<S: PollingSvc> Stream for Client<S>
+where
+    S::Error: fmt::Debug,
+    <S::Body as http_body::Body>::Error: fmt::Debug,
+{
+    type Item = Result<Packet, ClientError<S>>;
+
+    #[tracing::instrument(skip(cx))]
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        if self.should_send_pong {
+            //TODO: ret err
+            if let Err(e) = ready!(self.as_mut().heartbeat(cx)) {
+                return Poll::Ready(Some(Err(ClientError::Transport(e))));
+            }
+        }
+
+        if self.should_flush {
+            //TODO: ret err
+            if let Err(e) = ready!(self.as_mut().flush(cx)) {
+                return Poll::Ready(Some(Err(ClientError::Transport(e))));
+            }
+        }
+
+        match ready!(self.as_mut().project().transport_rx.poll_next(cx)) {
+            Some(Ok(Packet::Ping)) => {
+                if self.as_mut().heartbeat(cx).is_pending() {
+                    self.should_send_pong = true;
+                }
+                self.poll_next(cx)
+            }
+            Some(Ok(packet)) => Poll::Ready(Some(Ok(packet))),
+            Some(Err(e)) => Poll::Ready(Some(Err(e.into()))),
+            None => Poll::Ready(None),
+        }
+    }
+}
+
+impl<S: PollingSvc> Sink<Packet> for Client<S>
+where
+    S::Error: fmt::Debug,
+    <S::Body as http_body::Body>::Error: fmt::Debug,
+{
+    type Error = <Transport<S> as Sink<Packet>>::Error;
+
+    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.project().transport_tx.poll_ready(cx)
+    }
+
+    fn start_send(self: Pin<&mut Self>, item: Packet) -> Result<(), Self::Error> {
+        self.project().transport_tx.start_send(item)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.project().transport_tx.poll_flush(cx)
+    }
+
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.project().transport_tx.poll_close(cx)
+    }
+}
+
+impl<S: PollingSvc> fmt::Debug for Client<S> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Client")
+            .field("should_send_pong", &self.should_send_pong)
+            .field("should_flush", &self.should_flush)
+            .field("sid", &self.sid)
+            .finish()
+    }
+}

crates/engineioxide-client/src/io.rs

@@ -0,0 +1 @@
+

crates/engineioxide-client/src/lib.rs

@@ -0,0 +1,15 @@
+#![allow(missing_docs)]
+#![cfg_attr(docsrs, feature(doc_cfg))]
+#![doc(
+    html_logo_url = "https://raw.githubusercontent.com/Totodore/socketioxide/refs/heads/main/.github/logo_dark.svg"
+)]
+#![doc(
+    html_favicon_url = "https://raw.githubusercontent.com/Totodore/socketioxide/refs/heads/main/.github/logo_dark.ico"
+)]
+//! Engine.IO client library for Rust.
+
+mod client;
+mod io;
+mod transport;
+pub use crate::client::Client;
+pub use crate::transport::polling::HttpClient;

crates/engineioxide-client/src/transport/mod.rs

@@ -0,0 +1,81 @@
+use std::{
+    fmt,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use engineioxide_core::{Packet, PacketParseError, TransportType};
+use futures_core::Stream;
+use futures_util::Sink;
+
+use crate::{HttpClient, transport::polling::PollingSvc};
+
+pub mod polling;
+
+pin_project_lite::pin_project! {
+    #[project = TransportProj]
+    pub enum Transport<S: PollingSvc> {
+        Polling {
+            #[pin]
+            inner: HttpClient<S>
+        },
+        Websocket {
+            #[pin]
+            inner: HttpClient<S>
+        }
+    }
+}
+
+impl<S: PollingSvc> Transport<S> {
+    pub fn transport_type(&self) -> TransportType {
+        match self {
+            Transport::Polling { .. } => TransportType::Polling,
+            Transport::Websocket { .. } => TransportType::Websocket,
+        }
+    }
+}
+
+impl<S: PollingSvc> Stream for Transport<S>
+where
+    S::Error: fmt::Debug,
+    <S::Body as http_body::Body>::Error: fmt::Debug,
+{
+    type Item = Result<Packet, PacketParseError>;
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        match self.as_mut().project() {
+            TransportProj::Polling { inner } => inner.poll_next(cx),
+            TransportProj::Websocket { inner } => inner.poll_next(cx),
+        }
+    }
+}
+impl<S: PollingSvc> Sink<Packet> for Transport<S> {
+    type Error = <HttpClient<S> as Sink<Packet>>::Error;
+
+    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        match self.project() {
+            TransportProj::Polling { inner } => inner.poll_ready(cx),
+            TransportProj::Websocket { inner } => inner.poll_ready(cx),
+        }
+    }
+
+    fn start_send(self: Pin<&mut Self>, item: Packet) -> Result<(), Self::Error> {
+        match self.project() {
+            TransportProj::Polling { inner } => inner.start_send(item),
+            TransportProj::Websocket { inner } => inner.start_send(item),
+        }
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        match self.project() {
+            TransportProj::Polling { inner } => inner.poll_flush(cx),
+            TransportProj::Websocket { inner } => inner.poll_flush(cx),
+        }
+    }
+
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        match self.project() {
+            TransportProj::Polling { inner } => inner.poll_close(cx),
+            TransportProj::Websocket { inner } => inner.poll_close(cx),
+        }
+    }
+}