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

Delete unused code.

Browse Source 
PiperOrigin-RevId: 914475976
This commit is contained in:
Francis Tsui
2026-05-12 14:22:56 -07:00
committed by Copybara-Service
parent 24c6dd9015
commit 3dae0f119e
13 changed files with 0 additions and 2902 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Package.swift
-4
View File
@@ -321,7 +321,6 @@ let package = Package(
"connections/implementation/flags/BUILD",
"connections/implementation/mediums/advertisements/BUILD",
"connections/implementation/mediums/ble/BUILD",
"connections/implementation/mediums/multiplex/BUILD",
"connections/implementation/mediums/BUILD",
"connections/implementation/BUILD",
"connections/implementation/fuzzers",
@@ -392,9 +391,6 @@ let package = Package(
"connections/implementation/mediums/ble/discovered_peripheral_tracker_test.cc",
"connections/implementation/mediums/ble/instant_on_lost_advertisement_test.cc",
"connections/implementation/mediums/ble/instant_on_lost_manager_test.cc",
"connections/implementation/mediums/multiplex/multiplex_frames_test.cc",
"connections/implementation/mediums/multiplex/multiplex_socket_test.cc",
"connections/implementation/mediums/multiplex/multiplex_output_stream_test.cc",
"connections/implementation/mediums/webrtc_peer_id_test.cc",
"connections/implementation/mediums/wifi_lan_test.cc",
"connections/implementation/mediums/bluetooth_classic_test.cc",
connections/implementation/BUILD
-1
View File
@@ -145,7 +145,6 @@ cc_library(
"//connections:__pkg__",
"//connections:partners",
"//connections/implementation/fuzzers:__pkg__",
"//connections/implementation/mediums/multiplex:__pkg__",
"//sharing:__subpackages__",
],
deps = [
connections/implementation/mediums/BUILD
-1
View File
@@ -135,7 +135,6 @@ cc_library(
"//connections/implementation:__pkg__",
"//connections/implementation/mediums/advertisements:__pkg__",
"//connections/implementation/mediums/ble:__subpackages__",
"//connections/implementation/mediums/multiplex:__pkg__",
"//internal/platform/implementation/windows:__pkg__",
],
deps = [
connections/implementation/mediums/multiplex/BUILD
-73
View File
@@ -1,73 +0,0 @@
load("@rules_cc//cc:cc_library.bzl", "cc_library")
load("@rules_cc//cc:cc_test.bzl", "cc_test")
# Copyright 2024 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.
licenses(["notice"])
cc_library(
name = "multiplex",
srcs = [
"multiplex_frames.cc",
"multiplex_output_stream.cc",
"multiplex_socket.cc",
],
hdrs = [
"multiplex_frames.h",
"multiplex_output_stream.h",
"multiplex_socket.h",
],
visibility = [
"//connections/implementation:__subpackages__",
],
deps = [
"//connections:core_types",
"//connections/implementation/mediums:utils",
"//internal/platform:base",
"//internal/platform:logging",
"//internal/platform:types",
"//proto:connections_enums_cc_proto",
"//proto/mediums:multiplex_frames_cc_proto",
"@com_google_absl//absl/base:core_headers",
"@com_google_absl//absl/container:flat_hash_map",
"@com_google_absl//absl/functional:any_invocable",
"@com_google_absl//absl/strings",
"@com_google_absl//absl/time",
],
)
cc_test(
name = "multiplex_test",
srcs = [
"multiplex_frames_test.cc",
"multiplex_output_stream_test.cc",
"multiplex_socket_test.cc",
],
tags = ["notap"],
deps = [
":multiplex",
"//connections/implementation:internal",
"//internal/platform:base",
"//internal/platform:logging",
"//internal/platform:types",
"//internal/platform/implementation/g3", # buildcleaner: keep
"//proto:connections_enums_cc_proto",
"//proto/mediums:multiplex_frames_cc_proto",
"@com_github_protobuf_matchers//protobuf-matchers",
"@com_google_absl//absl/container:flat_hash_map",
"@com_google_absl//absl/strings:string_view",
"@com_google_absl//absl/time",
"@com_google_googletest//:gtest_main",
],
)
connections/implementation/mediums/multiplex/multiplex_frames.cc
-215
View File
@@ -1,215 +0,0 @@
// Copyright 2024 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.
#include "connections/implementation/mediums/multiplex/multiplex_frames.h"
#include <string>
#include <utility>
#include "absl/strings/string_view.h"
#include "connections/implementation/mediums/utils.h"
#include "internal/platform/base64_utils.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
#include "internal/platform/logging.h"
namespace nearby {
namespace connections {
namespace mediums {
namespace multiplex {
using ::location::nearby::mediums::ConnectionResponseFrame;
using ::location::nearby::mediums::MultiplexControlFrame;
using ::location::nearby::mediums::MultiplexFrame;
ByteArray GenerateServiceIdHash(const std::string& service_id) {
return Utils::Sha256Hash(service_id, kServiceIdHashLength);
}
ByteArray GenerateServiceIdHashWithSalt(const std::string& service_id,
std::string salt) {
if (salt.empty()) {
return GenerateServiceIdHash(service_id);
}
return Utils::Sha256Hash(service_id + salt, kServiceIdHashLength);
}
std::string GenerateServiceIdHashKey(const ByteArray& service_id_hash) {
return Base64Utils::Encode(service_id_hash);
}
std::string GenerateServiceIdHashKey(const std::string& service_id) {
return GenerateServiceIdHashKey(GenerateServiceIdHash(service_id));
}
std::string GenerateServiceIdHashKeyWithSalt(const std::string& service_id,
std::string salt) {
return GenerateServiceIdHashKey(
GenerateServiceIdHashWithSalt(service_id, salt));
}
ByteArray ToBytes(MultiplexFrame&& frame) {
ByteArray bytes(frame.ByteSizeLong());
frame.SerializeToArray(bytes.data(), bytes.size());
return bytes;
}
ByteArray ForConnectionRequest(const std::string& service_id,
const std::string& service_id_hash_salt) {
MultiplexFrame frame;
frame.set_frame_type(MultiplexFrame::CONTROL_FRAME);
auto* header = frame.mutable_header();
header->set_salted_service_id_hash(std::string(
GenerateServiceIdHashWithSalt(service_id, service_id_hash_salt)));
header->set_service_id_hash_salt(service_id_hash_salt);
auto* control_frame = frame.mutable_control_frame();
control_frame->set_control_frame_type(
MultiplexControlFrame::CONNECTION_REQUEST);
return ToBytes(std::move(frame));
}
ByteArray ForConnectionResponse(
const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt,
ConnectionResponseFrame::ConnectionResponseCode response_code) {
MultiplexFrame frame;
frame.set_frame_type(MultiplexFrame::CONTROL_FRAME);
auto* header = frame.mutable_header();
header->set_salted_service_id_hash(std::string(salted_service_id_hash));
header->set_service_id_hash_salt(service_id_hash_salt);
auto* control_frame = frame.mutable_control_frame();
control_frame->set_control_frame_type(
MultiplexControlFrame::CONNECTION_RESPONSE);
auto* response_frame = control_frame->mutable_connection_response_frame();
response_frame->set_connection_response_code(response_code);
return ToBytes(std::move(frame));
}
ByteArray ForDisconnection(const std::string& service_id,
const std::string& service_id_hash_salt) {
MultiplexFrame frame;
frame.set_frame_type(MultiplexFrame::CONTROL_FRAME);
auto* header = frame.mutable_header();
header->set_salted_service_id_hash(std::string(
GenerateServiceIdHashWithSalt(service_id, service_id_hash_salt)));
header->set_service_id_hash_salt(service_id_hash_salt);
auto* control_frame = frame.mutable_control_frame();
control_frame->set_control_frame_type(MultiplexControlFrame::DISCONNECTION);
return ToBytes(std::move(frame));
}
ByteArray ForData(const std::string& service_id,
const std::string& service_id_hash_salt,
bool should_pass_salt, absl::string_view data) {
MultiplexFrame frame;
frame.set_frame_type(MultiplexFrame::DATA_FRAME);
auto* header = frame.mutable_header();
header->set_salted_service_id_hash(std::string(
GenerateServiceIdHashWithSalt(service_id, service_id_hash_salt)));
if (should_pass_salt) {
header->set_service_id_hash_salt(service_id_hash_salt);
}
auto* data_frame = frame.mutable_data_frame();
data_frame->set_data(data);
return ToBytes(std::move(frame));
}
ExceptionOr<MultiplexFrame> FromBytes(const ByteArray& multiplex_frame_bytes) {
MultiplexFrame frame;
if (frame.ParseFromString(std::string(multiplex_frame_bytes))) {
if (!IsValid(frame)) {
return ExceptionOr<MultiplexFrame>(Exception::kInvalidProtocolBuffer);
}
return ExceptionOr<MultiplexFrame>(std::move(frame));
} else {
return ExceptionOr<MultiplexFrame>(Exception::kInvalidProtocolBuffer);
}
}
bool IsControlFrame(MultiplexFrame::MultiplexFrameType frame_type) {
return frame_type == MultiplexFrame::CONTROL_FRAME;
}
bool IsDataFrame(MultiplexFrame::MultiplexFrameType frame_type) {
return frame_type == MultiplexFrame::DATA_FRAME;
}
bool IsValid(const MultiplexFrame& frame) {
switch (frame.frame_type()) {
case MultiplexFrame::CONTROL_FRAME:
return IsValidControlFrame(frame);
case MultiplexFrame::DATA_FRAME:
return IsValidDataFrame(frame);
default:
return false;
}
}
bool IsValidControlFrame(const MultiplexFrame& frame) {
if (!frame.has_control_frame()) {
return false;
}
switch (frame.control_frame().control_frame_type()) {
case MultiplexControlFrame::CONNECTION_REQUEST:
case MultiplexControlFrame::CONNECTION_RESPONSE:
case MultiplexControlFrame::DISCONNECTION:
if (frame.header().salted_service_id_hash().size() ==
kServiceIdHashLength) {
return true;
}
break;
default:
break;
}
return false;
}
bool IsValidDataFrame(const MultiplexFrame& frame) {
return frame.has_data_frame() &&
frame.header().salted_service_id_hash().size() == kServiceIdHashLength;
}
bool IsMultiplexFrame(const ByteArray& data) {
ExceptionOr<MultiplexFrame> frame = FromBytes(data);
if (!frame.ok()) {
return false;
} else {
LOG(INFO) << "Checked data is a multiplex frame. Is Control ? "
<< frame.result().has_control_frame() << ", is data ? "
<< frame.result().has_data_frame();
return true;
}
}
} // namespace multiplex
} // namespace mediums
} // namespace connections
} // namespace nearby
connections/implementation/mediums/multiplex/multiplex_frames.h
-112
View File
@@ -1,112 +0,0 @@
// Copyright 2024 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_MULTIPLEX_MULTIPLEX_FRAMES_H_
#define CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_FRAMES_H_
#include <string>
#include "absl/strings/string_view.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
#include "proto/mediums/multiplex_frames.pb.h"
namespace nearby {
namespace connections {
namespace mediums {
namespace multiplex {
constexpr int kServiceIdHashLength = 4;
// Serialize/Deserialize MultiplexFrame messages.
// Parses incoming MultiplexFrame message.
// Returns MultiplexFrame if parser was able to understand it, or
// Exception::kInvalidProtocolBuffer, if parser failed.
// Generates a service ID hash bytes with {@link
// MultiplexFrames#SERVICE_ID_HASH_LENGTH}.
ByteArray GenerateServiceIdHash(const std::string& service_id);
// Generates a service ID hash bytes with salt and {@link
// MultiplexFrames#SERVICE_ID_HASH_LENGTH}.
ByteArray GenerateServiceIdHashWithSalt(const std::string& service_id,
std::string salt);
// Converts the service Id hash bytes to a Base64 encoded string to be used as a
// {@code Map} key.
std::string GenerateServiceIdHashKey(const ByteArray& service_id_hash);
// Generates a service ID hash bytes with {@link
// MultiplexFrames#SERVICE_ID_HASH_LENGTH} and converts to a Base64 encoded
// string to be used as a {@code Map} key.
std::string GenerateServiceIdHashKey(const std::string& service_id);
// Generates a service ID hash bytes with salt and {@link
// MultiplexFrames#SERVICE_ID_HASH_LENGTH} and converts to a Base64 encoded
// string to be used as a { @code Map } key.
std::string GenerateServiceIdHashKeyWithSalt(const std::string& service_id,
std::string salt);
// Build a MultiplexFrame Connection Request frame Bytes stream.
// @param service_id The service ID of the connection.
// @param service_id_hash_salt The salt used to generate the service ID hash.
ByteArray ForConnectionRequest(const std::string& service_id,
const std::string& service_id_hash_salt);
// Build a MultiplexFrame Connection Response frame Bytes stream.
// @param salted_service_id_hash The salted service ID hash.
// @param service_id_hash_salt The salt used to generate the service ID hash.
// @param response_code The response code of the connection.
ByteArray ForConnectionResponse(
const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt,
location::nearby::mediums::ConnectionResponseFrame::ConnectionResponseCode
response_code);
// Build a MultiplexFrame Disconnection frame Bytes stream.
// @param service_id The service ID of the connection.
// @param service_id_hash_salt The salt used to generate the service ID hash.
ByteArray ForDisconnection(const std::string& service_id,
const std::string& service_id_hash_salt);
// Build a MultiplexFrame Data frame Bytes stream.
// @param service_id The service ID of the connection.
// @param service_id_hash_salt The salt used to generate the service ID hash.
// @param should_pass_salt Whether to pass the salt in the data frame.
// @param data The data to send.
ByteArray ForData(const std::string& service_id,
const std::string& service_id_hash_salt,
bool should_pass_salt, absl::string_view data);
ExceptionOr<location::nearby::mediums::MultiplexFrame> FromBytes(
const ByteArray& multiplex_frame_bytes);
bool IsControlFrame(
location::nearby::mediums::MultiplexFrame::MultiplexFrameType frame_type);
bool IsDataFrame(
location::nearby::mediums::MultiplexFrame::MultiplexFrameType frame_type);
bool IsValid(const location::nearby::mediums::MultiplexFrame& frame);
bool IsValidControlFrame(
const location::nearby::mediums::MultiplexFrame& frame);
bool IsValidDataFrame(const location::nearby::mediums::MultiplexFrame& frame);
bool IsMultiplexFrame(const ByteArray& data);
} // namespace multiplex
} // namespace mediums
} // namespace connections
} // namespace nearby
#endif // CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_FRAMES_H_
connections/implementation/mediums/multiplex/multiplex_frames_test.cc
-170
View File
@@ -1,170 +0,0 @@
// Copyright 2024 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.
#include "connections/implementation/mediums/multiplex/multiplex_frames.h"
#include <string>
#include <utility>
#include "gtest/gtest.h"
#include "absl/strings/string_view.h"
#include "internal/platform/byte_array.h"
namespace nearby {
namespace connections {
namespace mediums {
namespace multiplex {
using ::location::nearby::mediums::MultiplexFrame;
using ::location::nearby::mediums::MultiplexControlFrame;
using ::location::nearby::mediums::ConnectionResponseFrame;
constexpr absl::string_view kServiceId_1 = "serviceId_1";
constexpr absl::string_view kServiceId_2 = "serviceId_2";
TEST(MultiplexFrameTest, FrameValidation) {
const ByteArray data("abcdefghijklmnopqrstuvwxyz");
MultiplexFrame frame;
EXPECT_FALSE(IsValid(frame));
frame.set_frame_type(MultiplexFrame::CONTROL_FRAME);
EXPECT_FALSE(IsValidControlFrame(frame));
auto* control_frame = frame.mutable_control_frame();
control_frame->set_control_frame_type(
MultiplexControlFrame::UNKNOWN_CONTROL_FRAME_TYPE);
EXPECT_FALSE(IsValidControlFrame(frame));
auto* header = frame.mutable_header();
header->set_salted_service_id_hash(std::string(
GenerateServiceIdHashWithSalt(std::string(kServiceId_1), "1234")));
control_frame->set_control_frame_type(
MultiplexControlFrame::CONNECTION_REQUEST);
EXPECT_TRUE(IsValidControlFrame(frame));
EXPECT_TRUE(IsValid(frame));
control_frame->set_control_frame_type(
MultiplexControlFrame::CONNECTION_RESPONSE);
EXPECT_TRUE(IsValidControlFrame(frame));
EXPECT_TRUE(IsValid(frame));
control_frame->set_control_frame_type(
MultiplexControlFrame::DISCONNECTION);
EXPECT_TRUE(IsValidControlFrame(frame));
EXPECT_TRUE(IsValid(frame));
EXPECT_FALSE(IsValidDataFrame(frame));
frame.set_frame_type(MultiplexFrame::DATA_FRAME);
auto* data_frame = frame.mutable_data_frame();
data_frame->set_data(std::string(std::move(data)));
EXPECT_TRUE(IsValidDataFrame(frame));
EXPECT_TRUE(IsValid(frame));
frame.set_frame_type(MultiplexFrame::UNKNOWN_FRAME_TYPE);
EXPECT_FALSE(IsValid(frame));
frame.set_frame_type(MultiplexFrame::DATA_FRAME);
auto serialized_bytes = ByteArray(frame.SerializeAsString());
EXPECT_TRUE(IsMultiplexFrame(std::move(serialized_bytes)));
EXPECT_TRUE(IsControlFrame(MultiplexFrame::CONTROL_FRAME));
EXPECT_FALSE(IsControlFrame(MultiplexFrame::DATA_FRAME));
EXPECT_TRUE(IsDataFrame(MultiplexFrame::DATA_FRAME));
EXPECT_FALSE(IsDataFrame(MultiplexFrame::UNKNOWN_FRAME_TYPE));
}
TEST(MultiplexFrameTest, HashValidtion) {
auto service_id_hash_1 = GenerateServiceIdHash(std::string(kServiceId_1));
EXPECT_EQ(service_id_hash_1.size(), kServiceIdHashLength);
auto service_id_hash_2 = GenerateServiceIdHash(std::string(kServiceId_2));
EXPECT_NE(service_id_hash_1, service_id_hash_2);
auto hash_key_1 = GenerateServiceIdHashKey(service_id_hash_1);
auto hash_key_2 = GenerateServiceIdHashKey(service_id_hash_2);
EXPECT_NE(hash_key_1, hash_key_2);
auto service_id_hash_with_salt_1 =
GenerateServiceIdHashWithSalt(std::string(kServiceId_1), "1234");
EXPECT_EQ(service_id_hash_with_salt_1.size(), kServiceIdHashLength);
auto service_id_hash_with_salt_2 =
GenerateServiceIdHashWithSalt(std::string(kServiceId_2), "1234");
EXPECT_NE(service_id_hash_with_salt_1, service_id_hash_with_salt_2);
service_id_hash_with_salt_2 =
GenerateServiceIdHashWithSalt(std::string(kServiceId_1), "abcd");
EXPECT_NE(service_id_hash_with_salt_1, service_id_hash_with_salt_2);
auto hash_key_with_salt_1 =
GenerateServiceIdHashKeyWithSalt(std::string(kServiceId_1), "1234");
auto hash_key_with_salt_2 =
GenerateServiceIdHashKeyWithSalt(std::string(kServiceId_2), "1234");
EXPECT_NE(hash_key_with_salt_1, hash_key_with_salt_2);
}
TEST(MultiplexFrameTest, CanGenerateConnectionRequest) {
ByteArray bytes = ForConnectionRequest(std::string(kServiceId_1), "1234");
auto request = FromBytes(bytes);
ASSERT_TRUE(request.ok());
auto frame = request.result();
EXPECT_EQ(frame.control_frame().control_frame_type(),
MultiplexControlFrame::CONNECTION_REQUEST);
EXPECT_EQ(frame.header().salted_service_id_hash(),
std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1),
"1234")));
}
TEST(MultiplexFrameTest, CanGenerateConnectionRespons) {
auto service_id_hash_with_salt_2 =
GenerateServiceIdHashWithSalt(std::string(kServiceId_2), "1234");
ByteArray bytes =
ForConnectionResponse(service_id_hash_with_salt_2, "1234",
ConnectionResponseFrame::CONNECTION_ACCEPTED);
auto response = FromBytes(bytes);
ASSERT_TRUE(response.ok());
auto frame = response.result();
EXPECT_EQ(frame.control_frame().control_frame_type(),
MultiplexControlFrame::CONNECTION_RESPONSE);
EXPECT_EQ(frame.header().salted_service_id_hash(),
std::string(service_id_hash_with_salt_2));
EXPECT_EQ(frame.control_frame()
.connection_response_frame()
.connection_response_code(),
ConnectionResponseFrame::CONNECTION_ACCEPTED);
}
TEST(MultiplexFrameTest, CanGenerateDisconnection) {
ByteArray bytes = ForDisconnection(std::string(kServiceId_1), "1234");
auto response = FromBytes(bytes);
ASSERT_TRUE(response.ok());
auto frame = response.result();
EXPECT_EQ(frame.control_frame().control_frame_type(),
MultiplexControlFrame::DISCONNECTION);
EXPECT_EQ(frame.header().salted_service_id_hash(),
std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1),
"1234")));
}
TEST(MultiplexFrameTest, CanGenerateData) {
absl::string_view data = "abcdefghijklmnopqrstuvwxyz";
ByteArray bytes =
ForData(std::string(kServiceId_1), "1234", true, data);
auto response = FromBytes(bytes);
ASSERT_TRUE(response.ok());
auto frame = response.result();
EXPECT_EQ(frame.frame_type(), MultiplexFrame::DATA_FRAME);
EXPECT_EQ(frame.header().salted_service_id_hash(),
std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1),
"1234")));
EXPECT_EQ(frame.data_frame().data(),
std::string("abcdefghijklmnopqrstuvwxyz"));
}
} // namespace multiplex
} // namespace mediums
} // namespace connections
} // namespace nearby
connections/implementation/mediums/multiplex/multiplex_output_stream.cc
-360
View File
@@ -1,360 +0,0 @@
// Copyright 2024 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.
#include "connections/implementation/mediums/multiplex/multiplex_output_stream.h"
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include "absl/strings/string_view.h"
#include "absl/time/time.h"
#include "connections/implementation/mediums/multiplex/multiplex_frames.h"
#include "internal/platform/array_blocking_queue.h"
#include "internal/platform/atomic_boolean.h"
#include "internal/platform/base64_utils.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/future.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex_lock.h"
#include "internal/platform/output_stream.h"
namespace nearby {
namespace connections {
namespace mediums {
namespace multiplex {
namespace {
using ::location::nearby::mediums::ConnectionResponseFrame;
constexpr absl::string_view kFakeSalt = "RECEIVER_CONDIMENT";
} // namespace
// Implementation for class MultiplexOutputStream
MultiplexOutputStream::MultiplexOutputStream(OutputStream* physical_writer,
AtomicBoolean& is_enabled)
: is_enabled_(is_enabled),
physical_writer_(physical_writer),
multiplex_writer_{physical_writer} {}
Exception MultiplexOutputStream::WaitForResult(const std::string& method_name,
Future<bool>* future) {
if (!future) {
LOG(INFO) << "No future to wait for; return with error.";
return {Exception::kFailed};
}
LOG(INFO) << "Waiting for future to complete: " << method_name;
ExceptionOr<bool> result =
future->Get(FeatureFlags::GetInstance()
.GetFlags()
.mediums_frame_write_timeout_millis);
if (!result.ok()) {
LOG(INFO) << "Future:[" << method_name
<< "] completed with exception:" << result.exception();
return {Exception::kFailed};
}
if (result.result()) {
LOG(INFO) << "Future:[" << method_name << "] completed with success.";
return {Exception::kSuccess};
}
LOG(INFO) << "Future:[" << method_name << "] completed with failure.";
return {Exception::kFailed};
}
bool MultiplexOutputStream::WriteConnectionRequestFrame(
const std::string& service_id, const std::string& service_id_hash_salt) {
if (!is_enabled_.Get()) {
return false;
}
Future<bool> future;
multiplex_writer_.EnqueueToSend(
&future, ForConnectionRequest(service_id, service_id_hash_salt),
"MultiplexFrame::CONNECTION_REQUEST");
if (WaitForResult("MultiplexFrame::CONNECTION_REQUEST", &future).Ok())
return true;
return false;
}
bool MultiplexOutputStream::WriteConnectionResponseFrame(
const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt,
ConnectionResponseFrame::ConnectionResponseCode response_code) {
if (!is_enabled_.Get()) {
return false;
}
Future<bool> future;
multiplex_writer_.EnqueueToSend(
&future,
ForConnectionResponse(salted_service_id_hash, service_id_hash_salt,
response_code),
"MultiplexFrame::CONNECTION_RESPONSE");
if (WaitForResult("MultiplexFrame::CONNECTION_RESPONSE", &future).Ok())
return true;
return false;
}
bool MultiplexOutputStream::Close(const std::string& service_id) {
auto item = virtual_output_streams_.find(service_id);
if (item == virtual_output_streams_.end()) {
LOG(INFO) << "Don't need to close VirtualOutputStream(" << service_id
<< ") because it's already gone.";
return false;
}
item->second->Close();
if (is_enabled_.Get()) {
Future<bool> future;
multiplex_writer_.EnqueueToSend(
&future,
ForDisconnection(service_id, item->second->GetServiceIdHashSalt()),
"MultiplexFrame::DISCONNECTION");
WaitForResult("MultiplexFrame::DISCONNECTION", &future);
}
virtual_output_streams_.erase(service_id);
if (virtual_output_streams_.empty()) {
physical_writer_->Close();
multiplex_writer_.Close();
}
return true;
}
void MultiplexOutputStream::CloseAll() {
for (auto& [service_id, virtual_output_stream] : virtual_output_streams_) {
if (is_enabled_.Get()) {
Future<bool> future;
multiplex_writer_.EnqueueToSend(
&future,
ForDisconnection(service_id,
virtual_output_stream->GetServiceIdHashSalt()),
"MultiplexFrame::DISCONNECTION");
WaitForResult("MultiplexFrame::DISCONNECTION", &future);
}
virtual_output_stream->Close();
}
virtual_output_streams_.clear();
physical_writer_->Close();
multiplex_writer_.Close();
}
OutputStream*
MultiplexOutputStream::CreateVirtualOutputStreamForFirstVirtualSocket(
const std::string& service_id, const std::string& service_id_hash_salt) {
return virtual_output_streams_
.emplace(service_id,
std::make_unique<VirtualOutputStream>(
service_id, service_id_hash_salt, physical_writer_,
multiplex_writer_,
VirtualOutputStreamType::kFirstVirtualSocket, *this))
.first->second.get();
}
OutputStream* MultiplexOutputStream::CreateVirtualOutputStream(
const std::string& service_id, const std::string& service_id_hash_salt) {
return virtual_output_streams_
.emplace(service_id,
std::make_unique<VirtualOutputStream>(
service_id, service_id_hash_salt, physical_writer_,
multiplex_writer_,
VirtualOutputStreamType::kNormalVirtualSocket, *this))
.first->second.get();
}
std::string MultiplexOutputStream::GetServiceIdHashSalt(
const std::string& service_id) {
auto item = virtual_output_streams_.find(service_id);
if (item != virtual_output_streams_.end()) {
return item->second->GetServiceIdHashSalt();
}
return {};
}
void MultiplexOutputStream::Shutdown() {
physical_writer_->Close();
multiplex_writer_.Close();
}
// Implementation for class MultiplexOutputStream::MultiplexWriter
MultiplexOutputStream::MultiplexWriter::MultiplexWriter(
OutputStream* physical_writer)
: physical_writer_(physical_writer) {}
MultiplexOutputStream::MultiplexWriter::~MultiplexWriter() {
Close();
physical_writer_ = nullptr;
}
void MultiplexOutputStream::MultiplexWriter::EnqueueToSend(
Future<bool>* future, const ByteArray& data,
const std::string& frame_name) {
MutexLock lock(&writing_mutex_);
data_queue_.Put(EnqueuedFrame(future, data));
if (is_writing_) {
return;
}
is_writing_ = true;
is_writing_cond_.Notify();
if (!is_write_loop_running_) {
is_write_loop_running_ = true;
writer_thread_.Execute("Start writing", [this] { StartWriting(); });
}
}
void MultiplexOutputStream::MultiplexWriter::StartWriting() {
LOG(INFO) << "Writing loop started.";
while (true) {
auto enqueued_frame = data_queue_.TryTake();
if (enqueued_frame != std::nullopt) {
Write(enqueued_frame.value());
continue;
}
{
MutexLock lock(&writing_mutex_);
if (data_queue_.Empty() && is_writing_ && !is_closed_) {
is_writing_ = false;
LOG(INFO) << "Waiting for data_queue_ has data.";
Exception wait_succeeded = is_writing_cond_.Wait();
if (!wait_succeeded.Ok()) {
LOG(WARNING) << "Failure waiting to wait: " << wait_succeeded.value;
return;
}
}
if (is_closed_) {
LOG(INFO) << "Notify to close_writing_thread";
MutexLock lock(&close_writing_thread_mutex_);
close_writing_thread_cond_.Notify();
break;
}
}
}
LOG(INFO) << "Writing loop stopped.";
}
void MultiplexOutputStream::MultiplexWriter::Write(
EnqueuedFrame& enqueued_frame) {
MutexLock lock(&writer_mutex_);
if (!Base64Utils::WriteInt(physical_writer_, enqueued_frame.data_.size())
.Ok()) {
enqueued_frame.future_->SetException({Exception::kIo});
return;
};
if (!physical_writer_->Write(enqueued_frame.data_.AsStringView()).Ok()) {
enqueued_frame.future_->SetException({Exception::kIo});
return;
};
if (!physical_writer_->Flush().Ok()) {
enqueued_frame.future_->SetException({Exception::kIo});
return;
};
enqueued_frame.future_->Set(true);
}
void MultiplexOutputStream::MultiplexWriter::Close() {
if (is_closed_) {
LOG(INFO) << "MultiplexWriter is already closed.";
return;
}
LOG(INFO) << "Stop writing loop and Shutdown writer thread.";
{
MutexLock lock(&writing_mutex_);
is_closed_ = true;
if (!is_write_loop_running_) {
writer_thread_.Shutdown();
return;
}
is_write_loop_running_ = false;
is_writing_cond_.Notify();
}
LOG(INFO) << "Wait to close_writing_thread";
{
MutexLock lock(&close_writing_thread_mutex_);
close_writing_thread_cond_.Wait(absl::Milliseconds(20));
LOG(INFO) << "Shutdown writer thread.";
writer_thread_.Shutdown();
}
}
MultiplexOutputStream::VirtualOutputStream::VirtualOutputStream(
std::string service_id, std::string service_id_hash_salt,
OutputStream* physical_writer, MultiplexWriter& multiplex_writer,
VirtualOutputStreamType virtual_output_stream_type,
MultiplexOutputStream& multiplex_output_stream)
: service_id_(service_id),
service_id_hash_salt_(service_id_hash_salt),
physical_writer_(physical_writer),
multiplex_writer_(multiplex_writer),
virtual_output_stream_type_(virtual_output_stream_type),
multiplex_output_stream_(multiplex_output_stream) {}
Exception MultiplexOutputStream::VirtualOutputStream::Write(
absl::string_view data) {
if (is_closed_.Get()) {
LOG(WARNING) << "Failed to write data because the VirtualOutputStream for "
<< service_id_ << " closed";
return {Exception::kIo};
}
if (multiplex_output_stream_.is_enabled_.Get()) {
bool should_pass_salt = false;
if (IsFirstVirtualOutputStream()) {
if (!first_frame_sent_for_first_virtual_output_stream_) {
first_frame_sent_for_first_virtual_output_stream_ = true;
should_pass_salt = true;
}
// Fixes b/290724590, b/290983930 which can't get the correct socket
// from the virtualSockets map. NS receiver side will pass 2
// DATA_FRAMEs continuously to the remote sender side but originally
// impl will only consider the 1st one. Add below fix to handle 2nd
// frame which the salt is still fake one and change shouldPassSalt to
// true to let the remote handle correctly.
if ((service_id_hash_salt_ == kFakeSalt) && !should_pass_salt) {
should_pass_salt = true;
LOG(INFO) << "service_idHashSalt is still a fake one and "
"not changed yet; continue to pass salt.";
}
}
ByteArray data_frame =
ForData(service_id_, service_id_hash_salt_, should_pass_salt, data);
Future<bool> future;
multiplex_writer_.EnqueueToSend(&future, data_frame,
"MultiplexFrame::DATA_FRAME");
return multiplex_output_stream_.WaitForResult("MultiplexFrame::DATA_FRAME",
&future);
} else {
if (!physical_writer_->Write(data).Ok()) {
return {Exception::kIo};
};
if (!physical_writer_->Flush().Ok()) {
return {Exception::kIo};
};
}
return {Exception::kSuccess};
}
Exception MultiplexOutputStream::VirtualOutputStream::Flush() {
return {Exception::kSuccess};
}
Exception MultiplexOutputStream::VirtualOutputStream::Close() {
LOG(INFO) << "MultiplexOutputStream::VirtualOutputStream::Close";
is_closed_.Set(true);
return {Exception::kSuccess};
}
} // namespace multiplex
} // namespace mediums
} // namespace connections
} // namespace nearby
connections/implementation/mediums/multiplex/multiplex_output_stream.h
-209
View File
@@ -1,209 +0,0 @@
// Copyright 2024 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_MULTIPLEX_MULTIPLEX_OUTPUT_STREAM_H_
#define CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_OUTPUT_STREAM_H_
#include <memory>
#include <string>
#include "absl/base/thread_annotations.h"
#include "absl/container/flat_hash_map.h"
#include "absl/strings/string_view.h"
#include "internal/platform/array_blocking_queue.h"
#include "internal/platform/atomic_boolean.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/future.h"
#include "internal/platform/mutex.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/single_thread_executor.h"
#include "proto/mediums/multiplex_frames.pb.h"
namespace nearby {
namespace connections {
namespace mediums {
namespace multiplex {
/**
* A helper class to send out the {@code MultiplexControlFrame} and the outgoing
* data from clients. It schedules control and data frames with priority below
*
* <p>{@link MultiplexControlFrameType#CONNECTION_REQUEST} and {@link
* MultiplexControlFrameType#CONNECTION_RESPONSE} have the highest priority
*
* <p>All {@link MultiplexDataFrame} has the medium priority. If there's
* multiple clients send data at the same time, should poll every client's
* outgoing data in sequence. For example, client A and B send data at the same
* time, the outgoing data sequence should like A-Frame-1, B-Frame-1, A-Frame-2,
* B-Frame-2,...
*
* <p>{@link MultiplexControlFrameType#DISCONNECTION} has the same priority with
* {@link MultiplexDataFrame} because the disconnect should not make the already
* enqueued data failed to send out, so put it in the same priority queue with
* the MultiplexDataFrame.
*/
class MultiplexOutputStream {
public:
enum class VirtualOutputStreamType {
// The type of virtual socket established for the physical socket is
// created.
kFirstVirtualSocket = 0,
// The others except FIRST_VIRTUAL_SCOKET type.
kNormalVirtualSocket = 1,
};
MultiplexOutputStream(OutputStream* physical_writer,
AtomicBoolean& is_enabled);
~MultiplexOutputStream() = default;
// Writes the connection request frame to the physical output stream.
bool WriteConnectionRequestFrame(const std::string& service_id,
const std::string& service_id_hash_salt);
// Writes the connection response frame to the physical output stream.
bool WriteConnectionResponseFrame(
const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt,
::location::nearby::mediums::ConnectionResponseFrame::
ConnectionResponseCode response_code);
// Closes the virtual output stream.
bool Close(const std::string& service_id);
// Closes all virtual output streams.
void CloseAll();
// Waits for the result of the future.
Exception WaitForResult(const std::string& method_name, Future<bool>* future);
// Creates the virtual output stream for the first virtual socket.
OutputStream* CreateVirtualOutputStreamForFirstVirtualSocket(
const std::string& service_id, const std::string& service_id_hash_salt);
// Creates the virtual output stream.
OutputStream* CreateVirtualOutputStream(
const std::string& service_id, const std::string& service_id_hash_salt);
// Gets the service id hash salt.
std::string GetServiceIdHashSalt(const std::string& service_id);
// Shuts down the multiplex output stream.
void Shutdown();
class EnqueuedFrame {
public:
EnqueuedFrame(Future<bool>* future, ByteArray data)
: future_(future), data_(data) {}
~EnqueuedFrame() = default;
Future<bool>* future_;
ByteArray data_;
};
class MultiplexWriter {
public:
explicit MultiplexWriter(OutputStream* physical_writer);
~MultiplexWriter();
// Enqueues the frame to be sent out.
void EnqueueToSend(Future<bool>* future, const ByteArray& data,
const std::string& frame_name);
// Closes the writer.
void Close();
private:
// Starts the writer thread.
void StartWriting();
// Writes the enqueued frame.
void Write(EnqueuedFrame& enqueued_frame);
Mutex writer_mutex_;
OutputStream* physical_writer_ ABSL_PT_GUARDED_BY(writer_mutex_);
ArrayBlockingQueue<EnqueuedFrame> data_queue_{
FeatureFlags::GetInstance()
.GetFlags()
.multiplex_socket_middle_priority_queue_capacity};
mutable Mutex writing_mutex_;
ConditionVariable is_writing_cond_{&writing_mutex_};
bool is_writing_ ABSL_GUARDED_BY(writing_mutex_) = false;
bool is_closed_ = false;
mutable Mutex close_writing_thread_mutex_;
ConditionVariable close_writing_thread_cond_{&close_writing_thread_mutex_};
// The single thread to write all enqueued frames.
SingleThreadExecutor writer_thread_;
bool is_write_loop_running_ = false;
};
class VirtualOutputStream : public OutputStream {
public:
VirtualOutputStream(std::string service_id,
std::string service_id_hash_salt,
OutputStream* physical_writer,
MultiplexWriter& multiplex_writer,
VirtualOutputStreamType virtual_output_stream_type,
MultiplexOutputStream& multiplex_output_stream);
~VirtualOutputStream() override = default;
// Returns true if the virtual output stream is the first virtual output
// stream.
bool IsFirstVirtualOutputStream() {
return virtual_output_stream_type_ ==
VirtualOutputStreamType::kFirstVirtualSocket;
}
// Returns the service id hash salt.
std::string GetServiceIdHashSalt() { return service_id_hash_salt_; }
// Sets the service id hash salt.
void SetserviceIdHashSalt(std::string service_id_hash_salt) {
service_id_hash_salt_ = service_id_hash_salt;
}
// Writes the data to the physical output stream.
Exception Write(absl::string_view data) override;
// Flushes the physical output stream.
Exception Flush() override;
// Closes the virtual output stream.
Exception Close() override;
private:
AtomicBoolean is_closed_{false};
std::string service_id_;
std::string service_id_hash_salt_;
OutputStream* physical_writer_;
MultiplexWriter& multiplex_writer_;
VirtualOutputStreamType virtual_output_stream_type_;
bool first_frame_sent_for_first_virtual_output_stream_ = false;
MultiplexOutputStream& multiplex_output_stream_;
};
private:
AtomicBoolean& is_enabled_;
OutputStream* physical_writer_;
absl::flat_hash_map<std::string, std::unique_ptr<VirtualOutputStream>>
virtual_output_streams_;
MultiplexWriter multiplex_writer_;
};
} // namespace multiplex
} // namespace mediums
} // namespace connections
} // namespace nearby
#endif // CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_OUTPUT_STREAM_H_
connections/implementation/mediums/multiplex/multiplex_output_stream_test.cc
-253
View File
@@ -1,253 +0,0 @@
// Copyright 2024 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.
#include "connections/implementation/mediums/multiplex/multiplex_output_stream.h"
#include <cstdint>
#include <memory>
#include <string>
#include <utility>
#include "gtest/gtest.h"
#include "absl/strings/string_view.h"
#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "connections/implementation/mediums/multiplex/multiplex_frames.h"
#include "internal/platform/atomic_boolean.h"
#include "internal/platform/base64_utils.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
#include "internal/platform/input_stream.h"
#include "internal/platform/logging.h"
#include "internal/platform/multi_thread_executor.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/pipe.h"
#include "proto/mediums/multiplex_frames.pb.h"
namespace nearby {
namespace connections {
namespace mediums {
namespace multiplex {
constexpr absl::string_view kServiceId_1 = "serviceId_1";
constexpr absl::string_view kServiceId_2 = "serviceId_2";
constexpr absl::string_view kNoSalt = "";
constexpr absl::string_view kSalt_1 = "DNFG";
constexpr absl::string_view kSalt_2 = "YFRT";
using ::location::nearby::mediums::ConnectionResponseFrame;
using ::location::nearby::mediums::MultiplexControlFrame;
using ::location::nearby::mediums::MultiplexFrame;
class MultiplexOutputStreamTest : public ::testing::Test {
protected:
ExceptionOr<MultiplexFrame> ReadFrame() {
ExceptionOr<std::int32_t> read_int = Base64Utils::ReadInt(reader_.get());
if (!read_int.ok()) return read_int.GetException();
if (read_int.result() <= 0) return {Exception::kFailed};
ExceptionOr<ByteArray> received_data =
reader_->ReadExactly(read_int.result());
if (!received_data.ok()) return received_data.GetException();
auto bytes = std::move(received_data.result());
return FromBytes(bytes);
}
AtomicBoolean enabled_{true};
std::pair<std::unique_ptr<InputStream>, std::unique_ptr<OutputStream>> pipe_ =
CreatePipe();
std::unique_ptr<InputStream> reader_ = std::move(pipe_.first);
std::unique_ptr<OutputStream> writer_ = std::move(pipe_.second);
std::unique_ptr<MultiplexOutputStream> multiplex_output_stream_;
};
TEST_F(MultiplexOutputStreamTest, SendConnectionRequestFrame) {
multiplex_output_stream_ =
std::make_unique<MultiplexOutputStream>(writer_.get(), enabled_);
EXPECT_TRUE(multiplex_output_stream_->WriteConnectionRequestFrame(
std::string(kServiceId_1), std::string(kNoSalt)));
auto request = ReadFrame();
ASSERT_TRUE(request.ok());
auto frame = request.result();
EXPECT_EQ(frame.control_frame().control_frame_type(),
MultiplexControlFrame::CONNECTION_REQUEST);
EXPECT_EQ(frame.header().salted_service_id_hash(),
std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1),
std::string(kNoSalt))));
multiplex_output_stream_->Shutdown();
}
TEST_F(MultiplexOutputStreamTest, SendConnectionRequestFrameDisabled) {
enabled_.Set(false);
multiplex_output_stream_ =
std::make_unique<MultiplexOutputStream>(writer_.get(), enabled_);
EXPECT_FALSE(multiplex_output_stream_->WriteConnectionRequestFrame(
std::string(kServiceId_1), std::string(kNoSalt)));
multiplex_output_stream_->Shutdown();
}
TEST_F(MultiplexOutputStreamTest, SendConnectionResponseFrame) {
multiplex_output_stream_ =
std::make_unique<MultiplexOutputStream>(writer_.get(), enabled_);
EXPECT_TRUE(multiplex_output_stream_->WriteConnectionResponseFrame(
GenerateServiceIdHash(std::string(kServiceId_1)), std::string(kNoSalt),
ConnectionResponseFrame::CONNECTION_ACCEPTED));
auto response = ReadFrame();
ASSERT_TRUE(response.ok());
auto frame = response.result();
EXPECT_EQ(frame.control_frame().control_frame_type(),
MultiplexControlFrame::CONNECTION_RESPONSE);
EXPECT_EQ(frame.header().salted_service_id_hash(),
std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1),
std::string(kNoSalt))));
EXPECT_EQ(frame.control_frame()
.connection_response_frame()
.connection_response_code(),
ConnectionResponseFrame::CONNECTION_ACCEPTED);
multiplex_output_stream_->Shutdown();
}
TEST_F(MultiplexOutputStreamTest, SendConnectionResponseFrameDisabled) {
enabled_.Set(false);
multiplex_output_stream_ =
std::make_unique<MultiplexOutputStream>(writer_.get(), enabled_);
EXPECT_FALSE(multiplex_output_stream_->WriteConnectionResponseFrame(
GenerateServiceIdHash(std::string(kServiceId_1)), std::string(kNoSalt),
ConnectionResponseFrame::CONNECTION_ACCEPTED));
multiplex_output_stream_->Shutdown();
}
TEST_F(MultiplexOutputStreamTest, CloseVirtualStreamFailed) {
multiplex_output_stream_ =
std::make_unique<MultiplexOutputStream>(writer_.get(), enabled_);
EXPECT_FALSE(multiplex_output_stream_->Close(std::string(kServiceId_1)));
multiplex_output_stream_->Shutdown();
}
TEST_F(MultiplexOutputStreamTest, CloseVirtualStreamSuccess) {
multiplex_output_stream_ =
std::make_unique<MultiplexOutputStream>(writer_.get(), enabled_);
EXPECT_FALSE(multiplex_output_stream_->Close(std::string(kServiceId_1)));
multiplex_output_stream_->CreateVirtualOutputStream(std::string(kServiceId_1),
std::string(kNoSalt));
EXPECT_TRUE(multiplex_output_stream_->Close(std::string(kServiceId_1)));
auto request = ReadFrame();
ASSERT_TRUE(request.ok());
auto frame = request.result();
EXPECT_EQ(frame.control_frame().control_frame_type(),
MultiplexControlFrame::DISCONNECTION);
EXPECT_EQ(frame.header().salted_service_id_hash(),
std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1),
std::string(kNoSalt))));
multiplex_output_stream_->Shutdown();
}
TEST_F(MultiplexOutputStreamTest, CreateVirtualStream_SendData) {
multiplex_output_stream_ =
std::make_unique<MultiplexOutputStream>(writer_.get(), enabled_);
auto virtual_output_stream =
multiplex_output_stream_->CreateVirtualOutputStream(
std::string(kServiceId_1), std::string(kSalt_1));
absl::string_view data = "abcdefghijklmnopqrstuvwxyz";
virtual_output_stream->Write(data);
virtual_output_stream->Flush();
auto frame_data = ReadFrame();
ASSERT_TRUE(frame_data.ok());
auto frame = frame_data.result();
EXPECT_EQ(frame.frame_type(), MultiplexFrame::DATA_FRAME);
EXPECT_EQ(frame.header().salted_service_id_hash(),
std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1),
std::string(kSalt_1))));
EXPECT_EQ(frame.data_frame().data(), std::string(data));
multiplex_output_stream_->Shutdown();
}
TEST_F(MultiplexOutputStreamTest, CreateTwoVirtualStreams_SendData) {
multiplex_output_stream_ =
std::make_unique<MultiplexOutputStream>(writer_.get(), enabled_);
auto virtual_output_stream_1 =
multiplex_output_stream_->CreateVirtualOutputStreamForFirstVirtualSocket(
std::string(kServiceId_1), std::string(kSalt_1));
auto virtual_output_stream_2 =
multiplex_output_stream_->CreateVirtualOutputStreamForFirstVirtualSocket(
std::string(kServiceId_2), std::string(kSalt_2));
absl::string_view data_1("abcdefg");
absl::string_view data_2("hijklmn");
MultiThreadExecutor executor(2);
CountDownLatch latch(2);
executor.Execute([&virtual_output_stream_1, &latch, &data_1]() {
absl::SleepFor(absl::Milliseconds(100));
virtual_output_stream_1->Write(data_1);
virtual_output_stream_1->Flush();
latch.CountDown();
});
executor.Execute([&virtual_output_stream_2, &latch, &data_2]() {
virtual_output_stream_2->Write(data_2);
virtual_output_stream_2->Flush();
latch.CountDown();
});
EXPECT_TRUE(latch.Await(absl::Milliseconds(5000)).result());
auto frame_data = ReadFrame();
ASSERT_TRUE(frame_data.ok());
auto frame = frame_data.result();
EXPECT_EQ(frame.frame_type(), MultiplexFrame::DATA_FRAME);
bool first_frame_is_data_1 = true;
if (frame.header().salted_service_id_hash() ==
std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1),
std::string(kSalt_1)))) {
EXPECT_EQ(frame.data_frame().data(), std::string(data_1));
LOG(INFO) << "Read first virtual stream frame first.";
} else {
EXPECT_EQ(frame.header().salted_service_id_hash(),
std::string(GenerateServiceIdHashWithSalt(
std::string(kServiceId_2), std::string(kSalt_2))));
EXPECT_EQ(frame.data_frame().data(), std::string(data_2));
first_frame_is_data_1 = false;
LOG(INFO) << "Read second virtual stream frame first.";
}
frame_data = ReadFrame();
ASSERT_TRUE(frame_data.ok());
frame = frame_data.result();
EXPECT_EQ(frame.frame_type(), MultiplexFrame::DATA_FRAME);
if (first_frame_is_data_1) {
EXPECT_EQ(frame.data_frame().data(), std::string(data_2));
} else {
EXPECT_EQ(frame.data_frame().data(), std::string(data_1));
}
multiplex_output_stream_->Shutdown();
}
} // namespace multiplex
} // namespace mediums
} // namespace connections
} // namespace nearby
connections/implementation/mediums/multiplex/multiplex_socket.cc
-818
View File
@@ -1,818 +0,0 @@
// Copyright 2024 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.
#include "connections/implementation/mediums/multiplex/multiplex_socket.h"
#include <cstdint>
#include <memory>
#include <string>
#include <type_traits>
#include <utility>
#include "absl/container/flat_hash_map.h"
#include "absl/functional/any_invocable.h"
#include "absl/strings/string_view.h"
#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "connections/implementation/mediums/multiplex/multiplex_frames.h"
#include "connections/implementation/mediums/multiplex/multiplex_output_stream.h"
#include "connections/implementation/mediums/utils.h"
#include "internal/platform/atomic_boolean.h"
#include "internal/platform/base64_utils.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/future.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex.h"
#include "internal/platform/mutex_lock.h"
#include "internal/platform/socket.h"
#include "internal/platform/types.h"
#include "proto/connections_enums.pb.h"
namespace nearby {
namespace connections {
namespace mediums {
namespace multiplex {
namespace {
// It is defined for the receiver which send the first packet to the sender
// without getting salt from it yet. The fake salt reminds sender to get the
// correct socket from `virtualSockets` without remapping it.
constexpr absl::string_view kFakeSalt = "RECEIVER_CONDIMENT";
// The max duration to wait for the reader thread to stop.
constexpr absl::Duration kTimeoutForReaderThreadStop = absl::Milliseconds(100);
} // namespace
using ::location::nearby::mediums::ConnectionResponseFrame;
using ::location::nearby::mediums::MultiplexControlFrame;
using ::location::nearby::mediums::MultiplexDataFrame;
using ::location::nearby::mediums::MultiplexFrame;
using ::location::nearby::proto::connections::Medium;
using ::location::nearby::proto::connections::Medium_Name;
using ConnectionResponseCode = ConnectionResponseFrame::ConnectionResponseCode;
// AtomicBoolean is trivial destructible, so it is safe to use it as a static
// variable.
AtomicBoolean MultiplexSocket::is_shutting_down_{false}; // NOLINT
void MultiplexSocket::ListenForIncomingConnection(
const std::string& service_id, Medium type,
MultiplexIncomingConnectionCb incoming_connection_cb) {
GetIncomingConnectionCallbacks().emplace(
std::pair<std::string, Medium>(service_id, type),
std::move(incoming_connection_cb));
}
void MultiplexSocket::StopListeningForIncomingConnection(
const std::string& service_id, Medium type) {
GetIncomingConnectionCallbacks().erase(
std::pair<std::string, Medium>(service_id, type));
}
MultiplexSocket::MultiplexSocket(std::shared_ptr<MediumSocket> physical_socket)
: physical_socket_ptr_(physical_socket),
multiplex_output_stream_{&physical_socket_ptr_->GetOutputStream(),
enabled_},
physical_reader_(&physical_socket_ptr_->GetInputStream()),
medium_(physical_socket_ptr_->GetMedium()) {}
absl::flat_hash_map<std::pair<std::string, Medium>,
MultiplexIncomingConnectionCb>&
MultiplexSocket::GetIncomingConnectionCallbacks() {
using MapType = absl::flat_hash_map<std::pair<std::string, Medium>,
MultiplexIncomingConnectionCb>;
alignas(MapType) static char storage[sizeof(MapType)];
static MapType* incoming_connection_callbacks = new (&storage) MapType();
return *incoming_connection_callbacks;
}
MultiplexSocket* MultiplexSocket::CreateIncomingSocket(
std::shared_ptr<MediumSocket> physical_socket,
const std::string& service_id, std::int32_t first_frame_len) {
while (is_shutting_down_.Get()) {
LOG(WARNING)
<< "Shutting down is going on, wait for 2ms to create incoming socket";
absl::SleepFor(absl::Milliseconds(2));
}
MultiplexSocket* multiplex_incoming_socket = nullptr;
static MultiplexSocket* multiplex_incoming_socket_bt = nullptr;
static MultiplexSocket* multiplex_incoming_socket_wlan = nullptr;
switch (physical_socket->GetMedium()) {
case Medium::BLUETOOTH:
if (multiplex_incoming_socket_bt != nullptr) {
LOG(INFO) << "Multiplex incoming socket already exists for BT";
return multiplex_incoming_socket_bt;
}
alignas(MultiplexSocket) static char storage_bt[sizeof(MultiplexSocket)];
multiplex_incoming_socket_bt =
new (&storage_bt) MultiplexSocket(physical_socket);
multiplex_incoming_socket = multiplex_incoming_socket_bt;
break;
case Medium::WIFI_LAN:
case Medium::AWDL:
if (multiplex_incoming_socket_wlan != nullptr) {
LOG(INFO) << "Multiplex incoming socket already exists for WLAN";
return multiplex_incoming_socket_wlan;
}
alignas(
MultiplexSocket) static char storage_wlan[sizeof(MultiplexSocket)];
multiplex_incoming_socket_wlan =
new (&storage_wlan) MultiplexSocket(physical_socket);
multiplex_incoming_socket = multiplex_incoming_socket_wlan;
break;
default:
LOG(ERROR) << __func__
<< "Unsupported medium: " << physical_socket->GetMedium();
multiplex_incoming_socket = nullptr;
return multiplex_incoming_socket;
}
LOG(INFO) << "CreateIncomingSocket with serviceId=" << service_id
<< ", serviceIdHashSalt=" << kFakeSalt
<< " for medium=" << Medium_Name(physical_socket->GetMedium());
multiplex_incoming_socket->CreateFirstVirtualSocket(service_id,
(std::string)kFakeSalt);
multiplex_incoming_socket->StartReaderThread(first_frame_len);
return multiplex_incoming_socket;
}
MultiplexSocket* MultiplexSocket::CreateOutgoingSocket(
std::shared_ptr<MediumSocket> physical_socket,
const std::string& service_id, const std::string& service_id_hash_salt) {
while (is_shutting_down_.Get()) {
LOG(WARNING)
<< "Shutting down is going on, wait for 2ms to create outgoing socket";
absl::SleepFor(absl::Milliseconds(2));
}
MultiplexSocket* multiplex_outgoing_socket = nullptr;
static MultiplexSocket* multiplex_outgoing_socket_bt = nullptr;
static MultiplexSocket* multiplex_outgoing_socket_wlan = nullptr;
switch (physical_socket->GetMedium()) {
case Medium::BLUETOOTH:
if (multiplex_outgoing_socket_bt != nullptr) {
LOG(INFO) << "Multiplex outgoing socket already exists for BT";
return multiplex_outgoing_socket_bt;
}
alignas(MultiplexSocket) static char storage_bt[sizeof(MultiplexSocket)];
multiplex_outgoing_socket_bt =
new (&storage_bt) MultiplexSocket(physical_socket);
multiplex_outgoing_socket = multiplex_outgoing_socket_bt;
break;
case Medium::WIFI_LAN:
case Medium::AWDL:
if (multiplex_outgoing_socket_wlan != nullptr) {
LOG(INFO) << "Multiplex outgoing socket already exists for WLAN";
return multiplex_outgoing_socket_wlan;
}
alignas(
MultiplexSocket) static char storage_wlan[sizeof(MultiplexSocket)];
multiplex_outgoing_socket_wlan =
new (&storage_wlan) MultiplexSocket(physical_socket);
multiplex_outgoing_socket = multiplex_outgoing_socket_wlan;
break;
default:
LOG(ERROR) << __func__
<< "Unsupported medium: " << physical_socket->GetMedium();
multiplex_outgoing_socket = nullptr;
return multiplex_outgoing_socket;
}
LOG(INFO) << "CreateOutgoingSocket with serviceId=" << service_id
<< ", serviceIdHashSalt=" << service_id_hash_salt
<< " for medium=" << Medium_Name(physical_socket->GetMedium());
multiplex_outgoing_socket->CreateFirstVirtualSocket(service_id,
service_id_hash_salt);
multiplex_outgoing_socket->StartReaderThread(0);
return multiplex_outgoing_socket;
}
MultiplexSocket* MultiplexSocket::CreateOutgoingSocket(
std::shared_ptr<MediumSocket> physical_socket,
const std::string& service_id) {
return CreateOutgoingSocket(physical_socket, service_id,
Utils::GenerateSalt());
}
std::shared_ptr<MediumSocket> MultiplexSocket::CreateFirstVirtualSocket(
const std::string& service_id, const std::string& service_id_hash_salt) {
auto output_stream =
multiplex_output_stream_.CreateVirtualOutputStreamForFirstVirtualSocket(
service_id, service_id_hash_salt);
MutexLock lock(&virtual_socket_mutex_);
std::string salted_service_id_hash_key =
GenerateServiceIdHashKeyWithSalt(service_id, service_id_hash_salt);
LOG(INFO) << __func__ << " for service_id=" << service_id
<< ", salt=" << service_id_hash_salt
<< ", salted_service_id_hash_key=" << salted_service_id_hash_key;
MediumSocket* virtual_socket_ptr = physical_socket_ptr_->CreateVirtualSocket(
salted_service_id_hash_key, output_stream, medium_, &virtual_sockets_);
if (virtual_socket_ptr == nullptr) {
return nullptr;
}
std::shared_ptr<MediumSocket> virtual_socket =
virtual_sockets_[salted_service_id_hash_key];
virtual_socket->AddOnSocketClosedListener(
std::make_unique<absl::AnyInvocable<void()>>(
[this, service_id]() { OnVirtualSocketClosed(service_id); }));
if (!IsEnabled()) {
LOG(INFO) << __func__ << ": Register multiplex enabled callback";
virtual_socket->RegisterMultiplexEnabledCallback(enable_cb_);
}
return virtual_socket;
}
std::shared_ptr<MediumSocket> MultiplexSocket::CreateVirtualSocket(
const std::string& service_id, const std::string& service_id_hash_salt) {
auto output_stream = multiplex_output_stream_.CreateVirtualOutputStream(
service_id, service_id_hash_salt);
MutexLock lock(&virtual_socket_mutex_);
std::string salted_service_id_hash_key =
GenerateServiceIdHashKeyWithSalt(service_id, service_id_hash_salt);
LOG(INFO) << __func__ << "service_id=" << service_id
<< ", salt=" << service_id_hash_salt
<< ", salted_service_id_hash_key=" << salted_service_id_hash_key;
MediumSocket* virtual_socket_ptr = physical_socket_ptr_->CreateVirtualSocket(
salted_service_id_hash_key, output_stream, medium_, &virtual_sockets_);
if (virtual_socket_ptr == nullptr) {
return nullptr;
}
std::shared_ptr<MediumSocket> virtual_socket =
virtual_sockets_[salted_service_id_hash_key];
virtual_socket->AddOnSocketClosedListener(
std::make_unique<absl::AnyInvocable<void()>>(
[this, service_id]() { OnVirtualSocketClosed(service_id); }));
return virtual_socket;
}
std::shared_ptr<MediumSocket> MultiplexSocket::GetVirtualSocket(
const std::string& service_id) {
MutexLock lock(&virtual_socket_mutex_);
LOG(INFO) << __func__ << " service_id=" << service_id << ", Salt="
<< multiplex_output_stream_.GetServiceIdHashSalt(service_id)
<< ", virtual_sockets_.size()=" << virtual_sockets_.size();
auto item = virtual_sockets_.find(GenerateServiceIdHashKeyWithSalt(
service_id, multiplex_output_stream_.GetServiceIdHashSalt(service_id)));
if (item == virtual_sockets_.end()) {
LOG(INFO) << "Not found!";
return nullptr;
}
return item->second;
}
int MultiplexSocket::GetVirtualSocketCount() {
MutexLock lock(&virtual_socket_mutex_);
return virtual_sockets_.size();
}
void MultiplexSocket::ListVirtualSocket() {
LOG(INFO) << __func__
<< " virtual_sockets_.size()=" << virtual_sockets_.size();
for (auto& [service_id_hash_key, virtual_socket] : virtual_sockets_) {
LOG(INFO) << __func__ << " service_id_hash_key=" << service_id_hash_key
<< ", virtual_socket=" << virtual_socket;
}
}
std::shared_ptr<Future<ConnectionResponseCode>>
MultiplexSocket::RegisterConnectionResponse(const std::string& service_id) {
auto future = std::make_shared<Future<ConnectionResponseCode>>();
connection_response_futures_.emplace(service_id, future);
return future;
}
void MultiplexSocket::UnRegisterConnectionResponse(
const std::string& service_id) {
connection_response_futures_.erase(service_id);
}
std::shared_ptr<MediumSocket> MultiplexSocket::EstablishVirtualSocket(
const std::string& service_id) {
if (!IsEnabled()) {
LOG(ERROR)
<< "MultiplexSocket is disabled, cannot establish virtual socket.";
return nullptr;
}
std::string service_id_hash_salt = Utils::GenerateSalt();
auto future = RegisterConnectionResponse(service_id);
multiplex_output_stream_.WriteConnectionRequestFrame(service_id,
service_id_hash_salt);
auto result =
future->Get(FeatureFlags::GetInstance()
.GetFlags()
.multiplex_socket_connection_response_timeout_millis);
if (!result.ok()) {
LOG(ERROR) << __func__
<< "EstablishVirtualSocket failed with response code="
<< result.exception();
return nullptr;
}
ConnectionResponseCode response_code = result.GetResult();
switch (response_code) {
case ConnectionResponseFrame::CONNECTION_ACCEPTED:
LOG(INFO) << "EstablishVirtualSocket after remote response to"
" accept the connection with service_id="
<< service_id
<< ", service_id_hash_salt=" << service_id_hash_salt;
return CreateVirtualSocket(service_id, service_id_hash_salt);
case ConnectionResponseFrame::NOT_LISTENING:
LOG(ERROR) << "EstablishVirtualSocket failed for service_id="
<< service_id
<< ", service_id_hash_salt=" << service_id_hash_salt
<< " with response code=NOT_LISTENING";
break;
default:
LOG(ERROR) << "EstablishVirtualSocket failed for service_id="
<< service_id
<< ", service_id_hash_salt=" << service_id_hash_salt
<< " with response code=UNKNOWN_RESPONSE_CODE";
break;
}
return nullptr;
}
void MultiplexSocket::StartReaderThread(std::int32_t first_frame_len) {
if (is_shutdown_) {
LOG(WARNING) << "Stop to start reader thread since socket is "
"shutdown.";
return;
}
reader_thread_shutdown_barrier_ = std::make_unique<CountDownLatch>(1);
physical_reader_thread_.Execute([this, first_frame_len]() {
LOG(INFO) << __func__ << " Reader thread starts.";
auto first_frame_len_copy = first_frame_len;
while (!is_shutdown_) {
bool fail = false;
ExceptionOr<ByteArray> bytes;
ExceptionOr<std::int32_t> read_int;
if (first_frame_len_copy > 0) {
read_int = ExceptionOr<std::int32_t>(first_frame_len);
first_frame_len_copy = 0;
} else {
read_int = Base64Utils::ReadInt(physical_reader_);
}
if (!read_int.ok()) {
LOG(WARNING) << __func__
<< "Failed to read. Exception:" << read_int.exception();
fail = true;
} else {
auto length = read_int.result();
VLOG(1) << __func__ << " length:" << length;
if (length < 0 || length > FeatureFlags::GetInstance()
.GetFlags()
.connection_max_frame_length) {
// Ignore the failure because not only one client use this
// connection.
LOG(WARNING) << __func__
<< "Failed to read because received a invalid length "
<< length << ", but continue to read.";
continue;
}
bytes = physical_reader_->ReadExactly(length);
if (!bytes.ok()) {
LOG(WARNING) << __func__
<< "Read data exception:" << bytes.exception();
fail = true;
}
}
if (fail) {
reader_thread_shutdown_barrier_->CountDown();
return;
}
ExceptionOr<MultiplexFrame> frame_exc =
multiplex::FromBytes(bytes.result());
if (!frame_exc.ok()) {
HandleOfflineFrame(bytes.result());
continue;
}
if (!IsEnabled()) {
// The reader thread will only be enabled when local device
// supports multiplex if we received a multiplex frame from
// the remote, it means that the remote and the local both
// support multiplex as well. So it is safe to just turn on
// the feature at this point.
LOG(INFO) << __func__
<< " Received a multiplex frame while not enabled, enable "
"multiplex.";
Enable();
}
const auto& frame = frame_exc.result();
auto salted_service_id_hash =
ByteArray{std::move(frame.header().salted_service_id_hash())};
auto service_id_hash_salt = frame.header().has_service_id_hash_salt()
? frame.header().service_id_hash_salt()
: "";
switch (frame.frame_type()) {
case MultiplexFrame::CONTROL_FRAME:
HandleControlFrame(salted_service_id_hash, service_id_hash_salt,
frame.control_frame());
break;
case MultiplexFrame::DATA_FRAME:
VLOG(1) << "service_id_hash_salt: " << service_id_hash_salt;
HandleDataFrame(salted_service_id_hash, service_id_hash_salt,
frame.data_frame());
break;
default:
LOG(WARNING) << __func__
<< " Received MultiplexFrame with unknown frame type "
<< frame.frame_type();
}
}
});
}
void MultiplexSocket::HandleOfflineFrame(const ByteArray& bytes) {
MutexLock lock(&virtual_socket_mutex_);
LOG(INFO) << __func__ << " Virtual_socket num:" << virtual_sockets_.size();
if (virtual_sockets_.size() == 1) {
auto item = virtual_sockets_.begin();
if (item->second == nullptr) {
LOG(WARNING) << "Expected one live socket, but found null.";
return;
}
LOG(INFO) << __func__ << "FeedIncomingData:" << std::string(bytes);
item->second->FeedIncomingData(Base64Utils::IntToBytes(bytes.size()));
item->second->FeedIncomingData(bytes);
}
}
void MultiplexSocket::HandleControlFrame(
const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt,
const MultiplexControlFrame& frame) {
switch (frame.control_frame_type()) {
case MultiplexControlFrame::CONNECTION_REQUEST:
RunOffloadThread("CONNECTION_REQUEST", [this, salted_service_id_hash,
service_id_hash_salt] {
HandleConnectionRequest(salted_service_id_hash, service_id_hash_salt);
});
break;
case MultiplexControlFrame::CONNECTION_RESPONSE:
LOG(INFO) << __func__ << "Received an CONNECTION_RESPONSE frame."
<< " salted_service_id_hash: "
<< std::string(salted_service_id_hash)
<< ", service_id_hash_salt: " << service_id_hash_salt
<< ", ConnectionResponseCode: "
<< frame.connection_response_frame().connection_response_code();
RunOffloadThread("CONNECTION_RESPONSE", [this, salted_service_id_hash,
service_id_hash_salt,
frame = frame] {
HandleConnectionResponse(salted_service_id_hash, service_id_hash_salt,
frame.connection_response_frame());
});
break;
case MultiplexControlFrame::DISCONNECTION:
// The virtual socket will be closed in the offload thread, so don't run
// the thread here.
HandleDisconnection(salted_service_id_hash);
break;
default:
LOG(WARNING) << __func__ << "Received an unknown frame type "
<< frame.control_frame_type();
break;
}
}
void MultiplexSocket::HandleConnectionRequest(
const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt) {
if (!IsEnabled()) {
LOG(WARNING) << "Received a CONNECTION_REQUEST frame on medium "
<< Medium_Name(medium_)
<< " but status is disabled, ignore it.";
return;
}
std::string salted_service_id_hash_key =
GenerateServiceIdHashKey(salted_service_id_hash);
MultiplexIncomingConnectionCb* incoming_connection_callback = nullptr;
std::string listening_service_id = "";
for (auto& [service_id_medium_pair, callback] :
GetIncomingConnectionCallbacks()) {
if (GenerateServiceIdHashWithSalt(service_id_medium_pair.first,
service_id_hash_salt) ==
salted_service_id_hash) {
incoming_connection_callback = &callback;
listening_service_id = service_id_medium_pair.first;
}
}
if (incoming_connection_callback == nullptr || listening_service_id.empty()) {
LOG(INFO) << "There's no client listening for hash salt : "
<< service_id_hash_salt
<< ", hash key : " << salted_service_id_hash_key << " on medium "
<< Medium_Name(medium_);
LOG(INFO) << "The size of incomingConnectionCallbacks : "
<< GetIncomingConnectionCallbacks().size();
if (!multiplex_output_stream_.WriteConnectionResponseFrame(
salted_service_id_hash, service_id_hash_salt,
ConnectionResponseFrame::NOT_LISTENING)) {
LOG(INFO) << __func__ << "Failed to write NOT_LISTENING frame.";
}
return;
}
LOG(INFO) << "Accept new virtual socket request service ID : "
<< listening_service_id << ", hash salt : " << service_id_hash_salt
<< ", hash key : " << salted_service_id_hash_key << " on medium "
<< Medium_Name(medium_);
if (!multiplex_output_stream_.WriteConnectionResponseFrame(
salted_service_id_hash, service_id_hash_salt,
ConnectionResponseFrame::CONNECTION_ACCEPTED)) {
LOG(INFO) << "Failed to write CONNECTION_ACCEPTED frame.";
return;
}
LOG(INFO)
<< "EstablishVirtualSocket after local device accept the connection "
"with serviceId="
<< listening_service_id;
std::shared_ptr<MediumSocket> virtual_socket =
CreateVirtualSocket(listening_service_id, service_id_hash_salt);
(*incoming_connection_callback)(std::move(listening_service_id),
virtual_socket);
}
void MultiplexSocket::HandleConnectionResponse(
const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt,
const ConnectionResponseFrame& frame) {
LOG(INFO) << __func__
<< "connection_response_code: " << frame.connection_response_code();
for (auto& [service_id, future] : connection_response_futures_) {
if (GenerateServiceIdHashWithSalt(service_id, service_id_hash_salt) ==
salted_service_id_hash) {
if (future != nullptr) {
future->Set(frame.connection_response_code());
LOG(INFO) << __func__ << "Set the future for serviceId=" << service_id
<< ", serviceIdHashSalt=" << service_id_hash_salt
<< " with response code=" << frame.connection_response_code();
return;
}
}
}
LOG(WARNING)
<< __func__
<< "Received a CONNECTION_RESPONSE frame but no client waiting for "
"service ID Hash Key"
<< GenerateServiceIdHashKey(salted_service_id_hash);
}
void MultiplexSocket::HandleDisconnection(
const ByteArray& salted_service_id_hash) {
std::string salted_service_id_hash_key =
GenerateServiceIdHashKey(salted_service_id_hash);
MediumSocket* virtual_socket_to_close = nullptr;
{
MutexLock lock(&virtual_socket_mutex_);
auto item = virtual_sockets_.find(salted_service_id_hash_key);
if (item != virtual_sockets_.end()) {
LOG(INFO)
<< "Received a DISCONNECTION frame to disconnect virtual socket for "
"salted service ID Hash Key "
<< salted_service_id_hash_key;
virtual_socket_to_close = item->second.get();
} else {
LOG(WARNING)
<< "Received a DISCONNECTION frame but there's no alive socket to "
"disconnect for service ID Hash Key "
<< salted_service_id_hash_key;
}
}
// Close the virtual socket outside of the mutex lock because
// OnVirtualSocketClosed will lock the mutex.
if (virtual_socket_to_close != nullptr) {
virtual_socket_to_close->Close();
}
}
void MultiplexSocket::HandleDataFrame(const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt,
const MultiplexDataFrame& frame) {
std::string salted_service_id_hash_key =
GenerateServiceIdHashKey(salted_service_id_hash);
std::shared_ptr<MediumSocket> virtual_socket = nullptr;
if (service_id_hash_salt.empty()) {
{
MutexLock lock(&virtual_socket_mutex_);
auto item = virtual_sockets_.find(salted_service_id_hash_key);
if (item != virtual_sockets_.end()) {
virtual_socket = item->second;
}
}
} else {
virtual_socket =
ReMapAndGetVirtualSocket(salted_service_id_hash, service_id_hash_salt);
}
if (virtual_socket != nullptr) {
VLOG(1)
<< "Received a DATA frame to feed virtual socket for salted service ID "
"Hash Key "
<< salted_service_id_hash_key;
virtual_socket->FeedIncomingData(ByteArray(frame.data()));
} else {
LOG(WARNING)
<< "Received a DATA frame but there's no alive socket to feed for "
"salted service ID Hash Key "
<< salted_service_id_hash_key;
}
}
void MultiplexSocket::OnPhysicalSocketClosed() {
RunOffloadThread("Shutdown", [this]() { Shutdown(); });
}
void MultiplexSocket::OnVirtualSocketClosed(const std::string& service_id) {
LOG(INFO) << __func__ << " for service_id:" << service_id;
CountDownLatch latch(1);
bool shutdown = false;
RunOffloadThread(
"VirtualSocketClosed", [this, service_id, &latch, &shutdown]() {
LOG(INFO) << "Try to close Virtual socket: " << service_id;
std::shared_ptr<MediumSocket> virtual_socket =
GetVirtualSocket(service_id);
{
MutexLock lock(&virtual_socket_mutex_);
LOG(INFO) << "virtual_socket:" << virtual_socket;
if (virtual_socket != nullptr) {
auto salted_service_id_hash_key = GenerateServiceIdHashKeyWithSalt(
service_id,
multiplex_output_stream_.GetServiceIdHashSalt(service_id));
multiplex_output_stream_.Close(service_id);
virtual_sockets_.erase(salted_service_id_hash_key);
LOG(INFO) << "Erase Virtual socket with service_id: " << service_id
<< ", hash_key: " << salted_service_id_hash_key;
ListVirtualSocket();
if (virtual_sockets_.empty()) {
LOG(INFO) << "Close the physical socket because all virtual "
"sockets disconnected.";
is_shutting_down_.Set(true);
Shutdown();
shutdown = true;
}
} else {
LOG(INFO) << "Virtual socket(" << service_id << ") not found";
}
}
latch.CountDown();
});
if (!latch.Await(absl::Milliseconds(1000)).result()) {
LOG(ERROR) << "Timeout to close virtual socket";
}
if (shutdown) {
LOG(INFO)
<< "Shutdown single_thread_offloader_ and physical_reader_thread_";
single_thread_offloader_.Shutdown();
physical_reader_thread_.Shutdown();
is_shutting_down_.Set(false);
}
}
std::shared_ptr<MediumSocket> MultiplexSocket::ReMapAndGetVirtualSocket(
const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt) {
std::string salted_service_id_hash_key =
GenerateServiceIdHashKey(salted_service_id_hash);
VLOG(1) << "ReMapAndGetVirtualSocket with serviceIdHashSalt="
<< service_id_hash_salt
<< ", saltedServiceIdHashKey=" << salted_service_id_hash_key;
{
MutexLock lock(&virtual_socket_mutex_);
for (auto& [hash_key, virtual_socket] : virtual_sockets_) {
auto output_stream =
down_cast<MultiplexOutputStream::VirtualOutputStream*>(
&(virtual_socket->GetOutputStream()));
if (output_stream == nullptr) {
continue;
}
if (!output_stream->IsFirstVirtualOutputStream()) {
continue;
}
if ((service_id_hash_salt == kFakeSalt) ||
(hash_key == salted_service_id_hash_key)) {
return virtual_socket;
} else {
LOG(INFO) << "Remap the virtualSockets.";
output_stream->SetserviceIdHashSalt(service_id_hash_salt);
auto virtual_socket_tmp = virtual_socket;
LOG(INFO) << "virtual_socket before:" << virtual_socket;
virtual_sockets_.erase(hash_key);
virtual_sockets_[salted_service_id_hash_key] = virtual_socket_tmp;
ListVirtualSocket();
return virtual_socket_tmp;
}
}
}
LOG(INFO) << "Failed to remap the virtualSockets.";
return nullptr;
}
void MultiplexSocket::RunOffloadThread(const std::string& name,
absl::AnyInvocable<void()> runnable) {
single_thread_offloader_.Execute(name, std::move(runnable));
}
void MultiplexSocket::Shutdown() {
LOG(INFO) << __func__ << " start";
if (is_shutdown_) {
LOG(INFO) << __func__ << " Already shutdown";
return;
}
multiplex_output_stream_.Shutdown();
physical_socket_ptr_->Close();
if (reader_thread_shutdown_barrier_) {
reader_thread_shutdown_barrier_->Await(kTimeoutForReaderThreadStop);
}
GetIncomingConnectionCallbacks().clear();
connection_response_futures_.clear();
is_shutdown_ = true;
enabled_.Set(false);
LOG(INFO) << __func__ << " end";
}
void MultiplexSocket::ShutdownAll() {
LOG(INFO) << __func__ << " start";
if (is_shutdown_) {
LOG(WARNING) << __func__ << " Already shutdown";
return;
}
CountDownLatch latch(1);
RunOffloadThread("VirtualSocketClosed", [this, &latch]() {
{
MutexLock lock(&virtual_socket_mutex_);
multiplex_output_stream_.CloseAll();
virtual_sockets_.clear();
Shutdown();
}
latch.CountDown();
});
if (!latch
.Await(FeatureFlags::GetInstance()
.GetFlags()
.mediums_frame_write_timeout_millis +
absl::Milliseconds(100))
.result()) {
LOG(ERROR) << "Timeout to close virtual socket";
}
LOG(INFO) << "Shutdown single_thread_offloader_ and physical_reader_thread_";
single_thread_offloader_.Shutdown();
physical_reader_thread_.Shutdown();
LOG(INFO) << __func__ << " end";
}
} // namespace multiplex
} // namespace mediums
} // namespace connections
} // namespace nearby
connections/implementation/mediums/multiplex/multiplex_socket.h
-223
View File
@@ -1,223 +0,0 @@
// Copyright 2024 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_MULTIPLEX_MULTIPLEX_SOCKET_H_
#define CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_SOCKET_H_
#include <cstdint>
#include <memory>
#include <string>
#include <utility>
#include "absl/base/thread_annotations.h"
#include "absl/container/flat_hash_map.h"
#include "absl/functional/any_invocable.h"
#include "connections/implementation/mediums/multiplex/multiplex_output_stream.h"
#include "connections/medium_selector.h"
#include "internal/platform/atomic_boolean.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/future.h"
#include "internal/platform/input_stream.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex.h"
#include "internal/platform/single_thread_executor.h"
#include "internal/platform/socket.h"
#include "proto/connections_enums.pb.h"
#include "proto/mediums/multiplex_frames.pb.h"
namespace nearby {
namespace connections {
namespace mediums {
namespace multiplex {
using MultiplexEnbaleCb = absl::AnyInvocable<void()>;
using MultiplexIncomingConnectionCb = absl::AnyInvocable<void(
const std::string& service_id, std::shared_ptr<MediumSocket> socket)>;
class MultiplexSocket {
public:
MultiplexSocket(const MultiplexSocket&) = delete;
MultiplexSocket& operator=(const MultiplexSocket&) = delete;
~MultiplexSocket() { ShutdownAll(); };
// Creates a new incoming MultiplexSocket.
static MultiplexSocket* CreateIncomingSocket(
std::shared_ptr<MediumSocket> physical_socket,
const std::string& service_id, std::int32_t first_frame_len);
// Creates a new outgoing MultiplexSocket.
static MultiplexSocket* CreateOutgoingSocket(
std::shared_ptr<MediumSocket> physical_socket,
const std::string& service_id, const std::string& service_id_hash_salt);
// Creates a new outgoing MultiplexSocket with default service_id_hash_salt.
static MultiplexSocket* CreateOutgoingSocket(
std::shared_ptr<MediumSocket> physical_socket,
const std::string& service_id);
// A Table of service Id as row key, medium type as column key, and
// MultiplexIncomingConnectionCb as value. Non-empty while the client starts
// listening for incoming virtual socket. The MultiplexIncomingConnectionCb
// will be called when the incoming virtual socket is established.
static absl::flat_hash_map<
std::pair<std::string, ::location::nearby::proto::connections::Medium>,
MultiplexIncomingConnectionCb>&
GetIncomingConnectionCallbacks();
// Listens for incoming connection through multiplex for specified {@code
// service_id} on medium
// {@code type}. Should register the callback before new the MultiplexSocket.
static void ListenForIncomingConnection(
const std::string& service_id,
::location::nearby::proto::connections::Medium type,
absl::AnyInvocable<void(const std::string& service_id,
std::shared_ptr<MediumSocket> socket)>
incoming_connection_cb);
// Stops listening for incoming multiplex connection for {@code service_id} on
// medium {@code type}.
static void StopListeningForIncomingConnection(
const std::string& service_id,
::location::nearby::proto::connections::Medium type);
bool IsEnabled() { return enabled_.Get(); }
void Enable() {
LOG(INFO) << "Enable the Multiplex MediumSocket.";
enabled_.Set(true);
}
// Gets the virtual socket by service id.
std::shared_ptr<MediumSocket> GetVirtualSocket(const std::string& service_id);
// Gets the virtual socket count.
int GetVirtualSocketCount();
void ListVirtualSocket()
ABSL_EXCLUSIVE_LOCKS_REQUIRED(virtual_socket_mutex_);
// Establishes the virtual socket by service id.
std::shared_ptr<MediumSocket> EstablishVirtualSocket(
const std::string& service_id);
// Shuts down the multiplex socket.
void Shutdown();
bool IsShutdown() { return is_shutdown_; }
void SetShutdown(bool is_shutdown) { is_shutdown_ = is_shutdown; }
void ShutdownAll();
private:
explicit MultiplexSocket(std::shared_ptr<MediumSocket> physical_socket);
// Creates the first virtual socket for the service id. The first virtual
// socket is created by the sender.
std::shared_ptr<MediumSocket> CreateFirstVirtualSocket(
const std::string& service_id, const std::string& service_id_hash_salt);
// Creates the virtual socket for the service id.
std::shared_ptr<MediumSocket> CreateVirtualSocket(
const std::string& service_id, const std::string& service_id_hash_salt);
// Registers the connection response future for the service id.
std::shared_ptr<Future<::location::nearby::mediums::ConnectionResponseFrame::
ConnectionResponseCode>>
RegisterConnectionResponse(const std::string& service_id);
// Unregisters the connection response future for the service id.
void UnRegisterConnectionResponse(const std::string& service_id);
// Starts the reader thread to read the incoming MultiplexFrame from the
// physical socket.
void StartReaderThread(std::int32_t first_frame_len);
// Handles the offline frame from the physical socket.
void HandleOfflineFrame(const ByteArray& bytes);
// Handles the control frame from the physical socket.
void HandleControlFrame(
const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt,
const ::location::nearby::mediums::MultiplexControlFrame& frame);
// Handles the connection request frame from the physical socket.
void HandleConnectionRequest(const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt);
// Handles the connection response frame from the physical socket.
void HandleConnectionResponse(
const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt,
const ::location::nearby::mediums::ConnectionResponseFrame& frame);
// Handles the disconnection frame from the physical socket.
void HandleDisconnection(const ByteArray& salted_service_id_hash);
// Handles the data frame from the physical socket.
void HandleDataFrame(
const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt,
const ::location::nearby::mediums::MultiplexDataFrame& frame);
// Handles the physical socket closed.
void OnPhysicalSocketClosed();
// Remaps and gets the virtual socket by service id hash.
std::shared_ptr<MediumSocket> ReMapAndGetVirtualSocket(
const ByteArray& salted_service_id_hash,
const std::string& service_id_hash_salt);
// Handles the virtual socket closed.
void OnVirtualSocketClosed(const std::string& service_id);
// Runs the offload thread.
void RunOffloadThread(const std::string& name,
absl::AnyInvocable<void()> runnable);
// The physical socket connect to the remote device.
std::shared_ptr<MediumSocket> physical_socket_ptr_;
// The output stream to manage all outgoing frames from all clients.
MultiplexOutputStream multiplex_output_stream_;
// The {@link InputStream} of the physical socket. It is used to read the
// incoming MultiplexFrame from the physical socket.
InputStream* physical_reader_;
// The medium type of the physical socket.
Medium medium_;
// The callback to enable the MultiplexSocket.
std::shared_ptr<absl::AnyInvocable<void()>> enable_cb_ =
std::make_shared<absl::AnyInvocable<void()>>([this]() { Enable(); });
// A map of service Id -> {@link SettableFuture} for waiting the
// ConnectionResponse. Non-empty while requesting the virtual socket.
absl::flat_hash_map<std::string,
std::shared_ptr<Future<
::location::nearby::mediums::ConnectionResponseFrame::
ConnectionResponseCode>>>
connection_response_futures_;
// A map of service Id hash key -> virtual socket. Non-empty while at least
// one virtual socket alive. Class derived from "MediumSocket" should define a
// pointer to the virtual sockets map. When here's any virtual socket
// operation, it will be reflected in both derived MediumSocket class and
// MultiplexSocket object
mutable Mutex virtual_socket_mutex_;
absl::flat_hash_map<std::string, std::shared_ptr<MediumSocket>>
virtual_sockets_ ABSL_GUARDED_BY(virtual_socket_mutex_);
// The thread to receive incoming MultiplexFrame from the physical socket.
SingleThreadExecutor physical_reader_thread_;
// The single thread we throw the potentially blocking work on to.
SingleThreadExecutor single_thread_offloader_;
// The status of the MultiplexSocket enabled or disabled, it depends on both
// Sender and Receiver supports MultiplexSocket or not. Default disabled and
// enable it once two devices negotiated finished.
AtomicBoolean enabled_{false};
// If the socket is already shutdown and no longer in use.
bool is_shutdown_ = false;
static AtomicBoolean is_shutting_down_;
std::unique_ptr<CountDownLatch> reader_thread_shutdown_barrier_;
};
} // namespace multiplex
} // namespace mediums
} // namespace connections
} // namespace nearby
#endif // CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_SOCKET_H_
connections/implementation/mediums/multiplex/multiplex_socket_test.cc
-463
View File
@@ -1,463 +0,0 @@
// Copyright 2024 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.
#include "connections/implementation/mediums/multiplex/multiplex_socket.h"
#include <cstdint>
#include <memory>
#include <string>
#include <utility>
#include "gtest/gtest.h"
#include "absl/container/flat_hash_map.h"
#include "absl/strings/string_view.h"
#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "connections/implementation/mediums/multiplex/multiplex_frames.h"
#include "connections/implementation/offline_frames.h"
#include "internal/platform/base64_utils.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/future.h"
#include "internal/platform/input_stream.h"
#include "internal/platform/logging.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/pipe.h"
#include "internal/platform/single_thread_executor.h"
#include "internal/platform/socket.h"
#include "internal/platform/types.h"
#include "proto/connections_enums.proto.h"
namespace nearby {
namespace connections {
namespace mediums {
namespace multiplex {
constexpr absl::string_view SERVICE_ID_1 = "serviceId_1";
constexpr absl::string_view SERVICE_ID_2 = "serviceId_2";
using location::nearby::mediums::ConnectionResponseFrame;
using location::nearby::mediums::MultiplexControlFrame;
using location::nearby::mediums::MultiplexFrame;
using location::nearby::proto::connections::Medium;
using location::nearby::proto::connections::Medium_Name;
// A fake socket for testing.
class FakeSocket : public MediumSocket {
public:
explicit FakeSocket(Medium medium) : MediumSocket(medium) {
pipe_1_ = CreatePipe();
reader_1_ = std::move(pipe_1_.first);
writer_1_ = std::move(pipe_1_.second);
pipe_2_ = CreatePipe();
reader_2_ = std::move(pipe_2_.first);
writer_2_ = std::move(pipe_2_.second);
LOG(WARNING) << "Physical Socket Medium:" << Medium_Name(GetMedium());
};
~FakeSocket() override = default;
FakeSocket(const FakeSocket&) = default;
FakeSocket& operator=(const FakeSocket&) = default;
/**
* The constructor for a virtual socket which own the virtual {@link
* OutputStream} and {@link InputStream}.
*/
explicit FakeSocket(Medium medium, OutputStream* virtualOutputStream)
: MediumSocket(medium),
is_virtual_socket_(true),
virtual_output_stream_(virtualOutputStream) {
pipe_1_ = CreatePipe();
reader_1_ = std::move(pipe_1_.first);
writer_1_ = std::move(pipe_1_.second);
pipe_2_ = CreatePipe();
reader_2_ = std::move(pipe_2_.first);
writer_2_ = std::move(pipe_2_.second);
}
InputStream& GetInputStream() override { return *reader_1_; }
OutputStream& GetOutputStream() override {
return IsVirtualSocket() ? *virtual_output_stream_ : *writer_2_;
}
Exception Close() override {
if (IsVirtualSocket()) {
LOG(INFO) << "Multiplex: Closing virtual socket: " << this;
CloseLocal();
return {Exception::kSuccess};
}
LOG(INFO) << "Multiplex: Closing physical socket: " << this;
reader_1_->Close();
reader_2_->Close();
writer_1_->Close();
writer_2_->Close();
return {Exception::kSuccess};
}
MediumSocket* CreateVirtualSocket(
const std::string& salted_service_id_hash_key, OutputStream* outputstream,
Medium medium,
absl::flat_hash_map<std::string, std::shared_ptr<MediumSocket>>*
virtual_sockets_ptr) override {
if (IsVirtualSocket()) {
LOG(WARNING)
<< "Creating the virtual socket on a virtual socket is not allowed.";
return nullptr;
}
auto virtual_socket = std::make_shared<FakeSocket>(medium, outputstream);
LOG(INFO) << "Created the virtual socket for Medium: "
<< Medium_Name(virtual_socket->GetMedium());
if (virtual_sockets_ptr_ == nullptr) {
virtual_sockets_ptr_ = virtual_sockets_ptr;
}
(*virtual_sockets_ptr_)[salted_service_id_hash_key] = virtual_socket;
LOG(INFO) << "virtual_sockets_ size: " << virtual_sockets_ptr_->size();
return virtual_socket.get();
}
void FeedIncomingData(ByteArray data) override {
bytes_read_future_.Set(data);
LOG(INFO) << "FeedIncomingData. Size of receive data: " << data.size()
<< ", bytes content:" << std::string(data);
}
bool IsVirtualSocket() override { return is_virtual_socket_; }
Future<ByteArray>& GetByteReadFuture() { return bytes_read_future_; }
std::pair<std::unique_ptr<InputStream>, std::unique_ptr<OutputStream>>
pipe_1_;
std::unique_ptr<InputStream> reader_1_;
std::unique_ptr<OutputStream> writer_1_;
std::pair<std::unique_ptr<InputStream>, std::unique_ptr<OutputStream>>
pipe_2_;
std::unique_ptr<InputStream> reader_2_;
std::unique_ptr<OutputStream> writer_2_;
private:
bool is_virtual_socket_ = false;
Future<ByteArray> bytes_read_future_;
absl::flat_hash_map<std::string, std::shared_ptr<MediumSocket>>*
virtual_sockets_ptr_ = nullptr;
OutputStream* virtual_output_stream_ = nullptr;
};
TEST(MultiplexSocketTest, CreateIncomingSocketSuccess) {
auto fake_socket_ptr = std::make_shared<FakeSocket>(Medium::BLUETOOTH);
MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_1),
Medium::BLUETOOTH);
MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_2),
Medium::BLUETOOTH);
MultiplexSocket::ListenForIncomingConnection(
std::string(SERVICE_ID_1), Medium::BLUETOOTH,
[](const std::string& service_id, std::shared_ptr<MediumSocket> socket) {
LOG(INFO) << "Incoming connection for service_id: " << service_id;
});
MultiplexSocket::ListenForIncomingConnection(
std::string(SERVICE_ID_2), Medium::BLUETOOTH,
[](const std::string& service_id, std::shared_ptr<MediumSocket> socket) {
LOG(INFO) << "Incoming connection for service_id: " << service_id;
});
MultiplexSocket* multiplex_socket_incoming =
MultiplexSocket::CreateIncomingSocket(
fake_socket_ptr, std::string(SERVICE_ID_1), /*first_frame_len*/ 0);
ASSERT_NE(multiplex_socket_incoming, nullptr);
MultiplexSocket* multiplex_socket_incoming_2 =
MultiplexSocket::CreateIncomingSocket(
fake_socket_ptr, std::string(SERVICE_ID_2), /*first_frame_len*/ 0);
ASSERT_EQ(multiplex_socket_incoming_2, multiplex_socket_incoming);
std::shared_ptr<MediumSocket> virtual_socket_shared =
multiplex_socket_incoming->GetVirtualSocket(std::string(SERVICE_ID_1));
ASSERT_NE(virtual_socket_shared, nullptr);
FakeSocket* virtual_socket =
down_cast<FakeSocket*>(virtual_socket_shared.get());
SingleThreadExecutor executor;
FakeSocket* socket = fake_socket_ptr.get();
executor.Execute([socket]() {
std::string connection_req_frame = parser::ForConnectionRequestConnections(
{}, {
.local_endpoint_id = "endpoint1",
.local_endpoint_info = ByteArray("endpoint1 info"),
});
auto& writer = socket->writer_1_;
LOG(INFO) << "writer_1_ Write start";
Base64Utils::WriteInt(writer.get(), connection_req_frame.size());
writer->Write(connection_req_frame);
writer->Flush();
LOG(INFO) << "writer_1_ Write end";
});
ExceptionOr<ByteArray> result = virtual_socket->GetByteReadFuture().Get();
if (!result.ok()) {
ADD_FAILURE() << "Read error: " << result.GetException().value;
}
ByteArray data = result.result();
LOG(INFO) << "Received " << data.size() << " bytes of data.";
EXPECT_NE(data.size(), 0);
absl::SleepFor(absl::Milliseconds(100));
EXPECT_EQ(multiplex_socket_incoming->GetVirtualSocketCount(), 1);
virtual_socket->Close();
EXPECT_EQ(multiplex_socket_incoming->GetVirtualSocketCount(), 0);
multiplex_socket_incoming->ShutdownAll();
}
TEST(MultiplexSocketTest, CreateFail_MediumNotSupport) {
auto fake_socket_ptr = std::make_shared<FakeSocket>(Medium::WEB_RTC);
MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_1),
Medium::WEB_RTC);
MultiplexSocket* multiplex_socket_incoming =
MultiplexSocket::CreateIncomingSocket(
fake_socket_ptr, std::string(SERVICE_ID_1), /*first_frame_len*/ 0);
ASSERT_EQ(multiplex_socket_incoming, nullptr);
}
TEST(MultiplexSocketTest, CreateIncomingVirtualSocketSuccess) {
auto fake_socket_ptr = std::make_shared<FakeSocket>(Medium::WIFI_LAN);
MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_1),
Medium::WIFI_LAN);
MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_2),
Medium::WIFI_LAN);
MultiplexSocket::ListenForIncomingConnection(
std::string(SERVICE_ID_1), Medium::WIFI_LAN,
[](const std::string& service_id, std::shared_ptr<MediumSocket> socket) {
LOG(INFO) << "Incoming connection for service_id: " << service_id;
});
MultiplexSocket::ListenForIncomingConnection(
std::string(SERVICE_ID_2), Medium::WIFI_LAN,
[](const std::string& service_id, std::shared_ptr<MediumSocket> socket) {
LOG(INFO) << "Incoming connection for service_id: " << service_id;
});
MultiplexSocket* multiplex_socket_incoming =
MultiplexSocket::CreateIncomingSocket(
fake_socket_ptr, std::string(SERVICE_ID_1), /*first_frame_len*/ 0);
ASSERT_NE(multiplex_socket_incoming, nullptr);
std::shared_ptr<MediumSocket> virtual_socket_shared =
multiplex_socket_incoming->GetVirtualSocket(std::string(SERVICE_ID_1));
ASSERT_NE(virtual_socket_shared, nullptr);
FakeSocket* virtual_socket =
down_cast<FakeSocket*>(virtual_socket_shared.get());
SingleThreadExecutor executor;
FakeSocket* socket = fake_socket_ptr.get();
executor.Execute([socket]() {
ByteArray connection_req_frame = ForConnectionRequest(
std::string(SERVICE_ID_2), "J7frzSmHK-VBTHjCKpf4ew");
auto& writer = socket->writer_1_;
LOG(INFO) << "writer_1_ Write start";
Base64Utils::WriteInt(writer.get(), connection_req_frame.size());
writer->Write(connection_req_frame.AsStringView());
writer->Flush();
LOG(INFO) << "writer_1_ Write end";
});
absl::SleepFor(absl::Milliseconds(100));
EXPECT_EQ(multiplex_socket_incoming->GetVirtualSocketCount(), 2);
virtual_socket->Close();
EXPECT_EQ(multiplex_socket_incoming->GetVirtualSocketCount(), 1);
multiplex_socket_incoming->ShutdownAll();
}
TEST(MultiplexSocketTest,
EstablishVirtualSocket_Timeout_BecauseNoConnectionResponse) {
auto fake_socket_ptr = std::make_shared<FakeSocket>(Medium::WIFI_LAN);
MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_1),
Medium::WIFI_LAN);
MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_2),
Medium::WIFI_LAN);
MultiplexSocket* multiplex_socket = MultiplexSocket::CreateOutgoingSocket(
fake_socket_ptr, std::string(SERVICE_ID_1));
ASSERT_NE(multiplex_socket, nullptr);
MultiplexSocket* multiplex_socket_2 = MultiplexSocket::CreateOutgoingSocket(
fake_socket_ptr, std::string(SERVICE_ID_2));
ASSERT_EQ(multiplex_socket_2, multiplex_socket);
multiplex_socket->Enable();
std::shared_ptr<MediumSocket> virtual_socket_shared =
multiplex_socket->GetVirtualSocket(std::string(SERVICE_ID_1));
ASSERT_NE(virtual_socket_shared, nullptr);
FakeSocket* virtual_socket =
down_cast<FakeSocket*>(virtual_socket_shared.get());
// This is a timeout test, the real timeout is 3s which is too long for a
// unit test, so we set a short timeout for flakiness test to avoid long wait
// time.
auto flags = FeatureFlags::GetInstance().GetFlags();
auto original_flags = flags;
flags.multiplex_socket_connection_response_timeout_millis =
absl::Milliseconds(200);
FeatureFlags::GetMutableInstanceForTesting().SetFlags(flags);
CountDownLatch latch(2);
SingleThreadExecutor establish_socket_executor;
establish_socket_executor.Execute([&multiplex_socket, &latch]() {
LOG(INFO) << "EstablishVirtualSocket";
std::shared_ptr<MediumSocket> socket =
multiplex_socket->EstablishVirtualSocket(std::string(SERVICE_ID_2));
LOG(INFO) << "EstablishVirtualSocket finished";
EXPECT_EQ(socket, nullptr);
latch.CountDown();
});
SingleThreadExecutor read_executor;
read_executor.Execute([&multiplex_socket, &fake_socket_ptr, &latch]() {
auto reader = fake_socket_ptr->reader_2_.get();
LOG(INFO) << "reader_2_ Read start";
ExceptionOr<std::int32_t> read_int = Base64Utils::ReadInt(reader);
if (!read_int.ok()) {
ADD_FAILURE() << "Failed to read. Exception:" << read_int.exception();
} else {
auto length = read_int.result();
LOG(INFO) << " length:" << length;
EXPECT_GT(length, 0);
}
EXPECT_EQ(multiplex_socket->GetVirtualSocket(std::string(SERVICE_ID_2)),
nullptr);
latch.CountDown();
});
EXPECT_TRUE(latch.Await(absl::Seconds(1)).result());
EXPECT_EQ(multiplex_socket->GetVirtualSocketCount(), 1);
virtual_socket->Close();
EXPECT_EQ(multiplex_socket->GetVirtualSocketCount(), 0);
multiplex_socket->ShutdownAll();
// Restore the original flags.
FeatureFlags::GetMutableInstanceForTesting().SetFlags(original_flags);
}
TEST(MultiplexSocketTest, EstablishVirtualSocket_RemoteAccepted) {
auto fake_socket_ptr = std::make_shared<FakeSocket>(Medium::BLUETOOTH);
MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_1),
Medium::BLUETOOTH);
MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_2),
Medium::BLUETOOTH);
MultiplexSocket* multiplex_socket = MultiplexSocket::CreateOutgoingSocket(
fake_socket_ptr, std::string(SERVICE_ID_1));
ASSERT_NE(multiplex_socket, nullptr);
MultiplexSocket* multiplex_socket_2 = MultiplexSocket::CreateOutgoingSocket(
fake_socket_ptr, std::string(SERVICE_ID_2));
ASSERT_EQ(multiplex_socket_2, multiplex_socket);
SingleThreadExecutor executor;
CountDownLatch latch(1);
executor.Execute([&multiplex_socket, &latch]() {
LOG(INFO) << "EstablishVirtualSocket";
std::shared_ptr<MediumSocket> socket =
multiplex_socket->EstablishVirtualSocket(std::string(SERVICE_ID_2));
EXPECT_EQ(socket, nullptr);
latch.CountDown();
});
latch.Await();
multiplex_socket->Enable();
executor.Execute([&multiplex_socket]() {
LOG(INFO) << "EstablishVirtualSocket";
std::shared_ptr<MediumSocket> socket =
multiplex_socket->EstablishVirtualSocket(std::string(SERVICE_ID_2));
EXPECT_NE(socket, nullptr);
});
auto reader = fake_socket_ptr->reader_2_.get();
LOG(INFO) << "reader_2_ Waiting for CONNECTION_REQUEST frame.";
ExceptionOr<std::int32_t> read_int = Base64Utils::ReadInt(reader);
if (!read_int.ok()) {
ADD_FAILURE() << "Failed to read length.Exception:" << read_int.exception();
}
auto length = read_int.result();
if (length < 0 ||
length >
FeatureFlags::GetInstance().GetFlags().connection_max_frame_length) {
ADD_FAILURE() << "Invalid length:" << length;
}
auto bytes = reader->ReadExactly(length);
if (!bytes.ok()) {
ADD_FAILURE() << "Failed to read frame. Exception:" << bytes.exception();
}
length = read_int.result();
if (length < 0 ||
length >
FeatureFlags::GetInstance().GetFlags().connection_max_frame_length) {
ADD_FAILURE() << "Invalid frame length:" << length;
}
ExceptionOr<MultiplexFrame> frame_exc = multiplex::FromBytes(bytes.result());
if (!frame_exc.ok()) {
ADD_FAILURE() << "Failed to parse MultiplexFrame. Exception:"
<< frame_exc.exception();
}
auto frame = frame_exc.result();
auto salted_service_id_hash =
ByteArray{std::move(frame.header().salted_service_id_hash())};
auto service_id_hash_salt = frame.header().has_service_id_hash_salt()
? frame.header().service_id_hash_salt()
: "";
ASSERT_EQ(frame.frame_type(), MultiplexFrame::CONTROL_FRAME);
auto control_frame = frame.control_frame();
ASSERT_EQ(control_frame.control_frame_type(),
MultiplexControlFrame::CONNECTION_REQUEST);
LOG(INFO) << "Recieved MultiplexControlFrame::CONNECTION_REQUEST "
"frame, now send CONNECTION_RESPONSE frame.";
ByteArray connection_response_frame =
ForConnectionResponse(salted_service_id_hash, service_id_hash_salt,
ConnectionResponseFrame::CONNECTION_ACCEPTED);
auto& writer = fake_socket_ptr->writer_1_;
LOG(INFO) << "writer_1_ Write start";
Base64Utils::WriteInt(writer.get(), connection_response_frame.size());
writer->Write(connection_response_frame.AsStringView());
writer->Flush();
LOG(INFO) << "writer_1_ Write end";
absl::SleepFor(absl::Milliseconds(100));
EXPECT_NE(multiplex_socket->GetVirtualSocket(std::string(SERVICE_ID_2)),
nullptr);
EXPECT_EQ(multiplex_socket->GetVirtualSocketCount(), 2);
LOG(INFO) << "Send Data frame on virtual socket for SERVICE_ID_2.";
ByteArray data_frame =
ForData(std::string(SERVICE_ID_2), service_id_hash_salt,
/*should_pass_salt=*/true, absl::string_view("data"));
Base64Utils::WriteInt(writer.get(), data_frame.size());
writer->Write(data_frame.AsStringView());
writer->Flush();
absl::SleepFor(absl::Milliseconds(100));
LOG(INFO) << "Send disconnection frame on virtual socket for SERVICE_ID_2.";
ByteArray disconnect_frame =
ForDisconnection(std::string(SERVICE_ID_2), service_id_hash_salt);
Base64Utils::WriteInt(writer.get(), disconnect_frame.size());
writer->Write(disconnect_frame.AsStringView());
writer->Flush();
absl::SleepFor(absl::Milliseconds(100));
EXPECT_EQ(multiplex_socket->GetVirtualSocketCount(), 1);
multiplex_socket->ShutdownAll();
}
} // namespace multiplex
} // namespace mediums
} // namespace connections
} // namespace nearby