diff --git a/connections/implementation/BUILD b/connections/implementation/BUILD index ba70a388..0ad6bbf1 100644 --- a/connections/implementation/BUILD +++ b/connections/implementation/BUILD @@ -459,6 +459,8 @@ cc_test( ], deps = [ ":internal", + "//connections/implementation/flags:connections_flags", + "//internal/flags:nearby_flags", "//internal/platform:base", "//internal/platform:logging", "//internal/platform:types", diff --git a/connections/implementation/base_endpoint_channel.cc b/connections/implementation/base_endpoint_channel.cc index b34785d0..b18db1d4 100644 --- a/connections/implementation/base_endpoint_channel.cc +++ b/connections/implementation/base_endpoint_channel.cc @@ -116,8 +116,15 @@ ExceptionOr BaseEndpointChannel::Read( if (NearbyFlags::GetInstance().GetBoolFlag( config_package_nearby::nearby_connections_feature:: kRefactorBleL2cap)) { - // TODO(edwinwu): Implement the new read logic. - return ExceptionOr(Exception::kFailed); + ExceptionOr read_control_block_bytes = DispatchPacket(); + if (!read_control_block_bytes.ok()) { + LOG(WARNING) << __func__ << ": Failed to dispatch packet: " + << read_control_block_bytes.exception(); + return ExceptionOr(read_control_block_bytes.exception()); + } + + read_int = + (GetMedium() == BLE_L2CAP) ? ReadPayloadLength() : ReadInt(reader_); } else { read_int = ReadInt(reader_); } @@ -244,8 +251,7 @@ Exception BaseEndpointChannel::Write(const ByteArray& data, config_package_nearby::nearby_connections_feature:: kRefactorBleL2cap) && (GetMedium() == BLE || GetMedium() == BLE_L2CAP)) { - // TODO(edwinwu): Implement the new write logic. - write_exception = {Exception::kFailed}; + write_exception = WritePayloadLength(data_size); } else { write_exception = WriteInt(writer_, static_cast(data_size)); } diff --git a/connections/implementation/base_endpoint_channel.h b/connections/implementation/base_endpoint_channel.h index 3a3f6dc8..66faf4eb 100644 --- a/connections/implementation/base_endpoint_channel.h +++ b/connections/implementation/base_endpoint_channel.h @@ -25,8 +25,8 @@ #include "connections/implementation/analytics/analytics_recorder.h" #include "connections/implementation/analytics/packet_meta_data.h" #include "connections/implementation/endpoint_channel.h" -#include "internal/platform/condition_variable.h" #include "internal/platform/byte_array.h" +#include "internal/platform/condition_variable.h" #include "internal/platform/exception.h" #include "internal/platform/input_stream.h" #include "internal/platform/mutex.h" @@ -90,6 +90,23 @@ class BaseEndpointChannel : public EndpointChannel { void SetAnalyticsRecorder(analytics::AnalyticsRecorder* analytics_recorder, const std::string& endpoint_id) override; + // Reads a complete packet from the underlying medium. + virtual ExceptionOr DispatchPacket() { + return ExceptionOr{}; + } + + // Reads the length of the next incoming data packet from the underlying + // medium. + virtual ExceptionOr ReadPayloadLength() { + return ExceptionOr{0}; + } + + // Writes the length of a data packet to the underlying medium before writing + // the packet itself. + virtual Exception WritePayloadLength(int payload_length) { + return {Exception::kFailed}; + } + protected: virtual void CloseImpl() = 0; // For tests only. diff --git a/connections/implementation/base_endpoint_channel_test.cc b/connections/implementation/base_endpoint_channel_test.cc index 306bede6..4e8fa93e 100644 --- a/connections/implementation/base_endpoint_channel_test.cc +++ b/connections/implementation/base_endpoint_channel_test.cc @@ -31,7 +31,9 @@ #include "connections/implementation/client_proxy.h" #include "connections/implementation/encryption_runner.h" #include "connections/implementation/endpoint_channel.h" +#include "connections/implementation/flags/nearby_connections_feature_flags.h" #include "connections/implementation/offline_frames.h" +#include "internal/flags/nearby_flags.h" #include "internal/platform/byte_array.h" #include "internal/platform/count_down_latch.h" #include "internal/platform/exception.h" @@ -58,6 +60,7 @@ class TestEndpointChannel : public BaseEndpointChannel { using BaseEndpointChannel::EncodeMessageForTests; + MOCK_METHOD(ExceptionOr, DispatchPacket, (), (override)); MOCK_METHOD(Medium, GetMedium, (), (const, override)); MOCK_METHOD(void, CloseImpl, (), (override)); }; @@ -163,13 +166,77 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a, return std::make_pair(std::move(context_a), std::move(context_b)); } -TEST(BaseEndpointChannelTest, ConstructorDestructorWorks) { +class BaseEndpointChannelTest : public ::testing::Test { + protected: + void TearDown() override { + // Restore any overridden flags after each test to ensure test isolation. + NearbyFlags::GetInstance().ResetOverridedValues(); + } + + const ByteArray kTestData{"test_data"}; +}; + +TEST_F(BaseEndpointChannelTest, ReadSucceedsWhenFlagDisabled) { + NearbyFlags::GetInstance().OverrideBoolFlagValue( + config_package_nearby::nearby_connections_feature::kRefactorBleL2cap, + false); + + auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. + auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. + TestEndpointChannel channel_a(pipe_b.first.get(), pipe_a.second.get()); + TestEndpointChannel channel_b(pipe_a.first.get(), pipe_b.second.get()); + + channel_a.Write(kTestData); + ByteArray rx_message = std::move(channel_b.Read().result()); + EXPECT_EQ(rx_message, kTestData); +} + +TEST_F(BaseEndpointChannelTest, ReadCallsDispatchPacketWhenFlagEnabled) { + NearbyFlags::GetInstance().OverrideBoolFlagValue( + config_package_nearby::nearby_connections_feature::kRefactorBleL2cap, + true); + + auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. + auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. + TestEndpointChannel channel_a(pipe_b.first.get(), pipe_a.second.get()); + TestEndpointChannel channel_b(pipe_a.first.get(), pipe_b.second.get()); + + EXPECT_CALL(channel_b, DispatchPacket) + .WillOnce(::testing::Return(ExceptionOr(kTestData))); + + channel_a.Write(kTestData); + + auto read_byte = channel_b.Read(); + EXPECT_TRUE(read_byte.ok()); + EXPECT_EQ(read_byte.result(), kTestData); +} + +TEST_F(BaseEndpointChannelTest, + ReadPropagatesFailureFromDispatchPacketWhenFlagEnabled) { + NearbyFlags::GetInstance().OverrideBoolFlagValue( + config_package_nearby::nearby_connections_feature::kRefactorBleL2cap, + true); + + auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. + auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. + TestEndpointChannel channel_a(pipe_b.first.get(), pipe_a.second.get()); + TestEndpointChannel channel_b(pipe_a.first.get(), pipe_b.second.get()); + + EXPECT_CALL(channel_b, DispatchPacket) + .WillOnce(::testing::Return(ExceptionOr(Exception::kIo))); + + auto read_byte = channel_b.Read(); + EXPECT_FALSE(read_byte.ok()); + EXPECT_EQ(read_byte.GetException().value, Exception::kIo); +} + +TEST_F(BaseEndpointChannelTest, ConstructorDestructorWorks) { auto [input, output] = CreatePipe(); TestEndpointChannel test_channel(input.get(), output.get()); } -TEST(BaseEndpointChannelTest, ReadWrite) { +TEST_F(BaseEndpointChannelTest, ReadWrite) { // Direct not-encrypted IO. auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. @@ -181,7 +248,7 @@ TEST(BaseEndpointChannelTest, ReadWrite) { EXPECT_EQ(rx_message, tx_message); } -TEST(BaseEndpointChannelTest, ChannelUnencryptedByDefault) { +TEST_F(BaseEndpointChannelTest, ChannelUnencryptedByDefault) { auto pipe = CreatePipe(); TestEndpointChannel channel(pipe.first.get(), pipe.second.get()); @@ -192,7 +259,7 @@ TEST(BaseEndpointChannelTest, ChannelUnencryptedByDefault) { EXPECT_EQ(result.exception(), Exception::kFailed); } -TEST(BaseEndpointChannelTest, TryDecrypt) { +TEST_F(BaseEndpointChannelTest, TryDecrypt) { absl::string_view kMessage = "message"; auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. @@ -214,7 +281,7 @@ TEST(BaseEndpointChannelTest, TryDecrypt) { EXPECT_EQ(decrypted_message.result().AsStringView(), kMessage); } -TEST(BaseEndpointChannelTest, TryDecryptFailsWhenDecryptionFails) { +TEST_F(BaseEndpointChannelTest, TryDecryptFailsWhenDecryptionFails) { auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. TestEndpointChannel channel_a(pipe_b.first.get(), pipe_a.second.get()); @@ -231,7 +298,7 @@ TEST(BaseEndpointChannelTest, TryDecryptFailsWhenDecryptionFails) { EXPECT_EQ(result.exception(), Exception::kExecution); } -TEST(BaseEndpointChannelTest, NotEncryptedReadWriteCanBeIntercepted) { +TEST_F(BaseEndpointChannelTest, NotEncryptedReadWriteCanBeIntercepted) { // Not encrypted IO; MITM scenario. // Setup test communication environment. @@ -282,7 +349,7 @@ TEST(BaseEndpointChannelTest, NotEncryptedReadWriteCanBeIntercepted) { channel_b.Close(DisconnectionReason::REMOTE_DISCONNECTION); } -TEST(BaseEndpointChannelTest, EncryptedReadWriteCanNotBeIntercepted) { +TEST_F(BaseEndpointChannelTest, EncryptedReadWriteCanNotBeIntercepted) { // Encrypted IO; MITM scenario. // Setup test communication environment. @@ -346,7 +413,7 @@ TEST(BaseEndpointChannelTest, EncryptedReadWriteCanNotBeIntercepted) { channel_b.Close(DisconnectionReason::REMOTE_DISCONNECTION); } -TEST(BaseEndpointChannelTest, CanBesuspendedAndResumed) { +TEST_F(BaseEndpointChannelTest, CanBesuspendedAndResumed) { // Setup test communication environment. auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. @@ -398,7 +465,7 @@ TEST(BaseEndpointChannelTest, CanBesuspendedAndResumed) { channel_b.Close(DisconnectionReason::REMOTE_DISCONNECTION); } -TEST(BaseEndpointChannelTest, ReadAfterInputStreamClosed) { +TEST_F(BaseEndpointChannelTest, ReadAfterInputStreamClosed) { auto [input, output] = CreatePipe(); TestEndpointChannel test_channel(input.get(), output.get()); @@ -413,7 +480,7 @@ TEST(BaseEndpointChannelTest, ReadAfterInputStreamClosed) { ASSERT_TRUE(read_data.GetException().Raised(Exception::kIo)); } -TEST(BaseEndpointChannelTest, ReadUnencryptedFrameOnEncryptedChannel) { +TEST_F(BaseEndpointChannelTest, ReadUnencryptedFrameOnEncryptedChannel) { // Setup test communication environment. auto pipe_a = CreatePipe(); // channel_a writes to pipe_a, reads from pipe_b. auto pipe_b = CreatePipe(); // channel_b writes to pipe_b, reads from pipe_a. diff --git a/connections/implementation/ble_endpoint_channel.cc b/connections/implementation/ble_endpoint_channel.cc index b25cad35..4146b7ac 100644 --- a/connections/implementation/ble_endpoint_channel.cc +++ b/connections/implementation/ble_endpoint_channel.cc @@ -21,6 +21,7 @@ #include "connections/implementation/base_endpoint_channel.h" #include "connections/implementation/mediums/ble/ble_socket.h" #include "internal/platform/ble.h" +#include "internal/platform/byte_array.h" #include "internal/platform/exception.h" #include "internal/platform/input_stream.h" #include "internal/platform/logging.h" @@ -116,5 +117,13 @@ void BleEndpointChannel::CloseImpl() { LOG(INFO) << "BleEndpointChannel " << GetName() << " is already closed."; } +ExceptionOr BleEndpointChannel::DispatchPacket() { + return ble_socket_2_->DispatchPacket(); +} + +Exception BleEndpointChannel::WritePayloadLength(int payload_length) { + return ble_socket_2_->WritePayloadLength(payload_length); +} + } // namespace connections } // namespace nearby diff --git a/connections/implementation/ble_endpoint_channel.h b/connections/implementation/ble_endpoint_channel.h index 8c2e0fd4..7ad7fc4e 100644 --- a/connections/implementation/ble_endpoint_channel.h +++ b/connections/implementation/ble_endpoint_channel.h @@ -21,6 +21,8 @@ #include "connections/implementation/base_endpoint_channel.h" #include "connections/implementation/mediums/ble/ble_socket.h" #include "internal/platform/ble.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" namespace nearby { namespace connections { @@ -40,6 +42,9 @@ class BleEndpointChannel final : public BaseEndpointChannel { int GetMaxTransmitPacketSize() const override; + ExceptionOr DispatchPacket() override; + Exception WritePayloadLength(int payload_length) override; + private: static constexpr int kDefaultBleMaxTransmitPacketSize = 512; // 512 bytes diff --git a/connections/implementation/ble_l2cap_endpoint_channel.cc b/connections/implementation/ble_l2cap_endpoint_channel.cc index 4734bb62..33056d69 100644 --- a/connections/implementation/ble_l2cap_endpoint_channel.cc +++ b/connections/implementation/ble_l2cap_endpoint_channel.cc @@ -14,6 +14,7 @@ #include "connections/implementation/ble_l2cap_endpoint_channel.h" +#include #include #include #include @@ -25,6 +26,7 @@ #include "internal/platform/input_stream.h" #include "internal/platform/logging.h" #include "internal/platform/output_stream.h" +#include "internal/platform/byte_array.h" namespace nearby { namespace connections { @@ -118,5 +120,17 @@ void BleL2capEndpointChannel::CloseImpl() { LOG(INFO) << "BleL2capEndpointChannel " << GetName() << " is already closed."; } +ExceptionOr BleL2capEndpointChannel::DispatchPacket() { + return ble_l2cap_socket_2_->DispatchPacket(); +} + +ExceptionOr BleL2capEndpointChannel::ReadPayloadLength() { + return ble_l2cap_socket_2_->ReadPayloadLength(); +} + +Exception BleL2capEndpointChannel::WritePayloadLength(int payload_length) { + return ble_l2cap_socket_2_->WritePayloadLength(payload_length); +} + } // namespace connections } // namespace nearby diff --git a/connections/implementation/ble_l2cap_endpoint_channel.h b/connections/implementation/ble_l2cap_endpoint_channel.h index 23e3ca3f..292d67ea 100644 --- a/connections/implementation/ble_l2cap_endpoint_channel.h +++ b/connections/implementation/ble_l2cap_endpoint_channel.h @@ -15,12 +15,15 @@ #ifndef CORE_INTERNAL_BLE_L2CAP_ENDPOINT_CHANNEL_H_ #define CORE_INTERNAL_BLE_L2CAP_ENDPOINT_CHANNEL_H_ +#include #include #include #include "connections/implementation/base_endpoint_channel.h" #include "connections/implementation/mediums/ble/ble_socket.h" #include "internal/platform/ble.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" namespace nearby { namespace connections { @@ -43,6 +46,10 @@ class BleL2capEndpointChannel final : public BaseEndpointChannel { // Returns the maximum transmit packet size of this endpoint channel. int GetMaxTransmitPacketSize() const override; + ExceptionOr DispatchPacket() override; + ExceptionOr ReadPayloadLength() override; + Exception WritePayloadLength(int payload_length) override; + private: void CloseImpl() override; diff --git a/connections/implementation/mediums/ble/BUILD b/connections/implementation/mediums/ble/BUILD index 7de679c8..c822edc5 100644 --- a/connections/implementation/mediums/ble/BUILD +++ b/connections/implementation/mediums/ble/BUILD @@ -61,11 +61,18 @@ cc_library( "//internal/platform/implementation/windows:__pkg__", ], deps = [ + ":ble", + "//connections/implementation/flags:connections_flags", + "//internal/flags:nearby_flags", "//internal/platform:base", "//internal/platform:comm", "//internal/platform:logging", "//internal/platform:types", + "//internal/platform:util", "@com_google_absl//absl/base:core_headers", + "@com_google_absl//absl/status:statusor", + "@com_google_absl//absl/strings", + "@com_google_absl//absl/synchronization", ], ) @@ -73,6 +80,7 @@ cc_test( name = "ble_socket_test", srcs = ["ble_socket_test.cc"], deps = [ + ":ble", ":ble_socket", "//internal/platform:base", "//internal/platform:comm", diff --git a/connections/implementation/mediums/ble/ble_socket.cc b/connections/implementation/mediums/ble/ble_socket.cc index 5b5072b9..3c258a9f 100644 --- a/connections/implementation/mediums/ble/ble_socket.cc +++ b/connections/implementation/mediums/ble/ble_socket.cc @@ -16,10 +16,17 @@ #include #include +#include #include +#include "absl/status/statusor.h" +#include "absl/strings/str_cat.h" +#include "connections/implementation/flags/nearby_connections_feature_flags.h" +#include "connections/implementation/mediums/ble/ble_packet.h" +#include "internal/flags/nearby_flags.h" #include "internal/platform/ble.h" #include "internal/platform/byte_array.h" +#include "internal/platform/byte_utils.h" #include "internal/platform/exception.h" #include "internal/platform/input_stream.h" #include "internal/platform/logging.h" @@ -39,14 +46,44 @@ ExceptionOr BleInputStream::Read(std::int64_t size) { Exception BleInputStream::Close() { return source_.Close(); } Exception BleOutputStream::Write(const ByteArray& data) { - // TODO(b/419654808): Implement this method. - return {Exception::kFailed}; + if (NearbyFlags::GetInstance().GetBoolFlag( + config_package_nearby::nearby_connections_feature:: + kRefactorBleL2cap)) { + if (!payload_length_) { + return {Exception::kFailed}; + } + // Prepend the packet length to the data. + std::string packet_str = + absl::StrCat(std::string(byte_utils::IntToBytes(payload_length_)), + std::string(data)); + payload_length_ = 0; + + // Prepend the service id hash to the data with the payload length. + absl::StatusOr ble_packet_status_or = + BlePacket::CreateDataPacket(service_id_hash_, + ByteArray(std::move(packet_str))); + if (!ble_packet_status_or.ok()) { + return {Exception::kFailed}; + } + return source_.Write(ByteArray(ble_packet_status_or.value())); + } else { + return source_.Write(data); + } } Exception BleOutputStream::Flush() { return source_.Flush(); } Exception BleOutputStream::Close() { return source_.Close(); } +Exception BleOutputStream::WritePayloadLength(int payload_length) { + if (payload_length_ != 0) { + return {Exception::kFailed}; + } + // Store the payload length to be prepended to the data later. + payload_length_ = payload_length; + return {Exception::kSuccess}; +} + BleSocket::BleSocket(const ByteArray& service_id_hash, std::unique_ptr ble_input_stream, std::unique_ptr ble_output_stream, @@ -164,13 +201,27 @@ ExceptionOr BleSocket::DispatchPacket() { } ExceptionOr BleSocket::ReadPayloadLength() { - // TODO(b/419654808): Implement this method. - return {Exception::kFailed}; + MutexLock lock(&mutex_); + if (!ble_input_stream_) { + return {Exception::kIo}; + } + + ExceptionOr read_bytes = + ble_input_stream_->Read(sizeof(std::int32_t)); + if (!read_bytes.ok()) { + return read_bytes.exception(); + } + + int payload_length = byte_utils::BytesToInt(std::move(read_bytes.result())); + return ExceptionOr(payload_length); } Exception BleSocket::WritePayloadLength(int payload_length) { - // TODO(b/419654808): Implement this method. - return {Exception::kFailed}; + MutexLock lock(&mutex_); + if (!ble_output_stream_) { + return {Exception::kIo}; + } + return ble_output_stream_->WritePayloadLength(payload_length); } Exception BleSocket::SendIntroduction() { diff --git a/connections/implementation/mediums/ble/ble_socket.h b/connections/implementation/mediums/ble/ble_socket.h index da7041e1..542a0458 100644 --- a/connections/implementation/mediums/ble/ble_socket.h +++ b/connections/implementation/mediums/ble/ble_socket.h @@ -105,6 +105,21 @@ class BleOutputStream : public OutputStream { Exception Flush() override; Exception Close() override; + /** + * Sends the length of a data payload to the remote endpoint. + * + * This method prepares the socket for an upcoming data payload by sending its + * length, including the necessary `service_id_hash` prefix. + * + * This method must be called immediately before sending the corresponding + * payload. + * + * @param payload_length The length, in bytes, of the upcoming payload. + * @return An `Exception` object indicating the status of the write + * operation. `{Exception::kSuccess}` on success. + */ + Exception WritePayloadLength(int payload_length); + private: OutputStream& source_; const ByteArray service_id_hash_;