Implement Bluetooth Multiplex.

PiperOrigin-RevId: 651921709

internal/platform/BUILD

@@ -280,6 +280,7 @@ cc_test(
 cc_library(
     name = "types",
     srcs = [
+        "blocking_queue_stream.cc",
         "clock_impl.cc",
         "device_info_impl.cc",
         "monitored_runnable.cc",
@@ -289,8 +290,10 @@ cc_library(
         "timer_impl.cc",
     ],
     hdrs = [
+        "array_blocking_queue.h",
         "atomic_boolean.h",
         "atomic_reference.h",
+        "blocking_queue_stream.h",
         "borrowable.h",
         "cancelable.h",
         "cancelable_alarm.h",

internal/platform/array_blocking_queue.h

@@ -0,0 +1,104 @@
+// 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 PLATFORM_PUBLIC_ARRAY_BLOCKING_QUEUE_H_
+#define PLATFORM_PUBLIC_ARRAY_BLOCKING_QUEUE_H_
+
+#include <cstddef>
+#include <optional>
+#include <queue>
+
+#include "internal/platform/condition_variable.h"
+#include "internal/platform/logging.h"
+#include "internal/platform/mutex.h"
+#include "internal/platform/mutex_lock.h"
+
+namespace nearby {
+
+/**
+ * Payload from different services/clients will be put into an
+ * ArrayBlockingQueue before sending to ensure each client has equal chance to
+ * send its data. Since C++ doesn't provide ArrayBlockingQueue as Java, we
+ * implement one here.
+ */
+template <typename T>
+class ArrayBlockingQueue {
+ public:
+  explicit ArrayBlockingQueue(size_t capacity) : capacity_(capacity) {}
+
+  void Put(const T& value) {
+    MutexLock lock(&queue_mutex_);
+    if (queue_.size() >= capacity_) {
+      has_space_.Wait();
+    }
+    queue_.push(value);
+    NEARBY_LOGS(INFO) << "ArrayBlockingQueue::Put()";
+    has_data_.Notify();
+  }
+
+  T Take() {
+    MutexLock lock(&queue_mutex_);
+    if (queue_.empty()) {
+      has_data_.Wait();
+    }
+    T front = queue_.front();
+    queue_.pop();
+    NEARBY_LOGS(INFO) << "ArrayBlockingQueue::Take()";
+    has_space_.Notify();
+    return front;
+  }
+
+  bool TryPut(const T& value) {
+    MutexLock lock(&queue_mutex_);
+    if (queue_.size() < capacity_) {
+      queue_.push(value);
+      has_data_.Notify();
+      return true;
+    }
+    return false;
+  }
+
+  // Returns std::nullopt if the queue is empty.
+  std::optional<T> TryTake() {
+    MutexLock lock(&queue_mutex_);
+    if (!queue_.empty()) {
+      T front = queue_.front();
+      queue_.pop();
+      has_space_.Notify();
+      return front;
+    }
+    return std::nullopt;
+  }
+
+  size_t Size() const {
+    MutexLock lock(&queue_mutex_);
+    return queue_.size();
+  }
+
+  bool Empty() const {
+    MutexLock lock(&queue_mutex_);
+    return queue_.empty();
+  }
+
+ private:
+  std::queue<T> queue_;
+  mutable Mutex queue_mutex_;
+  ConditionVariable has_data_{&queue_mutex_};
+  ConditionVariable has_space_{&queue_mutex_};
+  const size_t capacity_;
+};
+
+}  // namespace nearby
+
+#endif  // PLATFORM_PUBLIC_ARRAY_BLOCKING_QUEUE_H_

internal/platform/blocking_queue_stream.cc

@@ -0,0 +1,72 @@
+// 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 "internal/platform/blocking_queue_stream.h"
+
+#include <cstdint>
+
+#include "internal/platform/byte_array.h"
+#include "internal/platform/exception.h"
+#include "internal/platform/feature_flags.h"
+#include "internal/platform/logging.h"
+
+namespace nearby {
+
+BlockingQueueStream::BlockingQueueStream() {
+  NEARBY_LOGS(INFO) << "Create a BlockingQueueStream with size "
+                    << FeatureFlags::GetInstance()
+                           .GetFlags()
+                           .blocking_queue_stream_queue_capacity;
+}
+
+ExceptionOr<ByteArray> BlockingQueueStream::Read(std::int64_t size) {
+  if (is_closed_) {
+    NEARBY_LOGS(INFO)
+        << "Failed to read BlockingQueueStream because it was closed.";
+    return ExceptionOr<ByteArray>(Exception::kInterrupted);
+  }
+  NEARBY_LOGS(INFO) << "BlockingQueueStream read " << size << " bytes";
+  return ExceptionOr<ByteArray>(blocking_queue_.Take());
+}
+
+void BlockingQueueStream::Write(const ByteArray& bytes) {
+  if (is_closed_) {
+    NEARBY_LOGS(INFO)
+        << "Failed to write BlockingQueueStream because it was closed.";
+    return;
+  }
+  is_writing_ = true;
+  blocking_queue_.Put(bytes);
+  is_writing_ = false;
+  NEARBY_LOGS(VERBOSE) << "BlockingQueueStream wrote " << bytes.size()
+                       << " bytes";
+}
+
+Exception BlockingQueueStream::Close() {
+  if (is_closed_) {
+    NEARBY_LOGS(INFO) << "InputBlockingQueueStream has already been closed.";
+    return {Exception::kSuccess};
+  }
+  if (is_writing_) {
+    NEARBY_LOGS(INFO)
+        << "BlockingQueueStream is waiting for writing, read first to unblock";
+    blocking_queue_.TryTake();
+  }
+  blocking_queue_.TryPut(queue_end_);
+  is_closed_ = true;
+  NEARBY_LOGS(INFO) << "InputBlockingQueueStream is closed.";
+  return {Exception::kSuccess};
+}
+
+}  // namespace nearby

internal/platform/blocking_queue_stream.h

@@ -0,0 +1,52 @@
+// 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 PLATFORM_PUBLIC_BLOCKING_QUEUE_STREAM_H_
+#define PLATFORM_PUBLIC_BLOCKING_QUEUE_STREAM_H_
+
+#include <cstdint>
+
+#include "internal/platform/array_blocking_queue.h"
+#include "internal/platform/byte_array.h"
+#include "internal/platform/exception.h"
+#include "internal/platform/feature_flags.h"
+#include "internal/platform/input_stream.h"
+#include "internal/platform/mutex.h"
+
+namespace nearby {
+class BlockingQueueStream : public InputStream {
+ public:
+  BlockingQueueStream();
+  ~BlockingQueueStream() override = default;
+
+  ExceptionOr<ByteArray> Read(std::int64_t size) override;
+  void Write(const ByteArray& bytes);
+  Exception Close() override;
+  bool IsWriting() const {
+    return is_writing_;
+  }
+
+ private:
+  mutable Mutex mutex_;
+  ArrayBlockingQueue<ByteArray> blocking_queue_{FeatureFlags::GetInstance()
+                 .GetFlags()
+                 .blocking_queue_stream_queue_capacity};
+  ByteArray queue_end_{0};
+  bool is_writing_ = false;
+  bool is_closed_ = false;
+};
+}  // namespace nearby
+
+#endif  // #ifndef PLATFORM_PUBLIC_BLOCKING_QUEUE_STREAM_H_
+

internal/platform/bluetooth_classic.cc

@@ -14,10 +14,57 @@
 
 #include "internal/platform/bluetooth_classic.h"
 
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "absl/container/flat_hash_map.h"
+#include "internal/platform/bluetooth_adapter.h"
+#include "internal/platform/cancellation_flag.h"
+#include "internal/platform/implementation/bluetooth_classic.h"
 #include "internal/platform/logging.h"
 #include "internal/platform/mutex_lock.h"
+#include "internal/platform/output_stream.h"
+#include "internal/platform/socket.h"
 
 namespace nearby {
+using location::nearby::proto::connections::Medium;
+
+MediumSocket* BluetoothSocket::CreateVirtualSocket(OutputStream* outputstream) {
+  if (IsVirtualSocket()) {
+    NEARBY_LOGS(WARNING)
+        << "Creating the virtual socket on a virtual socket is not allowed.";
+    return nullptr;
+  }
+  auto virtual_socket = std::make_shared<BluetoothSocket>(outputstream);
+  return virtual_socket.get();
+}
+
+MediumSocket* BluetoothSocket::CreateVirtualSocket(
+    const std::string& salted_service_id_hash_key, OutputStream* outputstream,
+    Medium medium,
+    absl::flat_hash_map<std::string, std::shared_ptr<MediumSocket>>*
+        virtual_sockets_ptr) {
+  if (IsVirtualSocket()) {
+    NEARBY_LOGS(WARNING)
+        << "Creating the virtual socket on a virtual socket is not allowed.";
+    return nullptr;
+  }
+
+  auto virtual_socket = std::make_shared<BluetoothSocket>(outputstream);
+  virtual_socket->impl_ = this->impl_;
+  NEARBY_LOGS(WARNING) << "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;
+  NEARBY_LOGS(INFO) << "virtual_sockets_ size: "
+                    << virtual_sockets_ptr_->size();
+  return virtual_socket.get();
+}
 
 BluetoothClassicMedium::~BluetoothClassicMedium() {
   NEARBY_LOG(INFO, "~BluetoothClassicMedium: observer_list_ size: %d",
@@ -54,7 +101,7 @@ bool BluetoothClassicMedium::StartDiscovery(DiscoveryCallback callback) {
                        device.GetName().c_str());
             MutexLock lock(&mutex_);
             auto pair = devices_.emplace(
-                &device, absl::make_unique<DeviceDiscoveryInfo>());
+                &device, std::make_unique<DeviceDiscoveryInfo>());
             auto& context = *pair.first->second;
             if (!pair.second) {
               NEARBY_LOG(INFO, "Adding (again) device=%p, impl=%p",

internal/platform/bluetooth_classic.h

@@ -15,14 +15,15 @@
 #ifndef PLATFORM_PUBLIC_BLUETOOTH_CLASSIC_H_
 #define PLATFORM_PUBLIC_BLUETOOTH_CLASSIC_H_
 
+#include <stdbool.h>
 #include <memory>
 #include <optional>
 #include <string>
 #include <utility>
 
 #include "absl/container/flat_hash_map.h"
-#include "absl/container/flat_hash_set.h"
 #include "internal/base/observer_list.h"
+#include "internal/platform/blocking_queue_stream.h"
 #include "internal/platform/bluetooth_adapter.h"
 #include "internal/platform/byte_array.h"
 #include "internal/platform/cancellation_flag.h"
@@ -34,29 +35,79 @@
 #include "internal/platform/logging.h"
 #include "internal/platform/mutex.h"
 #include "internal/platform/output_stream.h"
+#include "internal/platform/socket.h"
 
 namespace nearby {
 
 // https://developer.android.com/reference/android/bluetooth/BluetoothSocket.html.
-class BluetoothSocket final {
+class BluetoothSocket : public MediumSocket {
  public:
-  BluetoothSocket() = default;
+  BluetoothSocket()
+      : MediumSocket(location::nearby::proto::connections::Medium::BLUETOOTH) {
+        };
   BluetoothSocket(const BluetoothSocket&) = default;
   BluetoothSocket& operator=(const BluetoothSocket&) = default;
+
+// Creates a physical BluetoothSocket from a platform implementation.
   explicit BluetoothSocket(std::unique_ptr<api::BluetoothSocket> socket)
-      : impl_(socket.release()) {}
-  ~BluetoothSocket() = default;
+      : MediumSocket(location::nearby::proto::connections::Medium::BLUETOOTH),
+        impl_(socket.release()) {}
+
+// Creates a virtual BluetoothSocket from a virtual output stream.
+  explicit BluetoothSocket(OutputStream* virtual_output_stream)
+      : MediumSocket(location::nearby::proto::connections::Medium::BLUETOOTH),
+        blocking_queue_input_stream_(std::make_shared<BlockingQueueStream>()),
+        virtual_output_stream_(virtual_output_stream),
+        is_virtual_socket_(true) {}
+
+  ~BluetoothSocket() override = default;
 
   // Returns the InputStream of this connected BluetoothSocket.
-  InputStream& GetInputStream() { return impl_->GetInputStream(); }
+  InputStream& GetInputStream() override {
+    return IsVirtualSocket() ? *blocking_queue_input_stream_
+                             : impl_->GetInputStream();
+  }
 
   // Returns the OutputStream of this connected BluetoothSocket.
-  OutputStream& GetOutputStream() { return impl_->GetOutputStream(); }
+  OutputStream& GetOutputStream() override {
+    return IsVirtualSocket() ? *virtual_output_stream_
+                             : impl_->GetOutputStream();
+  }
 
   // Closes both input and output streams, marks Socket as closed.
   // After this call object should be treated as not connected.
   // Returns Exception::kIo on error, Exception::kSuccess otherwise.
-  Exception Close() { return impl_->Close(); }
+  Exception Close() override {
+    if (IsVirtualSocket()) {
+      NEARBY_LOGS(INFO) << "Multiplex: Closing virtual socket: " << this;
+      blocking_queue_input_stream_->Close();
+      virtual_output_stream_->Close();
+      CloseLocal();
+      return {Exception::kSuccess};
+    }
+    NEARBY_LOGS(INFO) << "Multiplex: Closing physical socket: " << this;
+    return impl_->Close();
+  }
+
+  // Returns true if this is a virtual socket.
+  bool IsVirtualSocket() override { return is_virtual_socket_; }
+
+  // Creates a virtual socket only with outputstream.
+  MediumSocket* CreateVirtualSocket(OutputStream* outputstream) override;
+  MediumSocket* CreateVirtualSocket(
+      const std::string& salted_service_id_hash_key, OutputStream* outputstream,
+      location::nearby::proto::connections::Medium medium,
+      absl::flat_hash_map<std::string, std::shared_ptr<MediumSocket>>*
+          virtual_sockets_ptr) override;
+
+  /** Feeds the received incoming data to the client. */
+  void FeedIncomingData(ByteArray data) override {
+    if (!IsVirtualSocket()) {
+      NEARBY_LOGS(INFO) << "Feeding data on a physical socket is not allowed.";
+      return;
+    }
+    blocking_queue_input_stream_->Write(data);
+  }
 
   // https://developer.android.com/reference/android/bluetooth/BluetoothSocket.html#getRemoteDevice()
   BluetoothDevice GetRemoteDevice() {
@@ -73,7 +124,10 @@ class BluetoothSocket final {
   // BluetoothServerSocket::Accept().
   // These methods may also return an invalid socket if connection failed for
   // any reason.
-  bool IsValid() const { return impl_ != nullptr; }
+  bool IsValid() const {
+    if (is_virtual_socket_) return true;
+    return impl_ != nullptr;
+  }
 
   // Returns reference to platform implementation.
   // This is used to communicate with platform code, and for debugging purposes.
@@ -84,6 +138,11 @@ class BluetoothSocket final {
 
  private:
   std::shared_ptr<api::BluetoothSocket> impl_;
+  absl::flat_hash_map<std::string, std::shared_ptr<MediumSocket>>*
+      virtual_sockets_ptr_ = nullptr;
+  std::shared_ptr<BlockingQueueStream> blocking_queue_input_stream_ = nullptr;
+  OutputStream* virtual_output_stream_ = nullptr;
+  bool is_virtual_socket_ = false;
 };
 
 // https://developer.android.com/reference/android/bluetooth/BluetoothServerSocket.html.

internal/platform/feature_flags.h

@@ -110,6 +110,7 @@ class FeatureFlags {
     // The maximum size of frame we'll attempt to read, to avoid a remote device
     // from triggering an OutOfMemory error.
     std::uint32_t connection_max_frame_length = 1048576;
+    std::uint32_t blocking_queue_stream_queue_capacity = 10;
   };
 
   static const FeatureFlags& GetInstance() {

internal/platform/socket.h

@@ -23,6 +23,7 @@
 #include "absl/container/flat_hash_set.h"
 #include "absl/functional/any_invocable.h"
 #include "internal/platform/byte_array.h"
+#include "internal/platform/exception.h"
 #include "internal/platform/input_stream.h"
 #include "internal/platform/output_stream.h"
 #include "proto/connections_enums.pb.h"
@@ -38,7 +39,7 @@ class Socket {
 
   virtual InputStream& GetInputStream() = 0;
   virtual OutputStream& GetOutputStream() = 0;
-  virtual void Close() = 0;
+  virtual Exception Close() = 0;
 };
 
 class MediumSocket : public Socket {
@@ -52,6 +53,11 @@ class MediumSocket : public Socket {
     return medium_;
   }
 
+  /** Creates a virtual socket only with outputstream. */
+  virtual MediumSocket* CreateVirtualSocket(OutputStream* outputstream) {
+    return this;
+  }
+
   /** Creates a virtual socket. */
   virtual MediumSocket* CreateVirtualSocket(
       const std::string& salted_service_id_hash_key, OutputStream* outputstream,
@@ -62,7 +68,9 @@ class MediumSocket : public Socket {
   }
 
   /** Feeds the received incoming data to the client. */
-  virtual void FeedIncomingData(ByteArray data) {}
+  virtual void FeedIncomingData(ByteArray data) {
+//    NEARBY_LOGS(INFO) << "FeedIncomingData: do nothing";
+  }
 
   /** Returns true if the socket is a virtual socket. */
   virtual bool IsVirtualSocket() {
@@ -86,16 +94,15 @@ class MediumSocket : public Socket {
     if (!IsVirtualSocket()) {
       return;
     }
-
     for (auto& callback : multiplex_socket_enabled_cbs_) {
-      callback.get();
+      (*callback)();
     }
   }
 
   /** Closes the local socket. */
   void CloseLocal() {
     for (auto& listener : socket_closed_listeners_) {
-      listener.get();
+      (*listener)();
     }
   }