Refactor socketioxide parsing

.github/workflows/bencher.yml

@@ -31,7 +31,6 @@ jobs:
           BENCHER_API_TOKEN: ${{ secrets.BENCHER_API_TOKEN }}
           BENCHER_CMD: cargo bench --all-features
           BENCHER_PROJECT: socketioxide
-          RUSTFLAGS: --cfg=socketioxide_test
 
   benchmark_pr:
     if: github.event_name == 'workflow_dispatch' && github.event.pull_request.head.repo.full_name == github.repository
@@ -62,4 +61,3 @@ jobs:
           BENCHER_API_TOKEN: ${{ secrets.BENCHER_API_TOKEN }}
           BENCHER_CMD: cargo bench --all-features
           BENCHER_PROJECT: socketioxide
-          RUSTFLAGS: --cfg=socketioxide_test

.github/workflows/fuzzing.yml

@@ -0,0 +1,73 @@
+name: fuzzing
+
+on:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  fuzzing_decode_value:
+    name: Fuzzing decode value
+    strategy:
+      fail-fast: false
+      matrix:
+        crate: [parser-common, parser-msgpack]
+    runs-on: ubuntu-latest
+    env:
+      TEST_DATA_VERSION: 0.0.2
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: nightly
+          components: rustfmt
+      - uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cargo/bin/
+            ~/.cargo/registry/index/
+            ~/.cargo/registry/cache/
+            ~/.cargo/git/db/
+            target/
+          key: ${{ runner.os }}-cargo-fuzzing
+      - run: cargo install cargo-fuzz
+      - name: download corpus data
+        run: |
+          cd crates/${{ matrix.crate }}
+          wget https://github.com/Totodore/socketioxide-fuzzing-corpus/archive/refs/tags/v$TEST_DATA_VERSION.zip
+          unzip v$TEST_DATA_VERSION.zip
+          rm v$TEST_DATA_VERSION.zip
+      - name: cargo fuzz run decode_value
+        run: |
+          cd crates/${{ matrix.crate }}
+          cargo fuzz run decode_value -- -timeout=5 -max_len=2048 -runs=2000000 -only_ascii=1 \
+            socketioxide-fuzzing-corpus-$TEST_DATA_VERSION/corpus-${{ matrix.crate }}
+  fuzzing_decode_packet:
+    name: Fuzzing decode packet
+    strategy:
+      fail-fast: false
+      matrix:
+        crate: [parser-common, parser-msgpack]
+        target: [decode_packet_bin, decode_packet_str]
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: nightly
+          components: rustfmt
+      - uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cargo/bin/
+            ~/.cargo/registry/index/
+            ~/.cargo/registry/cache/
+            ~/.cargo/git/db/
+            target/
+          key: ${{ runner.os }}-cargo-fuzzing
+      - run: cargo install cargo-fuzz
+      - name: cargo fuzz run decode_packet
+        run: |
+          cd crates/${{ matrix.crate }}
+          cargo fuzz run ${{ matrix.target }} -- -timeout=5 -max_len=2048 -runs=2000000 -only_ascii=1

.github/workflows/github-ci.yml

@@ -18,8 +18,6 @@ jobs:
           toolchain: stable
           components: rustfmt
       - run: cargo fmt --all -- --check
-        env:
-          RUSTFLAGS: --cfg=socketioxide_test
 
   test:
     runs-on: ubuntu-latest
@@ -39,8 +37,6 @@ jobs:
             target/
           key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}
       - run: cargo test --tests --all-features --workspace
-        env:
-          RUSTFLAGS: --cfg=socketioxide_test
   udeps:
     runs-on: ubuntu-latest
     steps:
@@ -90,8 +86,6 @@ jobs:
 
       - name: check crates
         run: cargo check -p socketioxide -p engineioxide --all-features
-        env:
-          RUSTFLAGS: --cfg=socketioxide_test
 
   feature_set:
     runs-on: ubuntu-latest
@@ -179,16 +173,15 @@ jobs:
             target/
           key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}-clippy
 
-      - name: Install required cargo
+      - name: Install required cargo                                                                                                                                                                                                      │
         run: cargo install clippy-sarif sarif-fmt || true
 
       - name: Run rust-clippy
         run: cargo clippy
           --all-features
+          --tests
           --message-format=json | clippy-sarif | tee rust-clippy-results.sarif | sarif-fmt
         continue-on-error: true
