[BLEREFACTOR]: Implement the BLE/L2CAP read/write logic.

PiperOrigin-RevId: 831637265

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",

connections/implementation/base_endpoint_channel.cc

@@ -116,8 +116,15 @@ ExceptionOr<ByteArray> BaseEndpointChannel::Read(
     if (NearbyFlags::GetInstance().GetBoolFlag(
             config_package_nearby::nearby_connections_feature::
                 kRefactorBleL2cap)) {
-      // TODO(edwinwu): Implement the new read logic.
-      return ExceptionOr<ByteArray>(Exception::kFailed);
+      ExceptionOr<ByteArray> 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<ByteArray>(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<std::int32_t>(data_size));
     }

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<ByteArray> DispatchPacket() {
+    return ExceptionOr<ByteArray>{};
+  }
+
+  // Reads the length of the next incoming data packet from the underlying
+  // medium.
+  virtual ExceptionOr<std::int32_t> ReadPayloadLength() {
+    return ExceptionOr<std::int32_t>{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.

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<ByteArray>, 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<ByteArray>(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<ByteArray>(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.

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<ByteArray> BleEndpointChannel::DispatchPacket() {
+  return ble_socket_2_->DispatchPacket();
+}
+
+Exception BleEndpointChannel::WritePayloadLength(int payload_length) {
+  return ble_socket_2_->WritePayloadLength(payload_length);
+}
+
 }  // namespace connections
 }  // namespace nearby

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<ByteArray> DispatchPacket() override;
+  Exception WritePayloadLength(int payload_length) override;
+
  private:
   static constexpr int kDefaultBleMaxTransmitPacketSize = 512;  // 512 bytes
 

connections/implementation/ble_l2cap_endpoint_channel.cc

@@ -14,6 +14,7 @@
 
 #include "connections/implementation/ble_l2cap_endpoint_channel.h"
 
+#include <cstdint>
 #include <memory>
 #include <string>
 #include <utility>
@@ -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<ByteArray> BleL2capEndpointChannel::DispatchPacket() {
+  return ble_l2cap_socket_2_->DispatchPacket();
+}
+
+ExceptionOr<std::int32_t> 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

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 <cstdint>
 #include <memory>
 #include <string>
 
 #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<ByteArray> DispatchPacket() override;
+  ExceptionOr<std::int32_t> ReadPayloadLength() override;
+  Exception WritePayloadLength(int payload_length) override;
+
  private:
   void CloseImpl() override;
 

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",

connections/implementation/mediums/ble/ble_socket.cc

@@ -16,10 +16,17 @@
 
 #include <cstdint>
 #include <memory>
+#include <string>
 #include <utility>
 
+#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<ByteArray> 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<BlePacket> 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<BleInputStream> ble_input_stream,
                      std::unique_ptr<BleOutputStream> ble_output_stream,
@@ -164,13 +201,27 @@ ExceptionOr<ByteArray> BleSocket::DispatchPacket() {
 }
 
 ExceptionOr<std::int32_t> BleSocket::ReadPayloadLength() {
-  // TODO(b/419654808): Implement this method.
-  return {Exception::kFailed};
+  MutexLock lock(&mutex_);
+  if (!ble_input_stream_) {
+    return {Exception::kIo};
+  }
+
+  ExceptionOr<ByteArray> 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<std::int32_t>(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() {

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_;