From 10f606ee5175ea89eda8dc4b012b698c6efb670f Mon Sep 17 00:00:00 2001 From: Vibhav Pant Date: Fri, 1 Sep 2023 19:58:05 +0530 Subject: [PATCH] Add BluetoothInputStream, BluetoothOutputStream. --- internal/platform/implementation/linux/BUILD | 39 +++--- .../linux/bluetooth_classic_socket.cc | 123 ++++++++++++++++++ .../linux/bluetooth_classic_socket.h | 57 +++++++- .../platform/implementation/linux/stream.cc | 12 +- 4 files changed, 203 insertions(+), 28 deletions(-) create mode 100644 internal/platform/implementation/linux/bluetooth_classic_socket.cc diff --git a/internal/platform/implementation/linux/BUILD b/internal/platform/implementation/linux/BUILD index b387c3e9..60e094a7 100644 --- a/internal/platform/implementation/linux/BUILD +++ b/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", diff --git a/internal/platform/implementation/linux/bluetooth_classic_socket.cc b/internal/platform/implementation/linux/bluetooth_classic_socket.cc new file mode 100644 index 00000000..325cce9f --- /dev/null +++ b/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 +#include +#include +#include +#include +#include +#include + +#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 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 diff --git a/internal/platform/implementation/linux/bluetooth_classic_socket.h b/internal/platform/implementation/linux/bluetooth_classic_socket.h index b7e49606..7833aaa2 100644 --- a/internal/platform/implementation/linux/bluetooth_classic_socket.h +++ b/internal/platform/implementation/linux/bluetooth_classic_socket.h @@ -19,17 +19,66 @@ #include #include +#include #include #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 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 diff --git a/internal/platform/implementation/linux/stream.cc b/internal/platform/implementation/linux/stream.cc index 2f389fab..edfb1da2 100644 --- a/internal/platform/implementation/linux/stream.cc +++ b/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 #include #include #include @@ -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 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; }