Revert "removed bluetooth_classic_socket specific stream semantics. all linux sockets should use stream.h. improved stream semantics in stream.h"

This reverts commit 18970db598.

internal/platform/implementation/linux/BUILD

@@ -12,24 +12,25 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-load("@hedron_compile_commands//:refresh_compile_commands.bzl", "refresh_compile_commands")
+#load("@hedron_compile_commands//:refresh_compile_commands.bzl", "refresh_compile_commands")
+
 load("@rules_cc//cc:defs.bzl", "cc_library", "cc_test")
-
-refresh_compile_commands(
-    name = "refresh_compile_commands_linux",
-
-    # Specify the targets of interest.
-    # For example, specify a dict of targets and any flags required to build.
-    targets = {
-        ":linux": "-s --check_visibility=false --spawn_strategy=standalone --verbose_failures --cxxopt=-std=c++20 --host_cxxopt=-std=c++20  --cxxopt='-fvisibility-inlines-hidden'",
-    },
-    # No need to add flags already in .bazelrc. They're automatically picked up.
-    # If you don't need flags, a list of targets is also okay, as is a single target string.
-    # Wildcard patterns, like //... for everything, *are* allowed here, just like a build.
-    # As are additional targets (+) and subtractions (-), like in bazel query https://docs.bazel.build/versions/main/query.html#expressions
-    # And if you're working on a header-only library, specify a test or binary target that compiles it.
-)
-
+#refresh_compile_commands(
+#    name = "refresh_compile_commands",
+#
+#    # Specify the targets of interest.
+#    # For example, specify a dict of targets and any flags required to build.
+#    targets = {
+#        ":linux": "-s --check_visibility=false --spawn_strategy=standalone --verbose_failures --strip=never --copt=-O0 --copt=-g --copt=-fno-omit-frame-pointer",
+#        "//connections:core": "-s --check_visibility=false --spawn_strategy=standalone --verbose_failures --strip=never --copt=-O0 --copt=-g --copt=-fno-omit-frame-pointer",
+#        "//connections/file_share:file_share": "-s --check_visibility=false --spawn_strategy=standalone --verbose_failures --strip=never --copt=-O0 --copt=-g --copt=-fno-omit-frame-pointer",
+#    },
+#    # No need to add flags already in .bazelrc. They're automatically picked up.
+#    # If you don't need flags, a list of targets is also okay, as is a single target string.
+#    # Wildcard patterns, like //... for everything, *are* allowed here, just like a build.
+#    # As are additional targets (+) and subtractions (-), like in bazel query https://docs.bazel.build/versions/main/query.html#expressions
+#    # And if you're working on a header-only library, specify a test or binary target that compiles it.
+#)
 
 licenses(["notice"])
 
