From f92212b725217313ce014c75f7d381bce5784e72 Mon Sep 17 00:00:00 2001 From: Francis Tsui Date: Thu, 12 Feb 2026 16:13:56 -0800 Subject: [PATCH] Add is_timeout param to ReadFrame and callback. PiperOrigin-RevId: 869430080 --- sharing/incoming_frames_reader.cc | 55 ++++---- sharing/incoming_frames_reader.h | 33 +++-- sharing/incoming_frames_reader_test.cc | 124 +++++++++++------- sharing/incoming_share_session.cc | 11 +- sharing/incoming_share_session.h | 3 +- sharing/incoming_share_session_test.cc | 80 ++++++----- sharing/nearby_connection_impl_test.cc | 12 +- sharing/nearby_sharing_service_impl.cc | 10 +- sharing/nearby_sharing_service_impl.h | 2 +- sharing/outgoing_share_session.cc | 9 +- sharing/outgoing_share_session.h | 3 +- sharing/outgoing_share_session_test.cc | 6 +- sharing/paired_key_verification_runner.cc | 6 +- .../paired_key_verification_runner_test.cc | 51 +++---- 14 files changed, 239 insertions(+), 166 deletions(-) diff --git a/sharing/incoming_frames_reader.cc b/sharing/incoming_frames_reader.cc index a1b9423d..73d6d49e 100644 --- a/sharing/incoming_frames_reader.cc +++ b/sharing/incoming_frames_reader.cc @@ -56,30 +56,31 @@ std::unique_ptr DecodeV1Frame(const std::vector& 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)> callback) { - ProcessReadRequest(std::nullopt, std::move(callback), absl::ZeroDuration()); + std::function)> callback, + absl::Duration timeout) { + ProcessReadRequest(std::nullopt, std::move(callback), timeout); } void IncomingFramesReader::ReadFrame( - FrameType frame_type, std::function)> callback, + FrameType frame_type, + std::function)> callback, absl::Duration timeout) { ProcessReadRequest(frame_type, std::move(callback), timeout); } void IncomingFramesReader::ProcessReadRequest( std::optional frame_type, - std::function)> callback, + std::function)> callback, absl::Duration timeout) { std::unique_ptr 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( - 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( + 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 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 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 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); } } diff --git a/sharing/incoming_frames_reader.h b/sharing/incoming_frames_reader.h index 41e10640..cd3ba339 100644 --- a/sharing/incoming_frames_reader.h +++ b/sharing/incoming_frames_reader.h @@ -25,8 +25,8 @@ #include #include "absl/base/thread_annotations.h" -#include "absl/time/time.h" #include "absl/synchronization/mutex.h" +#include "absl/time/time.h" #include "internal/platform/task_runner.h" #include "sharing/nearby_connection.h" #include "sharing/proto/wire_format.pb.h" @@ -45,27 +45,34 @@ class IncomingFramesReader IncomingFramesReader(const IncomingFramesReader&) = delete; IncomingFramesReader& operator=(IncomingFramesReader&) = delete; - // Reads an incoming frame from |connection|. |callback| is called + // Reads an incoming frame from connection. `callback` is called // with the frame read from connection or nullopt if connection socket is - // closed. + // closed or timeout has occurred. If timeout has occurred, the `is_timeout` + // parameter will be true. Set `timeout` to absl::ZeroDuration() to disable + // timeout. // - // Note: Callers are expected wait for |callback| to be run before scheduling + // Note: Callers are expected wait for `callback` to be run before scheduling // subsequent calls to ReadFrame(..). virtual void ReadFrame( std::function< - void(std::optional)> - callback) ABSL_LOCKS_EXCLUDED(mutex_); + void(bool is_timeout, + std::optional)> + callback, + absl::Duration timeout) ABSL_LOCKS_EXCLUDED(mutex_); - // Reads a frame of type |frame_type| from |connection|. |callback| is called + // Reads a frame of type `frame_type` from `connection`. `callback` is called // with the frame read from connection or nullopt if connection socket is - // closed or |timeout| units of time have passed. + // closed or `timeout` units of time have passed. If timeout has occurred, + // the `is_timeout` parameter will be true. Set `timeout` to + // absl::ZeroDuration() to disable timeout. // // Note: Callers are expected wait for |callback| to be run before scheduling // subsequent calls to ReadFrame(..). virtual void ReadFrame( nearby::sharing::service::proto::V1Frame::FrameType frame_type, std::function< - void(std::optional)> + void(bool is_timeout, + std::optional)> callback, absl::Duration timeout) ABSL_LOCKS_EXCLUDED(mutex_); @@ -77,7 +84,8 @@ class IncomingFramesReader struct ReadFrameInfo { std::optional frame_type = std::nullopt; - std::function)> + std::function)> callback = nullptr; absl::Duration timeout = absl::ZeroDuration(); }; @@ -86,10 +94,11 @@ class IncomingFramesReader std::optional frame_type, std::function< - void(std::optional)> + void(bool is_timeout, + std::optional)> callback, absl::Duration timeout) ABSL_LOCKS_EXCLUDED(mutex_); - void CloseAllPendingReads() ABSL_LOCKS_EXCLUDED(mutex_); + void CloseAllPendingReads(bool is_timeout) ABSL_LOCKS_EXCLUDED(mutex_); void ReadNextFrame() ABSL_LOCKS_EXCLUDED(mutex_); void OnDataReadFromConnection(const std::vector& bytes) ABSL_LOCKS_EXCLUDED(mutex_); diff --git a/sharing/incoming_frames_reader_test.cc b/sharing/incoming_frames_reader_test.cc index d0f78654..34a23314 100644 --- a/sharing/incoming_frames_reader_test.cc +++ b/sharing/incoming_frames_reader_test.cc @@ -117,22 +117,16 @@ class IncomingFramesReaderTest : public testing::Test { IncomingFramesReader* frames_reader() { return frames_reader_.get(); } - void FastForward(absl::Duration delta) { - fake_clock_.FastForward(delta); - } + void FastForward(absl::Duration delta) { fake_clock_.FastForward(delta); } - void Sync() { - EXPECT_TRUE(fake_task_runner_.SyncWithTimeout(kTimeout)); - } + void Sync() { EXPECT_TRUE(fake_task_runner_.SyncWithTimeout(kTimeout)); } void ReleaseFrameReader() { frames_reader_.reset(); } - void CloseConnection() { - nearby_connection_ = nullptr; - } + void CloseConnection() { nearby_connection_ = nullptr; } private: FakeClock fake_clock_; - FakeTaskRunner fake_task_runner_ {&fake_clock_, 1}; + FakeTaskRunner fake_task_runner_{&fake_clock_, 1}; FakeDeviceInfo fake_device_info_; std::unique_ptr nearby_connection_; std::shared_ptr frames_reader_ = nullptr; @@ -142,7 +136,8 @@ TEST_F(IncomingFramesReaderTest, ReadTimedOut) { absl::Notification notification; frames_reader()->ReadFrame( service::proto::V1Frame::INTRODUCTION, - [&](std::optional frame) { + [&](bool is_timeout, std::optional frame) { + EXPECT_TRUE(is_timeout); EXPECT_EQ(frame, std::nullopt); notification.Notify(); }, @@ -168,10 +163,13 @@ TEST_F(IncomingFramesReaderTest, ReadNonV1FrameSkipped) { connection().WriteMessage(*introduction_frame); absl::Notification notification; - frames_reader()->ReadFrame([&](std::optional frame) { - EXPECT_EQ(frame->type(), service::proto::V1Frame::INTRODUCTION); - notification.Notify(); - }); + frames_reader()->ReadFrame( + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); + EXPECT_EQ(frame->type(), service::proto::V1Frame::INTRODUCTION); + notification.Notify(); + }, + absl::ZeroDuration()); EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTimeout)); } @@ -182,10 +180,13 @@ TEST_F(IncomingFramesReaderTest, ReadAnyFrameSuccessful) { connection().WriteMessage(*introduction_frame); absl::Notification notification; - frames_reader()->ReadFrame([&](std::optional frame) { - EXPECT_EQ(frame->type(), service::proto::V1Frame::INTRODUCTION); - notification.Notify(); - }); + frames_reader()->ReadFrame( + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); + EXPECT_EQ(frame->type(), service::proto::V1Frame::INTRODUCTION); + notification.Notify(); + }, + absl::ZeroDuration()); EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTimeout)); } @@ -198,7 +199,8 @@ TEST_F(IncomingFramesReaderTest, ReadSuccessful) { absl::Notification notification; frames_reader()->ReadFrame( service::proto::V1Frame::INTRODUCTION, - [&](std::optional frame) { + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); EXPECT_EQ(frame->type(), service::proto::V1Frame::INTRODUCTION); notification.Notify(); }, @@ -219,7 +221,8 @@ TEST_F(IncomingFramesReaderTest, ReadSuccessful_JumbledFramesOrdering) { absl::Notification notification; frames_reader()->ReadFrame( service::proto::V1Frame::INTRODUCTION, - [&](std::optional frame) { + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); EXPECT_EQ(frame->type(), service::proto::V1Frame::INTRODUCTION); notification.Notify(); }, @@ -243,7 +246,8 @@ TEST_F(IncomingFramesReaderTest, JumbledFramesOrdering_ReadFromCache) { absl::Notification notification; frames_reader()->ReadFrame( service::proto::V1Frame::INTRODUCTION, - [&](std::optional frame) { + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); EXPECT_EQ(frame->type(), service::proto::V1Frame::INTRODUCTION); notification.Notify(); }, @@ -252,18 +256,24 @@ TEST_F(IncomingFramesReaderTest, JumbledFramesOrdering_ReadFromCache) { EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTimeout)); // Reading any frame should return cancel frame, then response frame. absl::Notification cancel_notification; - frames_reader()->ReadFrame([&](std::optional frame) { - ASSERT_NE(frame, std::nullopt); - EXPECT_EQ(frame->type(), service::proto::V1Frame::CANCEL); - cancel_notification.Notify(); - }); + frames_reader()->ReadFrame( + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); + ASSERT_NE(frame, std::nullopt); + EXPECT_EQ(frame->type(), service::proto::V1Frame::CANCEL); + cancel_notification.Notify(); + }, + absl::ZeroDuration()); EXPECT_TRUE(cancel_notification.WaitForNotificationWithTimeout(kTimeout)); absl::Notification response_notification; - frames_reader()->ReadFrame([&](std::optional frame) { - ASSERT_NE(frame, std::nullopt); - EXPECT_EQ(frame->type(), service::proto::V1Frame::RESPONSE); - response_notification.Notify(); - }); + frames_reader()->ReadFrame( + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); + ASSERT_NE(frame, std::nullopt); + EXPECT_EQ(frame->type(), service::proto::V1Frame::RESPONSE); + response_notification.Notify(); + }, + absl::ZeroDuration()); EXPECT_TRUE(response_notification.WaitForNotificationWithTimeout(kTimeout)); } @@ -271,7 +281,8 @@ TEST_F(IncomingFramesReaderTest, ReadAfterConnectionClosed) { absl::Notification notification; frames_reader()->ReadFrame( service::proto::V1Frame::INTRODUCTION, - [&](std::optional frame) { + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); EXPECT_EQ(frame, std::nullopt); notification.Notify(); }, @@ -285,13 +296,15 @@ TEST_F(IncomingFramesReaderTest, ReadTwoFramesWithTimeoutSuccessfully) { absl::Notification notification; frames_reader()->ReadFrame( service::proto::V1Frame::INTRODUCTION, - [&](std::optional frame) { + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); EXPECT_EQ(frame->type(), service::proto::V1Frame::INTRODUCTION); }, kTimeout); frames_reader()->ReadFrame( service::proto::V1Frame::CANCEL, - [&](std::optional frame) { + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); EXPECT_EQ(frame->type(), service::proto::V1Frame::CANCEL); notification.Notify(); }, @@ -312,13 +325,19 @@ TEST_F(IncomingFramesReaderTest, ReadTwoFramesWithTimeoutSuccessfully) { TEST_F(IncomingFramesReaderTest, ReadTwoFramesWithoutTimeoutSuccessfully) { absl::Notification notification; - frames_reader()->ReadFrame([&](std::optional frame) { - EXPECT_EQ(frame->type(), service::proto::V1Frame::INTRODUCTION); - }); - frames_reader()->ReadFrame([&](std::optional frame) { - EXPECT_EQ(frame->type(), service::proto::V1Frame::CANCEL); - notification.Notify(); - }); + frames_reader()->ReadFrame( + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); + EXPECT_EQ(frame->type(), service::proto::V1Frame::INTRODUCTION); + }, + absl::ZeroDuration()); + frames_reader()->ReadFrame( + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); + EXPECT_EQ(frame->type(), service::proto::V1Frame::CANCEL); + notification.Notify(); + }, + absl::ZeroDuration()); std::optional> introduction_frame = GetIntroductionFrame(); @@ -336,11 +355,17 @@ TEST_F(IncomingFramesReaderTest, ReadTwoFramesWithoutTimeoutSuccessfully) { TEST_F(IncomingFramesReaderTest, ReleaseFrameReaderDuringRead) { frames_reader()->ReadFrame( service::proto::V1Frame::INTRODUCTION, - [&](std::optional frame) { EXPECT_EQ(frame, std::nullopt); }, + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); + EXPECT_EQ(frame, std::nullopt); + }, kTimeout); frames_reader()->ReadFrame( service::proto::V1Frame::INTRODUCTION, - [&](std::optional frame) { EXPECT_EQ(frame, std::nullopt); }, + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); + EXPECT_EQ(frame, std::nullopt); + }, kTimeout); ReleaseFrameReader(); EXPECT_EQ(frames_reader(), nullptr); @@ -348,10 +373,13 @@ TEST_F(IncomingFramesReaderTest, ReleaseFrameReaderDuringRead) { TEST_F(IncomingFramesReaderTest, SkipInvalidFrame) { absl::Notification notification; - frames_reader()->ReadFrame([&](std::optional frame) { - EXPECT_EQ(frame->type(), service::proto::V1Frame::CANCEL); - notification.Notify(); - }); + frames_reader()->ReadFrame( + [&](bool is_timeout, std::optional frame) { + EXPECT_FALSE(is_timeout); + EXPECT_EQ(frame->type(), service::proto::V1Frame::CANCEL); + notification.Notify(); + }, + absl::ZeroDuration()); std::optional> invalid_frame = GetInvalidFrame(); ASSERT_TRUE(invalid_frame.has_value()); diff --git a/sharing/incoming_share_session.cc b/sharing/incoming_share_session.cc index ac8a1f94..9d74403d 100644 --- a/sharing/incoming_share_session.cc +++ b/sharing/incoming_share_session.cc @@ -26,6 +26,7 @@ #include "absl/container/flat_hash_map.h" #include "absl/functional/any_invocable.h" +#include "absl/time/time.h" #include "internal/base/file_path.h" #include "internal/platform/clock.h" #include "internal/platform/task_runner.h" @@ -203,8 +204,8 @@ bool IncomingShareSession::ProcessKeyVerificationResult( frames_reader()->ReadFrame( V1Frame::INTRODUCTION, - [callback = - std::move(introduction_callback)](std::optional frame) { + [callback = std::move(introduction_callback)]( + bool is_timeout, std::optional frame) { if (!frame.has_value()) { callback(std::nullopt); } else { @@ -217,7 +218,8 @@ bool IncomingShareSession::ProcessKeyVerificationResult( bool IncomingShareSession::ReadyForTransfer( std::function accept_timeout_callback, - std::function frame)> frame_read_callback) { + std::function frame)> + frame_read_callback) { if (!IsConnected()) { LOG(WARNING) << "ReadyForTransfer called when not connected"; return false; @@ -228,7 +230,8 @@ bool IncomingShareSession::ReadyForTransfer( mutual_acceptance_timeout_ = std::make_unique( service_thread(), "incoming_mutual_acceptance_timeout", kReadResponseFrameTimeout, std::move(accept_timeout_callback)); - frames_reader()->ReadFrame(std::move(frame_read_callback)); + frames_reader()->ReadFrame(std::move(frame_read_callback), + absl::ZeroDuration()); if (!self_share()) { TransferMetadataBuilder transfer_metadata_builder; diff --git a/sharing/incoming_share_session.h b/sharing/incoming_share_session.h index 0f145d99..3728d4cf 100644 --- a/sharing/incoming_share_session.h +++ b/sharing/incoming_share_session.h @@ -78,7 +78,8 @@ class IncomingShareSession : public ShareSession { bool ReadyForTransfer( std::function accept_timeout_callback, std::function< - void(std::optional frame)> + void(bool is_timeout, + std::optional frame)> frame_read_callback); // Accept the transfer and begin listening for payload transfer updates. diff --git a/sharing/incoming_share_session_test.cc b/sharing/incoming_share_session_test.cc index 596697f6..70fd119d 100644 --- a/sharing/incoming_share_session_test.cc +++ b/sharing/incoming_share_session_test.cc @@ -297,24 +297,25 @@ TEST_F(IncomingShareSessionTest, ProcessIntroductionSuccess) { TEST_F(IncomingShareSessionTest, ProcessIntroductionWithApkSuccess) { IntroductionFrame introduction_frame; - CHECK(proto2::TextFormat::ParseFromString(R"pb( - app_metadata { - app_name: "MyApp" - size: 300 - payload_id: 9876 - payload_id: 9877 - payload_id: 9878 - id: 1234 - file_name: "MyApp.apk" - file_name: "MyApp1.apk" - file_name: "MyApp2.apk" - file_size: 100 - file_size: 100 - file_size: 100 - package_name: "com.example.myapp" - } - )pb", - &introduction_frame)); + CHECK( + proto2::TextFormat::ParseFromString(R"pb( + app_metadata { + app_name: "MyApp" + size: 300 + payload_id: 9876 + payload_id: 9877 + payload_id: 9878 + id: 1234 + file_name: "MyApp.apk" + file_name: "MyApp1.apk" + file_name: "MyApp2.apk" + file_size: 100 + file_size: 100 + file_size: 100 + package_name: "com.example.myapp" + } + )pb", + &introduction_frame)); service::proto::AppMetadata app_metadata = introduction_frame.app_metadata(0); int64_t payload_id1 = app_metadata.payload_id(0); int64_t payload_id2 = app_metadata.payload_id(1); @@ -394,7 +395,8 @@ TEST_F(IncomingShareSessionTest, HasEventType(EventType::RECEIVE_ATTACHMENTS_START), Property(&SharingLog::receive_attachments_start, HasSessionId(1234))))))); - session_.ReadyForTransfer([]() {}, [](std::optional frame) {}); + session_.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}); session_.AcceptTransfer([]() {}); session_.PushPayloadTransferUpdateForTest( @@ -496,7 +498,8 @@ TEST_F(IncomingShareSessionTest, HasEventType(EventType::RECEIVE_ATTACHMENTS_START), Property(&SharingLog::receive_attachments_start, HasSessionId(1234))))))); - session_.ReadyForTransfer([]() {}, [](std::optional frame) {}); + session_.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}); session_.AcceptTransfer([]() {}); session_.PushPayloadTransferUpdateForTest( @@ -597,7 +600,8 @@ TEST_F(IncomingShareSessionTest, HasEventType(EventType::RECEIVE_ATTACHMENTS_START), Property(&SharingLog::receive_attachments_start, HasSessionId(1234))))))); - session_.ReadyForTransfer([]() {}, [](std::optional frame) {}); + session_.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}); session_.AcceptTransfer([]() {}); session_.PushPayloadTransferUpdateForTest( std::make_unique( @@ -697,7 +701,8 @@ TEST_F(IncomingShareSessionTest, HasEventType(EventType::RECEIVE_ATTACHMENTS_START), Property(&SharingLog::receive_attachments_start, HasSessionId(1234))))))); - session_.ReadyForTransfer([]() {}, [](std::optional frame) {}); + session_.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}); session_.AcceptTransfer([]() {}); session_.PushPayloadTransferUpdateForTest( @@ -798,7 +803,8 @@ TEST_F(IncomingShareSessionTest, GetPayloadFilePaths) { HasEventType(EventType::RECEIVE_ATTACHMENTS_START), Property(&SharingLog::receive_attachments_start, HasSessionId(1234))))))); - session_.ReadyForTransfer([]() {}, [](std::optional frame) {}); + session_.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}); session_.AcceptTransfer([]() {}); session_.PushPayloadTransferUpdateForTest( std::make_unique( @@ -855,7 +861,8 @@ TEST_F(IncomingShareSessionTest, PayloadTransferUpdateCompleteWithSuccess) { HasEventType(EventType::RECEIVE_ATTACHMENTS_START), Property(&SharingLog::receive_attachments_start, HasSessionId(1234))))))); - session_.ReadyForTransfer([]() {}, [](std::optional frame) {}); + session_.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}); session_.AcceptTransfer([]() {}); session_.PushPayloadTransferUpdateForTest( std::make_unique( @@ -953,7 +960,8 @@ TEST_F(IncomingShareSessionTest, PayloadTransferUpdateCancelled) { HasEventType(EventType::RECEIVE_ATTACHMENTS_START), Property(&SharingLog::receive_attachments_start, HasSessionId(1234))))))); - session_.ReadyForTransfer([]() {}, [](std::optional frame) {}); + session_.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}); session_.AcceptTransfer([]() {}); session_.PushPayloadTransferUpdateForTest( std::make_unique( @@ -997,7 +1005,8 @@ TEST_F(IncomingShareSessionTest, PayloadTransferUpdateFailed) { connections_manager_.SetIncomingPayload( wifi_payload_id2_, CreateWifiCredentialsPayload(wifi_payload_id2_, "password2", true)); - session_.ReadyForTransfer([]() {}, [](std::optional frame) {}); + session_.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}); session_.AcceptTransfer([]() {}); session_.PushPayloadTransferUpdateForTest( std::make_unique( @@ -1048,7 +1057,8 @@ TEST_F(IncomingShareSessionTest, PayloadTransferUpdateInProgress) { HasEventType(EventType::RECEIVE_ATTACHMENTS_START), Property(&SharingLog::receive_attachments_start, HasSessionId(1234))))))); - session_.ReadyForTransfer([]() {}, [](std::optional frame) {}); + session_.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}); session_.AcceptTransfer([]() {}); session_.PushPayloadTransferUpdateForTest( std::make_unique( @@ -1066,7 +1076,8 @@ TEST_F(IncomingShareSessionTest, PayloadTransferUpdateInProgress) { TEST_F(IncomingShareSessionTest, ReadyForTransferNotConnected) { EXPECT_THAT( - session_.ReadyForTransfer([]() {}, [](std::optional frame) {}), + session_.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}), IsFalse()); } @@ -1077,7 +1088,8 @@ TEST_F(IncomingShareSessionTest, ReadyForTransferNotSelfShare) { Call(_, HasStatus(TransferMetadata::Status::kAwaitingLocalConfirmation))); EXPECT_THAT( - session_.ReadyForTransfer([]() {}, [](std::optional frame) {}), + session_.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}), IsFalse()); } @@ -1096,7 +1108,8 @@ TEST_F(IncomingShareSessionTest, ReadyForTransferSelfShare) { .Times(0); EXPECT_THAT( - session.ReadyForTransfer([]() {}, [](std::optional frame) {}), + session.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}), IsTrue()); } @@ -1109,7 +1122,7 @@ TEST_F(IncomingShareSessionTest, ReadyForTransferTimeout) { EXPECT_THAT(session_.ReadyForTransfer( [&accept_timeout_called]() { accept_timeout_called = true; }, - [](std::optional frame) {}), + [](bool is_timeout, std::optional frame) {}), IsFalse()); clock_.FastForward(absl::Seconds(60)); task_runner_.SyncWithTimeout(absl::Milliseconds(100)); @@ -1142,7 +1155,7 @@ TEST_F(IncomingShareSessionTest, ReadyForTransferTimeoutCancelled) { bool accept_timeout_called = false; EXPECT_THAT(session_.ReadyForTransfer( [&accept_timeout_called]() { accept_timeout_called = true; }, - [](std::optional frame) {}), + [](bool is_timeout, std::optional frame) {}), IsFalse()); session_.AcceptTransfer([]() {}); session_.PushPayloadTransferUpdateForTest( @@ -1177,7 +1190,8 @@ TEST_F(IncomingShareSessionTest, AcceptTransferSuccess) { EXPECT_THAT(session_.ProcessIntroduction(introduction_frame_), Eq(std::nullopt)); EXPECT_THAT( - session_.ReadyForTransfer([]() {}, [](std::optional frame) {}), + session_.ReadyForTransfer( + []() {}, [](bool is_timeout, std::optional frame) {}), IsFalse()); EXPECT_CALL( transfer_metadata_callback_, diff --git a/sharing/nearby_connection_impl_test.cc b/sharing/nearby_connection_impl_test.cc index 3ea55c63..f2a70f37 100644 --- a/sharing/nearby_connection_impl_test.cc +++ b/sharing/nearby_connection_impl_test.cc @@ -41,10 +41,12 @@ TEST(NearbyConnectionImpl, DestructorBeforeReaderDestructor) { absl::Notification notification; frames_reader->ReadFrame( - [&](std::optional frame) { + [&](bool is_timeout, + std::optional frame) { called = true; notification.Notify(); - }); + }, + absl::ZeroDuration()); EXPECT_TRUE(fake_task_runner.SyncWithTimeout(absl::Seconds(1))); connection.reset(); EXPECT_TRUE(notification.WaitForNotificationWithTimeout(absl::Seconds(1))); @@ -63,10 +65,12 @@ TEST(NearbyConnectionImpl, DestructorAfterReaderDestructor) { absl::Notification notification; frames_reader->ReadFrame( - [&](std::optional frame) { + [&](bool is_timeout, + std::optional frame) { frame_result = frame; notification.Notify(); - }); + }, + absl::ZeroDuration()); EXPECT_TRUE(fake_task_runner.SyncWithTimeout(absl::Seconds(1))); frames_reader.reset(); diff --git a/sharing/nearby_sharing_service_impl.cc b/sharing/nearby_sharing_service_impl.cc index ebbcb722..7f6b69c8 100644 --- a/sharing/nearby_sharing_service_impl.cc +++ b/sharing/nearby_sharing_service_impl.cc @@ -2563,8 +2563,9 @@ void NearbySharingServiceImpl::OnReceiveConnectionResponse( } session->SendPayloads( [this, share_target_id]( + bool is_timeout, std::optional frame) { - OnFrameRead(share_target_id, std::move(frame)); + OnFrameRead(share_target_id, is_timeout, std::move(frame)); }, absl::bind_front( &NearbySharingServiceImpl::OnOutgoingPayloadTransferUpdates, this, @@ -2597,7 +2598,7 @@ void NearbySharingServiceImpl::OnStorageCheckCompleted( } void NearbySharingServiceImpl::OnFrameRead( - int64_t share_target_id, + int64_t share_target_id, bool is_timeout, std::optional frame) { if (!frame.has_value()) { // This is the case when the connection has been closed since we wait @@ -2640,9 +2641,10 @@ void NearbySharingServiceImpl::OnFrameRead( session->frames_reader()->ReadFrame( [this, share_target_id]( + bool is_timeout, std::optional frame) { - OnFrameRead(share_target_id, std::move(frame)); - }); + OnFrameRead(share_target_id, is_timeout, std::move(frame)); + }, absl::ZeroDuration()); } void NearbySharingServiceImpl::OnConnectionDisconnected( diff --git a/sharing/nearby_sharing_service_impl.h b/sharing/nearby_sharing_service_impl.h index d2fbec13..f1e7b211 100644 --- a/sharing/nearby_sharing_service_impl.h +++ b/sharing/nearby_sharing_service_impl.h @@ -320,7 +320,7 @@ class NearbySharingServiceImpl frame); void OnStorageCheckCompleted(IncomingShareSession& session); void OnFrameRead( - int64_t share_target_id, + int64_t share_target_id, bool is_timeout, std::optional frame); void OnConnectionDisconnected(int64_t share_target_id); diff --git a/sharing/outgoing_share_session.cc b/sharing/outgoing_share_session.cc index 228f8244..6d89c6f2 100644 --- a/sharing/outgoing_share_session.cc +++ b/sharing/outgoing_share_session.cc @@ -342,7 +342,8 @@ bool OutgoingShareSession::AcceptTransfer( VLOG(1) << "Waiting for response frame from " << share_target().id; frames_reader()->ReadFrame( nearby::sharing::service::proto::V1Frame::RESPONSE, - [callback = std::move(response_callback)](std::optional frame) { + [callback = std::move(response_callback)](bool is_timeout, + std::optional frame) { if (!frame.has_value()) { callback(std::nullopt); return; @@ -355,14 +356,16 @@ bool OutgoingShareSession::AcceptTransfer( void OutgoingShareSession::SendPayloads( std::function< - void(std::optional frame)> + void(bool is_tiumeout, + std::optional frame)> frame_read_callback, std::function payload_transder_update_callback) { if (!IsConnected()) { LOG(WARNING) << "SendPayloads invoked for unconnected share target"; return; } - frames_reader()->ReadFrame(std::move(frame_read_callback)); + frames_reader()->ReadFrame(std::move(frame_read_callback), + absl::ZeroDuration()); // Log analytics event of sending attachment start. analytics_recorder().NewSendAttachmentsStart( diff --git a/sharing/outgoing_share_session.h b/sharing/outgoing_share_session.h index 27b7d8e7..d4742c6c 100644 --- a/sharing/outgoing_share_session.h +++ b/sharing/outgoing_share_session.h @@ -102,7 +102,8 @@ class OutgoingShareSession : public ShareSession { // Any other frames received will be passed to `frame_read_callback`. void SendPayloads( std::function< - void(std::optional frame)> + void(bool is_timeout, + std::optional frame)> frame_read_callback, std::function payload_transder_update_callback); // Send the next payload to NearbyConnectionManager. diff --git a/sharing/outgoing_share_session_test.cc b/sharing/outgoing_share_session_test.cc index e627b12d..b8f902d6 100644 --- a/sharing/outgoing_share_session_test.cc +++ b/sharing/outgoing_share_session_test.cc @@ -673,7 +673,7 @@ TEST_F(OutgoingShareSessionTest, SendPayloads) { NearbyConnectionImpl connection(device_info_); ConnectionSuccess(&connection); - session_.SendPayloads([](std::optional frame) {}, + session_.SendPayloads([](bool is_timeout, std::optional frame) {}, payload_transder_update_callback.AsStdFunction()); auto payload_listener = session_.payload_tracker().lock(); @@ -714,7 +714,7 @@ TEST_F(OutgoingShareSessionTest, SendPayloadsSetsAdvancedProtectionFlags) { session_.SetAdvancedProtectionStatus(/*advanced_protection_enabled=*/true, /*advanced_protection_mismatch=*/true); - session_.SendPayloads([](std::optional frame) {}, + session_.SendPayloads([](bool is_timeout, std::optional frame) {}, payload_transder_update_callback.AsStdFunction()); auto payload_listener = session_.payload_tracker().lock(); @@ -753,7 +753,7 @@ TEST_F(OutgoingShareSessionTest, SendNextPayload) { NearbyConnectionImpl connection(device_info_); ConnectionSuccess(&connection); - session_.SendPayloads([](std::optional frame) {}, + session_.SendPayloads([](bool is_timeout, std::optional frame) {}, payload_transder_update_callback.AsStdFunction()); EXPECT_CALL(send_payload_callback, Call(_, _)) diff --git a/sharing/paired_key_verification_runner.cc b/sharing/paired_key_verification_runner.cc index 7952d47c..e8cd74f3 100644 --- a/sharing/paired_key_verification_runner.cc +++ b/sharing/paired_key_verification_runner.cc @@ -137,7 +137,8 @@ void PairedKeyVerificationRunner::Run( SendPairedKeyEncryptionFrame(); frames_reader_->ReadFrame( V1Frame::PAIRED_KEY_ENCRYPTION, - [&, runner = GetWeakPtr()](std::optional frame) { + [&, runner = GetWeakPtr()](bool is_timeout, + std::optional frame) { auto verification_runner = runner.lock(); if (verification_runner == nullptr) { LOG(WARNING) << "PairedKeyVerificationRunner is released before."; @@ -182,7 +183,8 @@ void PairedKeyVerificationRunner::OnReadPairedKeyEncryptionFrame( frames_reader_->ReadFrame( V1Frame::PAIRED_KEY_RESULT, - [this, runner = GetWeakPtr()](std::optional frame) { + [this, runner = GetWeakPtr()](bool is_timeout, + std::optional frame) { auto verification_runner = runner.lock(); if (verification_runner == nullptr) { LOG(WARNING) << "PairedKeyVerificationRunner is released before."; diff --git a/sharing/paired_key_verification_runner_test.cc b/sharing/paired_key_verification_runner_test.cc index 1c293832..c022e972 100644 --- a/sharing/paired_key_verification_runner_test.cc +++ b/sharing/paired_key_verification_runner_test.cc @@ -142,15 +142,18 @@ class MockIncomingFramesReader : public IncomingFramesReader { NearbyConnection* connection) : IncomingFramesReader(service_thread, connection) {} - MOCK_METHOD(void, ReadFrame, - (std::function)> callback), - (override)); + MOCK_METHOD( + void, ReadFrame, + (std::function)> callback, + absl::Duration timeout), + (override)); - MOCK_METHOD(void, ReadFrame, - (service::proto::V1Frame_FrameType frame_type, - std::function)> callback, - absl::Duration timeout), - (override)); + MOCK_METHOD( + void, ReadFrame, + (service::proto::V1Frame_FrameType frame_type, + std::function)> callback, + absl::Duration timeout), + (override)); }; PairedKeyVerificationRunner::PairedKeyVerificationResult Merge( @@ -201,9 +204,7 @@ class PairedKeyVerificationRunnerTest : public testing::Test { }); } - void SetUp() override { - GetFakeClock()->FastForward(absl::Minutes(15)); - } + void SetUp() override { GetFakeClock()->FastForward(absl::Minutes(15)); } void RunVerification( bool is_incoming, bool use_valid_public_certificate, @@ -218,7 +219,8 @@ class PairedKeyVerificationRunnerTest : public testing::Test { auto runner = std::make_shared( &fake_clock_, OSType::WINDOWS, is_incoming, visibility_history, - GetAuthToken(), [this](const Frame& frame) { + GetAuthToken(), + [this](const Frame& frame) { frames_data_.push(std::make_unique(frame)); }, std::move(public_certificate), &certificate_manager_, &frames_reader_, @@ -237,10 +239,12 @@ class PairedKeyVerificationRunnerTest : public testing::Test { EXPECT_CALL(frames_reader_, ReadFrame(testing::Eq(V1Frame::PAIRED_KEY_ENCRYPTION), testing::_, testing::Eq(kTimeout))) - .WillOnce(testing::WithArg<1>(testing::Invoke( - [frame_type](std::function)> callback) { + .WillOnce(testing::WithArg<1>( + [frame_type]( + std::function)> + callback) { if (frame_type == ReturnFrameType::kNull) { - std::move(callback)(std::nullopt); + std::move(callback)(/*is_timeout=*/false, std::nullopt); return; } @@ -286,8 +290,8 @@ class PairedKeyVerificationRunnerTest : public testing::Test { encryption_frame->clear_secret_id_hash(); } - std::move(callback)(std::move(frame)); - }))); + std::move(callback)(/*is_timeout=*/false, std::move(frame)); + })); } void SetUpPairedKeyResultFrame( @@ -297,10 +301,11 @@ class PairedKeyVerificationRunnerTest : public testing::Test { EXPECT_CALL(frames_reader_, ReadFrame(testing::Eq(V1Frame::PAIRED_KEY_RESULT), testing::_, testing::Eq(kTimeout))) - .WillOnce(testing::WithArg<1>(testing::Invoke( - [=](std::function)> callback) { + .WillOnce(testing::WithArg<1>( + [=](std::function)> + callback) { if (frame_type == ReturnFrameType::kNull) { - std::move(callback)(std::nullopt); + std::move(callback)(/*is_timeout=*/false, std::nullopt); return; } @@ -312,8 +317,8 @@ class PairedKeyVerificationRunnerTest : public testing::Test { result_frame->set_status(status); result_frame->set_os_type(os_type); - std::move(callback)(std::move(frame)); - }))); + std::move(callback)(/*is_timeout=*/false, std::move(frame)); + })); } std::unique_ptr GetWrittenFrame() { std::unique_ptr frame = std::move(frames_data_.front()); @@ -338,7 +343,7 @@ class PairedKeyVerificationRunnerTest : public testing::Test { private: FakeClock fake_clock_; - FakeTaskRunner fake_task_runner_ {&fake_clock_, 1}; + FakeTaskRunner fake_task_runner_{&fake_clock_, 1}; FakeDeviceInfo fake_device_info_; FakeNearbyConnectionsManager fake_connections_manager_; NearbyConnectionImpl connection_;