Chunk text along UTF-8 boundaries

.nanpa/text-chunking.kdl

@@ -0,0 +1 @@
+minor type="added" "Data streams: chunk text along UTF-8 boundaries"

Sources/LiveKit/DataStream/Outgoing/OutgoingStreamManager.swift

@@ -171,7 +171,7 @@ actor OutgoingStreamManager: Loggable {
         log("Opened stream '\(info.id)'", .debug)
     }
 
-    private func send(_ data: Data, to id: String) async throws {
+    private func send(_ data: some StreamData, to id: String) async throws {
         for chunk in data.chunks(of: Self.chunkSize) {
             try await sendChunk(chunk, to: id)
         }
@@ -227,7 +227,7 @@ actor OutgoingStreamManager: Loggable {
             }
         }
 
-        func write(_ data: Data) async throws {
+        func write(_ data: some StreamData) async throws {
             guard let manager else { throw StreamError.terminated }
             try await manager.send(data, to: streamID)
         }

Sources/LiveKit/DataStream/Outgoing/StreamData.swift

@@ -0,0 +1,59 @@
+/*
+ * 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.
+ */
+
+import Foundation
+
+protocol StreamData {
+    func chunks(of size: Int) -> [Data]
+}
+
+extension Data: StreamData {
+    func chunks(of size: Int) -> [Data] {
+        guard size > 0, !isEmpty else { return [] }
+        return stride(from: startIndex, to: endIndex, by: size).map {
+            let end = index($0, offsetBy: size, limitedBy: endIndex) ?? endIndex
+            return self[$0 ..< end]
+        }
+    }
+}
+
+extension String: StreamData {
+    /// Chunk along valid UTF-8 bounderies.
+    ///
+    /// Uses the same algorithm as in the LiveKit JS SDK.
+    ///
+    func chunks(of size: Int) -> [Data] {
+        guard size > 0, !isEmpty else { return [] }
+
+        var chunks: [Data] = []
+        var encoded = Data(utf8)[...]
+
+        while encoded.count > size {
+            var k = size
+            while k > 0 {
+                guard encoded.indices.contains(k),
+                      encoded[k] & 0xC0 == 0x80 else { break }
+                k -= 1
+            }
+            chunks.append(encoded.subdata(in: 0 ..< k))
+            encoded = encoded.subdata(in: k ..< encoded.count)
+        }
+        if !encoded.isEmpty {
+            chunks.append(encoded)
+        }
+        return chunks
+    }
+}

Sources/LiveKit/DataStream/Outgoing/StreamWriterDestination.swift

@@ -18,6 +18,6 @@ import Foundation
 
 protocol StreamWriterDestination: Sendable {
     var isOpen: Bool { get async }
-    func write(_ data: Data) async throws
+    func write<T: StreamData>(_ data: T) async throws
     func close(reason: String?) async throws
 }

Sources/LiveKit/DataStream/Outgoing/TextStreamWriter.swift

@@ -37,7 +37,7 @@ public final class TextStreamWriter: NSObject, Sendable {
     ///   cannot be sent to remote participants.
     ///
     public func write(_ text: String) async throws {
-        try await destination.write(Data(text.utf8))
+        try await destination.write(text)
     }
 
     /// Close the stream.

Tests/LiveKitTests/DataStream/StreamDataTests.swift

@@ -0,0 +1,81 @@
+/*
+ * 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
+import XCTest
+
+class StreamDataTests: LKTestCase {
+    // MARK: - Data chunking
+
+    func testDataChunking() {
+        let testData = Data([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
+
+        let chunks = testData.chunks(of: 3)
+        XCTAssertEqual(chunks.count, 4)
+        XCTAssertEqual(chunks[0], Data([1, 2, 3]))
+        XCTAssertEqual(chunks[1], Data([4, 5, 6]))
+        XCTAssertEqual(chunks[2], Data([7, 8, 9]))
+        XCTAssertEqual(chunks[3], Data([10]))
+
+        let fullChunk = testData.chunks(of: 10)
+        XCTAssertEqual(fullChunk.count, 1)
+        XCTAssertEqual(fullChunk[0], testData)
+
+        let largeChunk = testData.chunks(of: 20)
+        XCTAssertEqual(largeChunk.count, 1)
+        XCTAssertEqual(largeChunk[0], testData)
+    }
+
+    func testEmptyDataChunking() {
+        XCTAssertTrue(Data().chunks(of: 5).isEmpty)
+    }
+
+    func testSingleByteDataChunking() {
+        let singleByteData = Data([42])
+        let chunks = singleByteData.chunks(of: 1)
+        XCTAssertEqual(chunks, [singleByteData])
+    }
+
+    func testDataInvalidChunkSize() {
+        let testData = Data([1, 2, 3, 4, 5])
+        XCTAssertTrue(testData.chunks(of: 0).isEmpty)
+        XCTAssertTrue(testData.chunks(of: -1).isEmpty)
+    }
+
+    // MARK: - String chunking
+
+    func testStringChunking() {
+        let testString = "Hello, World!"
+        let chunks = testString.chunks(of: 4)
+            .map { [UInt8]($0) }
+        XCTAssertEqual(chunks, [[72, 101, 108, 108], [111, 44, 32, 87], [111, 114, 108, 100], [33]])
+    }
+
+    func testEmptyStringChunking() {
+        XCTAssertTrue("".chunks(of: 5).isEmpty)
+    }
+
+    func testSingleCharacterStringChunking() {
+        XCTAssertEqual("X".chunks(of: 5).map { [UInt8]($0) }, [[88]])
+    }
+
+    func testMixedStringChunking() {
+        let mixedString = "Hello 👋"
+        let chunks = mixedString.chunks(of: 4)
+            .map { [UInt8]($0) }
+        XCTAssertEqual(chunks, [[0x48, 0x65, 0x6C, 0x6C], [0x6F, 0x20], [0xF0, 0x9F, 0x91, 0x8B]])
+    }
+}