From 24db7485baf7e79f6fc6cec6b91894ce659085a9 Mon Sep 17 00:00:00 2001 From: Kohio Deflesselle Date: Wed, 8 Jul 2026 10:37:57 -1000 Subject: [PATCH 1/5] feat: Add missing reply_to_stream_id field to TextStreamInfo and forward it from proto::DataStream.text_header().reply_to_stream_id() --- include/livekit/data_stream.h | 4 +- src/room_proto_converter.cpp | 3 + src/tests/integration/test_data_streams.cpp | 214 ++++++++++++++++++++ 3 files changed, 220 insertions(+), 1 deletion(-) create mode 100644 src/tests/integration/test_data_streams.cpp diff --git a/include/livekit/data_stream.h b/include/livekit/data_stream.h index 318a29ff..92ec44e7 100644 --- a/include/livekit/data_stream.h +++ b/include/livekit/data_stream.h @@ -61,8 +61,10 @@ struct BaseStreamInfo { /// Metadata for a text stream. struct TextStreamInfo : BaseStreamInfo { - /// IDs of any attached streams (for replies / threads). + /// IDs of any attached streams (for attached files). std::vector attachments; + /// If this stream is a reply to another stream, this field holds its ID + std::optional reply_to_stream_id = std::nullopt; }; /// Metadata for a byte stream. diff --git a/src/room_proto_converter.cpp b/src/room_proto_converter.cpp index 6e2fdcd0..c7039a22 100644 --- a/src/room_proto_converter.cpp +++ b/src/room_proto_converter.cpp @@ -615,6 +615,9 @@ TextStreamInfo makeTextInfo(const proto::DataStream::Header& header) { info.mime_type = header.mime_type(); info.topic = header.topic(); info.timestamp = header.timestamp(); + if (header.text_header().has_reply_to_stream_id()) { + info.reply_to_stream_id = header.text_header().reply_to_stream_id(); + } if (header.has_total_length()) { info.size = static_cast(header.total_length()); diff --git a/src/tests/integration/test_data_streams.cpp b/src/tests/integration/test_data_streams.cpp new file mode 100644 index 00000000..80c280be --- /dev/null +++ b/src/tests/integration/test_data_streams.cpp @@ -0,0 +1,214 @@ +/* + * Copyright 2026 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. + */ + +// End-to-end coverage for text/byte data streams (registerTextStreamHandler / +// registerByteStreamHandler + TextStreamWriter / ByteStreamWriter). Requires a +// local SFU; see test_data_track.cpp for setup instructions. Run with: +// ./build-debug/bin/livekit_integration_tests --gtest_filter=*DataStream* + +#include + +#include +#include +#include +#include +#include +#include + +#include "../common/test_common.h" + +namespace livekit::test { + +using namespace std::chrono_literals; + +namespace { + +constexpr auto kStreamWaitTimeout = 10s; + +std::string makeTopic(const std::string& suffix) { + return "test_topic_" + suffix + "_" + std::to_string(getTimestampUs()); +} + +/// Waits for a single incoming text stream on `topic` and captures its info +/// plus fully-read content. Registers itself as the topic's handler. +class TextStreamCollector { +public: + void registerOn(Room& room, const std::string& topic) { + // Handlers run on the Room event thread and must not block (per + // registerTextStreamHandler's docs), since later chunk/close events for + // this same reader are dispatched on that same thread. readAll() blocks + // until close, so it has to happen on a separate thread. + room.registerTextStreamHandler( + topic, [this](std::shared_ptr reader, const std::string& participant_identity) { + std::thread([this, reader, participant_identity] { + auto info = reader->info(); + auto text = reader->readAll(); + std::lock_guard lock(mutex_); + info_ = std::move(info); + text_ = std::move(text); + sender_identity_ = participant_identity; + done_ = true; + cv_.notify_all(); + }).detach(); + }); + } + + bool wait(std::chrono::milliseconds timeout) { + std::unique_lock lock(mutex_); + return cv_.wait_for(lock, timeout, [this] { return done_; }); + } + + const TextStreamInfo& info() const { return info_; } + const std::string& text() const { return text_; } + const std::string& senderIdentity() const { return sender_identity_; } + +private: + std::mutex mutex_; + std::condition_variable cv_; + bool done_ = false; + TextStreamInfo info_; + std::string text_; + std::string sender_identity_; +}; + +/// Same idea as TextStreamCollector, for byte streams. +class ByteStreamCollector { +public: + void registerOn(Room& room, const std::string& topic) { + // See TextStreamCollector::registerOn: the blocking readNext() loop must + // not run on the Room event thread. + room.registerByteStreamHandler( + topic, [this](std::shared_ptr reader, const std::string& participant_identity) { + std::thread([this, reader, participant_identity] { + auto info = reader->info(); + std::vector content; + std::vector chunk; + while (reader->readNext(chunk)) { + content.insert(content.end(), chunk.begin(), chunk.end()); + } + std::lock_guard lock(mutex_); + info_ = std::move(info); + content_ = std::move(content); + sender_identity_ = participant_identity; + done_ = true; + cv_.notify_all(); + }).detach(); + }); + } + + bool wait(std::chrono::milliseconds timeout) { + std::unique_lock lock(mutex_); + return cv_.wait_for(lock, timeout, [this] { return done_; }); + } + + const ByteStreamInfo& info() const { return info_; } + const std::vector& content() const { return content_; } + const std::string& senderIdentity() const { return sender_identity_; } + +private: + std::mutex mutex_; + std::condition_variable cv_; + bool done_ = false; + ByteStreamInfo info_; + std::vector content_; + std::string sender_identity_; +}; + +} // namespace + +class DataStreamE2ETest : public LiveKitTestBase {}; + +TEST_F(DataStreamE2ETest, TextStreamRoundTripEndToEnd) { + const auto topic = makeTopic("text"); + + auto rooms = testRooms(2); + auto& sender_room = rooms[0]; + auto& receiver_room = rooms[1]; + const auto sender_identity = lockLocalParticipant(*sender_room)->identity(); + + TextStreamCollector collector; + collector.registerOn(*receiver_room, topic); + + { + TextStreamWriter writer(*lockLocalParticipant(*sender_room), topic); + writer.write("hello, "); + writer.write("world!"); + writer.close(); + } + + ASSERT_TRUE(collector.wait(kStreamWaitTimeout)) << "Timed out waiting for text stream"; + EXPECT_EQ(collector.text(), "hello, world!"); + EXPECT_EQ(collector.senderIdentity(), sender_identity); + EXPECT_EQ(collector.info().topic, topic); + EXPECT_EQ(collector.info().mime_type, "text/plain"); +} + +// Regression coverage for the `text_header.reply_to_stream_id` field: writing +// a text stream with a reply-to id set should surface that id on the +// receiving side's TextStreamInfo. If this fails while the C++-side +// conversion (room_proto_converter.cpp: makeTextInfo) looks correct, the gap +// is upstream in the Rust FFI layer not forwarding the field. +TEST_F(DataStreamE2ETest, TextStreamReplyToStreamIdIsRoutedEndToEnd) { + const auto topic = makeTopic("text_reply"); + const std::string reply_to_id = "original-stream-" + std::to_string(getTimestampUs()); + + auto rooms = testRooms(2); + auto& sender_room = rooms[0]; + auto& receiver_room = rooms[1]; + + TextStreamCollector collector; + collector.registerOn(*receiver_room, topic); + + { + TextStreamWriter writer(*lockLocalParticipant(*sender_room), topic, /*attributes=*/{}, /*stream_id=*/"", + /*total_size=*/std::nullopt, reply_to_id); + writer.write("reply payload"); + writer.close(); + } + + ASSERT_TRUE(collector.wait(kStreamWaitTimeout)) << "Timed out waiting for text stream"; + EXPECT_EQ(collector.text(), "reply payload"); + ASSERT_TRUE(collector.info().reply_to_stream_id.has_value()) + << "reply_to_stream_id was not routed through FFI from the Rust SDK"; + EXPECT_EQ(collector.info().reply_to_stream_id.value(), reply_to_id); +} + +TEST_F(DataStreamE2ETest, ByteStreamRoundTripEndToEnd) { + const auto topic = makeTopic("bytes"); + const std::vector payload{0x00, 0x01, 0x02, 0xFE, 0xFF, 'h', 'i'}; + + auto rooms = testRooms(2); + auto& sender_room = rooms[0]; + auto& receiver_room = rooms[1]; + const auto sender_identity = lockLocalParticipant(*sender_room)->identity(); + + ByteStreamCollector collector; + collector.registerOn(*receiver_room, topic); + + { + ByteStreamWriter writer(*lockLocalParticipant(*sender_room), /*name=*/"payload.bin", topic); + writer.write(payload); + writer.close(); + } + + ASSERT_TRUE(collector.wait(kStreamWaitTimeout)) << "Timed out waiting for byte stream"; + EXPECT_EQ(collector.content(), payload); + EXPECT_EQ(collector.senderIdentity(), sender_identity); + EXPECT_EQ(collector.info().topic, topic); + EXPECT_EQ(collector.info().name, "payload.bin"); +} + +} // namespace livekit::test From 80cab4ccc49cb7f0ddddefd4fea23c37a4f7145d Mon Sep 17 00:00:00 2001 From: Kohio Deflesselle Date: Wed, 8 Jul 2026 10:53:39 -1000 Subject: [PATCH 2/5] Fill reply_to_stream_id field in TextStreamInfo on writer side as well --- src/data_stream.cpp | 1 + 1 file changed, 1 insertion(+) diff --git a/src/data_stream.cpp b/src/data_stream.cpp index d75a6c37..708b3b07 100644 --- a/src/data_stream.cpp +++ b/src/data_stream.cpp @@ -287,6 +287,7 @@ TextStreamWriter::TextStreamWriter(LocalParticipant& local_participant, const st reply_to_id_ = reply_to_id; // ✅ Canonical user-facing metadata comes from BaseStreamWriter fields. fillBaseInfo(info_, stream_id_, mime_type_, topic_, timestamp_ms_, total_size_, attributes_); + info_.reply_to_stream_id = reply_to_id; } void TextStreamWriter::write(const std::string& text) { From 259f6a3a1178108f205c3a07203fbd830e80355b Mon Sep 17 00:00:00 2001 From: Kohio Deflesselle <90938133+Soralsei@users.noreply.github.com> Date: Wed, 8 Jul 2026 10:59:33 -1000 Subject: [PATCH 3/5] Add empty reply_to_stream_id guard in TextStreamWriter constructor Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- src/data_stream.cpp | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/data_stream.cpp b/src/data_stream.cpp index 708b3b07..eb5ed11e 100644 --- a/src/data_stream.cpp +++ b/src/data_stream.cpp @@ -287,7 +287,9 @@ TextStreamWriter::TextStreamWriter(LocalParticipant& local_participant, const st reply_to_id_ = reply_to_id; // ✅ Canonical user-facing metadata comes from BaseStreamWriter fields. fillBaseInfo(info_, stream_id_, mime_type_, topic_, timestamp_ms_, total_size_, attributes_); - info_.reply_to_stream_id = reply_to_id; + if (!reply_to_id.empty()) { + info_.reply_to_stream_id = reply_to_id; + } } void TextStreamWriter::write(const std::string& text) { From 18812a17c029bd958d4a470e5458ca0c8e5d03cc Mon Sep 17 00:00:00 2001 From: Kohio Deflesselle Date: Wed, 8 Jul 2026 11:16:07 -1000 Subject: [PATCH 4/5] Fix potential crash if detached threads outlive Text/BytesStreamCollector in data stream tests --- src/tests/integration/test_data_streams.cpp | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/src/tests/integration/test_data_streams.cpp b/src/tests/integration/test_data_streams.cpp index 80c280be..c13d2c0e 100644 --- a/src/tests/integration/test_data_streams.cpp +++ b/src/tests/integration/test_data_streams.cpp @@ -46,6 +46,7 @@ std::string makeTopic(const std::string& suffix) { /// plus fully-read content. Registers itself as the topic's handler. class TextStreamCollector { public: + ~TextStreamCollector() { recv_thread_.join(); } void registerOn(Room& room, const std::string& topic) { // Handlers run on the Room event thread and must not block (per // registerTextStreamHandler's docs), since later chunk/close events for @@ -53,7 +54,7 @@ class TextStreamCollector { // until close, so it has to happen on a separate thread. room.registerTextStreamHandler( topic, [this](std::shared_ptr reader, const std::string& participant_identity) { - std::thread([this, reader, participant_identity] { + recv_thread_ = std::thread([this, reader, participant_identity] { auto info = reader->info(); auto text = reader->readAll(); std::lock_guard lock(mutex_); @@ -62,7 +63,8 @@ class TextStreamCollector { sender_identity_ = participant_identity; done_ = true; cv_.notify_all(); - }).detach(); + }); + recv_thread_.detach(); }); } @@ -76,6 +78,7 @@ class TextStreamCollector { const std::string& senderIdentity() const { return sender_identity_; } private: + std::thread recv_thread_{}; std::mutex mutex_; std::condition_variable cv_; bool done_ = false; @@ -87,12 +90,13 @@ class TextStreamCollector { /// Same idea as TextStreamCollector, for byte streams. class ByteStreamCollector { public: + ~ByteStreamCollector() { recv_thread_.join(); } void registerOn(Room& room, const std::string& topic) { // See TextStreamCollector::registerOn: the blocking readNext() loop must // not run on the Room event thread. room.registerByteStreamHandler( topic, [this](std::shared_ptr reader, const std::string& participant_identity) { - std::thread([this, reader, participant_identity] { + recv_thread_ = std::thread([this, reader, participant_identity] { auto info = reader->info(); std::vector content; std::vector chunk; @@ -105,7 +109,8 @@ class ByteStreamCollector { sender_identity_ = participant_identity; done_ = true; cv_.notify_all(); - }).detach(); + }); + recv_thread_.detach(); }); } @@ -119,6 +124,7 @@ class ByteStreamCollector { const std::string& senderIdentity() const { return sender_identity_; } private: + std::thread recv_thread_{}; std::mutex mutex_; std::condition_variable cv_; bool done_ = false; From 93647837a3ba4a31bb7e6e5b3814913f78702b4d Mon Sep 17 00:00:00 2001 From: Kohio Deflesselle Date: Wed, 8 Jul 2026 11:23:43 -1000 Subject: [PATCH 5/5] Remove .detach() calls in Text/BytesStreamCollecton recv_thread_ + add .joinable() guard before trying to join the thread --- src/tests/integration/test_data_streams.cpp | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/src/tests/integration/test_data_streams.cpp b/src/tests/integration/test_data_streams.cpp index c13d2c0e..ea133335 100644 --- a/src/tests/integration/test_data_streams.cpp +++ b/src/tests/integration/test_data_streams.cpp @@ -46,7 +46,11 @@ std::string makeTopic(const std::string& suffix) { /// plus fully-read content. Registers itself as the topic's handler. class TextStreamCollector { public: - ~TextStreamCollector() { recv_thread_.join(); } + ~TextStreamCollector() { + if (recv_thread_.joinable()) { + recv_thread_.join(); + } + } void registerOn(Room& room, const std::string& topic) { // Handlers run on the Room event thread and must not block (per // registerTextStreamHandler's docs), since later chunk/close events for @@ -64,7 +68,6 @@ class TextStreamCollector { done_ = true; cv_.notify_all(); }); - recv_thread_.detach(); }); } @@ -90,7 +93,11 @@ class TextStreamCollector { /// Same idea as TextStreamCollector, for byte streams. class ByteStreamCollector { public: - ~ByteStreamCollector() { recv_thread_.join(); } + ~ByteStreamCollector() { + if (recv_thread_.joinable()) { + recv_thread_.join(); + } + } void registerOn(Room& room, const std::string& topic) { // See TextStreamCollector::registerOn: the blocking readNext() loop must // not run on the Room event thread. @@ -110,7 +117,6 @@ class ByteStreamCollector { done_ = true; cv_.notify_all(); }); - recv_thread_.detach(); }); }