Add local participant identity to all outgoing data packets (#602)

Fixes #601

Author: ladvoc

.nanpa/data-packet-identity.kdl

@@ -0,0 +1 @@
+minor type="fixed" "Outgoing data packets missing local participant identity"

Sources/LiveKit/Core/Room+Engine.swift

@@ -102,6 +102,11 @@ extension Room {
             log("publisher data channel is not .open", .error)
         }
 
+        var packet = packet
+        if let identity = localParticipant.identity?.stringValue {
+            packet.participantIdentity = identity
+        }
+
         // Should return true if successful
         try publisherDataChannel.send(dataPacket: packet)
     }

Tests/LiveKitTests/Broadcast/IPCChannelTests.swift

@@ -93,6 +93,7 @@ final class IPCChannelTests: LKTestCase {
         await fulfillment(of: [cancelThrowsError], timeout: 5.0)
     }
 
+    // swiftformat:disable redundantSelf hoistAwait
     func testConnectorCancelDuringInit() async throws {
         try await assertInitCancellationThrows(
             await IPCChannel(connectingTo: self.socketPath)
@@ -105,6 +106,8 @@ final class IPCChannelTests: LKTestCase {
         )
     }
 
+    // swiftformat:enable all
+
     private struct TestHeader: Codable, Equatable {
         let someField: Int
     }

Tests/LiveKitTests/RoomTests.swift

@@ -40,6 +40,24 @@ class RoomTests: LKTestCase {
             // Nothing to do here
         }
     }
+
+    func testSendDataPacket() async throws {
+        try await withRooms([RoomTestingOptions()]) { rooms in
+            let room = rooms[0]
+
+            let expectDataPacket = self.expectation(description: "Should send data packet")
+
+            let mockDataChannel = MockDataChannelPair { packet in
+                XCTAssertEqual(packet.participantIdentity, room.localParticipant.identity?.stringValue ?? "")
+                expectDataPacket.fulfill()
+            }
+            room.publisherDataChannel = mockDataChannel
+
+            try await room.send(dataPacket: Livekit_DataPacket())
+
+            await self.fulfillment(of: [expectDataPacket], timeout: 5)
+        }
+    }
 }
 
 extension RoomTests: RoomDelegate {

Tests/LiveKitTests/RpcTests.swift

@@ -18,19 +18,6 @@
 import XCTest
 
 class RpcTests: LKTestCase {
-    // Mock DataChannelPair to intercept outgoing packets
-    class MockDataChannelPair: DataChannelPair {
-        var packetHandler: (Livekit_DataPacket) -> Void
-
-        init(packetHandler: @escaping (Livekit_DataPacket) -> Void) {
-            self.packetHandler = packetHandler
-        }
-
-        override func send(dataPacket packet: Livekit_DataPacket) throws {
-            packetHandler(packet)
-        }
-    }
-
     // Test performing RPC calls and verifying outgoing packets
     func testPerformRpc() async throws {
         try await withRooms([RoomTestingOptions()]) { rooms in

Tests/LiveKitTests/Support/MockDataChannelPair.swift

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2025 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+@testable import LiveKit
+
+/// Mock ``DataChannelPair`` to intercept outgoing packets.
+class MockDataChannelPair: DataChannelPair {
+    var packetHandler: (Livekit_DataPacket) -> Void
+
+    init(packetHandler: @escaping (Livekit_DataPacket) -> Void) {
+        self.packetHandler = packetHandler
+    }
+
+    override func send(dataPacket packet: Livekit_DataPacket) throws {
+        packetHandler(packet)
+    }
+}