diff --git a/cpp/core/internal/base_pcp_handler.cc b/cpp/core/internal/base_pcp_handler.cc index dcad3554..c41176a3 100644 --- a/cpp/core/internal/base_pcp_handler.cc +++ b/cpp/core/internal/base_pcp_handler.cc @@ -54,7 +54,7 @@ BasePcpHandler::BasePcpHandler(Mediums* mediums, BasePcpHandler::~BasePcpHandler() { NEARBY_LOGS(INFO) << "BasePcpHandler: going down; strategy=" - << strategy_.GetName() << "; handle=" << handle_; + << strategy_.GetName(); DisconnectFromEndpointManager(); // Stop all the ongoing Runnables (as gracefully as possible). NEARBY_LOGS(INFO) << "BasePcpHandler: bringing down executors; strategy=" @@ -68,10 +68,10 @@ BasePcpHandler::~BasePcpHandler() { void BasePcpHandler::DisconnectFromEndpointManager() { if (stop_.Set(true)) return; NEARBY_LOGS(INFO) << "BasePcpHandler: Unregister from EPM; strategy=" - << strategy_.GetName() << "; handle=" << handle_; + << strategy_.GetName(); // Unregister ourselves from EPM message dispatcher. endpoint_manager_->UnregisterFrameProcessor(V1Frame::CONNECTION_RESPONSE, - handle_, true); + this); } Status BasePcpHandler::StartAdvertising(ClientProxy* client, @@ -284,9 +284,7 @@ void BasePcpHandler::OnEncryptionSuccessRunnable( endpoint_id.c_str()); // Set ourselves up so that we receive all acceptance/rejection messages - handle_ = endpoint_manager_->RegisterFrameProcessor( - V1Frame::CONNECTION_RESPONSE, - static_cast(this)); + endpoint_manager_->RegisterFrameProcessor(V1Frame::CONNECTION_RESPONSE, this); // Now we register our endpoint so that we can listen for both sides to // accept. diff --git a/cpp/core/internal/base_pcp_handler.h b/cpp/core/internal/base_pcp_handler.h index 9ffcebb7..9508cc1c 100644 --- a/cpp/core/internal/base_pcp_handler.h +++ b/cpp/core/internal/base_pcp_handler.h @@ -511,7 +511,6 @@ class BasePcpHandler : public PcpHandler, Prng prng_; EncryptionRunner encryption_runner_; BwuManager* bwu_manager_; - EndpointManager::FrameProcessor::Handle handle_ = nullptr; }; } // namespace connections diff --git a/cpp/core/internal/endpoint_manager.cc b/cpp/core/internal/endpoint_manager.cc index a5d04132..5a30ff35 100644 --- a/cpp/core/internal/endpoint_manager.cc +++ b/cpp/core/internal/endpoint_manager.cc @@ -269,9 +269,8 @@ EndpointManager::~EndpointManager() { NEARBY_LOG(INFO, "EndpointManager is down"); } -EndpointManager::FrameProcessor::Handle EndpointManager::RegisterFrameProcessor( +void EndpointManager::RegisterFrameProcessor( V1Frame::FrameType frame_type, EndpointManager::FrameProcessor* processor) { - const FrameProcessor::Handle handle = processor; CountDownLatch latch(1); RunOnEndpointManagerThread([this, frame_type, &latch, processor]() { auto it = frame_processors_.find(frame_type); @@ -287,40 +286,41 @@ EndpointManager::FrameProcessor::Handle EndpointManager::RegisterFrameProcessor( latch.CountDown(); }); latch.Await(); - return handle; } -void EndpointManager::UnregisterFrameProcessor(V1Frame::FrameType frame_type, - const void* handle, bool sync) { - NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [enter]: handle=" << handle; - if (handle == nullptr) return; +void EndpointManager::UnregisterFrameProcessor( + V1Frame::FrameType frame_type, + const EndpointManager::FrameProcessor* processor) { + NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [enter]: processor =" + << processor; + if (processor == nullptr) return; CountDownLatch latch(1); - RunOnEndpointManagerThread([this, frame_type, handle, &latch, sync]() { + RunOnEndpointManagerThread([this, frame_type, processor, &latch]() { auto it = frame_processors_.find(frame_type); if (it == frame_processors_.end()) { - NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [not found]: handle=" - << handle; - if (sync) latch.CountDown(); + NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [not found]: processor=" + << processor; + latch.CountDown(); return; } - NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [found]: handle=" << handle; - if (it->second == handle) { + NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [found]: processor=" + << processor; + if (it->second == processor) { frame_processors_.erase(it); NEARBY_LOGS(INFO) << "Unregistered: type=" << frame_type - << "; processor=" << handle << "; self=" << this; + << "; processor=" << processor << "; self=" << this; } else { - NEARBY_LOG(INFO, - "Failed to unregister: type=%d; handle mismatch: passed=%p, " - "expected=%p", - frame_type, handle, it->second); + NEARBY_LOG( + INFO, + "Failed to unregister: type=%d; processor mismatch: passed=%p, " + "expected=%p", + frame_type, processor, it->second); } - if (sync) latch.CountDown(); + latch.CountDown(); }); - if (sync) { - latch.Await(); - NEARBY_LOGS(INFO) << "Unregistered [sync done]: type=" << frame_type - << "; processor=" << handle << "; self=" << this; - } + latch.Await(); + NEARBY_LOGS(INFO) << "Unregistered [sync done]: type=" << frame_type + << "; processor=" << processor << "; self=" << this; } EndpointManager::FrameProcessor* EndpointManager::GetFrameProcessor( diff --git a/cpp/core/internal/endpoint_manager.h b/cpp/core/internal/endpoint_manager.h index 34bb39e3..37d37be0 100644 --- a/cpp/core/internal/endpoint_manager.h +++ b/cpp/core/internal/endpoint_manager.h @@ -60,8 +60,6 @@ class EndpointManager { public: class FrameProcessor { public: - using Handle = void*; - virtual ~FrameProcessor() = default; // @EndpointManagerReaderThread @@ -93,12 +91,11 @@ class EndpointManager { // Invoked from the constructors of the various *Manager components that make // up the OfflineServiceController implementation. // FrameProcessor* instances are of dynamic duration and survive all sessions. - // returns unique handle to be used for unregistering. // Blocks until registration is complete. - FrameProcessor::Handle RegisterFrameProcessor(V1Frame::FrameType frame_type, - FrameProcessor* processor); + void RegisterFrameProcessor(V1Frame::FrameType frame_type, + FrameProcessor* processor); void UnregisterFrameProcessor(V1Frame::FrameType frame_type, - const void* handle, bool sync = false); + const FrameProcessor* processor); // Invoked from the different PcpHandler implementations (of which there can // be only one at a time). diff --git a/cpp/core/internal/endpoint_manager_test.cc b/cpp/core/internal/endpoint_manager_test.cc index 10fbcd3a..e5e1b7aa 100644 --- a/cpp/core/internal/endpoint_manager_test.cc +++ b/cpp/core/internal/endpoint_manager_test.cc @@ -188,10 +188,9 @@ TEST_F(EndpointManagerTest, RegisterFrameProcessorWorks) { // Register frame processor, then register endpoint. // Endpoint will read one frame, then fail to read more and terminate. // On disconnection, it will notify frame processor and we verify that. - const void* handle = em_.RegisterFrameProcessor(V1Frame::CONNECTION_REQUEST, - connect_request.get()); + em_.RegisterFrameProcessor(V1Frame::CONNECTION_REQUEST, + connect_request.get()); processors_.emplace_back(std::move(connect_request)); - EXPECT_NE(handle, nullptr); RegisterEndpoint(std::move(endpoint_channel)); } @@ -206,11 +205,12 @@ TEST_F(EndpointManagerTest, UnregisterFrameProcessorWorks) { auto connect_request = std::make_unique>(); // Register frame processor and immediately unregister it. - const void* handle = em_.RegisterFrameProcessor(V1Frame::CONNECTION_REQUEST, - connect_request.get()); + em_.RegisterFrameProcessor(V1Frame::CONNECTION_REQUEST, + connect_request.get()); + em_.UnregisterFrameProcessor(V1Frame::CONNECTION_REQUEST, + connect_request.get()); + processors_.emplace_back(std::move(connect_request)); - EXPECT_NE(handle, nullptr); - em_.UnregisterFrameProcessor(V1Frame::CONNECTION_REQUEST, handle); // Endpoint will not send OnDisconnect notification to frame processor. RegisterEndpoint(std::move(endpoint_channel), false); em_.UnregisterEndpoint(&client_, endpoint_id_); diff --git a/cpp/core/internal/payload_manager.cc b/cpp/core/internal/payload_manager.cc index 61bbb497..06fd25b8 100644 --- a/cpp/core/internal/payload_manager.cc +++ b/cpp/core/internal/payload_manager.cc @@ -217,8 +217,7 @@ Payload::Id PayloadManager::CreateOutgoingPayload( PayloadManager::PayloadManager(EndpointManager& endpoint_manager) : endpoint_manager_(&endpoint_manager) { - handle_ = endpoint_manager_->RegisterFrameProcessor(V1Frame::PAYLOAD_TRANSFER, - this); + endpoint_manager_->RegisterFrameProcessor(V1Frame::PAYLOAD_TRANSFER, this); } void PayloadManager::CancelAllPayloads() { @@ -249,8 +248,7 @@ void PayloadManager::CancelAllPayloads() { void PayloadManager::DisconnectFromEndpointManager() { if (shutdown_.Set(true)) return; // Unregister ourselves from the FrameProcessors. - endpoint_manager_->UnregisterFrameProcessor(V1Frame::PAYLOAD_TRANSFER, - handle_, true); + endpoint_manager_->UnregisterFrameProcessor(V1Frame::PAYLOAD_TRANSFER, this); } PayloadManager::~PayloadManager() { diff --git a/cpp/core/internal/payload_manager.h b/cpp/core/internal/payload_manager.h index fcc1115f..bf2dffaa 100644 --- a/cpp/core/internal/payload_manager.h +++ b/cpp/core/internal/payload_manager.h @@ -281,7 +281,6 @@ class PayloadManager : public EndpointManager::FrameProcessor { void CancelAllPayloads() ABSL_LOCKS_EXCLUDED(mutex_); mutable Mutex mutex_; - EndpointManager::FrameProcessor::Handle handle_; AtomicBoolean shutdown_{false}; std::unique_ptr shutdown_barrier_; int send_payload_count_ = 0;