diff --git a/Package.swift b/Package.swift index 6ac9729b..c0736224 100644 --- a/Package.swift +++ b/Package.swift @@ -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", diff --git a/connections/implementation/BUILD b/connections/implementation/BUILD index edc8fd47..61c37dc0 100644 --- a/connections/implementation/BUILD +++ b/connections/implementation/BUILD @@ -145,7 +145,6 @@ cc_library( "//connections:__pkg__", "//connections:partners", "//connections/implementation/fuzzers:__pkg__", - "//connections/implementation/mediums/multiplex:__pkg__", "//sharing:__subpackages__", ], deps = [ diff --git a/connections/implementation/mediums/BUILD b/connections/implementation/mediums/BUILD index df6b27fe..ad7736f8 100644 --- a/connections/implementation/mediums/BUILD +++ b/connections/implementation/mediums/BUILD @@ -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 = [ diff --git a/connections/implementation/mediums/multiplex/BUILD b/connections/implementation/mediums/multiplex/BUILD deleted file mode 100644 index 63c05094..00000000 --- a/connections/implementation/mediums/multiplex/BUILD +++ /dev/null @@ -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", - ], -) diff --git a/connections/implementation/mediums/multiplex/multiplex_frames.cc b/connections/implementation/mediums/multiplex/multiplex_frames.cc deleted file mode 100644 index 3eea2aa1..00000000 --- a/connections/implementation/mediums/multiplex/multiplex_frames.cc +++ /dev/null @@ -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 -#include - -#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 FromBytes(const ByteArray& multiplex_frame_bytes) { - MultiplexFrame frame; - - if (frame.ParseFromString(std::string(multiplex_frame_bytes))) { - if (!IsValid(frame)) { - return ExceptionOr(Exception::kInvalidProtocolBuffer); - } - return ExceptionOr(std::move(frame)); - } else { - return ExceptionOr(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 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 diff --git a/connections/implementation/mediums/multiplex/multiplex_frames.h b/connections/implementation/mediums/multiplex/multiplex_frames.h deleted file mode 100644 index 283697a4..00000000 --- a/connections/implementation/mediums/multiplex/multiplex_frames.h +++ /dev/null @@ -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 - -#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 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_ diff --git a/connections/implementation/mediums/multiplex/multiplex_frames_test.cc b/connections/implementation/mediums/multiplex/multiplex_frames_test.cc deleted file mode 100644 index 95a66803..00000000 --- a/connections/implementation/mediums/multiplex/multiplex_frames_test.cc +++ /dev/null @@ -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 -#include - -#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 diff --git a/connections/implementation/mediums/multiplex/multiplex_output_stream.cc b/connections/implementation/mediums/multiplex/multiplex_output_stream.cc deleted file mode 100644 index 0cdc2d52..00000000 --- a/connections/implementation/mediums/multiplex/multiplex_output_stream.cc +++ /dev/null @@ -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 -#include -#include -#include - -#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* 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 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 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 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 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 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( - 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( - 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* 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 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 diff --git a/connections/implementation/mediums/multiplex/multiplex_output_stream.h b/connections/implementation/mediums/multiplex/multiplex_output_stream.h deleted file mode 100644 index 22067d71..00000000 --- a/connections/implementation/mediums/multiplex/multiplex_output_stream.h +++ /dev/null @@ -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 -#include - -#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 - * - *

