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/data_stream.cpp b/src/data_stream.cpp index d75a6c37..eb5ed11e 100644 --- a/src/data_stream.cpp +++ b/src/data_stream.cpp @@ -287,6 +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_); + if (!reply_to_id.empty()) { + info_.reply_to_stream_id = reply_to_id; + } } void TextStreamWriter::write(const std::string& text) { 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..ea133335 --- /dev/null +++ b/src/tests/integration/test_data_streams.cpp @@ -0,0 +1,226 @@ +/* + * 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: + ~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 + // 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) { + recv_thread_ = 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(); + }); + }); + } + + 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::thread recv_thread_{}; + 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: + ~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. + room.registerByteStreamHandler( + topic, [this](std::shared_ptr reader, const std::string& participant_identity) { + recv_thread_ = 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(); + }); + }); + } + + 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::thread recv_thread_{}; + 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