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

Roll forward to cl/341966255

Browse Source 
Signed-off-by: hai007 <hais@google.com>
This commit is contained in:
hai007
2020-11-11 20:47:55 -08:00
parent 2526d46eaf
commit a00ffc5e0d
7 changed files with 73 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core/internal/mediums/webrtc.cc
+20 -25
View File
@@ -48,14 +48,14 @@ const std::string WebRtc::GetDefaultCountryCode() {
bool WebRtc::IsAvailable() { return medium_.IsValid(); }
bool WebRtc::IsAcceptingConnections() {
bool WebRtc::IsAcceptingConnections(const std::string& service_id) {
MutexLock lock(&mutex_);
// TODO(hais): refractor the implementation with maps.
return role_ == Role::kOfferer;
}
bool WebRtc::StartAcceptingConnections(const PeerId& self_id,
const std::string& service_id,
const std::string& local_endpoint_id,
const LocationHint& location_hint,
AcceptedConnectionCallback callback) {
if (!IsAvailable()) {
@@ -66,10 +66,17 @@ bool WebRtc::StartAcceptingConnections(const PeerId& self_id,
return false;
}
if (IsAcceptingConnections(service_id)) {
NEARBY_LOG(WARNING, "Already accepting WebRTC connections.");
return false;
}
{
MutexLock lock(&mutex_);
if (self_id_.GetId() == self_id.GetId()) {
NEARBY_LOG(WARNING, "Already accepting WebRTC connections.");
if (role_ != Role::kNone) {
NEARBY_LOG(WARNING,
"Cannot start accepting WebRTC connections, current role %d",
role_);
return false;
}
@@ -77,7 +84,8 @@ bool WebRtc::StartAcceptingConnections(const PeerId& self_id,
restart_receive_messages_alarm_ = CancelableAlarm(
"restart_receiving_messages_webrtc",
std::bind(&WebRtc::RestartReceiveMessages, this, location_hint),
std::bind(&WebRtc::RestartReceiveMessages, this, location_hint,
service_id),
kRestartReceiveMessagesDuration, &restart_receive_messages_executor_);
SessionDescriptionWrapper offer = connection_flow_->CreateOffer();
@@ -91,8 +99,6 @@ bool WebRtc::StartAcceptingConnections(const PeerId& self_id,
// the actual transport can begin.
ListenForWebRtcSocketFuture(connection_flow_->GetDataChannel(),
std::move(callback));
latest_service_id_ = service_id;
latest_local_endpoint_id_ = local_endpoint_id;
NEARBY_LOG(INFO, "Started listening for WebRtc connections as %s",
self_id.GetId().c_str());
}
@@ -151,11 +157,12 @@ bool WebRtc::SetLocalSessionDescription(SessionDescriptionWrapper sdp) {
return true;
}
void WebRtc::StopAcceptingConnections() {
if (!IsAcceptingConnections()) {
void WebRtc::StopAcceptingConnections(const std::string& service_id) {
if (!IsAcceptingConnections(service_id)) {
NEARBY_LOG(INFO,
"Skipped StopAcceptingConnections since we are not currently "
"accepting WebRTC connections");
"accepting WebRTC connections for %s",
service_id.c_str());
return;
}
@@ -166,19 +173,6 @@ void WebRtc::StopAcceptingConnections() {
NEARBY_LOG(INFO, "Stopped accepting WebRTC connections");
}
void WebRtc::StopAcceptingConnection(const std::string& service_id,
const std::string& local_endpoint_id) {
MutexLock lock(&mutex_);
if (service_id == latest_service_id_ &&
local_endpoint_id == latest_local_endpoint_id_) {
StopAcceptingConnections();
} else {
NEARBY_LOG(INFO,
"Skipped StopAcceptingConnection since we are not the latest"
"ongoing connection.");
}
}
Future<WebRtcSocketWrapper> WebRtc::ListenForWebRtcSocketFuture(
Future<rtc::scoped_refptr<webrtc::DataChannelInterface>>
data_channel_future,
@@ -486,8 +480,9 @@ void WebRtc::OffloadFromSignalingThread(Runnable runnable) {
single_thread_executor_.Execute(std::move(runnable));
}
void WebRtc::RestartReceiveMessages(const LocationHint& location_hint) {
if (!IsAcceptingConnections()) {
void WebRtc::RestartReceiveMessages(const LocationHint& location_hint,
const std::string& service_id) {
if (!IsAcceptingConnections(service_id)) {
NEARBY_LOG(INFO,
"Skipping restart since we are not accepting connections.");
return;
cpp/core/internal/mediums/webrtc.h
+7 -22
View File
@@ -53,14 +53,9 @@ class WebRtc {
// Runs on @MainThread.
bool IsAvailable();
// Returns if the device is ready to accept connections from remote devices.
// Returns if the device is accepting connection with specific service id.
// Runs on @MainThread.
bool IsAcceptingConnections() ABSL_LOCKS_EXCLUDED(mutex_);
// Returns if the device is accepting connection with specific service id and
// local endpoint id. Runs on @MainThread.
bool IsAcceptingConnection(const std::string& service_id,
const std::string& local_endpoint_id)
bool IsAcceptingConnections(const std::string& service_id)
ABSL_LOCKS_EXCLUDED(mutex_);
// Prepares the device to accept incoming WebRtc connections. Returns a
@@ -68,21 +63,13 @@ class WebRtc {
// Runs on @MainThread.
bool StartAcceptingConnections(const PeerId& self_id,
const std::string& service_id,
const std::string& local_endpoint_id,
const LocationHint& location_hint,
AcceptedConnectionCallback callback)
ABSL_LOCKS_EXCLUDED(mutex_);
// Prevents device from accepting future connections until
// StartAcceptingConnections() is called.
// Runs on @MainThread.
void StopAcceptingConnections() ABSL_LOCKS_EXCLUDED(mutex_);
// Try to stop (accepting) the specific connection with provided service id
// and local endpoint id. If the specific connection is not the latest one,
// then nothing will happen; if it's the latest one,
void StopAcceptingConnection(const std::string& service_id,
const std::string& local_endpoint_id)
// Try to stop (accepting) the specific connection with provided service id.
// Runs on @MainThread
void StopAcceptingConnections(const std::string& service_id)
ABSL_LOCKS_EXCLUDED(mutex_);
// Initiates a WebRtc connection with peer device identified by |peer_id|.
@@ -161,7 +148,8 @@ class WebRtc {
void OffloadFromSignalingThread(Runnable runnable);
// Runs on |restart_receive_messages_executor_|.
void RestartReceiveMessages(const LocationHint& location_hint)
void RestartReceiveMessages(const LocationHint& location_hint,
const std::string& service_id)
ABSL_LOCKS_EXCLUDED(mutex_);
Mutex mutex_;
@@ -184,9 +172,6 @@ class WebRtc {
// Restarts the signaling messenger for receiving messages.
ScheduledExecutor restart_receive_messages_executor_;
CancelableAlarm restart_receive_messages_alarm_;
std::string latest_service_id_ ABSL_GUARDED_BY(mutex_);
std::string latest_local_endpoint_id_ ABSL_GUARDED_BY(mutex_);
};
} // namespace mediums
cpp/core/internal/mediums/webrtc/BUILD
+1
View File
@@ -46,6 +46,7 @@ cc_test(
"signaling_frames_test.cc",
"webrtc_socket_test.cc",
],
tags = ["notsan"], # NOTE(b/139734036): known data race in usrsctplib.
deps = [
":webrtc",
"//platform/base",
cpp/core/internal/mediums/webrtc_test.cc
+34 -19
View File
@@ -26,7 +26,7 @@ class WebRtcTest : public ::testing::Test {
TEST_F(WebRtcTest, NotAcceptingConnections) {
WebRtc webrtc;
ASSERT_TRUE(webrtc.IsAvailable());
EXPECT_FALSE(webrtc.IsAcceptingConnections());
EXPECT_FALSE(webrtc.IsAcceptingConnections(std::string{}));
}
// Tests the flow when the device tries to accept connections twice. In this
@@ -38,14 +38,17 @@ TEST_F(WebRtcTest, StartAcceptingConnectionTwice) {
WebRtc webrtc;
PeerId self_id("peer_id");
const std::string service_id("NearbySharing");
LocationHint location_hint{};
ASSERT_TRUE(webrtc.IsAvailable());
ASSERT_TRUE(webrtc.StartAcceptingConnections(
self_id, location_hint, {mock_accepted_callback_.AsStdFunction()}));
self_id, service_id, location_hint,
{mock_accepted_callback_.AsStdFunction()}));
EXPECT_FALSE(webrtc.StartAcceptingConnections(
self_id, location_hint, {mock_accepted_callback_.AsStdFunction()}));
EXPECT_TRUE(webrtc.IsAcceptingConnections());
self_id, service_id, location_hint,
{mock_accepted_callback_.AsStdFunction()}));
EXPECT_TRUE(webrtc.IsAcceptingConnections(std::string{}));
}
// Tests the flow when the device tries to connect but the data channel times
@@ -53,14 +56,15 @@ TEST_F(WebRtcTest, StartAcceptingConnectionTwice) {
TEST_F(WebRtcTest, Connect_DataChannelTimeOut) {
WebRtc webrtc;
PeerId peer_id("peer_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
ASSERT_TRUE(webrtc.IsAvailable());
WebRtcSocketWrapper wrapper_1 = webrtc.Connect(peer_id, location_hint);
EXPECT_FALSE(wrapper_1.IsValid());
EXPECT_TRUE(webrtc.StartAcceptingConnections(peer_id, location_hint,
AcceptedConnectionCallback()));
EXPECT_TRUE(webrtc.StartAcceptingConnections(
peer_id, service_id, location_hint, AcceptedConnectionCallback()));
}
// Tests the flow when the device calls Connect() after calling
@@ -72,17 +76,20 @@ TEST_F(WebRtcTest, StartAcceptingConnection_ThenConnect) {
WebRtc webrtc;
PeerId self_id("peer_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
ASSERT_TRUE(webrtc.IsAvailable());
ASSERT_TRUE(webrtc.StartAcceptingConnections(
self_id, location_hint, {mock_accepted_callback_.AsStdFunction()}));
self_id, service_id, location_hint,
{mock_accepted_callback_.AsStdFunction()}));
WebRtcSocketWrapper wrapper =
webrtc.Connect(PeerId("random_peer_id"), location_hint);
EXPECT_TRUE(webrtc.IsAcceptingConnections());
EXPECT_TRUE(webrtc.IsAcceptingConnections(std::string{}));
EXPECT_FALSE(wrapper.IsValid());
EXPECT_FALSE(webrtc.StartAcceptingConnections(
self_id, location_hint, {mock_accepted_callback_.AsStdFunction()}));
self_id, service_id, location_hint,
{mock_accepted_callback_.AsStdFunction()}));
}
// Tests the flow when the device calls StartAcceptingConnections but the medium
@@ -94,13 +101,15 @@ TEST_F(WebRtcTest, StartAndStopAcceptingConnections) {
WebRtc webrtc;
PeerId self_id("peer_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
ASSERT_TRUE(webrtc.IsAvailable());
ASSERT_TRUE(webrtc.StartAcceptingConnections(
self_id, location_hint, {mock_accepted_callback_.AsStdFunction()}));
webrtc.StopAcceptingConnections();
EXPECT_FALSE(webrtc.IsAcceptingConnections());
self_id, service_id, location_hint,
{mock_accepted_callback_.AsStdFunction()}));
webrtc.StopAcceptingConnections(service_id);
EXPECT_FALSE(webrtc.IsAcceptingConnections(std::string{}));
}
// Tests the flow when the device tries to connect to two different peers
@@ -109,12 +118,13 @@ TEST_F(WebRtcTest, ConnectTwice) {
WebRtc receiver, sender, device_c;
WebRtcSocketWrapper receiver_socket, sender_socket;
const PeerId self_id("self_id"), other_id("other_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
Future<bool> connected;
ByteArray message("message xyz");
receiver.StartAcceptingConnections(
self_id, location_hint,
self_id, service_id, location_hint,
{[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
@@ -123,7 +133,7 @@ TEST_F(WebRtcTest, ConnectTwice) {
using MockAcceptedCallback =
testing::MockFunction<void(WebRtcSocketWrapper socket)>;
testing::StrictMock<MockAcceptedCallback> mock_accepted_callback_;
device_c.StartAcceptingConnections(other_id, location_hint,
device_c.StartAcceptingConnections(other_id, service_id, location_hint,
{mock_accepted_callback_.AsStdFunction()});
sender_socket = sender.Connect(self_id, location_hint);
@@ -154,12 +164,13 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndAbort) {
WebRtc receiver, sender;
WebRtcSocketWrapper receiver_socket, sender_socket;
const PeerId self_id("self_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
Future<bool> connected;
ByteArray message("message xyz");
receiver.StartAcceptingConnections(
self_id, location_hint,
self_id, service_id, location_hint,
{[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
@@ -181,12 +192,13 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndSendData) {
WebRtc receiver, sender;
WebRtcSocketWrapper receiver_socket, sender_socket;
const PeerId self_id("self_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
Future<bool> connected;
ByteArray message("message");
receiver.StartAcceptingConnections(
self_id, location_hint,
self_id, service_id, location_hint,
{[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
@@ -214,12 +226,13 @@ TEST_F(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) {
WebRtc receiver, sender;
WebRtcSocketWrapper receiver_socket, sender_socket;
const PeerId self_id("self_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
Future<bool> connected;
ByteArray message("message xyz");
receiver.StartAcceptingConnections(
self_id, location_hint,
self_id, service_id, location_hint,
{[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
@@ -233,7 +246,7 @@ TEST_F(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) {
EXPECT_TRUE(devices_connected.result());
// Only shuts down signaling channel.
receiver.StopAcceptingConnections();
receiver.StopAcceptingConnections(service_id);
sender_socket.GetOutputStream().Write(message);
ExceptionOr<ByteArray> received_msg =
@@ -252,11 +265,13 @@ TEST_F(WebRtcTest, StartAcceptingConnections_NullPeerConnection) {
WebRtc webrtc;
PeerId self_id("peer_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
ASSERT_TRUE(webrtc.IsAvailable());
EXPECT_FALSE(webrtc.StartAcceptingConnections(
self_id, location_hint, {mock_accepted_callback_.AsStdFunction()}));
self_id, service_id, location_hint,
{mock_accepted_callback_.AsStdFunction()}));
}
TEST_F(WebRtcTest, Connect_NullPeerConnection) {
cpp/core/internal/p2p_cluster_pcp_handler.cc
+5 -24
View File
@@ -78,17 +78,6 @@ BasePcpHandler::StartOperationResult P2pClusterPcpHandler::StartAdvertisingImpl(
std::vector<proto::connections::Medium> mediums_started_successfully;
WebRtcState web_rtc_state{WebRtcState::kUnconnectable};
if (options.allowed.web_rtc) {
proto::connections::Medium webrtc_medium =
StartListeningForWebRtcConnections(
client, service_id, local_endpoint_id, local_endpoint_info);
if (webrtc_medium != proto::connections::UNKNOWN_MEDIUM) {
NEARBY_LOG(INFO,
"P2pClusterPcpHandler::StartAdvertisingImpl: WebRtc added");
mediums_started_successfully.push_back(webrtc_medium);
web_rtc_state = WebRtcState::kConnectable;
}
}
if (options.allowed.wifi_lan) {
const ByteArray wifi_lan_hash =
@@ -149,9 +138,6 @@ Status P2pClusterPcpHandler::StopAdvertisingImpl(ClientProxy* client) {
ble_medium_.StopAdvertising(client->GetAdvertisingServiceId());
ble_medium_.StopAcceptingConnections(client->GetAdvertisingServiceId());
webrtc_medium_.StopAcceptingConnection(client->GetAdvertisingServiceId(),
client->GetLocalEndpointId());
wifi_lan_medium_.StopAdvertising(client->GetAdvertisingServiceId());
wifi_lan_medium_.StopAcceptingConnections(client->GetAdvertisingServiceId());
@@ -193,8 +179,8 @@ bool P2pClusterPcpHandler::IsRecognizedBluetoothEndpoint(
void P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler(
ClientProxy* client, const std::string& service_id,
BluetoothDevice& device) {
RunOnPcpHandlerThread([this, client, service_id, &device]() {
BluetoothDevice device) {
RunOnPcpHandlerThread([this, client, service_id, device]() {
// Make sure we are still discovering before proceeding.
if (!client->IsDiscovering()) {
NEARBY_LOG(INFO,
@@ -205,7 +191,7 @@ void P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler(
}
// Parse the Bluetooth device name.
const std::string& device_name_string = device.GetName();
const std::string device_name_string = device.GetName();
BluetoothDeviceName device_name(device_name_string);
// Make sure the Bluetooth device name points to a valid
@@ -690,10 +676,6 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::ConnectImpl(
break;
}
case proto::connections::Medium::WEB_RTC: {
auto* webrtc_endpoint = down_cast<WebRtcEndpoint*>(endpoint);
if (webrtc_endpoint) {
return WebRtcConnectImpl(client, webrtc_endpoint);
}
break;
}
default:
@@ -1155,13 +1137,12 @@ P2pClusterPcpHandler::StartListeningForWebRtcConnections(
return proto::connections::UNKNOWN_MEDIUM;
}
if (!webrtc_medium_.IsAcceptingConnections()) {
if (!webrtc_medium_.IsAcceptingConnections(service_id)) {
mediums::PeerId self_id = CreatePeerIdFromAdvertisement(
service_id, local_endpoint_id, local_endpoint_info);
std::string empty_country_code;
if (!webrtc_medium_.StartAcceptingConnections(
self_id, service_id, local_endpoint_id,
Utils::BuildLocationHint(empty_country_code),
self_id, service_id, Utils::BuildLocationHint(empty_country_code),
{[this, client,
local_endpoint_info](mediums::WebRtcSocketWrapper socket) {
if (!socket.IsValid()) {
cpp/core/internal/p2p_cluster_pcp_handler.h
+1 -1
View File
@@ -134,7 +134,7 @@ class P2pClusterPcpHandler : public BasePcpHandler {
const BluetoothDeviceName& name) const;
void BluetoothDeviceDiscoveredHandler(ClientProxy* client,
const std::string& service_id,
BluetoothDevice& device);
BluetoothDevice device);
void BluetoothDeviceLostHandler(ClientProxy* client,
const std::string& service_id,
BluetoothDevice& device);
cpp/core/internal/webrtc_bwu_handler.cc
+5 -6
View File
@@ -20,10 +20,10 @@ WebrtcBwuHandler::WebrtcBwuHandler(Mediums& mediums,
mediums_(mediums) {}
void WebrtcBwuHandler::Revert() {
if (!active_service_ids_.empty()) {
webrtc_.StopAcceptingConnections();
active_service_ids_.clear();
for (const auto& service_id : active_service_ids_) {
webrtc_.StopAcceptingConnections(service_id);
}
active_service_ids_.clear();
NEARBY_LOG(INFO, "WebrtcBwuHandler successfully reverted state.");
}
@@ -61,10 +61,9 @@ ByteArray WebrtcBwuHandler::InitializeUpgradedMediumForEndpoint(
Utils::BuildLocationHint(webrtc_.GetDefaultCountryCode());
mediums::PeerId self_id{mediums::PeerId::FromRandom()};
if (!webrtc_.IsAcceptingConnections()) {
if (!webrtc_.IsAcceptingConnections(service_id)) {
if (!webrtc_.StartAcceptingConnections(
self_id, upgrade_service_id, client->GetLocalEndpointId(),
location_hint,
self_id, upgrade_service_id, location_hint,
{
.accepted_cb = absl::bind_front(
&WebrtcBwuHandler::OnIncomingWebrtcConnection, this, client,