@@ -199,6 +200,7 @@ cc_library(
         "ble_l2cap_socket.cc",
         "bluetooth_adapter.cc",
         "bluetooth_bluez_profile.cc",
+        "bluetooth_classic_socket.cc",
         "bluetooth_classic_device.cc",
         "bluetooth_classic_medium.cc",
         "bluetooth_classic_server_socket.cc",
@@ -312,7 +314,7 @@ cc_test(
         # "bluetooth_adapter_test.cc",
         "crypto_test.cc",
         # "executor_test.cc",
-        #        "file_path_test.cc",
+        "file_path_test.cc",
         #        "http_loader_test.cc",
         # "preferences_manager_test.cc",
         # "preferences_repository_test.cc",

internal/platform/implementation/linux/ble_l2cap_socket.cc

@@ -34,9 +34,7 @@
 namespace nearby {
 namespace linux {
 
-BleL2capInputStream::~BleL2capInputStream() {
-  Close();
-}
+BleL2capInputStream::~BleL2capInputStream() { Close(); }
 
 ExceptionOr<ByteArray> BleL2capInputStream::Read(std::int64_t size) {
   std::vector<char> buffer(size);
@@ -56,24 +54,22 @@ ExceptionOr<ByteArray> BleL2capInputStream::Read(std::int64_t size) {
     }
     if (pfds[0].revents & POLLIN) {
       auto r = recv(fd_raw_->get(), buffer.data() + rcvd, size - rcvd, 0);
-      if (r < 0) {
-        return Exception{Exception::kIo};
-      }
+      if (r < 0){ return Exception{Exception::kIo};}
       rcvd += r;
     }
   }
 
+
   return ExceptionOr{ByteArray(std::string(buffer.begin(), buffer.end()))};
 }
 
 Exception BleL2capInputStream::Close() {
   if (!fd_raw_->isValid()) return {Exception::kSuccess};
-  fd_raw_->reset();
-  return {Exception::kSuccess};
-}
-BleL2capOutputStream::~BleL2capOutputStream() {
-  Close();
+  fd_raw_ -> reset();
+    return {Exception::kSuccess};
+
 }
+BleL2capOutputStream::~BleL2capOutputStream() { Close(); }
 
 Exception BleL2capOutputStream::Write(absl::string_view data) {
   pollfd pfds[1];
@@ -91,9 +87,7 @@ Exception BleL2capOutputStream::Write(absl::string_view data) {
     }
     if (pfds[0].revents & POLLOUT) {
       auto r = send(fd_raw_->get(), data.data() + sent, data.size(), 0);
-      if (r < 0) {
-        return Exception{Exception::kIo};
-      }
+      if (r < 0){ return Exception{Exception::kIo};}
       sent += r;
     }
   }
@@ -102,29 +96,32 @@ Exception BleL2capOutputStream::Write(absl::string_view data) {
 
 Exception BleL2capOutputStream::Close() {
   if (!fd_raw_->isValid()) return {Exception::kSuccess};
-  fd_raw_->reset();
-  return {Exception::kSuccess};
+  fd_raw_ -> reset();
+    return {Exception::kSuccess};
 }
 
 BleL2capSocket::BleL2capSocket(int fd,
                                api::ble::BlePeripheral::UniqueId peripheral_id,
-                               std::string service_id)
-    : fd_(std::make_shared<sdbus::UnixFd>(fd)),
-      input_stream_(std::make_unique<BleL2capInputStream>(fd_)),
+                               std::string service_id
+                               )
+    : fd_(std::make_shared<sdbus::UnixFd>(fd)), input_stream_(std::make_unique<BleL2capInputStream>(fd_)),
       output_stream_(std::make_unique<BleL2capOutputStream>(fd_)),
-      peripheral_id_(peripheral_id) {}
+      peripheral_id_(peripheral_id)
+      {}
+
+BleL2capSocket::~BleL2capSocket() { Close(); }
+
 
-BleL2capSocket::~BleL2capSocket() {
-  Close();
-}
 
 Exception BleL2capSocket::Close() {
   if (!fd_->isValid()) return {Exception::kIo};
-  fd_->reset();
-  return {Exception::kSuccess};
+  fd_ -> reset();
+    return {Exception::kSuccess};
 }
 
-void BleL2capSocket::SetCloseNotifier(absl::AnyInvocable<void()> notifier) {}
+
+void BleL2capSocket::SetCloseNotifier(absl::AnyInvocable<void()> notifier) {
+}
 
 bool BleL2capSocket::IsClosed() const {
   return closed_;

internal/platform/implementation/linux/ble_l2cap_socket.h

@@ -35,8 +35,6 @@
 namespace nearby {
 namespace linux {
 
-// TODO: use linux stream instead of bespoke l2cap input/output stream
-
 class BleL2capSocket;
 
 class BleL2capInputStream final : public InputStream {

internal/platform/implementation/linux/bluetooth_classic_socket.cc

@@ -0,0 +1,126 @@
+// Copyright 2023 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 <sys/poll.h>
+#include <sys/socket.h>
+#include <unistd.h>
+#include <array>
+#include <cerrno>
+#include <cstdint>
+#include <cstring>
+
+#include "internal/platform/byte_array.h"
+#include "internal/platform/exception.h"
+#include "internal/platform/implementation/linux/bluetooth_classic_socket.h"
+#include "internal/platform/logging.h"
+
+#include <sys/socket.h>
+#include <unistd.h>
+#include <cerrno>
+#include <cstddef>
+#include <algorithm>
+
+namespace nearby {
+namespace linux {
+
+// This method blocks until input data is available, end of file is detected, or an exception is thrown.
+ExceptionOr<ByteArray> BluetoothInputStream::Read(std::int64_t size) {
+  // Sanity: avoid negative / zero sizes
+  if (size <= 0) return ExceptionOr{ByteArray(std::string())};
+
+
+  std::vector<char> buffer(size);
+
+  // fd returned from bluez assumed to be stream type always
+
+  pollfd pfds[1];
+  pfds[0].fd = fd_raw_->get();
+  pfds[0].events = POLLIN;
+  ssize_t rcvd = 0;
+
+  while (rcvd < size) {
+    int r = poll(pfds, 1, -1);
+    if (r < 0) {
+      if (errno == EINTR) {
+        continue;
+      }
+      return Exception{Exception::kIo};
+    }
+    if (pfds[0].revents & POLLIN) {
+        while (rcvd < size) {
+
+          auto r = recv(fd_raw_->get(), buffer.data() + rcvd, size - rcvd, 0);
+          if (r > 0) {
+            rcvd += r;
+            continue;
+          }
+          if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {continue;}
+
+          LOG(ERROR) << __func__
+                     << ": error reading from fd: " << std::strerror(errno);
+          return {Exception::kIo};
+        }
+    }
+  }
+
+
+  return ExceptionOr{ByteArray(std::string(buffer.begin(), buffer.end()))};
+}
+
+Exception BluetoothInputStream::Close() {
+  if (!fd_raw_->isValid()) return {Exception::kSuccess};
+  fd_raw_ -> reset();
+    return {Exception::kSuccess};
+}
+
+Exception BluetoothOutputStream::Write(absl::string_view data) {
+  pollfd pfds[1];
+  pfds[0].fd = fd_raw_->get();
+  pfds[0].events = POLLOUT;
+  ssize_t sent = 0;
+
+  while (sent < data.size()) {
+    int r = poll(pfds, 1, -1);
+    if (r < 0) {
+      if (errno == EINTR) {
+        continue;
+      }
+      return Exception{Exception::kIo};
+    }
+    if (pfds[0].revents & POLLOUT) {
+      while (sent < data.size()) {
+        auto r = send(fd_raw_->get(), data.data() + sent, data.size(), 0);
+        if (r > 0) {
+          sent += r;
+          continue;
+        }
+        if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {continue;}
+
+        LOG(ERROR) << __func__
+                   << ": error reading from fd: " << std::strerror(errno);
+        return {Exception::kIo};
+      }
+    }
+  }
+  return {Exception::kSuccess};
+}
+
+Exception BluetoothOutputStream::Close() {
+  if (!fd_raw_->isValid()) return {Exception::kSuccess};
+  fd_raw_ -> reset();
+  return {Exception::kSuccess};
+}
+
+}  // namespace linux
+}  // namespace nearby

internal/platform/implementation/linux/bluetooth_classic_socket.h

@@ -27,21 +27,46 @@
 #include "internal/platform/exception.h"
 #include "internal/platform/implementation/bluetooth_classic.h"
 #include "internal/platform/implementation/linux/bluetooth_classic_device.h"
-
-#include "internal/platform/implementation/linux/stream.h"
+#include "internal/platform/input_stream.h"
+#include "internal/platform/output_stream.h"
 
 namespace nearby {
 namespace linux {
 
+class BluetoothInputStream final : public nearby::InputStream {
+ public:
+  explicit BluetoothInputStream(std::shared_ptr<sdbus::UnixFd> fd)
+      : fd_raw_(std::move(fd)){}
+
+  ExceptionOr<ByteArray> Read(std::int64_t size) override;
+  Exception Close() override;
+
+ private:
+  std::shared_ptr<sdbus::UnixFd> fd_raw_;
+};
+
+class BluetoothOutputStream : public nearby::OutputStream {
+ public:
+  explicit BluetoothOutputStream(std::shared_ptr<sdbus::UnixFd> fd)
+      : fd_raw_(std::move(fd)) {}
+
+  Exception Write(absl::string_view data) override;
+  Exception Flush() override { return {Exception::kSuccess}; }
+  Exception Close() override;
+
+ private:
+  mutable absl::Mutex fd_mutex_;
+  std::shared_ptr<sdbus::UnixFd> fd_raw_;
+};
 
 class BluetoothSocket final : public api::BluetoothSocket {
  public:
   BluetoothSocket(std::shared_ptr<BluetoothDevice> device,
                   sdbus::UnixFd fd)
-      :fd_(fd), device_(std::move(device)), output_stream_(fd_), input_stream_(fd_) {}
+      :fd_(std::make_shared<sdbus::UnixFd>(fd)), device_(std::move(device)), output_stream_(fd_), input_stream_(fd_) {}
 
-  InputStream &GetInputStream() override { return input_stream_; }
-  OutputStream &GetOutputStream() override { return output_stream_; }
+  nearby::InputStream &GetInputStream() override { return input_stream_; }
+  nearby::OutputStream &GetOutputStream() override { return output_stream_; }
   Exception Close() override {
     input_stream_.Close();
     output_stream_.Close();
@@ -51,10 +76,10 @@ class BluetoothSocket final : public api::BluetoothSocket {
   api::BluetoothDevice *GetRemoteDevice() override { return device_.get(); };
 
  private:
-  sdbus::UnixFd fd_;
+  std::shared_ptr<sdbus::UnixFd> fd_;
   std::shared_ptr<BluetoothDevice> device_;
-  OutputStream output_stream_;
-  InputStream input_stream_;
+  BluetoothOutputStream output_stream_;
+  BluetoothInputStream input_stream_;
 };
 }  // namespace linux
 }  // namespace nearby

internal/platform/implementation/linux/stream.cc

@@ -12,10 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include <sys/poll.h>
 #include <sys/socket.h>
 #include <unistd.h>
-#include <unistd.h>
 #include <array>
 #include <cerrno>
 #include <cstdint>
@@ -29,149 +27,52 @@ namespace nearby {
 namespace linux {
 
 ExceptionOr<ByteArray> InputStream::Read(std::int64_t size) {
-  if (size <= 0) {
-    return ExceptionOr<ByteArray>(ByteArray(std::string()));
-  }
-
-  if (!fd_ || !fd_->isValid()) {
-    return {Exception::kIo};
-  }
+  if (!fd_.isValid()) return {Exception::kIo};
 
   std::string buffer;
   buffer.resize(size);
-
-  while (true) {
-    pollfd pfd{};
-    pfd.fd = fd_->get();
-    pfd.events = POLLIN;
-
-    int poll_result = poll(&pfd, 1, -1);
-
-    if (poll_result < 0) {
-      if (errno == EINTR) {
-        continue;
-      }
-
-      LOG(ERROR) << __func__ << ": poll failed: " << std::strerror(errno);
-      return {Exception::kIo};
-    }
-
-    if (pfd.revents & POLLNVAL || pfd.revents & POLLERR) {
-      LOG(ERROR) << __func__ << ": Error reading from BluetoothSocket: "
-                 << std::strerror(errno);
-      return {Exception::kIo};
-    }
-
-    if (pfd.revents & (POLLIN | POLLHUP)) {
-      ssize_t bytes_read =
-          recv(fd_->get(), buffer.data(), buffer.size(), 0);
-
-      if (bytes_read > 0) {
-        buffer.resize(static_cast<std::size_t>(bytes_read));
-        return ExceptionOr<ByteArray>(ByteArray(std::move(buffer)));
-      }
-
-      if (bytes_read == 0) {
-        // EOF / peer closed.
-        return ExceptionOr<ByteArray>(ByteArray(std::string()));
-      }
-
-      if (errno == EINTR) {
-        continue;
-      }
-
-      if (errno == EAGAIN || errno == EWOULDBLOCK) {
-        // Non-blocking fd had no data by the time recv() ran.
-        // Go back to poll.
-        continue;
-      }
-
-      LOG(ERROR) << __func__ << ": recv failed: " << std::strerror(errno);
-      return {Exception::kIo};
-    }
+  ssize_t ret = recv(fd_.get(), buffer.data(), buffer.size(), MSG_WAITALL);
+  if (ret == 0) {
+    return ExceptionOr(ByteArray());
   }
+  if (ret < 0) {
+    LOG(ERROR) << __func__
+                       << ": error reading from fd: " << std::strerror(errno);
+    return {Exception::kIo};
+  }
+  buffer.resize(ret);
+
+  return ExceptionOr(ByteArray(std::move(buffer)));
 }
 
 Exception InputStream::Close() {
-  if (!fd_->isValid()) return Exception{Exception::kIo};
+  if (!fd_.isValid()) return Exception{Exception::kIo};
   fd_.reset();
   return {};
 }
 
 Exception OutputStream::Write(absl::string_view data) {
-  if (!fd_ || !fd_->isValid()) {
-    return {Exception::kIo};
-  }
+  if (!fd_.isValid()) return Exception{Exception::kIo};
 
-  const int fd = fd_->get();
-  size_t sent = 0;
-
-  while (sent < data.size()) {
-    pollfd pfd{};
-    pfd.fd = fd;
-    pfd.events = POLLOUT;
-
-    int poll_result;
-    do {
-      poll_result = poll(&pfd, 1, -1);
-    } while (poll_result < 0 && errno == EINTR);
-
-    if (poll_result < 0) {
+  size_t written = 0;
+  while (written < data.size()) {
+    ssize_t ret = write(fd_.get(), data.data(), data.size());
+    if (ret < 0) {
       LOG(ERROR) << __func__
-                 << ": poll failed: " << std::strerror(errno);
-      return {Exception::kIo};
+                         << ": error writing to fd: " << std::strerror(errno);
+      return Exception{Exception::kIo};
     }
-
-    if (pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
-      LOG(ERROR) << __func__
-                 << ": fd error/hangup during write, revents=" << pfd.revents;
-      return {Exception::kIo};
-    }
-
-    if (!(pfd.revents & POLLOUT)) {
-      continue;
-    }
-
-    ssize_t n = send(
-        fd,
-        data.data() + sent,
-        data.size() - sent,
-        MSG_NOSIGNAL);
-
-    if (n > 0) {
-      sent += static_cast<size_t>(n);
-      continue;
-    }
-
-    if (n == 0) {
-      LOG(ERROR) << __func__ << ": send returned 0";
-      return {Exception::kIo};
-    }
-
-    if (errno == EINTR) {
-      continue;
-    }
-
-    if (errno == EAGAIN || errno == EWOULDBLOCK) {
-      // Socket became not writable after poll said it was writable.
-      // Normal for non-blocking FDs. Go back to poll().
-      continue;
-    }
-
-    LOG(ERROR) << __func__
-               << ": error writing to fd: " << std::strerror(errno);
-    return {Exception::kIo};
+    written += ret;
   }
-
-  return {Exception::kSuccess};
+  return Exception{Exception::kSuccess};
 }
 
 Exception OutputStream::Flush() { return Exception{Exception::kSuccess}; }
 
 Exception OutputStream::Close() {
-  if (!fd_->isValid()) return Exception{Exception::kIo};
+  if (!fd_.isValid()) return Exception{Exception::kIo};
 
-  auto ret = close(fd_->get()) < 0 ? Exception{Exception::kIo}
+  auto ret = close(fd_.get()) < 0 ? Exception{Exception::kIo}
                                   : Exception{Exception::kSuccess};
   fd_.reset();
   return ret;

internal/platform/implementation/linux/stream.h

@@ -26,26 +26,26 @@ namespace nearby {
 namespace linux {
 class InputStream : public nearby::InputStream {
  public:
-  explicit InputStream(sdbus::UnixFd fd) : fd_(std::make_shared<sdbus::UnixFd>(fd)){};
+  explicit InputStream(sdbus::UnixFd fd) : fd_(std::move(fd)){};
 
   ExceptionOr<ByteArray> Read(std::int64_t size) override;
 
   Exception Close() override;
 
  private:
-  std::shared_ptr<sdbus::UnixFd> fd_;
+  sdbus::UnixFd fd_;
 };
 
 class OutputStream : public nearby::OutputStream {
  public:
-  explicit OutputStream(sdbus::UnixFd fd) : fd_(std::make_shared<sdbus::UnixFd>(fd)){};
+  explicit OutputStream(sdbus::UnixFd fd) : fd_(std::move(fd)){};
 
   Exception Write(absl::string_view data) override;
   Exception Flush() override;
   Exception Close() override;
 
  private:
-  std::shared_ptr<sdbus::UnixFd> fd_;
+  sdbus::UnixFd fd_;
 };
 
 }  // namespace linux

internal/platform/implementation/linux/timer_test.cc

@@ -14,11 +14,12 @@
 
 #include "internal/platform/implementation/timer.h"
 
+#include <chrono>  // NOLINT
+// NOLINT
 #include <memory>
+#include <thread>  // NOLINT
 
 #include "gtest/gtest.h"
-#include "absl/time/time.h"
-#include "internal/platform/count_down_latch.h"
 #include "internal/platform/implementation/platform.h"
 
 namespace nearby {
@@ -37,21 +38,29 @@ TEST(Timer, TestCreateTimer) {
 }
 
 // This test case cannot run on Google3
-TEST(Timer, TestRepeatTimer) {
-  CountDownLatch latch(3);
+TEST(Timer, DISABLED_TestRepeatTimer) {
   int count = 0;
+
   std::unique_ptr<nearby::api::Timer> timer =
       nearby::api::ImplementationPlatform::CreateTimer();
 
   ASSERT_TRUE(timer != nullptr);
-  EXPECT_TRUE(timer->Create(300, 300, [&]() {
-    ++count;
-    latch.CountDown();
-  }));
-
-  EXPECT_TRUE(latch.Await(absl::Seconds(2)));
-  EXPECT_EQ(count, 3);
+  EXPECT_TRUE(timer->Create(300, 300, [&]() { ++count; }));
+  std::this_thread::sleep_for(std::chrono::seconds(1));
   EXPECT_TRUE(timer->Stop());
+  EXPECT_EQ(count, 3);
+}
+
+TEST(Timer, DISABLED_TestFireNow) {
+  int count = 0;
+
+  auto timer = nearby::api::ImplementationPlatform::CreateTimer();
+
+  EXPECT_TRUE(timer != nullptr);
+  EXPECT_TRUE(timer->Create(3000, 3000, [&]() { ++count; }));
+  EXPECT_TRUE(timer->FireNow());
+  EXPECT_TRUE(timer->Stop());
+  EXPECT_EQ(count, 1);
 }
 
 }  // namespace