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

Remove the need for WebRtcSocket stub

Browse Source 
PiperOrigin-RevId: 915477372
This commit is contained in:
Nick Bourdakos
2026-05-14 09:48:59 -07:00
committed by Copybara-Service
parent 8e4f67912c
commit b66eebfef4
24 changed files with 315 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/implementation/BUILD
+3 -3
View File
@@ -155,7 +155,8 @@ cc_library(
"//connections/implementation/flags:connections_flags",
"//connections/implementation/mediums",
"//connections/implementation/mediums:utils",
"//connections/implementation/mediums:webrtc_utils",
"//connections/implementation/mediums:webrtc_peer_id",
"//connections/implementation/mediums:webrtc_socket",
"//connections/implementation/mediums/advertisements:dct_advertisement",
"//connections/implementation/mediums/advertisements:util",
"//connections/implementation/mediums/ble:ble_advertisement_header",
@@ -315,10 +316,9 @@ cc_test(
":internal_test",
":types",
"//connections:core_types",
"//connections/implementation/analytics",
"//connections/implementation/flags:connections_flags",
"//connections/implementation/mediums",
"//connections/implementation/mediums:webrtc_utils",
"//connections/implementation/mediums:webrtc_peer_id",
"//connections/implementation/proto:offline_wire_formats_cc_proto",
"//connections/v3:v3_types",
"//internal/analytics:mock_event_logger",
connections/implementation/mediums/BUILD
+17 -13
View File
@@ -50,7 +50,8 @@ cc_library(
],
deps = [
":utils",
":webrtc_utils",
":webrtc_peer_id",
":webrtc_socket",
"//connections:core_types",
"//connections/implementation:types",
"//connections/implementation/flags:connections_flags",
@@ -71,7 +72,6 @@ cc_library(
"//internal/platform/flags:platform_flags",
"//internal/platform/implementation:comm",
"//internal/platform/implementation:platform",
"//internal/platform/implementation:wifi_utils",
"//proto/mediums:web_rtc_signaling_frames_cc_proto",
# "//third_party/webrtc/files/stable/webrtc/api:create_peerconnection_factory", # buildcleaner: keep
# "//third_party/webrtc/files/stable/webrtc/api:jsep",
@@ -90,15 +90,9 @@ cc_library(
)
cc_library(
name = "webrtc_utils",
srcs = [
"webrtc_peer_id.cc",
],
hdrs = [
"webrtc_peer_id.h",
"webrtc_socket.h",
"webrtc_socket_stub.h",
],
name = "webrtc_peer_id",
srcs = ["webrtc_peer_id.cc"],
hdrs = ["webrtc_peer_id.h"],
visibility = [
"//connections/implementation:__pkg__",
"//connections/implementation/mediums:__pkg__",
@@ -106,7 +100,6 @@ cc_library(
],
deps = [
":utils",
"//connections/implementation/mediums/webrtc:data_types",
"//internal/platform:base",
"@com_google_absl//absl/strings",
],
@@ -141,6 +134,16 @@ cc_library(
],
)
cc_library(
name = "webrtc_socket",
hdrs = ["webrtc_socket.h"],
visibility = ["//connections/implementation:__subpackages__"],
deps = [
"//internal/platform:base",
"@com_google_absl//absl/strings:string_view",
],
)
cc_test(
name = "core_internal_mediums_test",
size = "small",
@@ -197,7 +200,8 @@ cc_test(
],
deps = [
":mediums",
":webrtc_utils",
":webrtc_peer_id",
":webrtc_socket",
"//internal/platform:base",
"//internal/platform:cancellation_flag",
"//internal/platform:comm",
connections/implementation/mediums/webrtc.cc
+11 -10
View File
@@ -207,13 +207,13 @@ void WebRtc::StopAcceptingConnections(const std::string& service_id) {
<< service_id;
}
ErrorOr<WebRtcSocketWrapper> WebRtc::Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> WebRtc::Connect(
const std::string& service_id, const WebrtcPeerId& remote_peer_id,
const LocationHint& location_hint, CancellationFlag* cancellation_flag,
bool non_cellular) {
service_id_to_connect_attempts_count_map_[service_id] = 1;
medium_->SetNonCellular(non_cellular);
ErrorOr<WebRtcSocketWrapper> wrapper_result = {
ErrorOr<std::shared_ptr<WebRtcSocket>> wrapper_result = {
Error(OperationResultCode::DETAIL_UNKNOWN)};
while (service_id_to_connect_attempts_count_map_[service_id] <=
kConnectAttemptsLimit) {
@@ -242,12 +242,12 @@ ErrorOr<WebRtcSocketWrapper> WebRtc::Connect(
return {Error(wrapper_result.error().operation_result_code().value())};
}
ErrorOr<WebRtcSocketWrapper> WebRtc::AttemptToConnect(
ErrorOr<std::shared_ptr<WebRtcSocket>> WebRtc::AttemptToConnect(
const std::string& service_id, const WebrtcPeerId& remote_peer_id,
const LocationHint& location_hint, CancellationFlag* cancellation_flag) {
ConnectionRequestInfo info = ConnectionRequestInfo();
info.self_peer_id = WebrtcPeerId::FromRandom();
Future<WebRtcSocketWrapper> socket_future = info.socket_future;
Future<std::shared_ptr<WebRtcSocket>> socket_future = info.socket_future;
// `listener` will go out of scope at the end of `AttemptToConnect`, and this
// is expected. This `listener` is tied to `socket_future` which we block on
@@ -329,7 +329,7 @@ ErrorOr<WebRtcSocketWrapper> WebRtc::AttemptToConnect(
// Wait for the connection to go through. Don't hold the mutex here so that
// we're not blocking necessary operations.
ExceptionOr<WebRtcSocketWrapper> socket_result =
ExceptionOr<std::shared_ptr<WebRtcSocket>> socket_result =
socket_future.Get(kDataChannelTimeout);
{
@@ -660,9 +660,9 @@ void WebRtc::RestartTachyonReceiveMessages(const std::string& service_id) {
<< service_id;
}
void WebRtc::ProcessDataChannelOpen(const std::string& service_id,
const WebrtcPeerId& remote_peer_id,
WebRtcSocketWrapper socket_wrapper) {
void WebRtc::ProcessDataChannelOpen(
const std::string& service_id, const WebrtcPeerId& remote_peer_id,
std::shared_ptr<WebRtcSocket> socket_wrapper) {
MutexLock lock(&mutex_);
// Notify the client of the newly formed socket.
@@ -683,7 +683,7 @@ void WebRtc::ProcessDataChannelOpen(const std::string& service_id,
}
// No one to handle the newly created DataChannel, so we'll just close it.
socket_wrapper.Close();
socket_wrapper->Close();
LOG(INFO) << "Ignoring new DataChannel because we are not accepting "
"connections for service "
<< service_id;
@@ -719,7 +719,8 @@ std::unique_ptr<ConnectionFlow> WebRtc::CreateConnectionFlow(
}}},
{
.data_channel_open_cb = {[this, service_id, remote_peer_id](
WebRtcSocketWrapper socket_wrapper) {
std::shared_ptr<WebRtcSocket>
socket_wrapper) {
OffloadFromThread(
"rtc-channel-created",
[this, service_id, remote_peer_id, socket_wrapper]() {
connections/implementation/mediums/webrtc.h
+5 -5
View File
@@ -50,7 +50,7 @@ class WebRtc {
public:
// Callback that is invoked when a new connection is accepted.
using AcceptedConnectionCallback = absl::AnyInvocable<void(
const std::string& service_id, WebRtcSocketWrapper socket)>;
const std::string& service_id, std::shared_ptr<WebRtcSocket> socket)>;
WebRtc();
~WebRtc();
@@ -85,7 +85,7 @@ class WebRtc {
// Initiates a WebRtc connection with peer device identified by |peer_id|
// with internal retry for maximum attempts of kConnectAttemptsLimit.
// Runs on @MainThread.
ErrorOr<WebRtcSocketWrapper> Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> Connect(
const std::string& service_id, const WebrtcPeerId& peer_id,
const location::nearby::connections::LocationHint& location_hint,
CancellationFlag* cancellation_flag, bool non_cellular)
@@ -145,13 +145,13 @@ class WebRtc {
// The pending DataChannel future. Our client will be blocked on this while
// they wait for us to set up the channel over Tachyon.
Future<WebRtcSocketWrapper> socket_future;
Future<std::shared_ptr<WebRtcSocket>> socket_future;
};
// Attempt to initiates a WebRtc connection with peer device identified by
// |peer_id|.
// Runs on @MainThread.
ErrorOr<WebRtcSocketWrapper> AttemptToConnect(
ErrorOr<std::shared_ptr<WebRtcSocket>> AttemptToConnect(
const std::string& service_id, const WebrtcPeerId& peer_id,
const location::nearby::connections::LocationHint& location_hint,
CancellationFlag* cancellation_flag) ABSL_LOCKS_EXCLUDED(mutex_);
@@ -214,7 +214,7 @@ class WebRtc {
// Runs on |single_thread_executor_|.
void ProcessDataChannelOpen(const std::string& service_id,
const WebrtcPeerId& remote_peer_id,
WebRtcSocketWrapper socket_wrapper)
std::shared_ptr<WebRtcSocket> socket_wrapper)
ABSL_LOCKS_EXCLUDED(mutex_);
// Runs on |single_thread_executor_|.
connections/implementation/mediums/webrtc/BUILD
+13 -15
View File
@@ -37,9 +37,10 @@ cc_library(
"//connections/implementation:__subpackages__",
],
deps = [
":data_types",
":webrtc_socket_impl",
"//connections:core_types",
"//connections/implementation/mediums:webrtc_utils",
"//connections/implementation/mediums:webrtc_peer_id",
"//connections/implementation/mediums:webrtc_socket",
"//internal/platform:base",
"//internal/platform:comm",
"//internal/platform:logging",
@@ -57,24 +58,20 @@ cc_library(
)
cc_library(
name = "data_types",
srcs = [
"webrtc_socket_impl.cc",
],
hdrs = [
"webrtc_socket_impl.h",
],
copts = [
"-DCORE_ADAPTER_DLL",
"-DNO_WEBRTC",
],
name = "webrtc_socket_impl",
srcs = ["webrtc_socket_impl.cc"],
hdrs = ["webrtc_socket_impl.h"],
visibility = [
"//connections/implementation:__subpackages__",
],
deps = [
"//connections/implementation/mediums:webrtc_socket",
"//internal/platform:base",
"//internal/platform:logging",
"//internal/platform:types",
# "//third_party/webrtc/files/stable/webrtc/api:data_channel_interface",
# "//third_party/webrtc/files/stable/webrtc/api:scoped_refptr",
"@com_google_absl//absl/functional:any_invocable",
"@com_google_absl//absl/strings:string_view",
],
)
@@ -92,9 +89,10 @@ cc_test(
"requires-net:external",
],
deps = [
":data_types",
":webrtc",
"//connections/implementation/mediums:webrtc_utils",
":webrtc_socket_impl",
"//connections/implementation/mediums:webrtc_peer_id",
"//connections/implementation/mediums:webrtc_socket",
"//internal/platform:base",
"//internal/platform:comm",
"//internal/platform:test_util",
connections/implementation/mediums/webrtc/connection_flow.cc
+7 -7
View File
@@ -412,10 +412,11 @@ void ConnectionFlow::OnSignalingStable() {
void ConnectionFlow::CreateSocketFromDataChannel(
webrtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) {
LOG(INFO) << "Creating data channel socket";
auto socket =
std::make_unique<WebRtcSocket>("WebRtcSocket", std::move(data_channel));
auto socket = std::make_shared<WebRtcSocketImpl>("WebRtcSocket",
std::move(data_channel));
socket_ = socket;
socket->SetSocketListener({
.socket_ready_cb = {[this](WebRtcSocket* socket) {
.socket_ready_cb = {[this](WebRtcSocketImpl* socket) {
CHECK(IsRunningOnSignalingThread());
if (!TransitionState(State::kWaitingToConnect, State::kConnected)) {
LOG(ERROR) << "Data channel socket is open but connection "
@@ -424,14 +425,13 @@ void ConnectionFlow::CreateSocketFromDataChannel(
return;
}
// Pass socket wrapper by copy on purpose
data_channel_listener_.data_channel_open_cb(socket_wrapper_);
data_channel_listener_.data_channel_open_cb(socket_);
}},
.socket_closed_cb =
[this](WebRtcSocket*) {
[this](WebRtcSocketImpl*) {
data_channel_listener_.data_channel_closed_cb();
},
});
socket_wrapper_ = WebRtcSocketWrapper(std::move(socket));
}
void ConnectionFlow::OnIceCandidate(const webrtc::IceCandidate* candidate) {
@@ -512,7 +512,7 @@ bool ConnectionFlow::CloseOnSignalingThread() {
// Close the socket wrapper before terminating the PeerConnection
// since the teardown process of the PC may close threads that are
// otherwise depended upon by objects kept alive by the socket_wrapper.
if (socket_wrapper_.IsValid()) socket_wrapper_.Close();
if (socket_ && socket_->IsValid()) socket_->Close();
// This prevents other tasks from queuing on the signaling thread for this
// object.
connections/implementation/mediums/webrtc/connection_flow.h
+1 -1
View File
@@ -222,7 +222,7 @@ class ConnectionFlow : public webrtc::PeerConnectionObserver {
// Used to hold a reference to the WebRtcSocket while the data channel is
// connecting.
WebRtcSocketWrapper socket_wrapper_;
std::shared_ptr<WebRtcSocket> socket_;
std::vector<std::unique_ptr<webrtc::IceCandidate>>
cached_remote_ice_candidates_;
connections/implementation/mediums/webrtc/connection_flow_test.cc
+24 -21
View File
@@ -61,7 +61,8 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) {
Future<ByteArray> message_received_future;
Future<WebRtcSocketWrapper> offerer_socket_future, answerer_socket_future;
Future<std::shared_ptr<WebRtcSocket>> offerer_socket_future,
answerer_socket_future;
std::unique_ptr<ConnectionFlow> offerer, answerer;
@@ -77,7 +78,7 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) {
answerer->OnRemoteIceCandidatesReceived(std::move(vec));
}},
{.data_channel_open_cb =
[&offerer_socket_future](WebRtcSocketWrapper socket) {
[&offerer_socket_future](std::shared_ptr<WebRtcSocket> socket) {
offerer_socket_future.Set(std::move(socket));
}},
{.adapter_type_changed_cb =
@@ -97,7 +98,7 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) {
offerer->OnRemoteIceCandidatesReceived(std::move(vec));
}},
{.data_channel_open_cb =
[&answerer_socket_future](WebRtcSocketWrapper socket) {
[&answerer_socket_future](std::shared_ptr<WebRtcSocket> socket) {
answerer_socket_future.Set(std::move(socket));
}},
{.adapter_type_changed_cb =
@@ -122,18 +123,18 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) {
EXPECT_TRUE(answerer->SetLocalSessionDescription(std::move(answer)));
// Retrieve Data Channels
ExceptionOr<WebRtcSocketWrapper> offerer_socket =
ExceptionOr<std::shared_ptr<WebRtcSocket>> offerer_socket =
offerer_socket_future.Get(absl::Seconds(1));
EXPECT_TRUE(offerer_socket.ok());
ExceptionOr<WebRtcSocketWrapper> answerer_socket =
ExceptionOr<std::shared_ptr<WebRtcSocket>> answerer_socket =
answerer_socket_future.Get(absl::Seconds(1));
EXPECT_TRUE(answerer_socket.ok());
// Send message on data channel
absl::string_view message = "Test";
offerer_socket.result().GetImpl().GetOutputStream().Write(message);
offerer_socket.result()->GetOutputStream().Write(message);
ExceptionOr<ByteArray> received_message =
answerer_socket.result().GetImpl().GetInputStream().Read(4);
answerer_socket.result()->GetInputStream().Read(4);
EXPECT_TRUE(received_message.ok());
EXPECT_EQ(received_message.result(), ByteArray{message.data()});
}
@@ -259,7 +260,8 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) {
Future<ByteArray> message_received_future;
Future<WebRtcSocketWrapper> offerer_socket_future, answerer_socket_future;
Future<std::shared_ptr<WebRtcSocket>> offerer_socket_future,
answerer_socket_future;
std::unique_ptr<ConnectionFlow> offerer, answerer;
@@ -275,7 +277,7 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) {
answerer->OnRemoteIceCandidatesReceived(std::move(vec));
}},
{.data_channel_open_cb =
[&offerer_socket_future](WebRtcSocketWrapper socket) {
[&offerer_socket_future](std::shared_ptr<WebRtcSocket> socket) {
offerer_socket_future.Set(std::move(socket));
}},
{.adapter_type_changed_cb =
@@ -295,7 +297,7 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) {
offerer->OnRemoteIceCandidatesReceived(std::move(vec));
}},
{.data_channel_open_cb =
[&answerer_socket_future](WebRtcSocketWrapper wrapper) {
[&answerer_socket_future](std::shared_ptr<WebRtcSocket> wrapper) {
answerer_socket_future.Set(std::move(wrapper));
}},
{.adapter_type_changed_cb =
@@ -321,10 +323,10 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) {
EXPECT_TRUE(answerer->SetLocalSessionDescription(std::move(answer)));
// Retrieve Data Channels
ExceptionOr<WebRtcSocketWrapper> offerer_socket =
ExceptionOr<std::shared_ptr<WebRtcSocket>> offerer_socket =
offerer_socket_future.Get(absl::Seconds(1));
EXPECT_TRUE(offerer_socket.ok());
ExceptionOr<WebRtcSocketWrapper> answerer_socket =
ExceptionOr<std::shared_ptr<WebRtcSocket>> answerer_socket =
answerer_socket_future.Get(absl::Seconds(1));
EXPECT_TRUE(offerer_socket.ok());
@@ -338,9 +340,9 @@ TEST_F(ConnectionFlowTest, TerminateAnswerer) {
// Send message on data channel
absl::string_view message = "Test";
offerer_socket.result().GetOutputStream().Write(message);
offerer_socket.result()->GetOutputStream().Write(message);
ExceptionOr<ByteArray> received_message =
answerer_socket.result().GetInputStream().Read(4);
answerer_socket.result()->GetInputStream().Read(4);
EXPECT_TRUE(received_message.GetResult().Empty());
}
@@ -349,7 +351,8 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) {
Future<ByteArray> message_received_future;
Future<WebRtcSocketWrapper> offerer_socket_future, answerer_socket_future;
Future<std::shared_ptr<WebRtcSocket>> offerer_socket_future,
answerer_socket_future;
std::unique_ptr<ConnectionFlow> offerer, answerer;
@@ -365,7 +368,7 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) {
answerer->OnRemoteIceCandidatesReceived(std::move(vec));
}},
{.data_channel_open_cb =
[&offerer_socket_future](WebRtcSocketWrapper socket) {
[&offerer_socket_future](std::shared_ptr<WebRtcSocket> socket) {
offerer_socket_future.Set(std::move(socket));
}},
{.adapter_type_changed_cb =
@@ -386,7 +389,7 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) {
offerer->OnRemoteIceCandidatesReceived(std::move(vec));
}},
{.data_channel_open_cb =
[&answerer_socket_future](WebRtcSocketWrapper wrapper) {
[&answerer_socket_future](std::shared_ptr<WebRtcSocket> wrapper) {
answerer_socket_future.Set(std::move(wrapper));
}},
{.adapter_type_changed_cb =
@@ -412,10 +415,10 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) {
EXPECT_TRUE(answerer->SetLocalSessionDescription(std::move(answer)));
// Retrieve Data Channels
ExceptionOr<WebRtcSocketWrapper> offerer_socket =
ExceptionOr<std::shared_ptr<WebRtcSocket>> offerer_socket =
offerer_socket_future.Get(absl::Seconds(1));
EXPECT_TRUE(offerer_socket.ok());
ExceptionOr<WebRtcSocketWrapper> answerer_socket =
ExceptionOr<std::shared_ptr<WebRtcSocket>> answerer_socket =
answerer_socket_future.Get(absl::Seconds(1));
EXPECT_TRUE(offerer_socket.ok());
@@ -429,9 +432,9 @@ TEST_F(ConnectionFlowTest, TerminateOfferer) {
// Send message on data channel
absl::string_view message = "Test";
offerer_socket.result().GetOutputStream().Write(message);
offerer_socket.result()->GetOutputStream().Write(message);
ExceptionOr<ByteArray> received_message =
answerer_socket.result().GetInputStream().Read(4);
answerer_socket.result()->GetInputStream().Read(4);
EXPECT_TRUE(received_message.GetResult().Empty());
}
connections/implementation/mediums/webrtc/data_channel_listener.h
+5 -3
View File
@@ -17,6 +17,8 @@
#ifndef NO_WEBRTC
#include <memory>
#include "absl/functional/any_invocable.h"
#include "connections/implementation/mediums/webrtc_socket.h"
@@ -28,8 +30,8 @@ namespace mediums {
struct DataChannelListener {
// Called when the data channel is open and the socket wrapper is ready to
// read and write.
absl::AnyInvocable<void(WebRtcSocketWrapper)> data_channel_open_cb =
[](WebRtcSocketWrapper) {};
absl::AnyInvocable<void(std::shared_ptr<WebRtcSocket>)> data_channel_open_cb =
[](std::shared_ptr<WebRtcSocket>) {};
// Called when the data channel is closed.
absl::AnyInvocable<void()> data_channel_closed_cb = []() {};
@@ -39,6 +41,6 @@ struct DataChannelListener {
} // namespace connections
} // namespace nearby
#endif
#endif // NO_WEBRTC
#endif // CORE_INTERNAL_MEDIUMS_WEBRTC_DATA_CHANNEL_LISTENER_H_
connections/implementation/mediums/webrtc/webrtc_socket_impl.cc
+29 -24
View File
@@ -12,6 +12,10 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#ifndef NO_WEBRTC
#include "connections/implementation/mediums/webrtc/webrtc_socket_impl.h"
#include <cstdint>
#include <string>
#include <tuple>
@@ -21,19 +25,20 @@
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
#include "internal/platform/input_stream.h"
#include "internal/platform/pipe.h"
#ifndef NO_WEBRTC
#include "connections/implementation/mediums/webrtc/webrtc_socket_impl.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex_lock.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/pipe.h"
#include "internal/platform/runnable.h"
#include "webrtc/api/data_channel_interface.h"
#include "webrtc/api/scoped_refptr.h"
namespace nearby {
namespace connections {
namespace mediums {
// OutputStreamImpl
Exception WebRtcSocket::OutputStreamImpl::Write(absl::string_view data) {
Exception WebRtcSocketImpl::OutputStreamImpl::Write(absl::string_view data) {
if (data.size() > kMaxDataSize) {
LOG(WARNING) << "Sending data larger than 1MB";
return {Exception::kIo};
@@ -53,18 +58,18 @@ Exception WebRtcSocket::OutputStreamImpl::Write(absl::string_view data) {
return {Exception::kSuccess};
}
Exception WebRtcSocket::OutputStreamImpl::Flush() {
Exception WebRtcSocketImpl::OutputStreamImpl::Flush() {
// Java implementation is empty.
return {Exception::kSuccess};
}
Exception WebRtcSocket::OutputStreamImpl::Close() {
Exception WebRtcSocketImpl::OutputStreamImpl::Close() {
socket_->Close();
return {Exception::kSuccess};
}
// WebRtcSocket
WebRtcSocket::WebRtcSocket(
WebRtcSocketImpl::WebRtcSocketImpl(
const std::string& name,
webrtc::scoped_refptr<webrtc::DataChannelInterface> data_channel)
: name_(name), data_channel_(std::move(data_channel)) {
@@ -73,7 +78,7 @@ WebRtcSocket::WebRtcSocket(
data_channel_->RegisterObserver(this);
}
WebRtcSocket::~WebRtcSocket() {
WebRtcSocketImpl::~WebRtcSocketImpl() {
LOG(INFO) << "WebRtcSocket::~WebRtcSocket(" << name_ << ") this: " << this;
if (!IsClosed()) {
@@ -85,16 +90,16 @@ WebRtcSocket::~WebRtcSocket() {
<< " done";
}
InputStream& WebRtcSocket::GetInputStream() { return *pipe_input_; }
InputStream& WebRtcSocketImpl::GetInputStream() { return *pipe_input_; }
OutputStream& WebRtcSocket::GetOutputStream() { return output_stream_; }
OutputStream& WebRtcSocketImpl::GetOutputStream() { return output_stream_; }
Exception WebRtcSocket::Close() {
Exception WebRtcSocketImpl::Close() {
LOG(INFO) << "WebRtcSocket::Close(" << name_ << ") this: " << this;
if (closed_.Set(true)) return {Exception::kSuccess};
ClosePipe();
// NOTE: This call blocks and triggers a state change on the siginaling thread
// NOTE: This call blocks and triggers a state change on the signaling thread
// to 'closing' but does not block until 'closed' is sent so the data channel
// is not fully closed when this call is done.
data_channel_->Close();
@@ -102,7 +107,7 @@ Exception WebRtcSocket::Close() {
return {Exception::kSuccess};
}
void WebRtcSocket::OnStateChange() {
void WebRtcSocketImpl::OnStateChange() {
// Running on the signaling thread right now.
LOG(ERROR) << "WebRtcSocket::OnStateChange() webrtc data channel state: "
<< webrtc::DataChannelInterface::DataStateString(
@@ -131,7 +136,7 @@ void WebRtcSocket::OnStateChange() {
break;
}
}
void WebRtcSocket::OnMessage(const webrtc::DataBuffer& buffer) {
void WebRtcSocketImpl::OnMessage(const webrtc::DataBuffer& buffer) {
// This is a data channel callback on the signaling thread, lets off load so
// we don't block signaling.
OffloadFromSignalingThread(
@@ -147,20 +152,20 @@ void WebRtcSocket::OnMessage(const webrtc::DataBuffer& buffer) {
});
}
void WebRtcSocket::OnBufferedAmountChange(uint64_t sent_data_size) {
void WebRtcSocketImpl::OnBufferedAmountChange(uint64_t sent_data_size) {
// This is a data channel callback on the signaling thread, lets off load so
// we don't block signaling.
OffloadFromSignalingThread([this] { WakeUpWriter(); });
}
bool WebRtcSocket::SendMessage(const ByteArray& data) {
bool WebRtcSocketImpl::SendMessage(const ByteArray& data) {
return data_channel_->Send(
webrtc::DataBuffer(std::string(data.data(), data.size())));
}
bool WebRtcSocket::IsClosed() { return closed_.Get(); }
bool WebRtcSocketImpl::IsClosed() { return closed_.Get(); }
void WebRtcSocket::ClosePipe() {
void WebRtcSocketImpl::ClosePipe() {
LOG(INFO) << "WebRtcSocket::ClosePipe(" << name_ << ") this: " << this;
// This is thread-safe to close these sockets even if a read or write is in
// process on another thread, Close will wait for the exclusive mutex before
@@ -173,16 +178,16 @@ void WebRtcSocket::ClosePipe() {
}
// Must not be called on signalling thread.
void WebRtcSocket::WakeUpWriter() {
void WebRtcSocketImpl::WakeUpWriter() {
MutexLock lock(&backpressure_mutex_);
buffer_variable_.Notify();
}
void WebRtcSocket::SetSocketListener(SocketListener&& listener) {
void WebRtcSocketImpl::SetSocketListener(SocketListener&& listener) {
socket_listener_ = std::move(listener);
}
void WebRtcSocket::BlockUntilSufficientSpaceInBuffer(int length) {
void WebRtcSocketImpl::BlockUntilSufficientSpaceInBuffer(int length) {
MutexLock lock(&backpressure_mutex_);
while (!IsClosed() &&
(data_channel_->buffered_amount() + length > kMaxDataSize)) {
@@ -191,7 +196,7 @@ void WebRtcSocket::BlockUntilSufficientSpaceInBuffer(int length) {
}
}
void WebRtcSocket::OffloadFromSignalingThread(Runnable runnable) {
void WebRtcSocketImpl::OffloadFromSignalingThread(Runnable runnable) {
single_thread_executor_.Execute(std::move(runnable));
}
@@ -199,4 +204,4 @@ void WebRtcSocket::OffloadFromSignalingThread(Runnable runnable) {
} // namespace connections
} // namespace nearby
#endif
#endif // NO_WEBRTC
connections/implementation/mediums/webrtc/webrtc_socket_impl.h
+30 -23
View File
@@ -15,23 +15,27 @@
#ifndef CORE_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_IMPL_H_
#define CORE_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_IMPL_H_
#include <cstdint>
#include <string>
#include <memory>
#include "absl/strings/string_view.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
#include "internal/platform/listeners.h"
#include "internal/platform/runnable.h"
#ifndef NO_WEBRTC
#include <cstdint>
#include <memory>
#include <string>
#include "absl/functional/any_invocable.h"
#include "absl/strings/string_view.h"
#include "connections/implementation/mediums/webrtc_socket.h"
#include "internal/platform/atomic_boolean.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/condition_variable.h"
#include "internal/platform/exception.h"
#include "internal/platform/input_stream.h"
#include "internal/platform/listeners.h"
#include "internal/platform/mutex.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/runnable.h"
#include "internal/platform/single_thread_executor.h"
#include "internal/platform/socket.h"
#include "webrtc/api/data_channel_interface.h"
#include "webrtc/api/scoped_refptr.h"
namespace nearby {
namespace connections {
@@ -45,20 +49,22 @@ constexpr int kMaxDataSize = 1 * 1024 * 1024;
//
// Messages are buffered here to prevent the data channel from overflowing,
// which could lead to data loss.
class WebRtcSocket : public Socket, public webrtc::DataChannelObserver {
class WebRtcSocketImpl : public WebRtcSocket,
public webrtc::DataChannelObserver {
public:
WebRtcSocket(
WebRtcSocketImpl(
const std::string& name,
webrtc::scoped_refptr<webrtc::DataChannelInterface> data_channel);
~WebRtcSocket() override;
~WebRtcSocketImpl() override;
WebRtcSocket(const WebRtcSocket& other) = delete;
WebRtcSocket& operator=(const WebRtcSocket& other) = delete;
WebRtcSocketImpl(const WebRtcSocketImpl& other) = delete;
WebRtcSocketImpl& operator=(const WebRtcSocketImpl& other) = delete;
// Overrides for nearby::Socket:
// Overrides for WebRtcSocket:
InputStream& GetInputStream() override;
OutputStream& GetOutputStream() override;
Exception Close() override;
bool IsValid() const override { return true; }
// webrtc::DataChannelObserver:
void OnStateChange() override;
@@ -67,10 +73,10 @@ class WebRtcSocket : public Socket, public webrtc::DataChannelObserver {
// Listener class the gets called when the socket is ready or closed
struct SocketListener {
absl::AnyInvocable<void(WebRtcSocket*)> socket_ready_cb =
DefaultCallback<WebRtcSocket*>();
absl::AnyInvocable<void(WebRtcSocket*)> socket_closed_cb =
DefaultCallback<WebRtcSocket*>();
absl::AnyInvocable<void(WebRtcSocketImpl*)> socket_ready_cb =
DefaultCallback<WebRtcSocketImpl*>();
absl::AnyInvocable<void(WebRtcSocketImpl*)> socket_closed_cb =
DefaultCallback<WebRtcSocketImpl*>();
};
void SetSocketListener(SocketListener&& listener);
@@ -78,7 +84,8 @@ class WebRtcSocket : public Socket, public webrtc::DataChannelObserver {
private:
class OutputStreamImpl : public OutputStream {
public:
explicit OutputStreamImpl(WebRtcSocket* const socket) : socket_(socket) {}
explicit OutputStreamImpl(WebRtcSocketImpl* const socket)
: socket_(socket) {}
~OutputStreamImpl() override = default;
OutputStreamImpl(const OutputStreamImpl& other) = delete;
@@ -91,7 +98,7 @@ class WebRtcSocket : public Socket, public webrtc::DataChannelObserver {
private:
// |this| OutputStreamImpl is owned by |socket_|.
WebRtcSocket* const socket_;
WebRtcSocketImpl* const socket_;
};
void WakeUpWriter();
@@ -124,6 +131,6 @@ class WebRtcSocket : public Socket, public webrtc::DataChannelObserver {
} // namespace connections
} // namespace nearby
#endif
#endif // NO_WEBRTC
#endif // CORE_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_IMPL_H_
connections/implementation/mediums/webrtc/webrtc_socket_impl_test.cc
+14 -14
View File
@@ -64,7 +64,7 @@ TEST(WebRtcSocketTest, ReadFromSocket) {
const char* message = "message";
webrtc::scoped_refptr<MockDataChannel> mock_data_channel(
new MockDataChannel());
WebRtcSocket webrtc_socket(kSocketName, mock_data_channel);
WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel);
webrtc_socket.OnMessage(webrtc::DataBuffer{message});
ExceptionOr<ByteArray> result = webrtc_socket.GetInputStream().Read(7);
@@ -75,7 +75,7 @@ TEST(WebRtcSocketTest, ReadFromSocket) {
TEST(WebRtcSocketTest, ReadMultipleMessages) {
webrtc::scoped_refptr<MockDataChannel> mock_data_channel(
new MockDataChannel());
WebRtcSocket webrtc_socket(kSocketName, mock_data_channel);
WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel);
webrtc_socket.OnMessage(webrtc::DataBuffer{"Me"});
webrtc_socket.OnMessage(webrtc::DataBuffer{"ssa"});
@@ -101,7 +101,7 @@ TEST(WebRtcSocketTest, WriteToSocket) {
absl::string_view kMessage{"Message"};
webrtc::scoped_refptr<MockDataChannel> mock_data_channel(
new MockDataChannel());
WebRtcSocket webrtc_socket(kSocketName, mock_data_channel);
WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel);
EXPECT_CALL(*mock_data_channel, Send(testing::_))
.WillRepeatedly(testing::Return(true));
@@ -112,7 +112,7 @@ TEST(WebRtcSocketTest, SendDataBiggerThanMax) {
std::string kMessage(kMaxDataSize + 1, '0');
webrtc::scoped_refptr<MockDataChannel> mock_data_channel(
new MockDataChannel());
WebRtcSocket webrtc_socket(kSocketName, mock_data_channel);
WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel);
EXPECT_CALL(*mock_data_channel, Send(testing::_)).Times(0);
EXPECT_EQ(webrtc_socket.GetOutputStream().Write(kMessage),
@@ -123,7 +123,7 @@ TEST(WebRtcSocketTest, WriteToDataChannelFails) {
absl::string_view kMessage{"Message"};
webrtc::scoped_refptr<MockDataChannel> mock_data_channel(
new MockDataChannel());
WebRtcSocket webrtc_socket(kSocketName, mock_data_channel);
WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel);
ON_CALL(*mock_data_channel, Send(testing::_))
.WillByDefault(testing::Return(false));
@@ -134,14 +134,14 @@ TEST(WebRtcSocketTest, WriteToDataChannelFails) {
TEST(WebRtcSocketTest, Close) {
webrtc::scoped_refptr<MockDataChannel> mock_data_channel(
new MockDataChannel());
WebRtcSocket webrtc_socket(kSocketName, mock_data_channel);
WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel);
EXPECT_CALL(*mock_data_channel, Close());
int socket_closed_cb_called = 0;
webrtc_socket.SetSocketListener(
{.socket_closed_cb = [&](WebRtcSocket* socket) {
{.socket_closed_cb = [&](WebRtcSocketImpl* socket) {
socket_closed_cb_called++;
}});
webrtc_socket.Close();
@@ -160,7 +160,7 @@ TEST(WebRtcSocketTest, WriteOnClosedChannel) {
absl::string_view kMessage{"Message"};
webrtc::scoped_refptr<MockDataChannel> mock_data_channel(
new MockDataChannel());
WebRtcSocket webrtc_socket(kSocketName, mock_data_channel);
WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel);
webrtc_socket.Close();
EXPECT_CALL(*mock_data_channel, Send(testing::_)).Times(0);
@@ -172,7 +172,7 @@ TEST(WebRtcSocketTest, ReadFromClosedChannel) {
absl::string_view kMessage{"Message"};
webrtc::scoped_refptr<MockDataChannel> mock_data_channel(
new MockDataChannel());
WebRtcSocket webrtc_socket(kSocketName, mock_data_channel);
WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel);
ON_CALL(*mock_data_channel, Send(testing::_))
.WillByDefault(testing::Return(true));
@@ -185,7 +185,7 @@ TEST(WebRtcSocketTest, ReadFromClosedChannel) {
TEST(WebRtcSocketTest, DataChannelCloseEventCleansUp) {
webrtc::scoped_refptr<MockDataChannel> mock_data_channel(
new MockDataChannel());
WebRtcSocket webrtc_socket(kSocketName, mock_data_channel);
WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel);
ON_CALL(*mock_data_channel, state())
.WillByDefault(
@@ -203,12 +203,12 @@ TEST(WebRtcSocketTest, DataChannelCloseEventCleansUp) {
TEST(WebRtcSocketTest, OpenStateTriggersCallback) {
webrtc::scoped_refptr<MockDataChannel> mock_data_channel(
new MockDataChannel());
WebRtcSocket webrtc_socket(kSocketName, mock_data_channel);
WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel);
int socket_ready_cb_called = 0;
webrtc_socket.SetSocketListener(
{.socket_ready_cb = [&](WebRtcSocket* socket) {
{.socket_ready_cb = [&](WebRtcSocketImpl* socket) {
socket_ready_cb_called++;
}});
@@ -224,12 +224,12 @@ TEST(WebRtcSocketTest, OpenStateTriggersCallback) {
TEST(WebRtcSocketTest, CloseStateTriggersCallback) {
webrtc::scoped_refptr<MockDataChannel> mock_data_channel(
new MockDataChannel());
WebRtcSocket webrtc_socket(kSocketName, mock_data_channel);
WebRtcSocketImpl webrtc_socket(kSocketName, mock_data_channel);
int socket_closed_cb_called = 0;
webrtc_socket.SetSocketListener(
{.socket_closed_cb = [&](WebRtcSocket* socket) {
{.socket_closed_cb = [&](WebRtcSocketImpl* socket) {
socket_closed_cb_called++;
}});
connections/implementation/mediums/webrtc_socket.h
+36 -23
View File
@@ -1,4 +1,4 @@
// Copyright 2020 Google LLC
// Copyright 2026 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
@@ -15,44 +15,57 @@
#ifndef CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_H_
#define CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_H_
#include <cstdint>
#include "absl/strings/string_view.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
#ifndef NO_WEBRTC
#include <memory>
#include "connections/implementation/mediums/webrtc/webrtc_socket_impl.h"
#include "internal/platform/input_stream.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/socket.h"
namespace nearby {
namespace connections {
namespace mediums {
class WebRtcSocketWrapper final {
// A base implementation that creates a non-working WebRtcSocket that can be
// used as a placeholder when WebRTC is disabled.
class WebRtcSocket : public Socket {
public:
WebRtcSocketWrapper() = default;
WebRtcSocketWrapper(const WebRtcSocketWrapper&) = default;
WebRtcSocketWrapper& operator=(const WebRtcSocketWrapper&) = default;
explicit WebRtcSocketWrapper(std::unique_ptr<WebRtcSocket> socket)
: impl_(socket.release()) {}
~WebRtcSocketWrapper() = default;
~WebRtcSocket() override = default;
InputStream& GetInputStream() { return impl_->GetInputStream(); }
InputStream& GetInputStream() override { return fake_input_stream_; }
OutputStream& GetOutputStream() { return impl_->GetOutputStream(); }
OutputStream& GetOutputStream() override { return fake_output_stream_; }
Exception Close() { return impl_->Close(); }
Exception Close() override { return {Exception::kSuccess}; }
bool IsValid() const { return impl_ != nullptr; }
WebRtcSocket& GetImpl() { return *impl_; }
virtual bool IsValid() const { return false; }
private:
std::shared_ptr<WebRtcSocket> impl_;
class FakeInputStream : public InputStream {
public:
ExceptionOr<ByteArray> Read(std::int64_t size) override {
return {Exception::kSuccess};
}
Exception Close() override { return {Exception::kSuccess}; }
};
class FakeOutputStream : public OutputStream {
public:
Exception Write(absl::string_view data) override {
return {Exception::kSuccess};
}
Exception Flush() override { return {Exception::kSuccess}; }
Exception Close() override { return {Exception::kSuccess}; }
};
FakeInputStream fake_input_stream_;
FakeOutputStream fake_output_stream_;
};
} // namespace mediums
} // namespace connections
} // namespace nearby
#endif
#endif // CORE_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_H_
#endif // CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_H_
connections/implementation/mediums/webrtc_socket_stub.h
-71
View File
@@ -1,71 +0,0 @@
// Copyright 2020 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#ifndef CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_STUB_H_
#define CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_STUB_H_
#ifdef NO_WEBRTC
#include <memory>
#include "internal/platform/input_stream.h"
#include "internal/platform/output_stream.h"
namespace nearby {
namespace connections {
namespace mediums {
class FakeInputStream : public InputStream {
public:
ExceptionOr<ByteArray> Read(std::int64_t size) {
return {Exception::kSuccess};
}
Exception Close() { return {Exception::kSuccess}; }
};
class FakeOutputStream : public OutputStream {
public:
Exception Write(absl::string_view data) override {
return {Exception::kSuccess};
}
Exception Flush() override { return {Exception::kSuccess}; }
Exception Close() override { return {Exception::kSuccess}; }
};
class WebRtcSocketWrapper final {
public:
WebRtcSocketWrapper() = default;
WebRtcSocketWrapper(const WebRtcSocketWrapper&) = default;
WebRtcSocketWrapper& operator=(const WebRtcSocketWrapper&) = default;
~WebRtcSocketWrapper() = default;
InputStream& GetInputStream() { return fake_input_stream_; }
OutputStream& GetOutputStream() { return fake_output_stream_; }
void Close() {}
bool IsValid() const { return false; }
private:
FakeInputStream fake_input_stream_;
FakeOutputStream fake_output_stream_;
};
} // namespace mediums
} // namespace connections
} // namespace nearby
#endif
#endif // CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_STUB_H_
connections/implementation/mediums/webrtc_stub.cc
+2 -2
View File
@@ -19,7 +19,7 @@
#include <functional>
#include <memory>
#include "connections/implementation/mediums/webrtc_socket_stub.h"
#include "connections/implementation/mediums/webrtc_socket.h"
#include "internal/platform/cancelable_alarm.h"
#include "internal/platform/expected.h"
#include "internal/platform/future.h"
@@ -52,7 +52,7 @@ bool WebRtc::StartAcceptingConnections(const std::string& service_id,
void WebRtc::StopAcceptingConnections(const std::string& service_id) {}
ErrorOr<WebRtcSocketWrapper> WebRtc::Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> WebRtc::Connect(
const std::string& service_id, const WebrtcPeerId& remote_peer_id,
const LocationHint& location_hint, CancellationFlag* cancellation_flag) {
return {Error(OperationResultCode::DETAIL_UNKNOWN)};
connections/implementation/mediums/webrtc_stub.h
+3 -3
View File
@@ -23,7 +23,7 @@
#include <string>
#include "connections/implementation/mediums/webrtc_peer_id.h"
#include "connections/implementation/mediums/webrtc_socket_stub.h"
#include "connections/implementation/mediums/webrtc_socket.h"
#include "connections/implementation/proto/offline_wire_formats.pb.h"
#include "internal/platform/cancellation_flag.h"
#include "internal/platform/expected.h"
@@ -38,7 +38,7 @@ class WebRtc {
public:
// Callback that is invoked when a new connection is accepted.
using AcceptedConnectionCallback =
absl::AnyInvocable<void(WebRtcSocketWrapper socket)>;
absl::AnyInvocable<void(std::shared_ptr<WebRtcSocket> socket)>;
WebRtc();
~WebRtc();
@@ -69,7 +69,7 @@ class WebRtc {
// Initiates a WebRtc connection with peer device identified by |peer_id|
// with internal retry for maximum attempts of kConnectAttemptsLimit.
// Runs on @MainThread.
ErrorOr<WebRtcSocketWrapper> Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> Connect(
const std::string& service_id, const WebrtcPeerId& peer_id,
const location::nearby::connections::LocationHint& location_hint,
CancellationFlag* cancellation_flag);
connections/implementation/mediums/webrtc_test.cc
+77 -68
View File
@@ -61,7 +61,7 @@ class TestWebRtc : public WebRtc {
class WebRtcTest : public ::testing::TestWithParam<WebRtcTestParams> {
protected:
using MockAcceptedCallback = testing::MockFunction<void(
const std::string& service_id, WebRtcSocketWrapper socket)>;
const std::string& service_id, std::shared_ptr<WebRtcSocket> socket)>;
MediumEnvironment& env_{MediumEnvironment::Instance()};
};
@@ -73,7 +73,7 @@ TEST_P(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) {
WebRtcTestParams params = GetParam();
env_.SetFeatureFlags(params.feature_flags);
WebRtc receiver, sender;
WebRtcSocketWrapper receiver_socket;
std::shared_ptr<WebRtcSocket> receiver_socket;
const WebrtcPeerId self_id("self_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
@@ -82,18 +82,19 @@ TEST_P(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) {
receiver.StartAcceptingConnections(
service_id, self_id, location_hint,
[&receiver_socket, connected](const std::string& service_id,
WebRtcSocketWrapper wrapper) mutable {
[&receiver_socket, connected](
const std::string& service_id,
std::shared_ptr<WebRtcSocket> wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
connected.Set(receiver_socket->IsValid());
},
params.non_cellular);
CancellationFlag flag;
ErrorOr<WebRtcSocketWrapper> sender_socket_result = sender.Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> sender_socket_result = sender.Connect(
service_id, self_id, location_hint, &flag, params.non_cellular);
EXPECT_TRUE(sender_socket_result.has_value());
EXPECT_TRUE(sender_socket_result.value().IsValid());
EXPECT_TRUE(sender_socket_result.value()->IsValid());
ExceptionOr<bool> devices_connected = connected.Get();
ASSERT_TRUE(devices_connected.ok());
@@ -102,9 +103,9 @@ TEST_P(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) {
// Only shuts down signaling channel.
receiver.StopAcceptingConnections(service_id);
sender_socket_result.value().GetOutputStream().Write(message);
sender_socket_result.value()->GetOutputStream().Write(message);
ExceptionOr<ByteArray> received_msg =
receiver_socket.GetInputStream().Read(/*size=*/32);
receiver_socket->GetInputStream().Read(/*size=*/32);
ASSERT_TRUE(received_msg.ok());
EXPECT_EQ(message, received_msg.result().AsStringView());
env_.Stop();
@@ -115,7 +116,7 @@ TEST_P(WebRtcTest, CanCancelConnect) {
WebRtcTestParams params = GetParam();
env_.SetFeatureFlags(params.feature_flags);
WebRtc receiver, sender;
WebRtcSocketWrapper receiver_socket;
std::shared_ptr<WebRtcSocket> receiver_socket;
const WebrtcPeerId self_id("self_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
@@ -124,32 +125,33 @@ TEST_P(WebRtcTest, CanCancelConnect) {
receiver.StartAcceptingConnections(
service_id, self_id, location_hint,
[&receiver_socket, connected](const std::string& service_id,
WebRtcSocketWrapper wrapper) mutable {
[&receiver_socket, connected](
const std::string& service_id,
std::shared_ptr<WebRtcSocket> wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
connected.Set(receiver_socket->IsValid());
},
params.non_cellular);
CancellationFlag flag(true);
ErrorOr<WebRtcSocketWrapper> sender_socket_result = sender.Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> sender_socket_result = sender.Connect(
service_id, self_id, location_hint, &flag, params.non_cellular);
// If FeatureFlag is disabled, Cancelled is false as no-op.
if (!params.feature_flags.enable_cancellation_flag) {
EXPECT_TRUE(sender_socket_result.has_value());
EXPECT_TRUE(sender_socket_result.value().IsValid());
EXPECT_TRUE(sender_socket_result.value()->IsValid());
ExceptionOr<bool> devices_connected = connected.Get();
ASSERT_TRUE(devices_connected.ok());
EXPECT_TRUE(devices_connected.result());
sender_socket_result.value().GetOutputStream().Write(message);
sender_socket_result.value()->GetOutputStream().Write(message);
ExceptionOr<ByteArray> received_msg =
receiver_socket.GetInputStream().Read(/*size=*/32);
receiver_socket->GetInputStream().Read(/*size=*/32);
ASSERT_TRUE(received_msg.ok());
EXPECT_EQ(message, received_msg.result().AsStringView());
receiver_socket.Close();
receiver_socket->Close();
} else {
EXPECT_TRUE(sender_socket_result.has_error());
}
@@ -200,7 +202,7 @@ TEST_P(WebRtcTest, Connect_NoPeer) {
ASSERT_TRUE(webrtc.IsAvailable());
CancellationFlag flag;
ErrorOr<WebRtcSocketWrapper> wrapper_1_result = webrtc.Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> wrapper_1_result = webrtc.Connect(
service_id, peer_id, location_hint, &flag, params.non_cellular);
EXPECT_TRUE(wrapper_1_result.has_error());
@@ -225,7 +227,7 @@ TEST_P(WebRtcTest, StartAcceptingConnection_ThenConnect) {
service_id, self_id, location_hint,
mock_accepted_callback_.AsStdFunction(), params.non_cellular));
CancellationFlag flag;
ErrorOr<WebRtcSocketWrapper> wrapper_result =
ErrorOr<std::shared_ptr<WebRtcSocket>> wrapper_result =
webrtc.Connect(service_id, WebrtcPeerId("random_peer_id"), location_hint,
&flag, params.non_cellular);
EXPECT_TRUE(webrtc.IsAcceptingConnections(service_id));
@@ -262,7 +264,7 @@ TEST_P(WebRtcTest, StartAndStopAcceptingConnections) {
TEST_P(WebRtcTest, ConnectTwice) {
env_.Start({.webrtc_enabled = true});
WebRtc receiver, sender, device_c;
WebRtcSocketWrapper receiver_socket;
std::shared_ptr<WebRtcSocket> receiver_socket;
WebRtcTestParams params = GetParam();
const WebrtcPeerId self_id("self_id"), other_id("other_id");
const std::string service_id("NearbySharing");
@@ -272,45 +274,47 @@ TEST_P(WebRtcTest, ConnectTwice) {
receiver.StartAcceptingConnections(
service_id, self_id, location_hint,
[&receiver_socket, connected](const std::string& service_id,
WebRtcSocketWrapper wrapper) mutable {
[&receiver_socket, connected](
const std::string& service_id,
std::shared_ptr<WebRtcSocket> wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
connected.Set(receiver_socket->IsValid());
},
params.non_cellular);
device_c.StartAcceptingConnections(
service_id, other_id, location_hint,
[](const std::string& service_id, WebRtcSocketWrapper wrapper) {},
[](const std::string& service_id, std::shared_ptr<WebRtcSocket> wrapper) {
},
params.non_cellular);
CancellationFlag flag;
ErrorOr<WebRtcSocketWrapper> sender_socket_result = sender.Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> sender_socket_result = sender.Connect(
service_id, self_id, location_hint, &flag, params.non_cellular);
EXPECT_TRUE(sender_socket_result.has_value());
EXPECT_TRUE(sender_socket_result.value().IsValid());
EXPECT_TRUE(sender_socket_result.value()->IsValid());
ExceptionOr<bool> devices_connected = connected.Get();
ASSERT_TRUE(devices_connected.ok());
EXPECT_TRUE(devices_connected.result());
ErrorOr<WebRtcSocketWrapper> socket_result = sender.Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> socket_result = sender.Connect(
service_id, other_id, location_hint, &flag, params.non_cellular);
EXPECT_TRUE(socket_result.has_value());
EXPECT_TRUE(socket_result.value().IsValid());
socket_result.value().Close();
EXPECT_TRUE(socket_result.value()->IsValid());
socket_result.value()->Close();
EXPECT_TRUE(receiver_socket.IsValid());
EXPECT_TRUE(receiver_socket->IsValid());
EXPECT_TRUE(sender_socket_result.has_value());
EXPECT_TRUE(sender_socket_result.value().IsValid());
EXPECT_TRUE(sender_socket_result.value()->IsValid());
sender_socket_result.value().GetOutputStream().Write(message);
sender_socket_result.value()->GetOutputStream().Write(message);
ExceptionOr<ByteArray> received_msg =
receiver_socket.GetInputStream().Read(/*size=*/32);
receiver_socket->GetInputStream().Read(/*size=*/32);
ASSERT_TRUE(received_msg.ok());
EXPECT_EQ(message, received_msg.result().AsStringView());
receiver_socket.Close();
receiver_socket->Close();
env_.Stop();
}
@@ -319,7 +323,7 @@ TEST_P(WebRtcTest, ConnectTwice) {
TEST_P(WebRtcTest, ConnectBothDevicesAndAbort) {
env_.Start({.webrtc_enabled = true});
WebRtc receiver, sender;
WebRtcSocketWrapper receiver_socket, sender_socket;
std::shared_ptr<WebRtcSocket> receiver_socket, sender_socket;
WebRtcTestParams params = GetParam();
const WebrtcPeerId self_id("self_id");
const std::string service_id("NearbySharing");
@@ -328,24 +332,25 @@ TEST_P(WebRtcTest, ConnectBothDevicesAndAbort) {
receiver.StartAcceptingConnections(
service_id, self_id, location_hint,
[&receiver_socket, connected](const std::string& service_id,
WebRtcSocketWrapper wrapper) mutable {
[&receiver_socket, connected](
const std::string& service_id,
std::shared_ptr<WebRtcSocket> wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
connected.Set(receiver_socket->IsValid());
},
params.non_cellular);
CancellationFlag flag;
ErrorOr<WebRtcSocketWrapper> sender_socket_result = sender.Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> sender_socket_result = sender.Connect(
service_id, self_id, location_hint, &flag, params.non_cellular);
EXPECT_TRUE(sender_socket_result.has_value());
EXPECT_TRUE(sender_socket_result.value().IsValid());
EXPECT_TRUE(sender_socket_result.value()->IsValid());
ExceptionOr<bool> devices_connected = connected.Get();
ASSERT_TRUE(devices_connected.ok());
EXPECT_TRUE(devices_connected.result());
receiver_socket.Close();
receiver_socket->Close();
env_.Stop();
}
@@ -354,7 +359,7 @@ TEST_P(WebRtcTest, ConnectBothDevicesAndAbort) {
TEST_P(WebRtcTest, ConnectBothDevicesAndSendData) {
env_.Start({.webrtc_enabled = true});
WebRtc receiver, sender;
WebRtcSocketWrapper receiver_socket;
std::shared_ptr<WebRtcSocket> receiver_socket;
WebRtcTestParams params = GetParam();
const WebrtcPeerId self_id("self_id");
const std::string service_id("NearbySharing");
@@ -364,30 +369,31 @@ TEST_P(WebRtcTest, ConnectBothDevicesAndSendData) {
receiver.StartAcceptingConnections(
service_id, self_id, location_hint,
[&receiver_socket, connected](const std::string& service_id,
WebRtcSocketWrapper wrapper) mutable {
[&receiver_socket, connected](
const std::string& service_id,
std::shared_ptr<WebRtcSocket> wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
connected.Set(receiver_socket->IsValid());
},
params.non_cellular);
CancellationFlag flag;
ErrorOr<WebRtcSocketWrapper> sender_socket_result = sender.Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> sender_socket_result = sender.Connect(
service_id, self_id, location_hint, &flag, params.non_cellular);
EXPECT_TRUE(sender_socket_result.has_value());
EXPECT_TRUE(sender_socket_result.value().IsValid());
EXPECT_TRUE(sender_socket_result.value()->IsValid());
ExceptionOr<bool> devices_connected = connected.Get();
ASSERT_TRUE(devices_connected.ok());
EXPECT_TRUE(devices_connected.result());
sender_socket_result.value().GetOutputStream().Write(message);
sender_socket_result.value()->GetOutputStream().Write(message);
ExceptionOr<ByteArray> received_msg =
receiver_socket.GetInputStream().Read(/*size=*/32);
receiver_socket->GetInputStream().Read(/*size=*/32);
ASSERT_TRUE(received_msg.ok());
EXPECT_EQ(message, received_msg.result().AsStringView());
receiver_socket.Close();
receiver_socket->Close();
env_.Stop();
}
@@ -405,7 +411,7 @@ TEST_P(WebRtcTest, Connect_NullPeerConnection) {
ASSERT_TRUE(webrtc.IsAvailable());
CancellationFlag flag;
ErrorOr<WebRtcSocketWrapper> wrapper_result =
ErrorOr<std::shared_ptr<WebRtcSocket>> wrapper_result =
webrtc.Connect(service_id, WebrtcPeerId("random_peer_id"), location_hint,
&flag, params.non_cellular);
EXPECT_TRUE(wrapper_result.has_error());
@@ -456,7 +462,7 @@ TEST_P(WebRtcTest, CancelDuringConnect) {
.enable_cancellation_flag = true,
});
WebRtcSocketWrapper receiver_socket, sender_socket;
std::shared_ptr<WebRtcSocket> receiver_socket, sender_socket;
const WebrtcPeerId self_id("self_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
@@ -479,14 +485,15 @@ TEST_P(WebRtcTest, CancelDuringConnect) {
receiver->StartAcceptingConnections(
service_id, self_id, location_hint,
[&receiver_socket, connected](const std::string& service_id,
WebRtcSocketWrapper wrapper) mutable {
[&receiver_socket, connected](
const std::string& service_id,
std::shared_ptr<WebRtcSocket> wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
connected.Set(receiver_socket->IsValid());
},
params.non_cellular);
ErrorOr<WebRtcSocketWrapper> sender_socket_result = sender->Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> sender_socket_result = sender->Connect(
service_id, self_id, location_hint, &sender_flag, params.non_cellular);
// Since the flag was cancelled during the initial `AttemptToConnect`, except
@@ -512,7 +519,7 @@ TEST_P(WebRtcTest, CancelBeforeConnect) {
.enable_cancellation_flag = true,
});
WebRtcSocketWrapper receiver_socket;
std::shared_ptr<WebRtcSocket> receiver_socket;
const WebrtcPeerId self_id("self_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
@@ -528,14 +535,15 @@ TEST_P(WebRtcTest, CancelBeforeConnect) {
receiver->StartAcceptingConnections(
service_id, self_id, location_hint,
[&receiver_socket, connected](const std::string& service_id,
WebRtcSocketWrapper wrapper) mutable {
[&receiver_socket, connected](
const std::string& service_id,
std::shared_ptr<WebRtcSocket> wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
connected.Set(receiver_socket->IsValid());
},
params.non_cellular);
ErrorOr<WebRtcSocketWrapper> sender_socket_result = sender->Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> sender_socket_result = sender->Connect(
service_id, self_id, location_hint, &sender_flag, params.non_cellular);
// Expect an invalid socket from stopping during the first attempt to connect,
@@ -558,7 +566,7 @@ TEST_P(WebRtcTest, CancelDuringConnect_MultipleConnect) {
.enable_cancellation_flag = true,
});
WebRtcSocketWrapper receiver_socket;
std::shared_ptr<WebRtcSocket> receiver_socket;
const WebrtcPeerId self_id("self_id");
const std::string ns_service_id("NearbySharing");
const std::string ph_service_id("PhoneHub");
@@ -576,18 +584,19 @@ TEST_P(WebRtcTest, CancelDuringConnect_MultipleConnect) {
receiver->StartAcceptingConnections(
ns_service_id, self_id, location_hint,
[&receiver_socket, connected](const std::string& ns_service_id,
WebRtcSocketWrapper wrapper) mutable {
[&receiver_socket, connected](
const std::string& ns_service_id,
std::shared_ptr<WebRtcSocket> wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
connected.Set(receiver_socket->IsValid());
},
params.non_cellular);
// Simulate a successful connect for the endpoint of NearbySharing.
ErrorOr<WebRtcSocketWrapper> sender_socket_result = sender->Connect(
ErrorOr<std::shared_ptr<WebRtcSocket>> sender_socket_result = sender->Connect(
ns_service_id, self_id, location_hint, &flag, params.non_cellular);
EXPECT_TRUE(sender_socket_result.has_value());
EXPECT_TRUE(sender_socket_result.value().IsValid());
EXPECT_TRUE(sender_socket_result.value()->IsValid());
// Calls `CancellationFlag::Cancel` during a call to `GetSignalingMessenger`
// to simulate the cancellation occuring during an `AttemptToConnect` for the
connections/implementation/p2p_cluster_pcp_handler.h
-1
View File
@@ -56,7 +56,6 @@
#include "internal/platform/nsd_service_info.h"
#include "internal/platform/wifi_lan.h"
#ifdef NO_WEBRTC
#include "connections/implementation/mediums/webrtc_socket_stub.h"
#include "connections/implementation/mediums/webrtc_stub.h"
#else
#include "connections/implementation/mediums/webrtc.h"
connections/implementation/webrtc_bwu_handler.cc
+11 -10
View File
@@ -60,10 +60,10 @@ LocationHint BuildLocationHint(const std::string& location) {
} // namespace
WebrtcBwuHandler::WebrtcIncomingSocket::WebrtcIncomingSocket(
const std::string& name, mediums::WebRtcSocketWrapper socket)
: name_(name), socket_(socket) {}
const std::string& name, std::shared_ptr<mediums::WebRtcSocket> socket)
: name_(name), socket_(std::move(socket)) {}
void WebrtcBwuHandler::WebrtcIncomingSocket::Close() { socket_.Close(); }
void WebrtcBwuHandler::WebrtcIncomingSocket::Close() { socket_->Close(); }
std::string WebrtcBwuHandler::WebrtcIncomingSocket::ToString() { return name_; }
@@ -93,9 +93,10 @@ WebrtcBwuHandler::CreateUpgradedEndpointChannel(
<< peer_id.GetId() << ", location hint "
<< location_hint.location();
ErrorOr<mediums::WebRtcSocketWrapper> socket_result = webrtc_.Connect(
service_id, peer_id, location_hint,
client->GetCancellationFlag(endpoint_id), client->GetWebRtcNonCellular());
ErrorOr<std::shared_ptr<mediums::WebRtcSocket>> socket_result =
webrtc_.Connect(service_id, peer_id, location_hint,
client->GetCancellationFlag(endpoint_id),
client->GetWebRtcNonCellular());
if (socket_result.has_error()) {
LOG(ERROR) << "WebRtcBwuHandler failed to connect to remote peer ("
<< peer_id.GetId() << ") on endpoint " << endpoint_id
@@ -111,7 +112,7 @@ WebrtcBwuHandler::CreateUpgradedEndpointChannel(
auto channel = std::make_unique<WebRtcEndpointChannel>(
service_id, /*channel_name=*/service_id, socket_result.value());
if (channel == nullptr) {
socket_result.value().Close();
socket_result.value()->Close();
LOG(ERROR) << "WebRtcBwuHandler failed to create new EndpointChannel for "
"outgoing socket, aborting upgrade.";
return {Error(
@@ -164,11 +165,11 @@ std::string WebrtcBwuHandler::HandleInitializeUpgradedMediumForEndpoint(
// for this socket.
void WebrtcBwuHandler::OnIncomingWebrtcConnection(
ClientProxy* client, const std::string& upgrade_service_id,
mediums::WebRtcSocketWrapper socket) {
std::shared_ptr<mediums::WebRtcSocket> socket) {
auto channel = std::make_unique<WebRtcEndpointChannel>(
upgrade_service_id, /*channel_name=*/upgrade_service_id, socket);
auto webrtc_socket =
std::make_unique<WebrtcIncomingSocket>(upgrade_service_id, socket);
auto webrtc_socket = std::make_unique<WebrtcIncomingSocket>(
upgrade_service_id, std::move(socket));
std::unique_ptr<IncomingSocketConnection> connection(
new IncomingSocketConnection{std::move(webrtc_socket),
std::move(channel)});
connections/implementation/webrtc_bwu_handler.h
+6 -6
View File
@@ -44,15 +44,15 @@ class WebrtcBwuHandler : public BaseBwuHandler {
private:
class WebrtcIncomingSocket : public BwuHandler::IncomingSocket {
public:
explicit WebrtcIncomingSocket(const std::string& name,
mediums::WebRtcSocketWrapper socket);
explicit WebrtcIncomingSocket(
const std::string& name, std::shared_ptr<mediums::WebRtcSocket> socket);
std::string ToString() override;
void Close() override;
private:
std::string name_;
mediums::WebRtcSocketWrapper socket_;
std::shared_ptr<mediums::WebRtcSocket> socket_;
};
// BwuHandler implementation:
@@ -74,9 +74,9 @@ class WebrtcBwuHandler : public BaseBwuHandler {
void HandleRevertInitiatorStateForService(
const std::string& upgrade_service_id) final;
void OnIncomingWebrtcConnection(ClientProxy* client,
const std::string& upgrade_service_id,
mediums::WebRtcSocketWrapper socket);
void OnIncomingWebrtcConnection(
ClientProxy* client, const std::string& upgrade_service_id,
std::shared_ptr<mediums::WebRtcSocket> socket);
Mediums& mediums_;
mediums::WebRtc& webrtc_{mediums_.GetWebRtc()};
connections/implementation/webrtc_bwu_handler_stub.cc
+3 -3
View File
@@ -36,8 +36,8 @@ using ::location::nearby::proto::connections::OperationResultCode;
} // namespace
WebrtcBwuHandler::WebrtcIncomingSocket::WebrtcIncomingSocket(
const std::string& name, mediums::WebRtcSocketWrapper socket)
: name_(name), socket_(socket) {}
const std::string& name, std::shared_ptr<mediums::WebRtcSocket> socket)
: name_(name), socket_(std::move(socket)) {}
void WebrtcBwuHandler::WebrtcIncomingSocket::Close() {}
@@ -76,7 +76,7 @@ std::string WebrtcBwuHandler::HandleInitializeUpgradedMediumForEndpoint(
// for this socket.
void WebrtcBwuHandler::OnIncomingWebrtcConnection(
ClientProxy* client, const std::string& upgrade_service_id,
mediums::WebRtcSocketWrapper socket) {}
std::shared_ptr<mediums::WebRtcSocket> socket) {}
} // namespace connections
} // namespace nearby
connections/implementation/webrtc_bwu_handler_stub.h
+6 -10
View File
@@ -23,11 +23,7 @@
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/mediums/mediums.h"
#ifdef NO_WEBRTC
#include "connections/implementation/mediums/webrtc_socket_stub.h"
#else
#include "connections/implementation/mediums/webrtc_socket.h"
#endif
#include "internal/platform/expected.h"
namespace nearby {
@@ -44,15 +40,15 @@ class WebrtcBwuHandler : public BaseBwuHandler {
private:
class WebrtcIncomingSocket : public BwuHandler::IncomingSocket {
public:
explicit WebrtcIncomingSocket(const std::string& name,
mediums::WebRtcSocketWrapper socket);
explicit WebrtcIncomingSocket(
const std::string& name, std::shared_ptr<mediums::WebRtcSocket> socket);
std::string ToString() override;
void Close() override;
private:
std::string name_;
mediums::WebRtcSocketWrapper socket_;
std::shared_ptr<mediums::WebRtcSocket> socket_;
};
// BwuHandler implementation:
@@ -74,9 +70,9 @@ class WebrtcBwuHandler : public BaseBwuHandler {
void HandleRevertInitiatorStateForService(
const std::string& upgrade_service_id) final;
void OnIncomingWebrtcConnection(ClientProxy* client,
const std::string& upgrade_service_id,
mediums::WebRtcSocketWrapper socket);
void OnIncomingWebrtcConnection(
ClientProxy* client, const std::string& upgrade_service_id,
std::shared_ptr<mediums::WebRtcSocket> socket);
Mediums& mediums_;
mediums::WebRtc& webrtc_{mediums_.GetWebRtc()};
connections/implementation/webrtc_endpoint_channel.cc
+9 -4
View File
@@ -14,16 +14,21 @@
#include "connections/implementation/webrtc_endpoint_channel.h"
#include <memory>
#include <string>
#include <utility>
#include "connections/implementation/base_endpoint_channel.h"
#include "connections/implementation/mediums/webrtc_socket.h"
namespace nearby {
namespace connections {
WebRtcEndpointChannel::WebRtcEndpointChannel(
const std::string& service_id, const std::string& channel_name,
mediums::WebRtcSocketWrapper socket)
: BaseEndpointChannel(service_id, channel_name, &socket.GetInputStream(),
&socket.GetOutputStream()),
std::shared_ptr<mediums::WebRtcSocket> socket)
: BaseEndpointChannel(service_id, channel_name, &socket->GetInputStream(),
&socket->GetOutputStream()),
webrtc_socket_(std::move(socket)) {}
location::nearby::proto::connections::Medium WebRtcEndpointChannel::GetMedium()
@@ -31,7 +36,7 @@ location::nearby::proto::connections::Medium WebRtcEndpointChannel::GetMedium()
return location::nearby::proto::connections::Medium::WEB_RTC;
}
void WebRtcEndpointChannel::CloseImpl() { webrtc_socket_.Close(); }
void WebRtcEndpointChannel::CloseImpl() { webrtc_socket_->Close(); }
} // namespace connections
} // namespace nearby
connections/implementation/webrtc_endpoint_channel.h
+3 -6
View File
@@ -15,14 +15,11 @@
#ifndef CORE_INTERNAL_WEBRTC_ENDPOINT_CHANNEL_H_
#define CORE_INTERNAL_WEBRTC_ENDPOINT_CHANNEL_H_
#include <memory>
#include <string>
#include "connections/implementation/base_endpoint_channel.h"
#ifdef NO_WEBRTC
#include "connections/implementation/mediums/webrtc_socket_stub.h"
#else
#include "connections/implementation/mediums/webrtc_socket.h"
#endif
namespace nearby {
namespace connections {
@@ -31,14 +28,14 @@ class WebRtcEndpointChannel final : public BaseEndpointChannel {
public:
WebRtcEndpointChannel(const std::string& service_id,
const std::string& channel_name,
mediums::WebRtcSocketWrapper webrtc_socket);
std::shared_ptr<mediums::WebRtcSocket> socket);
location::nearby::proto::connections::Medium GetMedium() const override;
private:
void CloseImpl() override;
mediums::WebRtcSocketWrapper webrtc_socket_;
std::shared_ptr<mediums::WebRtcSocket> webrtc_socket_;
};
} // namespace connections