-        env:
-          RUSTFLAGS: --cfg=socketioxide_test
 
       - name: Upload analysis results to GitHub
         uses: github/codeql-action/upload-sarif@v3
@@ -242,7 +235,7 @@ jobs:
     needs: [test]
     strategy:
       matrix:
-        socketio-version: [v4, v5]
+        socketio-version: [v4, v5, v5-msgpack]
     steps:
       - uses: actions/checkout@v4
       - uses: dtolnay/rust-toolchain@master
@@ -267,9 +260,11 @@ jobs:
           key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}-release
       - name: Install deps & run tests
         run: |
+          PARSER=$(echo ${{ matrix.socketio-version }} | cut -d'-' -f2 -s)
+          VERSION=$(echo ${{ matrix.socketio-version }} | cut -d'-' -f1)
           cd socket.io-protocol/test-suite && npm install && cd ../..
-          cargo build -p socketioxide-e2e --bin socketioxide-e2e --features ${{ matrix.socketio-version }} --release
-          cargo run -p socketioxide-e2e --bin socketioxide-e2e --features ${{ matrix.socketio-version }} --release > server.txt & npm --prefix socket.io-protocol/test-suite test > client.txt
+          cargo build -p socketioxide-e2e --bin socketioxide-e2e --features $VERSION,$PARSER --release
+          cargo run -p socketioxide-e2e --bin socketioxide-e2e --features $VERSION,$PARSER --release > server.txt & npm --prefix socket.io-protocol/test-suite test > client.txt
       - name: Server output
         if: always()
         run: cat server.txt

.github/workflows/release.yml

@@ -16,15 +16,15 @@ jobs:
       - name: Get releasing version
         working-directory: .
         run: |
-            NEXT_VERSION=$(sed -nE 's/^\s*version = "(.*?)"/\1/p' Cargo.toml) >> $GITHUB_ENV
-            echo NEXT_VERSION=$NEXT_VERSION >> $GITHUB_ENV
-            echo $NEXT_VERSION
+          NEXT_VERSION=$(sed -nE 's/^\s*version = "(.*?)"/\1/p' Cargo.toml) >> $GITHUB_ENV
+          echo NEXT_VERSION=$NEXT_VERSION >> $GITHUB_ENV
+          echo $NEXT_VERSION
 
       - name: Check published version
         run: |
-            PREV_VERSION=$(cargo search socketioxide --limit 1 | sed -nE 's/^[^"]*"//; s/".*//1p' -)
-            echo PREV_VERSION=$PREV_VERSION >> $GITHUB_ENV
-            echo $PREV_VERSION
+          PREV_VERSION=$(cargo search socketioxide --limit 1 | sed -nE 's/^[^"]*"//; s/".*//1p' -)
+          echo PREV_VERSION=$PREV_VERSION >> $GITHUB_ENV
+          echo $PREV_VERSION
 
       - name: Cancel run
         if: env.NEXT_VERSION == env.PREV_VERSION
@@ -44,17 +44,23 @@ jobs:
 
   publish:
     runs-on: ubuntu-latest
-    needs: 
+    needs:
       - git-tag
     strategy:
       fail-fast: false
       max-parallel: 1
       matrix:
         package:
-            - crate: engineioxide
-              path: engineioxide
-            - crate: socketioxide
-              path: socketioxide
+          - crate: engineioxide
+            path: crates/engineioxide
+          - crate: socketioxide_core
+            path: crates/socketioxide_core
+          - crate: socketioxide_parser_common
+            path: crates/parser_common
+          - crate: socketioxide_parser_msgpack
+            path: crates/parser_msgpack
+          - crate: socketioxide
+            path: crates/socketioxide
     steps:
       - uses: dtolnay/rust-toolchain@stable
         with:
@@ -89,4 +95,4 @@ jobs:
         run: |
           echo "Cargo Publishing..."
           cargo publish --no-verify
-          echo "New version of $NEXT_VERSION has been published"
+          echo "New version of $NEXT_VERSION has been published"

CONTRIBUTING.md

@@ -16,7 +16,7 @@ today! As a contributor, here are the guidelines we would like you to follow:
 
 ## <a name="question"></a> Got a Question or Problem?
 
