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

Merge branch 'master' into release to roll forward up to cl/358328147.

Browse Source
This commit is contained in:
hai007
2021-02-19 11:36:09 -08:00
parent 481ed7e744 dcf45ccbc6
commit cd2cc7f19a
20 changed files with 276 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core/internal/BUILD
+1
View File
@@ -185,6 +185,7 @@ cc_test(
":internal_test",
"//core:core_types",
"//core/internal/mediums",
"//core/internal/mediums:utils",
"//proto/connections:offline_wire_formats_portable_proto",
"//platform/base",
"//platform/base:test_util",
cpp/core/internal/base_endpoint_channel_test.cc
+10 -6
View File
@@ -107,9 +107,10 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a,
{
.on_success_cb =
[&latch, &context_a](
const string& endpoint_id,
const std::string& endpoint_id,
std::unique_ptr<securegcm::UKey2Handshake> ukey2,
const string& auth_token, const ByteArray& raw_auth_token) {
const std::string& auth_token,
const ByteArray& raw_auth_token) {
NEARBY_LOG(INFO, "client-A side key negotiation done");
EXPECT_TRUE(ukey2->VerifyHandshake());
auto context = ukey2->ToConnectionContext();
@@ -118,7 +119,8 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a,
latch.CountDown();
},
.on_failure_cb =
[&latch](const string& endpoint_id, EndpointChannel* channel) {
[&latch](const std::string& endpoint_id,
EndpointChannel* channel) {
NEARBY_LOG(INFO, "client-A side key negotiation failed");
latch.CountDown();
},
@@ -128,9 +130,10 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a,
{
.on_success_cb =
[&latch, &context_b](
const string& endpoint_id,
const std::string& endpoint_id,
std::unique_ptr<securegcm::UKey2Handshake> ukey2,
const string& auth_token, const ByteArray& raw_auth_token) {
const std::string& auth_token,
const ByteArray& raw_auth_token) {
NEARBY_LOG(INFO, "client-B side key negotiation done");
EXPECT_TRUE(ukey2->VerifyHandshake());
auto context = ukey2->ToConnectionContext();
@@ -139,7 +142,8 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a,
latch.CountDown();
},
.on_failure_cb =
[&latch](const string& endpoint_id, EndpointChannel* channel) {
[&latch](const std::string& endpoint_id,
EndpointChannel* channel) {
NEARBY_LOG(INFO, "client-B side key negotiation failed");
latch.CountDown();
},
cpp/core/internal/base_pcp_handler_test.cc
+29 -1
View File
@@ -29,6 +29,7 @@
#include "proto/connections/offline_wire_formats.pb.h"
#include "platform/base/byte_array.h"
#include "platform/base/exception.h"
#include "platform/base/medium_environment.h"
#include "platform/public/count_down_latch.h"
#include "platform/public/pipe.h"
#include "proto/connections_enums.pb.h"
@@ -130,7 +131,7 @@ class MockPcpHandler : public BasePcpHandler {
(override));
MOCK_METHOD(Status, StopAdvertisingImpl, (ClientProxy * client), (override));
MOCK_METHOD(StartOperationResult, StartDiscoveryImpl,
(ClientProxy * client, const string& service_id,
(ClientProxy * client, const std::string& service_id,
const ConnectionOptions& options),
(override));
MOCK_METHOD(Status, StopDiscoveryImpl, (ClientProxy * client), (override));
@@ -399,9 +400,11 @@ class BasePcpHandlerTest
.endpoint_distance_changed_cb =
mock_discovery_listener_.endpoint_distance_changed_cb.AsStdFunction(),
};
MediumEnvironment& env_ = MediumEnvironment::Instance();
};
TEST_P(BasePcpHandlerTest, ConstructorDestructorWorks) {
env_.Start();
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
@@ -409,9 +412,11 @@ TEST_P(BasePcpHandlerTest, ConstructorDestructorWorks) {
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
SUCCEED();
bwu.Shutdown();
env_.Stop();
}
TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) {
env_.Start();
ClientProxy client;
Mediums m;
EndpointChannelManager ecm;
@@ -420,9 +425,11 @@ TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) {
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartAdvertising(&client, &pcp_handler);
bwu.Shutdown();
env_.Stop();
}
TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) {
env_.Start();
ClientProxy client;
Mediums m;
EndpointChannelManager ecm;
@@ -435,9 +442,11 @@ TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) {
pcp_handler.StopAdvertising(&client);
EXPECT_FALSE(client.IsAdvertising());
bwu.Shutdown();
env_.Stop();
}
TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) {
env_.Start();
ClientProxy client;
Mediums m;
EndpointChannelManager ecm;
@@ -446,9 +455,11 @@ TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) {
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
bwu.Shutdown();
env_.Stop();
}
TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) {
env_.Start();
ClientProxy client;
Mediums m;
EndpointChannelManager ecm;
@@ -461,9 +472,11 @@ TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) {
pcp_handler.StopDiscovery(&client);
EXPECT_FALSE(client.IsDiscovering());
bwu.Shutdown();
env_.Stop();
}
TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
env_.Start();
std::string endpoint_id{"1234"};
ClientProxy client;
Mediums m;
@@ -486,9 +499,11 @@ TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
channel_b->Close();
bwu.Shutdown();
pcp_handler.DisconnectFromEndpointManager();
env_.Stop();
}
TEST_P(BasePcpHandlerTest, IoError_RequestConnectionFails) {
env_.Start();
std::string endpoint_id{"1234"};
ClientProxy client;
Mediums m;
@@ -513,9 +528,11 @@ TEST_P(BasePcpHandlerTest, IoError_RequestConnectionFails) {
channel_b->Close();
bwu.Shutdown();
pcp_handler.DisconnectFromEndpointManager();
env_.Stop();
}
TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) {
env_.Start();
std::string endpoint_id{"1234"};
ClientProxy client;
Mediums m;
@@ -542,9 +559,11 @@ TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) {
channel_b->Close();
bwu.Shutdown();
pcp_handler.DisconnectFromEndpointManager();
env_.Stop();
}
TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) {
env_.Start();
std::string endpoint_id{"1234"};
ClientProxy client;
Mediums m;
@@ -567,9 +586,11 @@ TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) {
channel_b->Close();
bwu.Shutdown();
pcp_handler.DisconnectFromEndpointManager();
env_.Stop();
}
TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
env_.Start();
std::string endpoint_id{"1234"};
ClientProxy client;
Mediums m;
@@ -602,9 +623,11 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
channel_b->Close();
bwu.Shutdown();
pcp_handler.DisconnectFromEndpointManager();
env_.Stop();
}
TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
env_.Start();
std::atomic_int destroyed_flag = 0;
int mediums_count = 0;
{
@@ -638,9 +661,11 @@ TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
bwu.Shutdown();
}
EXPECT_EQ(destroyed_flag.load(), mediums_count);
env_.Stop();
}
TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) {
env_.Start();
BooleanMediumSelector allowed = GetParam();
if (allowed.Count(true) < 2) {
// Ignore single-medium test cases, and implicit "all mediums" case.
@@ -685,12 +710,14 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) {
pcp_handler.DisconnectFromEndpointManager();
}
EXPECT_EQ(destroyed_flag.load(), mediums_count);
env_.Stop();
}
INSTANTIATE_TEST_SUITE_P(ParameterizedBasePcpHandlerTest, BasePcpHandlerTest,
::testing::ValuesIn(kTestCases));
TEST_F(BasePcpHandlerTest, InjectEndpoint) {
env_.Start();
std::string service_id{"service"};
std::string endpoint_id{"ABCD"};
ClientProxy client;
@@ -742,6 +769,7 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) {
.remote_bluetooth_mac_address = ByteArray(kFakeMacAddress),
});
bwu.Shutdown();
env_.Stop();
}
} // namespace
cpp/core/internal/bluetooth_device_name.cc
-5
View File
@@ -62,12 +62,7 @@ BluetoothDeviceName::BluetoothDeviceName(
absl::string_view bluetooth_device_name_string) {
ByteArray bluetooth_device_name_bytes =
Base64Utils::Decode(bluetooth_device_name_string);
if (bluetooth_device_name_bytes.Empty()) {
NEARBY_LOG(
INFO,
"Cannot deserialize BluetoothDeviceName: failed Base64 decoding of %s",
std::string(bluetooth_device_name_string).c_str());
return;
}
cpp/core/internal/bwu_manager.cc
+15 -9
View File
@@ -205,12 +205,18 @@ void BwuManager::OnIncomingFrame(OfflineFrame& frame,
if (parser::GetFrameType(frame) != V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION)
return;
auto bwu_frame = frame.v1().bandwidth_upgrade_negotiation();
CountDownLatch latch(1);
RunOnBwuManagerThread([this, client, endpoint_id, &bwu_frame, &latch]() {
OnBwuNegotiationFrame(client, bwu_frame, endpoint_id);
latch.CountDown();
});
latch.Await();
if (FeatureFlags::GetInstance().GetFlags().enable_async_bandwidth_upgrade) {
RunOnBwuManagerThread([this, client, endpoint_id, bwu_frame]() {
OnBwuNegotiationFrame(client, bwu_frame, endpoint_id);
});
} else {
CountDownLatch latch(1);
RunOnBwuManagerThread([this, client, endpoint_id, bwu_frame, &latch]() {
OnBwuNegotiationFrame(client, bwu_frame, endpoint_id);
latch.CountDown();
});
latch.Await();
}
}
void BwuManager::OnEndpointDisconnect(ClientProxy* client,
@@ -274,7 +280,7 @@ void BwuManager::Revert() {
}
void BwuManager::OnBwuNegotiationFrame(ClientProxy* client,
const BwuNegotiationFrame& frame,
const BwuNegotiationFrame frame,
const string& endpoint_id) {
NEARBY_LOG(INFO, "OnBwuNegotiationFrame for endpoint %s",
endpoint_id.c_str());
@@ -426,7 +432,7 @@ void BwuManager::ProcessBwuPathAvailableEvent(
std::shared_ptr<EndpointChannel> previous_endpoint_channel =
item.mapped();
if (previous_endpoint_channel) {
previous_endpoint_channel->Close(DisconnectionReason::IO_ERROR);
previous_endpoint_channel->Close(DisconnectionReason::UNFINISHED);
}
}
std::shared_ptr<EndpointChannel> new_channel =
@@ -435,7 +441,7 @@ void BwuManager::ProcessBwuPathAvailableEvent(
// The upgraded channel never finished upgrading, and therefore is still
// paused.
new_channel->Resume();
new_channel->Close(DisconnectionReason::IO_ERROR);
new_channel->Close(DisconnectionReason::UNFINISHED);
}
return;
cpp/core/internal/bwu_manager.h
+1 -1
View File
@@ -114,7 +114,7 @@ class BwuManager : public EndpointManager::FrameProcessor {
// Processes the BwuNegotiationFrames that come over the
// EndpointChannel on both initiator and responder side of the upgrade.
void OnBwuNegotiationFrame(ClientProxy* client,
const BwuNegotiationFrame& frame,
const BwuNegotiationFrame frame,
const string& endpoint_id);
// Called to revert any state changed by the Initiator or Responder in the
cpp/core/internal/bwu_manager_test.cc
+18
View File
@@ -20,6 +20,7 @@
#include "core/internal/endpoint_channel_manager.h"
#include "core/internal/endpoint_manager.h"
#include "core/internal/mediums/mediums.h"
#include "core/internal/mediums/utils.h"
#include "platform/public/system_clock.h"
#include "gmock/gmock.h"
#include "gtest/gtest.h"
@@ -56,6 +57,23 @@ TEST(BwuManagerTest, CanInitiateBwu) {
bwu_manager.Shutdown();
}
TEST(BwuManagerTest, CanProcessPathAvailableFrame) {
ClientProxy client;
std::string endpoint_id("EP_A");
Mediums mediums;
EndpointChannelManager ecm;
EndpointManager em{&ecm};
BwuManager bwu_manager{mediums, em, ecm, {}, {}};
LocationHint location_hint = Utils::BuildLocationHint("US");
ExceptionOr<OfflineFrame> wrapped_frame = parser::FromBytes(
parser::ForBwuWebrtcPathAvailable("my_id", location_hint));
bwu_manager.OnIncomingFrame(wrapped_frame.result(), endpoint_id, &client,
Medium::WEB_RTC);
bwu_manager.Shutdown();
}
} // namespace
} // namespace connections
} // namespace nearby
cpp/core/internal/encryption_runner_test.cc
+8 -6
View File
@@ -102,15 +102,16 @@ TEST(EncryptionRunnerTest, ReadWrite) {
&user_a.client, "endpoint_id", &user_a.channel,
{
.on_success_cb =
[&response](const string& endpoint_id,
[&response](const std::string& endpoint_id,
std::unique_ptr<securegcm::UKey2Handshake> ukey2,
const string& auth_token,
const std::string& auth_token,
const ByteArray& raw_auth_token) {
response.server_status = Response::Status::kDone;
response.latch.CountDown();
},
.on_failure_cb =
[&response](const string& endpoint_id, EndpointChannel* channel) {
[&response](const std::string& endpoint_id,
EndpointChannel* channel) {
response.server_status = Response::Status::kFailed;
response.latch.CountDown();
},
@@ -119,15 +120,16 @@ TEST(EncryptionRunnerTest, ReadWrite) {
&user_b.client, "endpoint_id", &user_b.channel,
{
.on_success_cb =
[&response](const string& endpoint_id,
[&response](const std::string& endpoint_id,
std::unique_ptr<securegcm::UKey2Handshake> ukey2,
const string& auth_token,
const std::string& auth_token,
const ByteArray& raw_auth_token) {
response.client_status = Response::Status::kDone;
response.latch.CountDown();
},
.on_failure_cb =
[&response](const string& endpoint_id, EndpointChannel* channel) {
[&response](const std::string& endpoint_id,
EndpointChannel* channel) {
response.client_status = Response::Status::kFailed;
response.latch.CountDown();
},
cpp/core/internal/endpoint_manager.cc
+25 -11
View File
@@ -20,10 +20,10 @@
#include "core/internal/endpoint_channel.h"
#include "core/internal/offline_frames.h"
#include "platform/base/exception.h"
#include "platform/base/feature_flags.h"
#include "platform/public/count_down_latch.h"
#include "platform/public/logging.h"
#include "proto/connections_enums.pb.h"
namespace location {
namespace nearby {
namespace connections {
@@ -301,21 +301,19 @@ void EndpointManager::UnregisterFrameProcessor(V1Frame::FrameType frame_type,
EndpointManager::FrameProcessor* EndpointManager::GetFrameProcessor(
V1Frame::FrameType frame_type) {
EndpointManager::FrameProcessor* processor = nullptr;
CountDownLatch latch(1);
RunOnEndpointManagerThread([this, frame_type, &processor, &latch]() {
auto it = frame_processors_.find(frame_type);
if (it != frame_processors_.end()) {
processor = it->second;
}
latch.CountDown();
});
latch.Await();
auto it = frame_processors_.find(frame_type);
if (it != frame_processors_.end()) {
processor = it->second;
}
return processor;
}
void EndpointManager::EnsureWorkersTerminated(const std::string& endpoint_id) {
NEARBY_LOG(ERROR, "EnsureWorkersTerminated for endpoint %s",
endpoint_id.c_str());
auto item = endpoints_.find(endpoint_id);
if (item != endpoints_.end()) {
NEARBY_LOGS(INFO) << "EndpointState found for id: " << endpoint_id;
// If another instance of data and keep-alive handlers is running, it will
// terminate soon; we should block until it happens.
EndpointState& endpoint_state = item->second;
@@ -346,13 +344,21 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client,
RunOnEndpointManagerThread([this, client, channel = channel.release(),
&endpoint_id, &info, &options, &listener,
&latch]() {
if (endpoints_.contains(endpoint_id)) {
NEARBY_LOG(WARNING, "Registing duplicate endpoint %s",
endpoint_id.c_str());
if (!FeatureFlags::GetInstance()
.GetFlags()
.endpoint_manager_ensure_workers_terminated_inside_remove) {
EnsureWorkersTerminated(endpoint_id);
}
}
// Pass ownership of channel to EndpointChannelManager
NEARBY_LOG(INFO, "Registering endpoint with channel manager: id=%s",
endpoint_id.c_str());
channel_manager_->RegisterChannelForEndpoint(
client, endpoint_id, std::unique_ptr<EndpointChannel>(channel));
EnsureWorkersTerminated(endpoint_id);
EndpointState& endpoint_state =
endpoints_.emplace(endpoint_id, EndpointState()).first->second;
endpoint_state.client = client;
@@ -406,6 +412,7 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client,
void EndpointManager::UnregisterEndpoint(ClientProxy* client,
const std::string& endpoint_id) {
NEARBY_LOG(ERROR, "UnregisterEndpoint for endpoint %s", endpoint_id.c_str());
CountDownLatch latch(1);
RunOnEndpointManagerThread([this, client, endpoint_id, &latch]() {
RemoveEndpoint(client, endpoint_id,
@@ -442,6 +449,7 @@ std::vector<std::string> EndpointManager::SendPayloadChunk(
// allow synchronous behavior here it will cause a live lock.
void EndpointManager::DiscardEndpoint(ClientProxy* client,
const std::string& endpoint_id) {
NEARBY_LOG(ERROR, "DiscardEndpoint for endpoint %s", endpoint_id.c_str());
RunOnEndpointManagerThread([this, client, endpoint_id]() {
RemoveEndpoint(client, endpoint_id,
/*notify=*/
@@ -464,6 +472,7 @@ std::vector<std::string> EndpointManager::SendControlMessage(
void EndpointManager::RemoveEndpoint(ClientProxy* client,
const std::string& endpoint_id,
bool notify) {
NEARBY_LOG(ERROR, "RemoveEndpoint for endpoint %s", endpoint_id.c_str());
// Unregistering from channel_manager_ will also serve to terminate
// the dedicated handler and KeepAlive threads we started when we registered
// this endpoint.
@@ -478,6 +487,11 @@ void EndpointManager::RemoveEndpoint(ClientProxy* client,
client->OnDisconnected(endpoint_id, notify);
NEARBY_LOG(INFO, "Removed endpoint; id=%s", endpoint_id.c_str());
}
if (FeatureFlags::GetInstance()
.GetFlags()
.endpoint_manager_ensure_workers_terminated_inside_remove) {
EnsureWorkersTerminated(endpoint_id);
}
}
// @EndpointManagerThread
cpp/core/internal/mediums/webrtc/connection_flow.cc
+4
View File
@@ -343,11 +343,15 @@ bool ConnectionFlow::TransitionState(State current_state, State new_state) {
}
bool ConnectionFlow::CloseLocked() {
NEARBY_LOG(INFO, "Closing WebRTC connection.");
if (state_ == State::kEnded) {
return false;
}
state_ = State::kEnded;
single_threaded_signaling_offloader_.Shutdown();
peer_connection_observer_.Shutdown();
if (peer_connection_) peer_connection_->Close();
data_channel_observer_.reset();
cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc
+19 -4
View File
@@ -28,6 +28,10 @@ PeerConnectionObserverImpl::PeerConnectionObserverImpl(
: connection_flow_(connection_flow),
local_ice_candidate_listener_(std::move(local_ice_candidate_listener)) {}
PeerConnectionObserverImpl::~PeerConnectionObserverImpl() {
Shutdown();
}
void PeerConnectionObserverImpl::OnIceCandidate(
const webrtc::IceCandidateInterface* candidate) {
local_ice_candidate_listener_.local_ice_candidate_found_cb(candidate);
@@ -38,8 +42,10 @@ void PeerConnectionObserverImpl::OnSignalingChange(
NEARBY_LOG(INFO, "OnSignalingChange: %d", new_state);
OffloadFromSignalingThread([this, new_state]() {
if (new_state == webrtc::PeerConnectionInterface::SignalingState::kStable)
if (new_state == webrtc::PeerConnectionInterface::SignalingState::kStable &&
connection_flow_) {
connection_flow_->OnSignalingStable();
}
});
}
@@ -47,8 +53,10 @@ void PeerConnectionObserverImpl::OnDataChannel(
rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) {
NEARBY_LOG(INFO, "OnDataChannel");
data_channel->RegisterObserver(
connection_flow_->CreateDataChannelObserver(data_channel));
if (connection_flow_) {
data_channel->RegisterObserver(
connection_flow_->CreateDataChannelObserver(data_channel));
}
}
void PeerConnectionObserverImpl::OnIceGatheringChange(
@@ -61,7 +69,9 @@ void PeerConnectionObserverImpl::OnConnectionChange(
NEARBY_LOG(INFO, "OnConnectionChange: %d", new_state);
OffloadFromSignalingThread([this, new_state]() {
connection_flow_->ProcessOnPeerConnectionChange(new_state);
if (connection_flow_) {
connection_flow_->ProcessOnPeerConnectionChange(new_state);
}
});
}
@@ -69,6 +79,11 @@ void PeerConnectionObserverImpl ::OnRenegotiationNeeded() {
NEARBY_LOG(INFO, "OnRenegotiationNeeded");
}
void PeerConnectionObserverImpl::Shutdown() {
single_threaded_signaling_offloader_.Shutdown();
connection_flow_ = nullptr;
}
void PeerConnectionObserverImpl::OffloadFromSignalingThread(Runnable runnable) {
single_threaded_signaling_offloader_.Execute(std::move(runnable));
}
cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h
+4 -2
View File
@@ -28,10 +28,10 @@ class ConnectionFlow;
class PeerConnectionObserverImpl : public webrtc::PeerConnectionObserver {
public:
~PeerConnectionObserverImpl() override = default;
PeerConnectionObserverImpl(
ConnectionFlow* connection_flow,
LocalIceCandidateListener local_ice_candidate_listener);
~PeerConnectionObserverImpl() override;
// webrtc::PeerConnectionObserver:
void OnIceCandidate(const webrtc::IceCandidateInterface* candidate) override;
@@ -45,10 +45,12 @@ class PeerConnectionObserverImpl : public webrtc::PeerConnectionObserver {
webrtc::PeerConnectionInterface::PeerConnectionState new_state) override;
void OnRenegotiationNeeded() override;
void Shutdown();
private:
void OffloadFromSignalingThread(Runnable runnable);
ConnectionFlow* connection_flow_;
ConnectionFlow* volatile connection_flow_;
LocalIceCandidateListener local_ice_candidate_listener_;
SingleThreadExecutor single_threaded_signaling_offloader_;
};
cpp/platform/base/feature_flags.h
+7
View File
@@ -35,6 +35,13 @@ class FeatureFlags {
// Ignore subsequent BWU Available events when we're still processing the
// first one.
bool disallow_out_of_order_bwu_avail_event = true;
bool enable_async_bandwidth_upgrade = true;
// Let endpoint_manager erase deleted endpoint from endpoints_ inside
// function RemoveEndpoint.
bool endpoint_manager_ensure_workers_terminated_inside_remove = true;
// If a scheduled runnable is already running, Cancel() will synchronously
// wait for the task to complete.
bool cancel_waits_for_running_tasks = true;
};
static const FeatureFlags& GetInstance() {
cpp/platform/impl/g3/bluetooth_classic.cc
+1 -2
View File
@@ -106,9 +106,8 @@ BluetoothDevice* BluetoothSocket::GetRemoteDevice() {
std::unique_ptr<api::BluetoothSocket> BluetoothServerSocket::Accept() {
absl::MutexLock lock(&mutex_);
while (pending_sockets_.empty()) {
while (!closed_ && pending_sockets_.empty()) {
cond_.Wait(&mutex_);
if (closed_) break;
}
// whether or not we were running in the wait loop, return early if closed.
if (closed_) return {};
cpp/platform/public/BUILD
+2
View File
@@ -22,6 +22,7 @@ cc_library(
"atomic_reference.h",
"cancelable.h",
"cancelable_alarm.h",
"cancellable_task.h",
"condition_variable.h",
"count_down_latch.h",
"crypto.h",
@@ -107,6 +108,7 @@ cc_library(
cc_test(
name = "public_test",
size = "small",
timeout = "moderate",
srcs = [
"atomic_boolean_test.cc",
"atomic_reference_test.cc",
cpp/platform/public/ble.cc
-3
View File
@@ -53,9 +53,6 @@ bool BleMedium::StartScanning(
auto& context = *pair.first->second;
if (pair.second) {
context.peripheral = BlePeripheral(&peripheral);
NEARBY_LOG(INFO,
"Discovered peripheral '%s'",
peripheral.GetName().c_str());
discovered_peripheral_callback_.peripheral_discovered_cb(
context.peripheral, service_id,
context.peripheral.GetAdvertisementBytes(service_id),
cpp/platform/public/cancelable.h
+11 -3
View File
@@ -19,6 +19,7 @@
#include <utility>
#include "platform/api/cancelable.h"
#include "platform/public/cancellable_task.h"
namespace location {
namespace nearby {
@@ -35,14 +36,21 @@ class Cancelable final {
// This constructor is used internally only,
// by other classes in "//platform/public/".
explicit Cancelable(std::shared_ptr<api::Cancelable> impl)
: impl_(std::move(impl)) {}
explicit Cancelable(std::shared_ptr<CancellableTask> task,
std::shared_ptr<api::Cancelable> impl)
: task_{task}, impl_(std::move(impl)) {}
bool Cancel() { return impl_ ? impl_->Cancel() : false; }
bool Cancel() {
if (!impl_) return false;
bool result = impl_->Cancel();
task_->CancelAndWaitIfStarted();
return result;
}
bool IsValid() { return impl_ != nullptr; }
private:
std::shared_ptr<CancellableTask> task_;
std::shared_ptr<api::Cancelable> impl_;
};
cpp/platform/public/cancellable_task.h
+56
View File
@@ -0,0 +1,56 @@
#ifndef PLATFORM_PUBLIC_CANCELLABLE_TASK_H_
#define PLATFORM_PUBLIC_CANCELLABLE_TASK_H_
#include <utility>
#include "platform/base/feature_flags.h"
#include "platform/base/runnable.h"
#include "platform/public/atomic_boolean.h"
#include "platform/public/future.h"
namespace location {
namespace nearby {
/**
* Runnable wrapper that allows one to wait for the task
* to complete if it is already running.
*/
class CancellableTask {
public:
explicit CancellableTask(Runnable&& runnable)
: runnable_{std::move(runnable)} {}
/**
* Try to cancel the task and wait until completion if the task is already
* running.
*/
void CancelAndWaitIfStarted() {
if (started_or_cancelled_.Set(true)) {
if (FeatureFlags::GetInstance()
.GetFlags()
.cancel_waits_for_running_tasks) {
// task could still be running, wait until finish
finished_.Get();
}
} else {
// mark as finished to support multiple calls to this method
finished_.Set(true);
}
}
void operator()() {
if (started_or_cancelled_.Set(true)) return;
runnable_();
finished_.Set(true);
}
private:
AtomicBoolean started_or_cancelled_;
Future<bool> finished_;
Runnable runnable_;
};
} // namespace nearby
} // namespace location
#endif // PLATFORM_PUBLIC_CANCELLABLE_TASK_H_
cpp/platform/public/scheduled_executor.h
+8 -2
View File
@@ -23,6 +23,7 @@
#include "platform/api/scheduled_executor.h"
#include "platform/base/runnable.h"
#include "platform/public/cancelable.h"
#include "platform/public/cancellable_task.h"
#include "platform/public/mutex.h"
#include "platform/public/mutex_lock.h"
#include "absl/time/time.h"
@@ -73,8 +74,13 @@ class ScheduledExecutor final {
Cancelable Schedule(Runnable&& runnable, absl::Duration duration)
ABSL_LOCKS_EXCLUDED(mutex_) {
MutexLock lock(&mutex_);
return impl_ ? Cancelable(impl_->Schedule(std::move(runnable), duration))
: Cancelable();
if (impl_) {
auto task = std::make_shared<CancellableTask>(std::move(runnable));
return Cancelable(task,
impl_->Schedule([task]() { (*task)(); }, duration));
} else {
return Cancelable();
}
}
private:
cpp/platform/public/scheduled_executor_test.cc
+57
View File
@@ -18,6 +18,7 @@
#include <functional>
#include "platform/base/exception.h"
#include "platform/public/count_down_latch.h"
#include "gtest/gtest.h"
#include "absl/synchronization/mutex.h"
#include "absl/time/clock.h"
@@ -95,6 +96,20 @@ TEST(ScheduledExecutorTest, CanCancel) {
EXPECT_EQ(value, 0);
}
TEST(ScheduledExecutorTest, CanCancelTwice) {
ScheduledExecutor executor;
std::atomic_int value = 0;
Cancelable cancelable =
executor.Schedule([&value]() { value += 1; }, kShortDelay);
EXPECT_EQ(value, 0);
cancelable.Cancel();
cancelable.Cancel();
absl::SleepFor(kLongDelay);
EXPECT_EQ(value, 0);
}
TEST(ScheduledExecutorTest, FailToCancel) {
absl::Mutex mutex;
absl::CondVar cond;
@@ -118,5 +133,47 @@ TEST(ScheduledExecutorTest, FailToCancel) {
EXPECT_EQ(value, 1);
}
TEST(ScheduledExecutorTest,
CancelWhileRunning_TaskCompletesBeforeCancelReturns) {
CountDownLatch start_latch(1);
ScheduledExecutor executor;
std::atomic_int value = 0;
// A task that takes a little bit of time to complete
Cancelable cancelable = executor.Schedule(
[&start_latch, &value]() {
start_latch.CountDown();
absl::SleepFor(kLongDelay);
value += 1;
},
absl::ZeroDuration());
start_latch.Await();
cancelable.Cancel();
EXPECT_EQ(value, 1);
}
TEST(ScheduledExecutorTest,
CancelTwiceWhileRunning_TaskCompletesBeforeCancelReturns) {
CountDownLatch start_latch(1);
ScheduledExecutor executor;
std::atomic_int value = 0;
// A task that takes a little bit of time to complete
Cancelable cancelable = executor.Schedule(
[&start_latch, &value]() {
start_latch.CountDown();
absl::SleepFor(kLongDelay);
value += 1;
},
absl::ZeroDuration());
start_latch.Await();
cancelable.Cancel();
cancelable.Cancel();
EXPECT_EQ(value, 1);
}
} // namespace nearby
} // namespace location