[FEAT]: Remove RwLock<HashMap<T>>, use dashmap instead, make PubSub generic.

.github/workflows/e2e-tests.yml

@@ -2,9 +2,9 @@ name: E2E Tests
 
 on:
   push:
-    branches: [ main, develop ]
+    branches: [main, develop]
   pull_request:
-    branches: [ main, develop ]
+    branches: [main, develop]
 
 jobs:
   e2e-tests:
@@ -14,115 +14,115 @@ jobs:
       fail-fast: false
       matrix:
         project: [chromium, firefox, webkit, "Mobile Chrome", "Mobile Safari"]
-    
+
     steps:
-    - uses: actions/checkout@v4
-    
-    - name: Setup Rust
-      uses: dtolnay/rust-toolchain@stable
-      with:
-        targets: wasm32-unknown-unknown
-        components: rustfmt, clippy
-    
-    - name: Cache Cargo registry and git trees
-      uses: actions/cache@v4
-      with:
-        path: |
-          ~/.cargo/registry/index/
-          ~/.cargo/registry/cache/
-          ~/.cargo/git/db/
-        key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }}
-        restore-keys: |
-          ${{ runner.os }}-cargo-registry-
-    
-    - name: Cache Cargo binaries
-      uses: actions/cache@v4
-      with:
-        path: ~/.cargo/bin/
-        key: ${{ runner.os }}-cargo-bin-trunk-${{ hashFiles('**/Cargo.lock') }}
-        restore-keys: |
-          ${{ runner.os }}-cargo-bin-trunk-
-          ${{ runner.os }}-cargo-bin-
-    
-    - name: Cache target directory
-      uses: actions/cache@v4
-      with:
-        path: target/
-        key: ${{ runner.os }}-target-${{ hashFiles('**/Cargo.lock') }}-${{ hashFiles('**/*.rs') }}
-        restore-keys: |
-          ${{ runner.os }}-target-${{ hashFiles('**/Cargo.lock') }}-
-          ${{ runner.os }}-target-
-    
-    - name: Install Rust dependencies
-      run: |
-        cargo install trunk --locked || echo "trunk already installed"
-        cargo install cargo-watch --locked || echo "cargo-watch already installed"
-    
-    - name: Setup Node.js
-      uses: actions/setup-node@v4
-      with:
-        node-version: '20'
-        cache: 'npm'
-        cache-dependency-path: 'tests/e2e/package-lock.json'
-    
-    - name: Cache global npm packages
-      uses: actions/cache@v4
-      with:
-        path: ~/.npm
-        key: ${{ runner.os }}-npm-global-${{ hashFiles('**/package-lock.json') }}
-        restore-keys: |
-          ${{ runner.os }}-npm-global-
-    
-    - name: Cache frontend node_modules
-      uses: actions/cache@v4
-      with:
-        path: crates/kiko-frontend/node_modules
-        key: ${{ runner.os }}-frontend-node-modules-tailwindcss
-        restore-keys: |
-          ${{ runner.os }}-frontend-node-modules-
-    
-    - name: Install Node.js dependencies
-      run: |
-        # Install tailwindcss globally and create local node_modules in frontend
-        npm install -g tailwindcss
-        cd crates/kiko-frontend
-        mkdir -p node_modules
-        npm install tailwindcss --no-save
-        cd ../../tests/e2e && npm install
-    
-    - name: Cache Playwright Browsers
-      uses: actions/cache@v4
-      with:
-        path: ~/.cache/ms-playwright
-        key: ${{ runner.os }}-playwright-${{ hashFiles('tests/e2e/package-lock.json') }}
-        restore-keys: |
-          ${{ runner.os }}-playwright-
-    
-    - name: Install Playwright Browsers
-      run: cd tests/e2e && npx playwright install --with-deps
-    
-    - name: Build Rust project
-      run: |
-        cargo build --release
-        cd crates/kiko-frontend && RUSTFLAGS="-C target-feature=-nontrapping-fptoint" trunk build --release
-    
-    - name: Run E2E tests
-      run: cd tests/e2e && npx playwright test --project="${{ matrix.project }}"
-      env:
-        CI: true
-    
-    - name: Upload Playwright Report
-      uses: actions/upload-artifact@v4
-      if: failure()
-      with:
-        name: playwright-report
-        path: tests/e2e/playwright-report/
-        retention-days: 30
-    
-    - name: Upload test screenshots
-      uses: actions/upload-artifact@v4
-      if: failure()
-      with:
-        name: test-screenshots
-        path: tests/e2e/test-screenshots/
-        retention-days: 30
+      - uses: actions/checkout@v4
+
+      - name: Setup Rust
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          targets: wasm32-unknown-unknown
+          components: rustfmt, clippy
+
+      - name: Cache Cargo registry and git trees
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cargo/registry/index/
+            ~/.cargo/registry/cache/
+            ~/.cargo/git/db/
+          key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-cargo-registry-
+
+      - name: Cache Cargo binaries
+        uses: actions/cache@v4
+        with:
+          path: ~/.cargo/bin/
+          key: ${{ runner.os }}-cargo-bin-trunk-${{ hashFiles('**/Cargo.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-cargo-bin-trunk-
+            ${{ runner.os }}-cargo-bin-
+
+      - name: Cache target directory
+        uses: actions/cache@v4
+        with:
+          path: target/
+          key: ${{ runner.os }}-target-${{ hashFiles('**/Cargo.lock') }}-${{ hashFiles('**/*.rs') }}
+          restore-keys: |
+            ${{ runner.os }}-target-${{ hashFiles('**/Cargo.lock') }}-
+            ${{ runner.os }}-target-
+
+      - name: Install Rust dependencies
+        run: |
+          cargo install trunk --locked || echo "trunk already installed"
+          cargo install cargo-watch --locked || echo "cargo-watch already installed"
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: "20"
+          cache: "npm"
+          cache-dependency-path: "tests/e2e/package-lock.json"
+
+      - name: Cache global npm packages
+        uses: actions/cache@v4
+        with:
+          path: ~/.npm
+          key: ${{ runner.os }}-npm-global-${{ hashFiles('**/package-lock.json') }}
+          restore-keys: |
+            ${{ runner.os }}-npm-global-
+
+      - name: Cache frontend node_modules
+        uses: actions/cache@v4
+        with:
+          path: crates/kiko-frontend/node_modules
+          key: ${{ runner.os }}-frontend-node-modules-tailwindcss
+          restore-keys: |
+            ${{ runner.os }}-frontend-node-modules-
+
+      - name: Install Node.js dependencies
+        run: |
+          # Install tailwindcss globally and create local node_modules in frontend
+          npm install -g tailwindcss
+          cd crates/kiko-frontend
+          mkdir -p node_modules
+          npm install tailwindcss --no-save
+          cd ../../tests/e2e && npm install
+
+      - name: Cache Playwright Browsers
+        uses: actions/cache@v4
+        with:
+          path: ~/.cache/ms-playwright
+          key: ${{ runner.os }}-playwright-${{ hashFiles('tests/e2e/package-lock.json') }}
+          restore-keys: |
+            ${{ runner.os }}-playwright-
+
+      - name: Install Playwright Browsers
+        run: cd tests/e2e && npx playwright install --with-deps
+
+      - name: Build Rust project
+        run: |
+          cargo build --release
+          cd crates/kiko-frontend && RUSTFLAGS="-C target-feature=-nontrapping-fptoint" trunk build --release
+
+      - name: Run E2E tests
+        run: cd tests/e2e && npx playwright test --project="${{ matrix.project }}"
+        env:
+          CI: true
+
+      - name: Upload Playwright Report
+        uses: actions/upload-artifact@v4
+        if: failure()
+        with:
+          name: playwright-report
+          path: tests/e2e/playwright-report/
+          retention-days: 30
+
+      - name: Upload test screenshots
+        uses: actions/upload-artifact@v4
+        if: failure()
+        with:
+          name: test-screenshots
+          path: tests/e2e/test-screenshots/
+          retention-days: 30

crates/kiko-backend/benches/pubsub.rs

@@ -6,13 +6,13 @@ use kiko::{data::SessionMessage, id::SessionId};
 use kiko_backend::messaging::PubSub;
 
 fn bench_pubsub_subscribe(c: &mut Criterion) {
-    let pubsub = Arc::new(PubSub::new());
+    let pubsub = Arc::new(PubSub::<SessionId, SessionMessage>::new());
 
     c.bench_function("pubsub_subscribe_single", |b| {
         b.to_async(tokio::runtime::Runtime::new().unwrap())
             .iter(|| async {
                 let session_id = SessionId::new();
-                black_box(pubsub.subscribe(session_id).await)
+                black_box(pubsub.subscribe(session_id))
             });
     });
 
@@ -30,7 +30,7 @@ fn bench_pubsub_subscribe(c: &mut Criterion) {
                                 let pubsub = pubsub.clone();
                                 async move {
                                     let session_id = SessionId::new();
-                                    pubsub.subscribe(session_id).await
+                                    pubsub.subscribe(session_id)
                                 }
                             })
                             .collect();
@@ -44,7 +44,7 @@ fn bench_pubsub_subscribe(c: &mut Criterion) {
 }
 
 fn bench_pubsub_publish(c: &mut Criterion) {
-    let pubsub = Arc::new(PubSub::new());
+    let pubsub = Arc::new(PubSub::<SessionId, SessionMessage>::new());
 
     let session_id = SessionId::new();
     let message = SessionMessage::CreateSession(kiko::data::CreateSession {
@@ -55,13 +55,13 @@ fn bench_pubsub_publish(c: &mut Criterion) {
     // Setup subscription first
     let rt = tokio::runtime::Runtime::new().unwrap();
     rt.block_on(async {
-        let _notifier = pubsub.subscribe(session_id.clone()).await;
+        let _notifier = pubsub.subscribe(session_id.clone());
     });
 
     c.bench_function("pubsub_publish_single", |b| {
         b.to_async(tokio::runtime::Runtime::new().unwrap())
             .iter(|| async {
-                pubsub.publish(session_id.clone(), message.clone()).await;
+                pubsub.publish(session_id.clone(), message.clone());
             });
     });
 
@@ -75,7 +75,7 @@ fn bench_pubsub_publish(c: &mut Criterion) {
                 b.to_async(tokio::runtime::Runtime::new().unwrap())
                     .iter(|| async {
                         for _ in 0..message_count {
-                            pubsub.publish(session_id.clone(), message.clone()).await;
+                            pubsub.publish(session_id.clone(), message.clone());
                         }
                     });
             },
@@ -85,7 +85,7 @@ fn bench_pubsub_publish(c: &mut Criterion) {
 }
 
 fn bench_pubsub_consume_events(c: &mut Criterion) {
-    let pubsub = Arc::new(PubSub::new());
+    let pubsub = Arc::new(PubSub::<SessionId, SessionMessage>::new());
 
     let session_id = SessionId::new();
     let message = SessionMessage::CreateSession(kiko::data::CreateSession {
@@ -96,13 +96,13 @@ fn bench_pubsub_consume_events(c: &mut Criterion) {
     // Setup subscription first
     let rt = tokio::runtime::Runtime::new().unwrap();
     rt.block_on(async {
-        let _notifier = pubsub.subscribe(session_id.clone()).await;
-        pubsub.publish(session_id.clone(), message).await;
+        let _notifier = pubsub.subscribe(session_id.clone());
+        pubsub.publish(session_id.clone(), message);
     });
 
     c.bench_function("pubsub_get_event", |b| {
         b.to_async(tokio::runtime::Runtime::new().unwrap())
-            .iter(|| async { black_box(pubsub.get_event(&session_id).await) });
+            .iter(|| async { black_box(pubsub.get_event(&session_id)) });
     });
 
     let mut group = c.benchmark_group("pubsub_consume_event_cycle");
@@ -119,8 +119,8 @@ fn bench_pubsub_consume_events(c: &mut Criterion) {
                                 name: format!("Benchmark Session {i}"),
                                 duration: Duration::from_secs(3600),
                             });
-                            pubsub.publish(session_id.clone(), msg).await;
-                            black_box(pubsub.consume_event(&session_id).await);
+                            pubsub.publish(session_id.clone(), msg);
+                            black_box(pubsub.consume_event(&session_id));
                         }
                     });
             },
@@ -130,7 +130,7 @@ fn bench_pubsub_consume_events(c: &mut Criterion) {
 }
 
 fn bench_pubsub_concurrent_operations(c: &mut Criterion) {
-    let pubsub = Arc::new(PubSub::new());
+    let pubsub = Arc::new(PubSub::<SessionId, SessionMessage>::new());
 
     let mut group = c.benchmark_group("pubsub_concurrent_pub_sub");
     for session_count in [10, 50, 100].iter() {
@@ -146,16 +146,16 @@ fn bench_pubsub_concurrent_operations(c: &mut Criterion) {
                                 let pubsub = pubsub.clone();
                                 async move {
                                     let session_id = SessionId::new();
-                                    let _notifier = pubsub.subscribe(session_id.clone()).await;
+                                    let _notifier = pubsub.subscribe(session_id.clone());
 
                                     let message =
                                         SessionMessage::CreateSession(kiko::data::CreateSession {
                                             name: format!("Concurrent Session {i}"),
                                             duration: Duration::from_secs(3600),
                                         });
 
-                                    pubsub.publish(session_id.clone(), message).await;
-                                    pubsub.consume_event(&session_id).await
+                                    pubsub.publish(session_id.clone(), message);
+                                    pubsub.consume_event(&session_id)
                                 }
                             })
                             .collect();
@@ -169,24 +169,24 @@ fn bench_pubsub_concurrent_operations(c: &mut Criterion) {
 }
 
 fn bench_pubsub_memory_efficiency(c: &mut Criterion) {
-    let pubsub = Arc::new(PubSub::new());
+    let pubsub = Arc::new(PubSub::<SessionId, SessionMessage>::new());
 
     c.bench_function("pubsub_session_cleanup", |b| {
         b.to_async(tokio::runtime::Runtime::new().unwrap())
             .iter(|| async {
                 let session_ids: Vec<SessionId> = (0..100).map(|_| SessionId::new()).collect();
 
                 for session_id in &session_ids {
-                    let _notifier = pubsub.subscribe(session_id.clone()).await;
+                    let _notifier = pubsub.subscribe(session_id.clone());
                     let message = SessionMessage::CreateSession(kiko::data::CreateSession {
                         name: "Cleanup Test Session".to_string(),
                         duration: Duration::from_secs(3600),
                     });
-                    pubsub.publish(session_id.clone(), message).await;
+                    pubsub.publish(session_id.clone(), message);
                 }
 
                 for session_id in &session_ids {
-                    pubsub.cleanup_session(session_id).await;
+                    pubsub.cleanup_session(session_id);
                 }
             });
     });