-Please open a discussion on the [Q&A discussions](https://github.com/totodore/socketioxide/discussions) page. 
+Please open a discussion on the [Q&A discussions](https://github.com/totodore/socketioxide/discussions) page.
 We want to keep GitHub Issues for bugs and feature requests. If you open an issue it will be moved to Discussions.
 
 ## <a name="issue"></a> Found a Bug?
@@ -139,11 +139,6 @@ You will need [rustc and cargo](www.rust-lang.org/tools/install).
     ```shell
     git clone https://github.com/totodore/socketioxide
     ```
-2. To test socketioxide don't forget to enable the flag `socketioxide_test` through the `RUSTFLAGS` environment variable:
-
-    ```shell
-    export RUSTFLAGS="--cfg socketioxide_test"
-    ```
 2. Depending on what you want to change, clone the [socketio/engine.io-protocol](https://github.com/socketio/engine.io-protocol) repo or the [socketio/socket.io-protocol](https://github.com/socketio/socket.io-protocol) repo or both
     ```shell
     git clone https://github.com/socketio/engine.io-protocol
@@ -261,4 +256,4 @@ The subject contains succinct description of the change:
 ### Body
 
 Just as in the **subject**, use the imperative, present tense: "change" not "changed" nor "changes".
-The body should include the motivation for the change and contrast this with previous behavior.
+The body should include the motivation for the change and contrast this with previous behavior.

Cargo.toml

@@ -14,8 +14,8 @@ categories = [
 license = "MIT"
 
 [workspace]
-members = ["engineioxide", "socketioxide", "e2e/*"]
-default-members = ["engineioxide", "socketioxide"]
+members = ["crates/*", "e2e/*", "crates/*/fuzz"]
+default-members = ["crates/*"]
 resolver = "2"
 
 [workspace.dependencies]
@@ -45,5 +45,6 @@ matchit = "0.8.2"
 tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
 criterion = { version = "0.5.1", features = [
     "rayon",
+    "html_reports",
 ], default-features = false }
 axum = "0.7.2"

engineioxide/Cargo.toml → crates/engineioxide/Cargo.toml

@@ -58,6 +58,7 @@ hyper-util = { workspace = true, features = ["tokio", "client-legacy"] }
 [features]
 v3 = ["memchr", "unicode-segmentation", "itoa"]
 tracing = ["dep:tracing"]
+__test_harness = []
 
 [[bench]]
 name = "packet_encode"

engineioxide/src/errors.rs → crates/engineioxide/src/errors.rs

@@ -8,8 +8,6 @@ use crate::sid::Sid;
 
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
-    #[error("error serializing json packet: {0:?}")]
-    Serialize(#[from] serde_json::Error),
     #[error("error decoding binary packet from polling request: {0:?}")]
     Base64(#[from] base64::DecodeError),
     #[error("error decoding packet: {0:?}")]

engineioxide/src/lib.rs → crates/engineioxide/src/lib.rs

@@ -10,7 +10,6 @@
     clippy::needless_borrow,
     clippy::match_wildcard_for_single_variants,
     clippy::if_let_mutex,
-    clippy::mismatched_target_os,
     clippy::await_holding_lock,
     clippy::match_on_vec_items,
     clippy::imprecise_flops,
@@ -36,7 +35,8 @@ pub use crate::str::Str;
 pub use service::{ProtocolVersion, TransportType};
 pub use socket::{DisconnectReason, Socket};
 
-#[cfg(any(test, socketioxide_test))]
+#[doc(hidden)]
+#[cfg(feature = "__test_harness")]
 pub use packet::*;
 
 pub mod config;

engineioxide/src/packet.rs → crates/engineioxide/src/packet.rs

@@ -118,7 +118,7 @@ impl TryInto<String> for Packet {
         match self {
             Packet::Open(open) => {
                 buffer.push('0');
-                buffer.push_str(&serde_json::to_string(&open)?);
+                buffer.push_str(&serde_json::to_string(&open).unwrap());
             }
             Packet::Close => buffer.push('1'),
             Packet::Ping => buffer.push('2'),

engineioxide/src/socket.rs → crates/engineioxide/src/socket.rs

@@ -55,6 +55,7 @@
 //! let svc = EngineIoService::new(Arc::new(MyHandler::default()));
 //! ```
 use std::{
+    collections::VecDeque,
     sync::{
         atomic::{AtomicU8, Ordering},
         Arc,
@@ -108,10 +109,8 @@ impl From<&Error> for Option<DisconnectReason> {
         match err {
             WsTransport(tungstenite::Error::ConnectionClosed) => None,
             WsTransport(_) | Io(_) => Some(DisconnectReason::TransportError),
-            BadPacket(_) | Serialize(_) | Base64(_) | StrUtf8(_) | PayloadTooLarge
-            | InvalidPacketLength | InvalidPacketType(_) => {
-                Some(DisconnectReason::PacketParsingError)
-            }
+            BadPacket(_) | Base64(_) | StrUtf8(_) | PayloadTooLarge | InvalidPacketLength
+            | InvalidPacketType(_) => Some(DisconnectReason::PacketParsingError),
             HeartbeatTimeout => Some(DisconnectReason::HeartbeatTimeout),
             _ => None,
         }
@@ -126,8 +125,8 @@ pub struct Permit<'a> {
 impl Permit<'_> {
     /// Consume the permit and emit a message to the client.
     #[inline]
-    pub fn emit(self, msg: String) {
-        self.inner.send(smallvec![Packet::Message(msg.into())]);
+    pub fn emit(self, msg: Str) {
+        self.inner.send(smallvec![Packet::Message(msg)]);
     }
     /// Consume the permit and emit a binary message to the client.
     #[inline]
@@ -138,9 +137,21 @@ impl Permit<'_> {
     /// Consume the permit and emit a message with multiple binary data to the client.
     ///
     /// It can be used to ensure atomicity when sending a string packet with adjacent binary packets.
-    pub fn emit_many(self, msg: String, data: Vec<Bytes>) {
+    pub fn emit_many(self, msg: Str, data: VecDeque<Bytes>) {
+        let mut packets = SmallVec::with_capacity(data.len() + 1);
+        packets.push(Packet::Message(msg));
+        for d in data {
+            packets.push(Packet::Binary(d));
+        }
+        self.inner.send(packets);
+    }
+
+    /// Consume the permit and emit a message with multiple binary data to the client.
+    ///
+    /// It can be used to ensure atomicity when sending a string packet with adjacent binary packets.
+    pub fn emit_many_binary(self, bin: Bytes, data: Vec<Bytes>) {
         let mut packets = SmallVec::with_capacity(data.len() + 1);
-        packets.push(Packet::Message(msg.into()));
+        packets.push(Packet::Binary(bin));
         for d in data {
             packets.push(Packet::Binary(d));
         }
@@ -155,7 +166,7 @@ pub(crate) type PacketBuf = SmallVec<[Packet; 10]>;
 ///
 /// It handles :
 /// * the packet communication between with the `Engine`
-/// and the user defined [`Handler`](crate::handler::EngineIoHandler).
+///   and the user defined [`Handler`](crate::handler::EngineIoHandler).
 /// * the user defined [`Data`](crate::handler::EngineIoHandler::Data) bound to the socket.
 /// * the heartbeat job that verify that the connection is still up by sending packets periodically.
 pub struct Socket<D>
@@ -185,7 +196,7 @@ where
     /// * From the [encoder](crate::service::encoder) if the transport is polling
     /// * From the fn [`on_ws_req_init`](crate::engine::EngineIo) if the transport is websocket
     /// * Automatically via the [`close_session fn`](crate::engine::EngineIo::close_session) as a fallback.
-    /// Because with polling transport, if the client is not currently polling then the encoder will never be able to close the channel
+    ///   Because with polling transport, if the client is not currently polling then the encoder will never be able to close the channel
     ///
     /// The channel is made of a [`SmallVec`] of [`Packet`]s so that adjacent packets can be sent atomically.
     pub(crate) internal_rx: Mutex<PeekableReceiver<PacketBuf>>,
@@ -469,7 +480,8 @@ impl<D: Default + Send + Sync + 'static> std::fmt::Debug for Socket<D> {
     }
 }
 
-#[cfg(socketioxide_test)]
+#[doc(hidden)]
+#[cfg(feature = "__test_harness")]
 impl<D> Drop for Socket<D>
 where
     D: Default + Send + Sync + 'static,
@@ -480,7 +492,8 @@ where
     }
 }
 
-#[cfg(any(socketioxide_test, test))]
+#[doc(hidden)]
+#[cfg(feature = "__test_harness")]
 impl<D> Socket<D>
 where
     D: Default + Send + Sync + 'static,