{@link MultiplexControlFrameType#CONNECTION_REQUEST} and {@link - * MultiplexControlFrameType#CONNECTION_RESPONSE} have the highest priority - * - *

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,... - * - *

{@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* 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* future, ByteArray data) - : future_(future), data_(data) {} - ~EnqueuedFrame() = default; - - Future* future_; - ByteArray data_; - }; - - class MultiplexWriter { - public: - explicit MultiplexWriter(OutputStream* physical_writer); - ~MultiplexWriter(); - - // Enqueues the frame to be sent out. - void EnqueueToSend(Future* 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 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> - virtual_output_streams_; - MultiplexWriter multiplex_writer_; -}; - -} // namespace multiplex -} // namespace mediums -} // namespace connections -} // namespace nearby - -#endif // CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_OUTPUT_STREAM_H_ diff --git a/connections/implementation/mediums/multiplex/multiplex_output_stream_test.cc b/connections/implementation/mediums/multiplex/multiplex_output_stream_test.cc deleted file mode 100644 index 4a5ae130..00000000 --- a/connections/implementation/mediums/multiplex/multiplex_output_stream_test.cc +++ /dev/null @@ -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 -#include -#include -#include - -#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 ReadFrame() { - ExceptionOr read_int = Base64Utils::ReadInt(reader_.get()); - if (!read_int.ok()) return read_int.GetException(); - if (read_int.result() <= 0) return {Exception::kFailed}; - - ExceptionOr 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> pipe_ = - CreatePipe(); - - std::unique_ptr reader_ = std::move(pipe_.first); - std::unique_ptr writer_ = std::move(pipe_.second); - std::unique_ptr multiplex_output_stream_; -}; - -TEST_F(MultiplexOutputStreamTest, SendConnectionRequestFrame) { - multiplex_output_stream_ = - std::make_unique(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(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(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(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(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(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(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(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 diff --git a/connections/implementation/mediums/multiplex/multiplex_socket.cc b/connections/implementation/mediums/multiplex/multiplex_socket.cc deleted file mode 100644 index 888efb82..00000000 --- a/connections/implementation/mediums/multiplex/multiplex_socket.cc +++ /dev/null @@ -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 -#include -#include -#include -#include - -#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(service_id, type), - std::move(incoming_connection_cb)); -} - -void MultiplexSocket::StopListeningForIncomingConnection( - const std::string& service_id, Medium type) { - GetIncomingConnectionCallbacks().erase( - std::pair(service_id, type)); -} - -MultiplexSocket::MultiplexSocket(std::shared_ptr 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, - MultiplexIncomingConnectionCb>& -MultiplexSocket::GetIncomingConnectionCallbacks() { - using MapType = absl::flat_hash_map, - 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 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 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 physical_socket, - const std::string& service_id) { - return CreateOutgoingSocket(physical_socket, service_id, - Utils::GenerateSalt()); -} - -std::shared_ptr 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 virtual_socket = - virtual_sockets_[salted_service_id_hash_key]; - virtual_socket->AddOnSocketClosedListener( - std::make_unique>( - [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 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 virtual_socket = - virtual_sockets_[salted_service_id_hash_key]; - virtual_socket->AddOnSocketClosedListener( - std::make_unique>( - [this, service_id]() { OnVirtualSocketClosed(service_id); })); - - return virtual_socket; -} - -std::shared_ptr 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> -MultiplexSocket::RegisterConnectionResponse(const std::string& service_id) { - auto future = std::make_shared>(); - 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 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(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 bytes; - ExceptionOr read_int; - if (first_frame_len_copy > 0) { - read_int = ExceptionOr(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 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 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 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 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 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( - &(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 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 diff --git a/connections/implementation/mediums/multiplex/multiplex_socket.h b/connections/implementation/mediums/multiplex/multiplex_socket.h deleted file mode 100644 index ed3fdd7d..00000000 --- a/connections/implementation/mediums/multiplex/multiplex_socket.h +++ /dev/null @@ -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 -#include -#include -#include - -#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; -using MultiplexIncomingConnectionCb = absl::AnyInvocable 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 physical_socket, - const std::string& service_id, std::int32_t first_frame_len); - // Creates a new outgoing MultiplexSocket. - static MultiplexSocket* CreateOutgoingSocket( - std::shared_ptr 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 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, - 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 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 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 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 physical_socket); - - // Creates the first virtual socket for the service id. The first virtual - // socket is created by the sender. - std::shared_ptr CreateFirstVirtualSocket( - const std::string& service_id, const std::string& service_id_hash_salt); - // Creates the virtual socket for the service id. - std::shared_ptr 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> - 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 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 runnable); - - // The physical socket connect to the remote device. - std::shared_ptr 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> enable_cb_ = - std::make_shared>([this]() { Enable(); }); - - // A map of service Id -> {@link SettableFuture} for waiting the - // ConnectionResponse. Non-empty while requesting the virtual socket. - absl::flat_hash_map>> - 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> - 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 reader_thread_shutdown_barrier_; -}; - -} // namespace multiplex -} // namespace mediums -} // namespace connections -} // namespace nearby - -#endif // CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_SOCKET_H_ diff --git a/connections/implementation/mediums/multiplex/multiplex_socket_test.cc b/connections/implementation/mediums/multiplex/multiplex_socket_test.cc deleted file mode 100644 index cf382eef..00000000 --- a/connections/implementation/mediums/multiplex/multiplex_socket_test.cc +++ /dev/null @@ -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 -#include -#include -#include - -#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>* - 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(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& GetByteReadFuture() { return bytes_read_future_; } - - std::pair, std::unique_ptr> - pipe_1_; - std::unique_ptr reader_1_; - std::unique_ptr writer_1_; - std::pair, std::unique_ptr> - pipe_2_; - std::unique_ptr reader_2_; - std::unique_ptr writer_2_; - - private: - bool is_virtual_socket_ = false; - Future bytes_read_future_; - absl::flat_hash_map>* - virtual_sockets_ptr_ = nullptr; - OutputStream* virtual_output_stream_ = nullptr; -}; - -TEST(MultiplexSocketTest, CreateIncomingSocketSuccess) { - auto fake_socket_ptr = std::make_shared(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 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 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 virtual_socket_shared = - multiplex_socket_incoming->GetVirtualSocket(std::string(SERVICE_ID_1)); - ASSERT_NE(virtual_socket_shared, nullptr); - FakeSocket* virtual_socket = - down_cast(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 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(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(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 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 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 virtual_socket_shared = - multiplex_socket_incoming->GetVirtualSocket(std::string(SERVICE_ID_1)); - ASSERT_NE(virtual_socket_shared, nullptr); - FakeSocket* virtual_socket = - down_cast(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(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 virtual_socket_shared = - multiplex_socket->GetVirtualSocket(std::string(SERVICE_ID_1)); - ASSERT_NE(virtual_socket_shared, nullptr); - FakeSocket* virtual_socket = - down_cast(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 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 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(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 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 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 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 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