Nearby Connection iOS L2CAP Socket Implementation

PiperOrigin-RevId: 750068733

internal/platform/implementation/apple/ble_l2cap_server_socket.h

@@ -12,18 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// Note: File language is detected using heuristics. Many Objective-C++ headers
-// are incorrectly classified as C++ resulting in invalid linter errors. The use
-// of "NSArray" and other Foundation classes like "NSData", "NSDictionary" and
-// "NSUUID" are highly weighted for Objective-C and Objective-C++ scores. Oddly,
-// "#import <Foundation/Foundation.h>" does not contribute any points. This
-// comment alone should be enough to trick the IDE in to believing this is
-// actually some sort of Objective-C file. See:
-// cs/google3/devtools/search/lang/recognize_language_classifiers_data
-
 #include <memory>
 
-#import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPServer.h"
+#include "absl/functional/any_invocable.h"
+#include "internal/platform/exception.h"
+#include "internal/platform/implementation/ble_v2.h"
 #import "internal/platform/implementation/apple/ble_l2cap_socket.h"
 
 namespace nearby {
@@ -33,14 +26,16 @@ namespace apple {
 class BleL2capServerSocket : public api::ble_v2::BleL2capServerSocket {
  public:
   BleL2capServerSocket() = default;
-  ~BleL2capServerSocket() override = default;
+  ~BleL2capServerSocket() override;
 
   // Gets PSM value has been published by the server.
   int GetPSM() const override;
-
+  
   // Sets PSM value has been published by the server.
-  void SetPSM(int PSM);
+  void SetPSM(int psm);
 
+  // Wait for an available socket.
+  //
   // Blocks until either:
   // - at least one incoming connection request is available, or
   // - ServerSocket is closed.
@@ -48,17 +43,25 @@ class BleL2capServerSocket : public api::ble_v2::BleL2capServerSocket {
   // Returns nullptr on error.
   // Once error is reported, it is permanent, and L2CAP ServerSocket has to be
   // closed.
-  std::unique_ptr<api::ble_v2::BleL2capSocket> Accept() override;
+  std::unique_ptr<api::ble_v2::BleL2capSocket> Accept() override
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Closes the L2CAP server socket.
   Exception Close() override;
 
-  // Connects to the L2CAP server socket.
-  bool Connect(std::unique_ptr<BleL2capSocket> socket);
+  // Adds a pending socket to the server socket.
+  bool AddPendingSocket(std::unique_ptr<BleL2capSocket> socket);
 
  private:
-  // The PSM value of the L2CAP server socket.
-  int PSM_ = 0;
+  Exception DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  mutable absl::Mutex mutex_;
+  absl::CondVar cond_;
+  absl::flat_hash_set<std::unique_ptr<BleL2capSocket>> pending_sockets_
+      ABSL_GUARDED_BY(mutex_);
+  absl::AnyInvocable<void()> close_notifier_ ABSL_GUARDED_BY(mutex_);
+  bool closed_ ABSL_GUARDED_BY(mutex_) = false;
+  int psm_ = 0;
 };
 
 }  // namespace apple

internal/platform/implementation/apple/ble_l2cap_server_socket.mm

@@ -20,28 +20,62 @@
 #include <utility>
 
 #import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPServer.h"
-#import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPStream.h"
-#import "internal/platform/implementation/apple/ble_l2cap_socket.h"
 #import "GoogleToolboxForMac/GTMLogger.h"
 
 namespace nearby {
 namespace apple {
 
-int BleL2capServerSocket::GetPSM() const { return PSM_; }
-
-void BleL2capServerSocket::SetPSM(int PSM) { PSM_ = PSM; }
-
-std::unique_ptr<api::ble_v2::BleL2capSocket> BleL2capServerSocket::Accept() {
-  // TODO: b/399815436 - Implement to accept incoming l2cap connection.
-  return nullptr;
+BleL2capServerSocket::~BleL2capServerSocket() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
 }
 
-bool BleL2capServerSocket::Connect(std::unique_ptr<BleL2capSocket> socket) {
-  // TODO: b/399815436 - Implement to connect to l2cap server socket.
-  return false;
+int BleL2capServerSocket::GetPSM() const { return psm_; }
+
+void BleL2capServerSocket::SetPSM(int psm) { psm_ = psm; }
+
+// TODO: b/399815436 - Refactor Accept() and AddPendingSocket() for better readability.
+std::unique_ptr<api::ble_v2::BleL2capSocket> BleL2capServerSocket::Accept() {
+  absl::MutexLock lock(&mutex_);
+  while (!closed_ && pending_sockets_.empty()) {
+    cond_.Wait(&mutex_);
+  }
+  if (closed_) return {};
+
+  std::unique_ptr<BleL2capSocket> remote_socket =
+      std::move(pending_sockets_.extract(pending_sockets_.begin()).value());
+  return std::move(remote_socket);
+}
+
+bool BleL2capServerSocket::AddPendingSocket(std::unique_ptr<BleL2capSocket> socket) {
+  absl::MutexLock lock(&mutex_);
+  if (closed_) {
+    return false;
+  }
+  pending_sockets_.insert(std::move(socket));
+  cond_.SignalAll();
+  return !closed_;
 }
 
 Exception BleL2capServerSocket::Close() {
+  absl::MutexLock lock(&mutex_);
+  return DoClose();
+}
+
+Exception BleL2capServerSocket::DoClose() {
+  bool should_notify = !closed_;
+  closed_ = true;
+  if (should_notify) {
+    cond_.SignalAll();
+    if (close_notifier_) {
+      auto notifier = std::move(close_notifier_);
+      mutex_.Unlock();
+      // Notifier may contain calls to public API, and may cause deadlock, if
+      // mutex_ is held during the call.
+      notifier();
+      mutex_.Lock();
+    }
+  }
   return {Exception::kSuccess};
 }
 

internal/platform/implementation/apple/ble_l2cap_socket.h

@@ -14,21 +14,21 @@
 
 #include <memory>
 
-#import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPStream.h"
-#import "internal/platform/implementation/apple/ble_peripheral.h"
 #include "internal/platform/implementation/ble_v2.h"
 
+#import "internal/platform/implementation/apple/ble_peripheral.h"
+
+@class GNCMConnectionHandlers;
+@class GNCBLEL2CAPConnection;
+
 namespace nearby {
 namespace apple {
 
 /** A readable stream of bytes. */
 class BleL2capInputStream : public InputStream {
  public:
-  // Creates a BleL2capInputStream.
-  //
-  // @param stream The underlying stream to use for reading and writing.
-  explicit BleL2capInputStream(GNCBLEL2CAPStream* stream);
-  ~BleL2capInputStream() override = default;
+  explicit BleL2capInputStream(GNCBLEL2CAPConnection* connection);
+  ~BleL2capInputStream() override;
 
   // Reads at most `size` bytes from the input stream.
   //
@@ -41,22 +41,24 @@ class BleL2capInputStream : public InputStream {
   Exception Close() override;
 
  private:
-  GNCBLEL2CAPStream* stream_;
+  GNCMConnectionHandlers *connectionHandlers_;
+  GNCBLEL2CAPConnection* connection_;
+  NSMutableArray<NSData *> *newDataPackets_;
+  NSMutableData *accumulatedData_;
+  NSCondition *condition_;
 };
 
 /** A writable stream of bytes. */
 class BleL2capOutputStream : public OutputStream {
  public:
-  // Creates a BleL2capOutputStream.
-  //
-  // @param stream The underlying stream to use for reading and writing.
-  explicit BleL2capOutputStream(GNCBLEL2CAPStream* stream);
-  ~BleL2capOutputStream() override = default;
+  explicit BleL2capOutputStream(GNCBLEL2CAPConnection* connection)
+      : connection_(connection), condition_([[NSCondition alloc] init]) {}
+  ~BleL2capOutputStream() override;
 
   // Write the provided bytes to the output stream.
   //
   // Returns Exception::kIo on error, otherwise Exception::kSuccess.
-  Exception Write(const ByteArray& data) override;
+  Exception Write(const ByteArray &data) override;
 
   // no-op
   //
@@ -69,7 +71,8 @@ class BleL2capOutputStream : public OutputStream {
   Exception Close() override;
 
  private:
-  GNCBLEL2CAPStream* stream_;
+  GNCBLEL2CAPConnection* connection_;
+  NSCondition *condition_;
 };
 
 /**
@@ -77,46 +80,47 @@ class BleL2capOutputStream : public OutputStream {
  */
 class BleL2capSocket : public api::ble_v2::BleL2capSocket {
  public:
-  explicit BleL2capSocket(GNCBLEL2CAPStream* stream);
+  explicit BleL2capSocket(GNCBLEL2CAPConnection* connection);
 
-  // The peripheral used to create the socket must outlive the socket or
-  // undefined behavior will occur.
-  BleL2capSocket(GNCBLEL2CAPStream* stream,
-                 api::ble_v2::BlePeripheral* peripheral);
-  ~BleL2capSocket() override = default;
+  // The peripheral used to create the socket must outlive the socket or undefined behavior will
+  // occur.
+  BleL2capSocket(GNCBLEL2CAPConnection* connection, api::ble_v2::BlePeripheral *peripheral);
+  ~BleL2capSocket() override;
 
   // Returns the InputStream of the BleL2capSocket.
   // On error, returned stream will report Exception::kIo on any operation.
   //
   // The returned object is not owned by the caller, and can be invalidated once
   // the BleL2capSocket object is destroyed.
-  InputStream& GetInputStream() override;
+  BleL2capInputStream &GetInputStream() override { return *input_stream_; }
 
   // Returns the OutputStream of the BleL2capSocket.
   // On error, returned stream will report Exception::kIo on any operation.
   //
   // The returned object is not owned by the caller, and can be invalidated once
   // the BleL2capSocket object is destroyed.
-  OutputStream& GetOutputStream() override;
+  BleL2capOutputStream &GetOutputStream() override { return *output_stream_; }
 
   // Returns Exception::kIo on error, Exception::kSuccess otherwise.
-  Exception Close() override;
-
+  Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_);
   // Sets the close notifier by client side.
-  void SetCloseNotifier(absl::AnyInvocable<void()> notifier) override;
+  void SetCloseNotifier(absl::AnyInvocable<void()> notifier) override {};
 
   // Returns valid BlePeripheral pointer if there is a connection, and
   // nullptr otherwise.
-  api::ble_v2::BlePeripheral* GetRemotePeripheral() override {
-    return peripheral_;
-  }
+  api::ble_v2::BlePeripheral *GetRemotePeripheral() override { return peripheral_; }
+
+  bool IsClosed() const ABSL_LOCKS_EXCLUDED(mutex_);
 
  private:
-  GNCBLEL2CAPStream* stream_;
+  void DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  mutable absl::Mutex mutex_;
+  bool closed_ ABSL_GUARDED_BY(mutex_) = false;
   std::unique_ptr<BleL2capInputStream> input_stream_;
   std::unique_ptr<BleL2capOutputStream> output_stream_;
-  api::ble_v2::BlePeripheral* peripheral_;
+  api::ble_v2::BlePeripheral *peripheral_;
 };
 
 }  // namespace apple
-}  // namespace nearby
+}  // namespace nearby

internal/platform/implementation/apple/ble_l2cap_socket.mm

@@ -14,9 +14,9 @@
 
 #import "internal/platform/implementation/apple/ble_l2cap_socket.h"
 
-#import <Foundation/Foundation.h>
-
-#import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPStream.h"
+#include "internal/platform/implementation/ble_v2.h"
+#import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPConnection.h"
+#import "internal/platform/implementation/apple/utils.h"
 #import "GoogleToolboxForMac/GTMLogger.h"
 
 namespace nearby {
@@ -24,55 +24,182 @@ namespace apple {
 
 #pragma mark - BleL2capInputStream
 
-BleL2capInputStream::BleL2capInputStream(GNCBLEL2CAPStream* stream) : stream_(stream) {
-  GTMLoggerInfo(@"BleL2capInputStream::BleL2capInputStream");
+BleL2capInputStream::BleL2capInputStream(GNCBLEL2CAPConnection *connection)
+    : connection_(connection),
+      newDataPackets_([NSMutableArray array]),
+      accumulatedData_([NSMutableData data]),
+      condition_([[NSCondition alloc] init]) {
+  // Create the handlers of incoming data from the remote endpoint.
+  connection.connectionHandlers = [GNCMConnectionHandlers
+      payloadHandler:^(NSData *data) {
+        [condition_ lock];
+        // Add the incoming data to the data packet array to be processed in read() below.
+        [newDataPackets_ addObject:data];
+        [condition_ broadcast];
+        [condition_ unlock];
+      }
+      disconnectedHandler:^{
+        [condition_ lock];
+        // Release the data packet array, meaning the stream has been closed or severed.
+        newDataPackets_ = nil;
+        [condition_ broadcast];
+        [condition_ unlock];
+      }];
+}
+
+BleL2capInputStream::~BleL2capInputStream() {
+  NSCAssert(!newDataPackets_, @"BleInputStream not closed before destruction");
 }
 
 ExceptionOr<ByteArray> BleL2capInputStream::Read(std::int64_t size) {
-  // TODO: edwinwu - Implement to read data from l2cap channel.
-  return {Exception::kIo};
+  // Block until either (a) the connection has been closed, (b) we have enough data to return.
+  NSData *dataToReturn;
+  [condition_ lock];
+  while (true) {
+    // Check if the stream has been closed or severed.
+    if (!newDataPackets_) break;
+
+    if (newDataPackets_.count > 0) {
+      // Add the packet data to the accumulated data.
+      for (NSData *data in newDataPackets_) {
+        if (data.length > 0) {
+          [accumulatedData_ appendData:data];
+        }
+      }
+      [newDataPackets_ removeAllObjects];
+    }
+
+    if (accumulatedData_.length > 0) {
+      // Return up to |size| bytes of the data.
+      std::int64_t sizeToReturn = (accumulatedData_.length < size) ? accumulatedData_.length : size;
+      NSRange range = NSMakeRange(0, (NSUInteger)sizeToReturn);
+      dataToReturn = [accumulatedData_ subdataWithRange:range];
+      [accumulatedData_ replaceBytesInRange:range withBytes:nil length:0];
+      break;
+    }
+
+    [condition_ wait];
+  }
+  [condition_ unlock];
+
+  if (dataToReturn) {
+    GTMLoggerInfo(@"[NEARBY] BleL2capInputStream: Received data of size: %lu",
+                  (unsigned long)dataToReturn.length);
+    return ExceptionOr<ByteArray>{ByteArray((const char *)dataToReturn.bytes, dataToReturn.length)};
+  } else {
+    return ExceptionOr<ByteArray>{Exception::kIo};
+  }
 }
 
 Exception BleL2capInputStream::Close() {
-  // The input stream reads directly from the connection. It can not be closed without closing the
-  // connection itself. A call to `BleL2capSocket::Close` will close the connection.
+  // Unblock pending read operation.
+  [condition_ lock];
+  newDataPackets_ = nil;
+  [condition_ broadcast];
+  [condition_ unlock];
   return {Exception::kSuccess};
 }
 
 #pragma mark - BleL2capOutputStream
 
-BleL2capOutputStream::BleL2capOutputStream(GNCBLEL2CAPStream* stream) : stream_(stream) {}
-
-Exception BleL2capOutputStream::Write(const ByteArray& data) {
-  // TODO: edwinwu - Implement to write data to l2cap channel.
-  return {Exception::kIo};
+BleL2capOutputStream::~BleL2capOutputStream() {
+  NSCAssert(!connection_, @"BleL2capOutputStream not closed before destruction");
 }
 
-Exception BleL2capOutputStream::Flush() { return {Exception::kSuccess}; }
+Exception BleL2capOutputStream::Write(const ByteArray &data) {
+  [condition_ lock];
+  GTMLoggerInfo(@"[NEARBY] BleL2capOutputStream: Sending data of size: %lu",
+                NSDataFromByteArray(data).length);
 
-Exception BleL2capOutputStream::Close() { return {Exception::kSuccess}; }
+  if (!connection_) {
+    [condition_ unlock];
+    return {Exception::kIo};
+  }
 
-#pragma mark - BleL2capSocket
+  NSMutableData *packet = [NSMutableData dataWithBytes:data.data() length:data.size()];
 
-BleL2capSocket::BleL2capSocket(GNCBLEL2CAPStream* stream)
-    : BleL2capSocket(stream, new EmptyBlePeripheral()) {}
+  // Send the data, blocking until the completion handler is called.
+  __block BOOL isComplete = NO;
+  __block BOOL sendResult = NO;
+  NSCondition *condition = condition_;  // don't capture |this| in completion
 
-BleL2capSocket::BleL2capSocket(GNCBLEL2CAPStream* stream, api::ble_v2::BlePeripheral* peripheral)
-    : stream_(stream),
-      input_stream_(std::make_unique<BleL2capInputStream>(stream)),
-      output_stream_(std::make_unique<BleL2capOutputStream>(stream)),
-      peripheral_(peripheral) {}
+  [connection_ sendData:packet
+             completion:^(BOOL result) {
+               [condition lock];
+               if (isComplete) {
+                 [condition unlock];
+                 return;
+               }
+               isComplete = YES;
+               sendResult = result;
+               [condition broadcast];
+               [condition unlock];
+             }];
 
-InputStream& BleL2capSocket::GetInputStream() { return *input_stream_; }
+  while (connection_ && !isComplete) {
+    [condition_ wait];
+  }
 
-OutputStream& BleL2capSocket::GetOutputStream() { return *output_stream_; }
+  if (sendResult == YES) {
+    [condition_ unlock];
+    return {Exception::kSuccess};
+  } else {
+    [condition_ unlock];
+    return {Exception::kIo};
+  }
+}
 
-Exception BleL2capSocket::Close() {
-  [stream_ close];
+Exception BleL2capOutputStream::Flush() {
+  // The write() function blocks until the data is received by the remote endpoint, so there's
+  // nothing to do here.
   return {Exception::kSuccess};
 }
 
-void BleL2capSocket::SetCloseNotifier(absl::AnyInvocable<void()> notifier) {}
+Exception BleL2capOutputStream::Close() {
+  GTMLoggerInfo(@"[NEARBY] edwin : BleL2capOutputStream Closing");
+  // Unblock pending write operation.
+  [condition_ lock];
+  connection_ = nil;
+  [condition_ broadcast];
+  [condition_ unlock];
+  return {Exception::kSuccess};
+}
+
+#pragma mark - BleL2capSocket
+
+BleL2capSocket::BleL2capSocket(GNCBLEL2CAPConnection *connection)
+    : BleL2capSocket(connection, new EmptyBlePeripheral()) {}
+
+BleL2capSocket::BleL2capSocket(GNCBLEL2CAPConnection *connection,
+                               api::ble_v2::BlePeripheral *peripheral)
+    : input_stream_(std::make_unique<BleL2capInputStream>(connection)),
+      output_stream_(std::make_unique<BleL2capOutputStream>(connection)),
+      peripheral_(peripheral) {}
+
+BleL2capSocket::~BleL2capSocket() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+}
+
+bool BleL2capSocket::IsClosed() const {
+  absl::MutexLock lock(&mutex_);
+  return closed_;
+}
+
+Exception BleL2capSocket::Close() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+  return {Exception::kSuccess};
+}
+
+void BleL2capSocket::DoClose() {
+  GTMLoggerInfo(@"[NEARBY] edwin : BleL2capSocket DoClose");
+  if (!closed_) {
+    input_stream_->Close();
+    output_stream_->Close();
+    closed_ = true;
+  }
+}
 
 }  // namespace apple
 }  // namespace nearby

internal/platform/implementation/apple/ble_medium.mm

@@ -406,7 +406,15 @@ std::unique_ptr<api::ble_v2::BleL2capServerSocket> BleMedium::OpenL2capServerSoc
           GTMLoggerError(@"Error opening L2CAP channel in L2CAP server: %@", error);
           return;
         }
-        // TODO: b/399815436 - Implement to create socket when stream is ready.
+        GNCBLEL2CAPConnection *connection =
+            [GNCBLEL2CAPConnection connectionWithStream:stream
+                                              serviceID:@(service_id_str.c_str())
+                                     incomingConnection:YES
+                                          callbackQueue:dispatch_get_main_queue()];
+        auto socket = std::make_unique<BleL2capSocket>(connection);
+        if (l2cap_server_socket_ptr) {
+          l2cap_server_socket_ptr->AddPendingSocket(std::move(socket));
+        }
       }
       peripheralManager:nil];
   dispatch_semaphore_wait(semaphore, DISPATCH_TIME_FOREVER);