diff --git a/sharing/BUILD b/sharing/BUILD index d31ae85e..5e27dd7a 100644 --- a/sharing/BUILD +++ b/sharing/BUILD @@ -144,6 +144,7 @@ cc_library( "//internal/platform:types", "//sharing/internal/public:logging", "//sharing/proto:wire_format_cc_proto", + "@com_google_absl//absl/base:core_headers", "@com_google_absl//absl/time", "@com_google_absl//absl/types:span", ], diff --git a/sharing/incoming_frames_reader.cc b/sharing/incoming_frames_reader.cc index 68e63de1..36679ea4 100644 --- a/sharing/incoming_frames_reader.cc +++ b/sharing/incoming_frames_reader.cc @@ -54,96 +54,92 @@ IncomingFramesReader::IncomingFramesReader(TaskRunner& service_thread, } IncomingFramesReader::~IncomingFramesReader() { - MutexLock lock(&mutex_); NL_LOG(INFO) << "~IncomingFramesReader is called"; - Done(std::nullopt); + CloseAllPendingReads(); } void IncomingFramesReader::ReadFrame( std::function)> callback) { - MutexLock lock(&mutex_); - if (!read_frame_info_queue_.empty()) { + { + MutexLock lock(&mutex_); + if (!read_frame_info_queue_.empty()) { + ReadFrameInfo read_fame_info{std::nullopt, std::move(callback), + std::nullopt}; + read_frame_info_queue_.push(std::move(read_fame_info)); + return; + } + + // Check in the cache for frame. + std::optional cached_frame = GetCachedFrame(std::nullopt); + if (cached_frame.has_value()) { + callback(std::move(cached_frame)); + return; + } + ReadFrameInfo read_fame_info{std::nullopt, std::move(callback), std::nullopt}; read_frame_info_queue_.push(std::move(read_fame_info)); - return; } - - // Check in the cache for frame. - std::optional cached_frame = GetCachedFrame(std::nullopt); - if (cached_frame.has_value()) { - callback(std::move(cached_frame)); - return; - } - - ReadFrameInfo read_fame_info{std::nullopt, std::move(callback), std::nullopt}; - read_frame_info_queue_.push(std::move(read_fame_info)); ReadNextFrame(); } void IncomingFramesReader::ReadFrame( FrameType frame_type, std::function)> callback, absl::Duration timeout) { - MutexLock lock(&mutex_); - if (!read_frame_info_queue_.empty()) { + { + MutexLock lock(&mutex_); + if (!read_frame_info_queue_.empty()) { + ReadFrameInfo read_fame_info{frame_type, std::move(callback), timeout}; + read_frame_info_queue_.push(std::move(read_fame_info)); + return; + } + + // Check in the cache for frame. + std::optional cached_frame = GetCachedFrame(frame_type); + if (cached_frame.has_value()) { + callback(std::move(cached_frame)); + return; + } + ReadFrameInfo read_fame_info{frame_type, std::move(callback), timeout}; read_frame_info_queue_.push(std::move(read_fame_info)); - return; + + timeout_timer_ = std::make_unique( + service_thread_, "frame_reader_timeout", timeout, + [reader = GetWeakPtr()]() { + auto frame_reader = reader.lock(); + if (frame_reader == nullptr) { + NL_LOG(WARNING) << "IncomingFramesReader is released before."; + return; + } + frame_reader->OnTimeout(); + }); } - - // Check in the cache for frame. - std::optional cached_frame = GetCachedFrame(frame_type); - if (cached_frame.has_value()) { - callback(std::move(cached_frame)); - return; - } - - ReadFrameInfo read_fame_info{frame_type, std::move(callback), timeout}; - read_frame_info_queue_.push(std::move(read_fame_info)); - - timeout_timer_ = std::make_unique( - service_thread_, "frame_reader_timeout", timeout, - [reader = GetWeakPtr()]() { - auto frame_reader = reader.lock(); - if (frame_reader == nullptr) { - NL_LOG(WARNING) << "IncomingFramesReader is released before."; - return; - } - frame_reader->OnTimeout(); - }); - ReadNextFrame(); } void IncomingFramesReader::ReadNextFrame() { connection_->Read( - [&, reader = GetWeakPtr()](std::optional> bytes) { + [reader = GetWeakPtr()](std::optional> bytes) { auto frame_reader = reader.lock(); if (frame_reader == nullptr) { NL_LOG(WARNING) << "IncomingFramesReader is released before."; return; } - - OnDataReadFromConnection(std::move(bytes)); + frame_reader->OnDataReadFromConnection(std::move(bytes)); }); } void IncomingFramesReader::OnTimeout() { - MutexLock lock(&mutex_); NL_LOG(WARNING) << __func__ << ": Timed out reading from NearbyConnection."; - Done(std::nullopt); + CloseAllPendingReads(); } void IncomingFramesReader::OnDataReadFromConnection( std::optional> bytes) { - MutexLock lock(&mutex_); - if (read_frame_info_queue_.empty()) { - return; - } - if (!bytes.has_value()) { NL_LOG(WARNING) << __func__ << ": Failed to read frame"; - Done(std::nullopt); + CloseAllPendingReads(); return; } @@ -153,74 +149,84 @@ void IncomingFramesReader::OnDataReadFromConnection( NL_LOG(WARNING) << __func__ << ": Cannot decode frame. Not currently bound to nearby process"; - Done(std::nullopt); + CloseAllPendingReads(); return; } - OnFrameDecoded(std::move(*frame)); + const V1Frame* v1_frame = OnFrameDecoded(*frame); + if (v1_frame != nullptr) { + Done(*v1_frame); + } } -void IncomingFramesReader::OnFrameDecoded(std::optional frame) { - if (!frame.has_value()) { - ReadNextFrame(); - return; - } - - if (frame->version() != Frame::V1) { +const V1Frame* IncomingFramesReader::OnFrameDecoded(const Frame& frame) { + if (frame.version() != Frame::V1) { NL_VLOG(1) << __func__ << ": Frame read does not have V1Frame"; ReadNextFrame(); - return; + return nullptr; } - - auto v1_frame = frame->v1(); + auto v1_frame = frame.v1(); FrameType v1_frame_type = v1_frame.type(); - - const ReadFrameInfo& frame_info = read_frame_info_queue_.front(); - if (frame_info.frame_type.has_value() && - *frame_info.frame_type != v1_frame_type) { - NL_LOG(WARNING) << __func__ << ": Failed to read frame of type " - << *frame_info.frame_type << ", but got frame of type " - << v1_frame_type << ". Cached for later."; - cached_frames_.insert({v1_frame_type, std::move(v1_frame)}); - ReadNextFrame(); - return; + bool cached_frame = false; + { + MutexLock lock(&mutex_); + if (read_frame_info_queue_.empty()) { + return nullptr; + } + const ReadFrameInfo& frame_info = read_frame_info_queue_.front(); + if (frame_info.frame_type.has_value() && + *frame_info.frame_type != v1_frame_type) { + NL_LOG(WARNING) << __func__ << ": Failed to read frame of type " + << *frame_info.frame_type << ", but got frame of type " + << v1_frame_type << ". Cached for later."; + cached_frames_.insert({v1_frame_type, v1_frame}); + cached_frame = true; + } } - - Done(std::move(v1_frame)); + if (cached_frame) { + ReadNextFrame(); + return nullptr; + } + return &frame.v1(); } -void IncomingFramesReader::Done(std::optional frame) { - if (read_frame_info_queue_.empty()) { - return; +void IncomingFramesReader::CloseAllPendingReads() { + std::queue queue; + { + MutexLock lock(&mutex_); + queue.swap(read_frame_info_queue_); } + while (!queue.empty()) { + ReadFrameInfo read_frame_info = std::move(queue.front()); + queue.pop(); + read_frame_info.callback(std::nullopt); + } +} - timeout_timer_.reset(); - - bool is_empty_frame = !frame.has_value(); - ReadFrameInfo read_frame_info = std::move(read_frame_info_queue_.front()); - read_frame_info_queue_.pop(); +void IncomingFramesReader::Done(const V1Frame& frame) { + ReadFrameInfo read_frame_info; + { + MutexLock lock(&mutex_); + timeout_timer_.reset(); + read_frame_info = std::move(read_frame_info_queue_.front()); + read_frame_info_queue_.pop(); + } read_frame_info.callback(std::move(frame)); - if (is_empty_frame) { - // should complete all pending readers. - while (!read_frame_info_queue_.empty()) { - read_frame_info = std::move(read_frame_info_queue_.front()); - read_frame_info_queue_.pop(); - read_frame_info.callback(std::nullopt); + { + MutexLock lock(&mutex_); + if (read_frame_info_queue_.empty()) { + return; } - return; + read_frame_info = std::move(read_frame_info_queue_.front()); + read_frame_info_queue_.pop(); } - if (!read_frame_info_queue_.empty()) { - ReadFrameInfo read_frame_info = std::move(read_frame_info_queue_.front()); - read_frame_info_queue_.pop(); - - if (read_frame_info.timeout.has_value()) { - ReadFrame(*read_frame_info.frame_type, - std::move(read_frame_info.callback), *read_frame_info.timeout); - } else { - ReadFrame(std::move(read_frame_info.callback)); - } + if (read_frame_info.timeout.has_value()) { + ReadFrame(*read_frame_info.frame_type, + std::move(read_frame_info.callback), *read_frame_info.timeout); + } else { + ReadFrame(std::move(read_frame_info.callback)); } } diff --git a/sharing/incoming_frames_reader.h b/sharing/incoming_frames_reader.h index 05cdfc7a..f383a199 100644 --- a/sharing/incoming_frames_reader.h +++ b/sharing/incoming_frames_reader.h @@ -24,6 +24,7 @@ #include #include +#include "absl/base/thread_annotations.h" #include "absl/time/time.h" #include "internal/platform/mutex.h" #include "internal/platform/task_runner.h" @@ -55,7 +56,7 @@ class IncomingFramesReader virtual void ReadFrame( std::function< void(std::optional)> - callback); + callback) ABSL_LOCKS_EXCLUDED(mutex_); // Reads a frame of type |frame_type| from |connection|. |callback| is called // with the frame read from connection or nullopt if connection socket is @@ -68,7 +69,7 @@ class IncomingFramesReader std::function< void(std::optional)> callback, - absl::Duration timeout); + absl::Duration timeout) ABSL_LOCKS_EXCLUDED(mutex_); std::weak_ptr GetWeakPtr() { return this->weak_from_this(); @@ -83,30 +84,33 @@ class IncomingFramesReader std::optional timeout = std::nullopt; }; - void ReadNextFrame(); - void OnDataReadFromConnection(std::optional> bytes); - void OnFrameDecoded( - std::optional frame); + void CloseAllPendingReads() ABSL_LOCKS_EXCLUDED(mutex_); + void ReadNextFrame() ABSL_LOCKS_EXCLUDED(mutex_); + void OnDataReadFromConnection(std::optional> bytes) + ABSL_LOCKS_EXCLUDED(mutex_); + const nearby::sharing::service::proto::V1Frame* OnFrameDecoded( + const nearby::sharing::service::proto::Frame& frame) + ABSL_LOCKS_EXCLUDED(mutex_); void OnTimeout(); - void Done(std::optional frame); + void Done(const nearby::sharing::service::proto::V1Frame& frame) + ABSL_LOCKS_EXCLUDED(mutex_); std::optional GetCachedFrame( std::optional - frame_type); + frame_type) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); TaskRunner& service_thread_; const NearbySharingDecoder& decoder_; - NearbyConnection* connection_; + NearbyConnection* const connection_; RecursiveMutex mutex_; - std::queue read_frame_info_queue_; - std::function timeout_callback_; + std::queue read_frame_info_queue_ ABSL_GUARDED_BY(mutex_); // Caches frames read from NearbyConnection which are not used immediately. std::map> - cached_frames_; + cached_frames_ ABSL_GUARDED_BY(mutex_); - std::unique_ptr timeout_timer_; + std::unique_ptr timeout_timer_ ABSL_GUARDED_BY(mutex_); }; } // namespace sharing diff --git a/sharing/nearby_sharing_service_impl_test.cc b/sharing/nearby_sharing_service_impl_test.cc index 32b6e88b..ce929502 100644 --- a/sharing/nearby_sharing_service_impl_test.cc +++ b/sharing/nearby_sharing_service_impl_test.cc @@ -3204,6 +3204,11 @@ TEST_F(NearbySharingServiceImplTest, connection_->AppendReadableData(bytes); FlushTesting(); EXPECT_CALL(*mock_app_info_, SetActiveFlag()); + EXPECT_CALL(callback, + OnTransferUpdate( + testing::_, testing::_, + nearby::sharing::HasStatus( + TransferMetadata::Status::kPairedKeyVerificationFailed))); service_->OnIncomingConnection(kEndpointId, GetValidV1EndpointInfo(), connection_.get());