Multiplex implementation - Create Virtual Output Stream

PiperOrigin-RevId: 623540454

Package.swift

@@ -461,6 +461,7 @@ let package = Package(
         "connections/implementation/mediums/ble_v2/discovered_peripheral_tracker_test.cc",
         "connections/implementation/mediums/ble_v2/instant_on_lost_advertisement_test.cc",
         "connections/implementation/mediums/multiplex/multiplex_frames_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",

connections/implementation/mediums/multiplex/BUILD

@@ -17,9 +17,11 @@ cc_library(
     name = "multiplex",
     srcs = [
         "multiplex_frames.cc",
+        "multiplex_output_stream.cc",
     ],
     hdrs = [
         "multiplex_frames.h",
+        "multiplex_output_stream.h",
     ],
     copts = ["-DCORE_ADAPTER_DLL"],
     visibility = [
@@ -35,6 +37,7 @@ cc_library(
         "//internal/platform:util",
         "//internal/platform:uuid",
         "//internal/platform/implementation:comm",
+        "//internal/platform/implementation:types",
         "//proto/mediums:multiplex_frames_cc_proto",
         "@aappleby_smhasher//:libmurmur3",
         "@com_google_absl//absl/base:core_headers",
@@ -55,6 +58,7 @@ cc_test(
     name = "multiplex_test",
     srcs = [
         "multiplex_frames_test.cc",
+        "multiplex_output_stream_test.cc",
     ],
     deps = [
         ":multiplex",

connections/implementation/mediums/multiplex/multiplex_frames.cc

@@ -66,8 +66,8 @@ ByteArray ToBytes(MultiplexFrame&& frame) {
   return bytes;
 }
 
-ByteArray ForConnectionRequest(std::string service_id,
-                               std::string service_id_hash_salt) {
+ByteArray ForConnectionRequest(const std::string& service_id,
+                               const std::string& service_id_hash_salt) {
   MultiplexFrame frame;
 
   frame.set_frame_type(MultiplexFrame::CONTROL_FRAME);
@@ -84,7 +84,8 @@ ByteArray ForConnectionRequest(std::string service_id,
 }
 
 ByteArray ForConnectionResponse(
-    ByteArray& salted_service_id_hash, std::string service_id_hash_salt,
+    const ByteArray& salted_service_id_hash,
+    const std::string& service_id_hash_salt,
     ConnectionResponseFrame::ConnectionResponseCode response_code) {
   MultiplexFrame frame;
 
@@ -103,8 +104,8 @@ ByteArray ForConnectionResponse(
   return ToBytes(std::move(frame));
 }
 
-ByteArray ForDisconnection(std::string service_id,
-                               std::string service_id_hash_salt) {
+ByteArray ForDisconnection(const std::string& service_id,
+                           const std::string& service_id_hash_salt) {
   MultiplexFrame frame;
 
   frame.set_frame_type(MultiplexFrame::CONTROL_FRAME);
@@ -120,8 +121,9 @@ ByteArray ForDisconnection(std::string service_id,
   return ToBytes(std::move(frame));
 }
 
-ByteArray ForData(std::string service_id, std::string service_id_hash_salt,
-                  bool should_pass_salt, ByteArray& data) {
+ByteArray ForData(const std::string& service_id,
+                  const std::string& service_id_hash_salt,
+                  bool should_pass_salt, const ByteArray& data) {
   MultiplexFrame frame;
 
   frame.set_frame_type(MultiplexFrame::DATA_FRAME);

connections/implementation/mediums/multiplex/multiplex_frames.h

@@ -62,31 +62,33 @@ std::string GenerateServiceIdHashKeyWithSalt(const std::string& service_id,
 // 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(std::string service_id,
-                               std::string service_id_hash_salt);
+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(
-    ByteArray& salted_service_id_hash, std::string service_id_hash_salt,
+    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(std::string service_id,
-                           std::string service_id_hash_salt);
+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(std::string service_id, std::string service_id_hash_salt,
-                  bool should_pass_salt, ByteArray& data);
+ByteArray ForData(const std::string& service_id,
+                  const std::string& service_id_hash_salt,
+                  bool should_pass_salt, const ByteArray& data);
 
 ExceptionOr<location::nearby::mediums::MultiplexFrame> FromBytes(
     const ByteArray& multiplex_frame_bytes);

connections/implementation/mediums/multiplex/multiplex_frames_test.cc

@@ -108,9 +108,9 @@ TEST(MultiplexFrameTest, HashValidtion) {
 
 TEST(MultiplexFrameTest, CanGenerateConnectionRequest) {
   ByteArray bytes = ForConnectionRequest(std::string(kServiceId_1), "1234");
-  auto response = FromBytes(bytes);
-  ASSERT_TRUE(response.ok());
-  auto frame = response.result();
+  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(),

connections/implementation/mediums/multiplex/multiplex_output_stream.cc

@@ -0,0 +1,374 @@
+// 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 <memory>
+#include <optional>
+#include <string>
+#include <utility>
+
+#include "absl/strings/string_view.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/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 TAG = "MultiplexOutputStream:";
+constexpr absl::string_view kFakeSalt = "RECEIVER_CONDIMENT";
+}  // namespace
+
+// Implementation for class ArrayBlockingQueue
+template <typename T>
+void ArrayBlockingQueue<T>::Put(const T& value) {
+  MutexLock lock(&queue_mutex_);
+  if (queue_.size() >= capacity_) {
+    has_space_.Wait();
+  }
+  queue_.push(value);
+  has_data_.Notify();
+}
+
+template <typename T>
+T ArrayBlockingQueue<T>::Take() {
+  MutexLock lock(&queue_mutex_);
+  if (queue_.empty()) {
+    has_data_.Wait();
+  }
+  T front = queue_.front();
+  queue_.pop();
+  has_space_.Notify();
+  return front;
+}
+
+template <typename T>
+bool ArrayBlockingQueue<T>::TryPut(const T& value) {
+  MutexLock lock(&queue_mutex_);
+  if (queue_.size() < capacity_) {
+    queue_.push(value);
+    has_data_.Notify();
+    return true;
+  }
+  return false;
+}
+
+template <typename T>
+std::optional<T> ArrayBlockingQueue<T>::TryTake() {
+  MutexLock lock(&queue_mutex_);
+  if (!queue_.empty()) {
+    T front = queue_.front();
+    queue_.pop();
+    has_space_.Notify();
+    return front;
+  }
+  return std::nullopt;
+}
+
+// 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<bool>* future) {
+  if (!future) {
+    NEARBY_LOGS(INFO) << TAG << "No future to wait for; return with error.";
+    return {Exception::kFailed};
+  }
+  NEARBY_LOGS(INFO) << TAG << "Waiting for future to complete: " << method_name;
+  ExceptionOr<bool> result =
+      future->Get(FeatureFlags::GetInstance()
+                      .GetFlags()
+                      .mediums_frame_write_timeout_millis);
+  if (!result.ok()) {
+    NEARBY_LOGS(INFO) << TAG << "Future:[" << method_name
+                      << "] completed with exception:" << result.exception();
+    return {Exception::kFailed};
+  }
+  if (result.result()) {
+    NEARBY_LOGS(INFO) << TAG << "Future:[" << method_name
+                      << "] completed with success.";
+    return {Exception::kSuccess};
+  }
+  NEARBY_LOGS(INFO) << TAG << "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<bool> 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<bool> 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()) {
+    NEARBY_LOGS(WARNING) << TAG << "Failed to close VirtualOutputStream("
+                         << service_id << ") because it's already gone.";
+    return false;
+  }
+
+  auto service_id_hash_salt = item->second->GetserviceIdHashSalt();
+  item->second->Close();
+  if (is_enabled_.Get()) {
+    Future<bool> 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;
+}
+
+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<VirtualOutputStream>(
+                   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<VirtualOutputStream>(
+                   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();
+  writer_thread_.Shutdown();
+  physical_writer_ = nullptr;
+}
+
+void MultiplexOutputStream::MultiplexWriter::EnqueueToSend(
+    Future<bool>* future, const ByteArray& data,
+    const std::string& frame_name) {
+  MutexLock lock(&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() {
+  NEARBY_LOGS(INFO) << TAG << "Writing loop started.";
+  while (true) {
+    auto enqueued_frame = data_queue_.TryTake();
+    if (enqueued_frame != std::nullopt) {
+      Write(enqueued_frame.value());
+      continue;
+    }
+
+    MutexLock lock(&mutex_);
+    if (data_queue_.Empty() && is_writing_) {
+      is_writing_ = false;
+      Exception wait_succeeded = is_writing_cond_.Wait();
+      if (!wait_succeeded.Ok()) {
+        NEARBY_LOGS(WARNING)
+            << TAG << __func__
+            << ": Failure waiting to wait: " << wait_succeeded.value;
+        return;
+      }
+      if (is_closed_) break;
+    }
+  }
+  NEARBY_LOGS(INFO) << TAG << "Writing loop stopped.";
+}
+
+void MultiplexOutputStream::MultiplexWriter::Write(
+    EnqueuedFrame& enqueued_frame) {
+  MutexLock lock(&writer_mutex_);
+  if (!physical_writer_
+           ->Write(Base64Utils::IntToBytes(enqueued_frame.data_.size()))
+           .Ok()) {
+    enqueued_frame.future_->SetException({Exception::kIo});
+    return;
+  };
+  if (!physical_writer_->Write(enqueued_frame.data_).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() {
+  MutexLock lock(&mutex_);
+  is_closed_ = true;
+  if (is_write_loop_running_) {
+    NEARBY_LOGS(INFO) << TAG << "Stop writing loop and Shutdown writer thread.";
+    is_write_loop_running_ = false;
+    is_writing_cond_.Notify();
+  }
+}
+
+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(
+    const ByteArray& data) {
+  if (is_closed_) {
+    NEARBY_LOGS(WARNING)
+        << TAG << "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;
+        NEARBY_LOGS(INFO) << TAG
+                          << "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<bool> 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() {
+  is_closed_.Set(true);
+  return {Exception::kSuccess};
+}
+
+}  // namespace multiplex
+}  // namespace mediums
+}  // namespace connections
+}  // namespace nearby

connections/implementation/mediums/multiplex/multiplex_output_stream.h

@@ -0,0 +1,240 @@
+// 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 <cstddef>
+#include <memory>
+#include <optional>
+#include <queue>
+#include <string>
+
+#include "absl/base/thread_annotations.h"
+#include "absl/container/flat_hash_map.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 {
+
+/**
+ * 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);
+  T Take();
+  bool TryPut(const T& value);
+  // Returns std::nullopt if the queue is empty.
+  std::optional<T> TryTake();
+
+  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_;
+};
+
+/**
+ * A helper class to send out the {@code MultiplexControlFrame} and the outgoing
+ * data from clients. It schedules control and data frames with priority below
+ *
+ * <p>{@link MultiplexControlFrameType#CONNECTION_REQUEST} and {@link
+ * MultiplexControlFrameType#CONNECTION_RESPONSE} have the highest priority
+ *
+ * <p>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,...
+ *
+ * <p>{@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() { Shutdown(); }
+
+  // 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);
+
+  // Waits for the result of the future.
+  Exception WaitForResult(const std::string& method_name, Future<bool>* 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<bool>* future, ByteArray data)
+        : future_(future), data_(data) {}
+    ~EnqueuedFrame() = default;
+
+    Future<bool>* future_;
+    ByteArray data_;
+  };
+
+  class MultiplexWriter {
+   public:
+    explicit MultiplexWriter(OutputStream* physical_writer);
+    ~MultiplexWriter();
+
+    // Enqueues the frame to be sent out.
+    void EnqueueToSend(Future<bool>* 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<EnqueuedFrame> data_queue_{
+        FeatureFlags::GetInstance()
+            .GetFlags()
+            .multiplex_socket_middle_priority_queue_capacity};
+
+    mutable Mutex mutex_;
+    ConditionVariable is_writing_cond_{&mutex_};
+    bool is_writing_ = false;
+    bool is_closed_ = false;
+
+    // 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(const ByteArray& 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<std::string, std::unique_ptr<VirtualOutputStream>>
+      virtual_output_streams_;
+  MultiplexWriter multiplex_writer_;
+};
+
+}  // namespace multiplex
+}  // namespace mediums
+}  // namespace connections
+}  // namespace nearby
+
+#endif  // CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_OUTPUT_STREAM_H_

connections/implementation/mediums/multiplex/multiplex_output_stream_test.cc

@@ -0,0 +1,253 @@
+// 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 <cstdint>
+#include <memory>
+#include <string>
+#include <utility>
+
+#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<MultiplexFrame> ReadFrame() {
+    ExceptionOr<std::int32_t> read_int = Base64Utils::ReadInt(reader_.get());
+    if (!read_int.ok()) return read_int.GetException();
+    if (read_int.result() <= 0) return {Exception::kFailed};
+
+    ExceptionOr<ByteArray> 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<InputStream>, std::unique_ptr<OutputStream>> pipe_ =
+      CreatePipe();
+
+  std::unique_ptr<InputStream> reader_ = std::move(pipe_.first);
+  std::unique_ptr<OutputStream> writer_ = std::move(pipe_.second);
+  std::unique_ptr<MultiplexOutputStream> multiplex_output_stream_;
+};
+
+TEST_F(MultiplexOutputStreamTest, SendConnectionRequestFrame) {
+  multiplex_output_stream_ = std::make_unique<MultiplexOutputStream>(
+      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<MultiplexOutputStream>(
+      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<MultiplexOutputStream>(
+      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<MultiplexOutputStream>(
+      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<MultiplexOutputStream>(
+      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<MultiplexOutputStream>(
+      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<MultiplexOutputStream>(
+      writer_.get(), enabled_);
+
+  auto virtual_output_stream =
+      multiplex_output_stream_->CreateVirtualOutputStream(
+          std::string(kServiceId_1), std::string(kSalt_1));
+
+  const ByteArray 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<MultiplexOutputStream>(
+      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));
+
+  const ByteArray data_1("abcdefg");
+  const ByteArray data_2("hijklmn");
+  MultiThreadExecutor executor(2);
+  CountDownLatch latch(2);
+  executor.Execute([&virtual_output_stream_1, &latch, &data_1]() {
+    absl::SleepFor(absl::Milliseconds(500));
+    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(2000)).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));
+    NEARBY_LOGS(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;
+    NEARBY_LOGS(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

internal/platform/base64_utils.cc

@@ -14,8 +14,16 @@
 
 #include "internal/platform/base64_utils.h"
 
+#include <cstdint>
+#include <string>
+#include <utility>
+
 #include "absl/strings/escaping.h"
+#include "absl/strings/string_view.h"
 #include "internal/platform/byte_array.h"
+#include "internal/platform/exception.h"
+#include "internal/platform/input_stream.h"
+#include "internal/platform/output_stream.h"
 
 namespace nearby {
 
@@ -36,4 +44,39 @@ ByteArray Base64Utils::Decode(absl::string_view base64_string) {
   return ByteArray(decoded_string.data(), decoded_string.size());
 }
 
+std::int32_t Base64Utils::BytesToInt(const ByteArray& bytes) {
+  const char* int_bytes = bytes.data();
+
+  std::int32_t result = 0;
+  result |= (static_cast<std::int32_t>(int_bytes[0]) & 0x0FF) << 24;
+  result |= (static_cast<std::int32_t>(int_bytes[1]) & 0x0FF) << 16;
+  result |= (static_cast<std::int32_t>(int_bytes[2]) & 0x0FF) << 8;
+  result |= (static_cast<std::int32_t>(int_bytes[3]) & 0x0FF);
+
+  return result;
+}
+
+ByteArray Base64Utils::IntToBytes(std::int32_t value) {
+  char int_bytes[sizeof(std::int32_t)];
+  int_bytes[0] = static_cast<char>((value >> 24) & 0x0FF);
+  int_bytes[1] = static_cast<char>((value >> 16) & 0x0FF);
+  int_bytes[2] = static_cast<char>((value >> 8) & 0x0FF);
+  int_bytes[3] = static_cast<char>((value) & 0x0FF);
+
+  return ByteArray(int_bytes, sizeof(int_bytes));
+}
+
+ExceptionOr<std::int32_t> Base64Utils::ReadInt(InputStream* reader) {
+  ExceptionOr<ByteArray> read_bytes = reader->ReadExactly(sizeof(std::int32_t));
+  if (!read_bytes.ok()) {
+    return ExceptionOr<std::int32_t>(read_bytes.exception());
+  }
+  return ExceptionOr<std::int32_t>(
+      BytesToInt(std::move(read_bytes.result())));
+}
+
+Exception Base64Utils::WriteInt(OutputStream* writer, std::int32_t value) {
+  return writer->Write(IntToBytes(value));
+}
+
 }  // namespace nearby

internal/platform/base64_utils.h

@@ -15,8 +15,13 @@
 #ifndef PLATFORM_BASE_BASE64_UTILS_H_
 #define PLATFORM_BASE_BASE64_UTILS_H_
 
+#include <cstdint>
+#include <string>
 #include "absl/strings/string_view.h"
 #include "internal/platform/byte_array.h"
+#include "internal/platform/exception.h"
+#include "internal/platform/input_stream.h"
+#include "internal/platform/output_stream.h"
 
 namespace nearby {
 
@@ -24,6 +29,10 @@ class Base64Utils {
  public:
   static std::string Encode(const ByteArray& bytes);
   static ByteArray Decode(absl::string_view base64_string);
+  static std::int32_t BytesToInt(const ByteArray& bytes);
+  static ByteArray IntToBytes(std::int32_t value);
+  static ExceptionOr<std::int32_t> ReadInt(InputStream* reader);
+  static Exception WriteInt(OutputStream* writer, std::int32_t value);
 };
 
 }  // namespace nearby

internal/platform/feature_flags.h

@@ -91,6 +91,21 @@ class FeatureFlags {
     // If the receiver doesn't ack with payload_received_ack frame in 1s, the
     // sender will timeout the waiting.
     absl::Duration wait_payload_received_ack_millis = absl::Milliseconds(1000);
+
+    // Multiplex related flags
+    // Timeout value for read frame operation in endpoint channel.
+    absl::Duration mediums_frame_read_timeout_millis =
+        absl::Milliseconds(15000);
+    // Timeout value for write frame operation in endpoint channel.
+    absl::Duration mediums_frame_write_timeout_millis =
+        absl::Milliseconds(15000);
+    // The timeout for waiting on connection request response.
+    absl::Duration multiplex_socket_connection_response_timeout_millis =
+        absl::Milliseconds(3000);
+    // The capacity of the middle priority queue inner MultiplexOutputStream.
+    // The new outgoing frame with the middle priority will wait for space to
+    // become available if the queue is full.'
+    std::uint32_t multiplex_socket_middle_priority_queue_capacity = 50;
   };
 
   static const FeatureFlags& GetInstance() {