Add BluetoothInputStream, BluetoothOutputStream.

internal/platform/implementation/linux/BUILD

@@ -127,9 +127,9 @@ cc_library(
         "avahi.cc",
         "bluetooth_adapter.cc",
         "bluetooth_bluez_profile.cc",
+        "bluetooth_classic_socket.cc",
         "bluetooth_classic_device.cc",
         "bluetooth_classic_medium.cc",
-        "bluetooth_classic_socket.cc",
         "bluetooth_classic_server_socket.cc",
         "bluetooth_devices.cc",
         "bluetooth_pairing.cc",
@@ -142,6 +142,7 @@ cc_library(
         "preferences_manager.cc",
         "preferences_repository.cc",
         "scheduled_executor.cc",
+        "stream.cc",
         "submittable_executor.cc",
         "system_clock.cc",
         "thread_pool.cc",
@@ -212,6 +213,7 @@ cc_library(
     deps = [
         "//internal/platform:base",
         "@nlohmann_json//:json",
+        ":types",
     ],
 )
 
@@ -221,25 +223,20 @@ cc_test(
     srcs = [
         "atomic_boolean_test.cc",
         "atomic_reference_test.cc",
-        "ble_gatt_server_test.cc",
-        "ble_medium_test.cc",
-        "ble_v2_peripheral_test.cc",
-        "ble_v2_test.cc",
-        "bluetooth_adapter_test.cc",
-        "count_down_latch_test.cc",
-        "crypto_test.cc",
-        "device_info_test.cc",
-        "executor_test.cc",
-        "file_path_test.cc",
-        "http_loader_test.cc",
-        "preferences_manager_test.cc",
-        "preferences_repository_test.cc",
-        "scheduled_executor_test.cc",
-        "submittable_executor_test.cc",
-        "thread_pool_test.cc",
-        "timer_test.cc",
-        "utils_test.cc",
-        "webrtc_test.cc",
+        "mutex_test.cc",
+        # "bluetooth_adapter_test.cc",
+        # "crypto_test.cc",
+        # "device_info_test.cc",
+        # "executor_test.cc",
+        # "file_path_test.cc",
+        # "http_loader_test.cc",
+        # "preferences_manager_test.cc",
+        # "preferences_repository_test.cc",
+        # "scheduled_executor_test.cc",
+        # "submittable_executor_test.cc",
+        # "thread_pool_test.cc",
+        # "timer_test.cc",
+        # "utils_test.cc",
     ],
     tags = ["notap"],
     deps = [
@@ -247,7 +244,7 @@ cc_test(
         ":crypto",
         ":test_utils",
         ":types",
-        ":windows",
+        ":linux",
         "//internal/platform:base",
         "//internal/platform/implementation:comm",
         "//internal/platform/implementation:platform",

internal/platform/implementation/linux/bluetooth_classic_socket.cc

@@ -0,0 +1,123 @@
+// 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"
+
+namespace nearby {
+namespace linux {
+Exception Poller::Ready() {
+  while (true) {
+    auto ret = poll(fds_, 1, -1);
+    if (ret < 0) {
+      if (errno == EAGAIN) continue;
+      NEARBY_LOGS(ERROR) << __func__ << ": error polling socket for I/O: "
+                         << std::strerror(errno);
+      return {Exception::kIo};
+    }
+    if ((fds_[0].revents & poll_event_) != 0) {
+      return {Exception::kSuccess};
+    }
+    if ((fds_[0].revents & POLLHUP) != 0) {
+      NEARBY_LOGS(ERROR) << __func__ << ": socket disconnected";
+      return {Exception::kIo};
+    }
+    if ((fds_[0].revents & (POLLERR | POLLNVAL)) != 0) {
+      NEARBY_LOGS(ERROR) << __func__ << ": an error occured on the socket";
+      return {Exception::kIo};
+    }
+  }
+}
+
+ExceptionOr<ByteArray> BluetoothInputStream::Read(std::int64_t size) {
+  if (!fd_.isValid()) return Exception{Exception::kIo};
+
+  auto poller = Poller::CreateInputPoller(fd_);
+
+  std::string buffer;
+  buffer.resize(size);
+  char *data = buffer.data();
+
+  size_t total_read = 0;
+
+  while (total_read < size) {
+    auto result = poller.Ready();
+    if (result.Raised()) return result;
+
+    auto bytes_read = read(fd_.get(), &data[total_read], (size - total_read));
+    if (bytes_read < 0) {
+      if (errno == EAGAIN || errno == EWOULDBLOCK) continue;
+      NEARBY_LOGS(ERROR) << __func__
+                         << ": error reading data on bluetooth socket: "
+                         << std::strerror(errno);
+      return {Exception::kIo};
+    }
+    total_read += bytes_read;
+  }
+
+  return ExceptionOr{ByteArray(std::move(buffer))};
+}
+
+Exception BluetoothInputStream::Close() {
+  if (!fd_.isValid()) return {Exception::kIo};
+  fd_.reset();
+  return {Exception::kSuccess};
+}
+
+Exception BluetoothOutputStream::Write(const ByteArray &data) {
+  if (!fd_.isValid()) return Exception{Exception::kIo};
+
+  auto poller = Poller::CreateOutputPoller(fd_);
+
+  size_t total_wrote = 0;
+
+  while (total_wrote < data.size()) {
+    auto result = poller.Ready();
+    if (result.Raised()) return result;
+
+    const char *buf = data.data();
+    auto wrote =
+        write(fd_.get(), &buf[total_wrote], (data.size() - total_wrote));
+    if (wrote < 0) {
+      if (errno == EAGAIN || errno == EWOULDBLOCK) continue;
+      NEARBY_LOGS(ERROR) << __func__
+                         << ": error writing data on bluetooth socket: "
+                         << std::strerror(errno);
+      return {Exception::kIo};
+    }
+
+    total_wrote += wrote;
+  }
+
+  return {Exception::kSuccess};
+}
+
+Exception BluetoothOutputStream::Close() {
+  if (!fd_.isValid()) return {Exception::kIo};
+  fd_.reset();
+  return {Exception::kSuccess};
+}
+
+}  // namespace linux
+}  // namespace nearby

internal/platform/implementation/linux/bluetooth_classic_socket.h

@@ -19,17 +19,66 @@
 #include <optional>
 
 #include <sdbus-c++/Types.h>
+#include <sys/poll.h>
 #include <systemd/sd-bus.h>
 
 #include "internal/platform/exception.h"
 #include "internal/platform/implementation/bluetooth_classic.h"
-#include "internal/platform/implementation/linux/stream.h"
+#include "internal/platform/input_stream.h"
+#include "internal/platform/output_stream.h"
 
 namespace nearby {
 namespace linux {
+// BlueZ's NewConnection gives us a non-blocking FD, so we need to poll
+// it to be able to write/read bytes.
+class Poller final {
+ public:
+  static Poller CreateInputPoller(const sdbus::UnixFd &fd) {
+    return Poller(fd, POLLIN);
+  }
+
+  static Poller CreateOutputPoller(const sdbus::UnixFd &fd) {
+    return Poller(fd, POLLOUT);
+  }
+
+  Exception Ready();
+
+ private:
+  Poller(const sdbus::UnixFd &fd, short event) : poll_event_(event) {
+    fds_[0].fd = fd.get();
+    fds_[0].events = event;
+  }
+
+  short poll_event_;
+  struct pollfd fds_[1];
+};
+
+class BluetoothInputStream final : public nearby::InputStream {
+ public:
+  explicit BluetoothInputStream(sdbus::UnixFd fd) : fd_(std::move(fd)){};
+
+  ExceptionOr<ByteArray> Read(std::int64_t size) override;
+  Exception Close() override;
+
+ private:
+  sdbus::UnixFd fd_;
+};
+
+class BluetoothOutputStream : public nearby::OutputStream {
+ public:
+  explicit BluetoothOutputStream(sdbus::UnixFd fd) : fd_(std::move(fd)){};
+
+  Exception Write(const ByteArray &data) override;
+  Exception Flush() override {return {Exception::kSuccess};}
+  Exception Close() override;
+
+ private:
+  sdbus::UnixFd fd_;
+};
+
 class BluetoothSocket final : public api::BluetoothSocket {
  public:
-  BluetoothSocket(api::BluetoothDevice &device, sdbus::UnixFd fd)
+  BluetoothSocket(api::BluetoothDevice &device, const sdbus::UnixFd &fd)
       : device_(device), output_stream_(fd), input_stream_(fd) {}
 
   nearby::InputStream &GetInputStream() override { return input_stream_; }
@@ -44,8 +93,8 @@ class BluetoothSocket final : public api::BluetoothSocket {
 
  private:
   api::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,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <sys/socket.h>
 #include <unistd.h>
 #include <array>
 #include <cerrno>
@@ -20,6 +21,7 @@
 #include "internal/platform/byte_array.h"
 #include "internal/platform/exception.h"
 #include "internal/platform/implementation/linux/stream.h"
+#include "internal/platform/logging.h"
 
 namespace nearby {
 namespace linux {
@@ -29,11 +31,13 @@ ExceptionOr<ByteArray> InputStream::Read(std::int64_t size) {
 
   std::string buffer;
   buffer.resize(size);
-  ssize_t ret = read(fd_.get(), buffer.data(), buffer.size());
+  ssize_t ret = recv(fd_.get(), buffer.data(), buffer.size(), MSG_WAITALL);
   if (ret == 0) {
     return ExceptionOr(ByteArray());
   }
   if (ret < 0) {
+    NEARBY_LOGS(ERROR) << __func__
+                       << ": error reading from fd: " << std::strerror(errno);
     return {Exception::kIo};
   }
   buffer.resize(ret);
@@ -53,7 +57,9 @@ Exception OutputStream::Write(const ByteArray &data) {
   size_t written = 0;
   while (written < data.size()) {
     ssize_t ret = write(fd_.get(), data.data(), data.size());
-    if (ret < 1) {
+    if (ret < 0) {
+      NEARBY_LOGS(ERROR) << __func__
+                         << ": error writing to fd: " << std::strerror(errno);
       return Exception{Exception::kIo};
     }
     written += ret;
@@ -67,7 +73,7 @@ Exception OutputStream::Close() {
   if (!fd_.isValid()) return Exception{Exception::kIo};
 
   auto ret = close(fd_.get()) < 0 ? Exception{Exception::kIo}
-                                   : Exception{Exception::kSuccess};
+                                  : Exception{Exception::kSuccess};
   fd_.reset();
   return ret;
 }