TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-16 15:36:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add is_timeout param to ReadFrame and callback.

Browse Source 
PiperOrigin-RevId: 869430080
This commit is contained in:
Francis Tsui
2026-02-12 16:14:57 -08:00
committed by Copybara-Service
parent 743093a55d
commit f92212b725
14 changed files with 239 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sharing/incoming_frames_reader.cc
+28 -27
View File
@@ -56,30 +56,31 @@ std::unique_ptr<V1Frame> DecodeV1Frame(const std::vector<uint8_t>& data) {
IncomingFramesReader::IncomingFramesReader(TaskRunner& service_thread,
NearbyConnection* connection)
: service_thread_(service_thread),
connection_(connection) {
: service_thread_(service_thread), connection_(connection) {
DCHECK(connection);
}
IncomingFramesReader::~IncomingFramesReader() {
VLOG(1) << "~IncomingFramesReader is called";
CloseAllPendingReads();
CloseAllPendingReads(/*is_timeout=*/false);
}
void IncomingFramesReader::ReadFrame(
std::function<void(std::optional<V1Frame>)> callback) {
ProcessReadRequest(std::nullopt, std::move(callback), absl::ZeroDuration());
std::function<void(bool is_timeout, std::optional<V1Frame>)> callback,
absl::Duration timeout) {
ProcessReadRequest(std::nullopt, std::move(callback), timeout);
}
void IncomingFramesReader::ReadFrame(
FrameType frame_type, std::function<void(std::optional<V1Frame>)> callback,
FrameType frame_type,
std::function<void(bool is_timeout, std::optional<V1Frame>)> callback,
absl::Duration timeout) {
ProcessReadRequest(frame_type, std::move(callback), timeout);
}
void IncomingFramesReader::ProcessReadRequest(
std::optional<FrameType> frame_type,
std::function<void(std::optional<V1Frame>)> callback,
std::function<void(bool is_timeout, std::optional<V1Frame>)> callback,
absl::Duration timeout) {
std::unique_ptr<V1Frame> cached_frame;
{
@@ -95,7 +96,7 @@ void IncomingFramesReader::ProcessReadRequest(
cached_frame = PopCachedFrame(frame_type);
}
if (cached_frame) {
callback(*cached_frame);
callback(/*is_timeout=*/false, std::move(*cached_frame));
return;
}
{
@@ -105,17 +106,17 @@ void IncomingFramesReader::ProcessReadRequest(
read_frame_info_queue_.push(std::move(read_frame_info));
if (timeout != absl::ZeroDuration()) {
timeout_timer_ = std::make_unique<ThreadTimer>(
service_thread_, "frame_reader_timeout", timeout,
[reader = GetWeakPtr()]() {
auto frame_reader = reader.lock();
if (frame_reader == nullptr) {
LOG(WARNING) << "IncomingFramesReader has already been released "
"before read timeout.";
return;
}
frame_reader->OnTimeout();
});
timeout_timer_ = std::make_unique<ThreadTimer>(
service_thread_, "frame_reader_timeout", timeout,
[reader = GetWeakPtr()]() {
auto frame_reader = reader.lock();
if (frame_reader == nullptr) {
LOG(WARNING) << "IncomingFramesReader has already been released "
"before read timeout.";
return;
}
frame_reader->OnTimeout();
});
}
}
ReadNextFrame();
@@ -131,7 +132,7 @@ void IncomingFramesReader::ReadNextFrame() {
}
if (!bytes.has_value()) {
LOG(WARNING) << __func__ << ": Failed to read frame";
frame_reader->CloseAllPendingReads();
frame_reader->CloseAllPendingReads(/*is_timeout=*/false);
return;
}
frame_reader->OnDataReadFromConnection(*bytes);
@@ -140,7 +141,7 @@ void IncomingFramesReader::ReadNextFrame() {
void IncomingFramesReader::OnTimeout() {
LOG(WARNING) << __func__ << ": Timed out reading from NearbyConnection.";
CloseAllPendingReads();
CloseAllPendingReads(/*is_timeout=*/true);
}
void IncomingFramesReader::OnDataReadFromConnection(
@@ -177,7 +178,7 @@ void IncomingFramesReader::OnDataReadFromConnection(
Done(std::move(frame));
}
void IncomingFramesReader::CloseAllPendingReads() {
void IncomingFramesReader::CloseAllPendingReads(bool is_timeout) {
std::queue<ReadFrameInfo> queue;
{
absl::MutexLock lock(mutex_);
@@ -186,7 +187,7 @@ void IncomingFramesReader::CloseAllPendingReads() {
while (!queue.empty()) {
ReadFrameInfo read_frame_info = std::move(queue.front());
queue.pop();
read_frame_info.callback(std::nullopt);
read_frame_info.callback(is_timeout, std::nullopt);
}
}
@@ -198,7 +199,7 @@ void IncomingFramesReader::Done(std::unique_ptr<V1Frame> frame) {
read_frame_info = std::move(read_frame_info_queue_.front());
read_frame_info_queue_.pop();
}
read_frame_info.callback(*frame);
read_frame_info.callback(/*is_timeout=*/false, *frame);
{
absl::MutexLock lock(mutex_);
@@ -210,10 +211,10 @@ void IncomingFramesReader::Done(std::unique_ptr<V1Frame> frame) {
}
if (read_frame_info.timeout != absl::ZeroDuration()) {
ReadFrame(*read_frame_info.frame_type,
std::move(read_frame_info.callback), read_frame_info.timeout);
ReadFrame(*read_frame_info.frame_type, std::move(read_frame_info.callback),
read_frame_info.timeout);
} else {
ReadFrame(std::move(read_frame_info.callback));
ReadFrame(std::move(read_frame_info.callback), read_frame_info.timeout);
}
}