diff --git a/internal/platform/implementation/linux/BUILD b/internal/platform/implementation/linux/BUILD index d0470bf9..eaa6b031 100644 --- a/internal/platform/implementation/linux/BUILD +++ b/internal/platform/implementation/linux/BUILD @@ -304,6 +304,7 @@ cc_test( "atomic_reference_test.cc", "mutex_test.cc", "utils_test.cc", + "ble_l2cap_socket_test.cc", # "bluetooth_adapter_test.cc", # "crypto_test.cc", # "device_info_test.cc", @@ -322,14 +323,12 @@ cc_test( ":comm", ":crypto", ":linux", - ":test_utils", ":types", "//internal/platform:base", "//internal/platform/implementation:comm", "//internal/platform/implementation:platform", "//internal/platform/implementation:types", "//internal/platform/implementation/shared:count_down_latch", - "@com_github_protobuf_matchers//protobuf-matchers", "@com_google_absl//absl/status", "@com_google_absl//absl/strings", "@com_google_absl//absl/synchronization", diff --git a/internal/platform/implementation/linux/ble_l2cap_server_socket.cc b/internal/platform/implementation/linux/ble_l2cap_server_socket.cc index 22c2cdbc..a3ef4b69 100644 --- a/internal/platform/implementation/linux/ble_l2cap_server_socket.cc +++ b/internal/platform/implementation/linux/ble_l2cap_server_socket.cc @@ -29,208 +29,86 @@ #include "internal/platform/logging.h" +#include "internal/platform/prng.h" + namespace nearby { namespace linux { -namespace { - -bool SetNonBlocking(int fd) { - int flags = fcntl(fd, F_GETFL, 0); - if (flags < 0) return false; - return fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0; -} - -void DrainFd(int fd) { - char buf[64]; - while (true) { - ssize_t read_count = read(fd, buf, sizeof(buf)); - if (read_count > 0) continue; - if (read_count < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) break; - break; - } -} - -} // namespace BleL2capServerSocket::BleL2capServerSocket() = default; -BleL2capServerSocket::BleL2capServerSocket( - int psm, std::string service_id) - : psm_(psm), - service_id_(std::move(service_id)) {} +BleL2capServerSocket::BleL2capServerSocket(int psm, std::string service_id) + : psm_(psm), service_id_(std::move(service_id)) {} -BleL2capServerSocket::~BleL2capServerSocket() { Close(); } +BleL2capServerSocket::~BleL2capServerSocket() { + Close(); +} void BleL2capServerSocket::SetPSM(int psm) { absl::MutexLock lock(&mutex_); psm_ = psm; } -bool BleL2capServerSocket::InitializeServerSocketLocked() { - if (closed_) { - errno = EINTR; - return false; - } - - if (server_fd_ >= 0) { - return true; - } - - if ((stop_pipe_[0] == -1) != (stop_pipe_[1] == -1)) { - if (stop_pipe_[0] != -1) close(stop_pipe_[0]); - if (stop_pipe_[1] != -1) close(stop_pipe_[1]); - stop_pipe_[0] = -1; - stop_pipe_[1] = -1; - } - - if (stop_pipe_[0] == -1) { - if (pipe(stop_pipe_) < 0) { - LOG(ERROR) << "Failed to create stop pipe: " << std::strerror(errno); - return false; - } - if (!SetNonBlocking(stop_pipe_[0]) || !SetNonBlocking(stop_pipe_[1])) { - LOG(ERROR) << "Failed to set non-blocking mode on stop pipe: " - << std::strerror(errno); - close(stop_pipe_[0]); - close(stop_pipe_[1]); - stop_pipe_[0] = -1; - stop_pipe_[1] = -1; - return false; - } - } - - server_fd_ = socket(AF_BLUETOOTH, SOCK_SEQPACKET, BTPROTO_L2CAP); - if (server_fd_ < 0) { - LOG(ERROR) << "Failed to create L2CAP server socket: " - << std::strerror(errno); - return false; - } - - sockaddr_l2 addr; - std::memset(&addr, 0, sizeof(addr)); - addr.l2_family = AF_BLUETOOTH; - addr.l2_psm = htobs(psm_); - addr.l2_bdaddr_type = BDADDR_LE_PUBLIC; - std::memset(&addr.l2_bdaddr, 0, sizeof(addr.l2_bdaddr)); - - if (bind(server_fd_, reinterpret_cast(&addr), sizeof(addr)) < 0) { - LOG(ERROR) << "Failed to bind L2CAP server socket: " << std::strerror(errno) - << " (errno: " << errno << ")"; - close(server_fd_); - server_fd_ = -1; - return false; - } - - constexpr uint16_t kReceiveMtu = 672; - if (setsockopt(server_fd_, SOL_BLUETOOTH, BT_RCVMTU, &kReceiveMtu, - sizeof(kReceiveMtu)) < 0) { - LOG(WARNING) << "Failed to set receive MTU on L2CAP server socket: " - << std::strerror(errno); - } - - if (listen(server_fd_, 5) < 0) { - LOG(ERROR) << "Failed to listen on L2CAP server socket: " - << std::strerror(errno); - close(server_fd_); - server_fd_ = -1; - return false; - } - - if (!SetNonBlocking(server_fd_)) { - LOG(ERROR) << "Failed to set non-blocking on L2CAP server socket: " - << std::strerror(errno); - close(server_fd_); - server_fd_ = -1; - return false; - } - - socklen_t addr_len = sizeof(addr); - if (getsockname(server_fd_, reinterpret_cast(&addr), &addr_len) == - 0) { - psm_ = btohs(addr.l2_psm); - LOG(INFO) << "L2CAP server socket listening on PSM: " << psm_; - } else { - LOG(WARNING) << "Failed to get socket name: " << std::strerror(errno); - } - - return true; -} - -void BleL2capServerSocket::AcceptPoll(int server_fd, int stop_fd, int& client_fd, - sockaddr_l2& client_addr, - socklen_t& client_len) { - while (true) { - pollfd fds[2]; - fds[0].fd = server_fd; - fds[0].events = POLLIN; - fds[0].revents = 0; - fds[1].fd = stop_fd; - fds[1].events = POLLIN; - fds[1].revents = 0; - - int result = poll(fds, 2, -1); - if (result < 0) { - if (errno == EINTR) continue; - LOG(ERROR) << "poll() failed: " << std::strerror(errno); - client_fd = -1; - return; - } - - if (fds[1].revents & (POLLIN | POLLERR | POLLHUP | POLLNVAL)) { - if (fds[1].revents & POLLIN) { - DrainFd(stop_fd); - } - client_fd = -1; - errno = EINTR; - return; - } - - if (fds[0].revents & (POLLERR | POLLHUP | POLLNVAL)) { - LOG(ERROR) << "poll() listen fd error revents=" << fds[0].revents; - client_fd = -1; - errno = EIO; - return; - } - - if ((fds[0].revents & POLLIN) == 0) continue; - - while (true) { - client_fd = accept(server_fd, reinterpret_cast(&client_addr), - &client_len); - if (client_fd >= 0) { - return; - } - - if (errno == EINTR) continue; - if (errno == EAGAIN || errno == EWOULDBLOCK) { - client_fd = -1; - break; - } - - LOG(ERROR) << "Failed to accept L2CAP connection: " << std::strerror(errno); - return; - } - } -} - std::unique_ptr BleL2capServerSocket::Accept() { - int server_fd = -1; - int stop_fd = -1; - int listening_psm = 0; { - absl::MutexLock lock(&mutex_); - if (!InitializeServerSocketLocked()) return nullptr; - server_fd = server_fd_; - stop_fd = stop_pipe_[0]; - listening_psm = psm_; + server_fd_ = socket(AF_BLUETOOTH, SOCK_SEQPACKET, BTPROTO_L2CAP); + if (server_fd_ < 0) { + LOG(ERROR) << "Failed to create L2CAP server socket: " + << std::strerror(errno); + return nullptr; + } + + // generating psm value for l2cap socket + Prng prng; + psm_ = 0x80 + (prng.NextUint32() % 0x80); + + sockaddr_l2 addr; + std::memset(&addr, 0, sizeof(addr)); + addr.l2_family = AF_BLUETOOTH; + addr.l2_psm = htobs(psm_); + addr.l2_bdaddr_type = BDADDR_LE_PUBLIC; + std::memset(&addr.l2_bdaddr, 0, sizeof(addr.l2_bdaddr)); + + if (bind(server_fd_, reinterpret_cast(&addr), sizeof(addr)) < + 0) { + LOG(ERROR) << "Failed to bind L2CAP server socket: " + << std::strerror(errno) << " (errno: " << errno << ")"; + close(server_fd_); + server_fd_ = -1; + return nullptr; + } + + constexpr uint16_t kReceiveMtu = 672; + if (setsockopt(server_fd_, SOL_BLUETOOTH, BT_RCVMTU, &kReceiveMtu, + sizeof(kReceiveMtu)) < 0) { + LOG(WARNING) << "Failed to set receive MTU on L2CAP server socket: " + << std::strerror(errno); + } + + if (listen(server_fd_, 5) < 0) { + LOG(ERROR) << "Failed to listen on L2CAP server socket: " + << std::strerror(errno); + close(server_fd_); + server_fd_ = -1; + return nullptr; + } + + socklen_t addr_len = sizeof(addr); + if (getsockname(server_fd_, reinterpret_cast(&addr), + &addr_len) == 0) { + psm_ = btohs(addr.l2_psm); + LOG(INFO) << "L2CAP server socket listening on PSM: " << psm_; + } else { + LOG(WARNING) << "Failed to get socket name: " << std::strerror(errno); + } } sockaddr_l2 client_addr; std::memset(&client_addr, 0, sizeof(client_addr)); socklen_t client_len = sizeof(client_addr); - LOG(INFO) << "Waiting for L2CAP connection on PSM " << listening_psm << "..."; - int client_fd = -1; - AcceptPoll(server_fd, stop_fd, client_fd, client_addr, client_len); + LOG(INFO) << "Waiting for L2CAP connection on PSM " << psm_ << "..."; + + int client_fd = accept(server_fd_, (struct sockaddr*) &client_addr, &client_len); if (client_fd < 0) { if (errno == EINTR || errno == EAGAIN) { @@ -249,8 +127,7 @@ std::unique_ptr BleL2capServerSocket::Accept() { api::ble::BlePeripheral::UniqueId peripheral_id = 0; for (int i = 0; i < 6; ++i) { peripheral_id = - (peripheral_id << 8) | - static_cast(client_addr.l2_bdaddr.b[i]); + (peripheral_id << 8) | static_cast(client_addr.l2_bdaddr.b[i]); } std::string service_id; @@ -258,67 +135,25 @@ std::unique_ptr BleL2capServerSocket::Accept() { absl::MutexLock lock(&mutex_); service_id = service_id_; } - return std::make_unique( - client_fd, peripheral_id, service_id, - /*incoming_connection=*/true); + return std::make_unique(client_fd, peripheral_id, service_id ); } Exception BleL2capServerSocket::Close() { - absl::AnyInvocable notifier; int server_fd = -1; - int stop_read_fd = -1; - int stop_write_fd = -1; { absl::MutexLock lock(&mutex_); if (closed_) { return {Exception::kSuccess}; } closed_ = true; - notifier = std::move(close_notifier_); server_fd = std::exchange(server_fd_, -1); - stop_read_fd = std::exchange(stop_pipe_[0], -1); - stop_write_fd = std::exchange(stop_pipe_[1], -1); } - - if (stop_write_fd != -1) { - char wake = 'x'; - ssize_t ignored = write(stop_write_fd, &wake, 1); - (void)ignored; - } - if (server_fd != -1 && close(server_fd) != 0) { - LOG(WARNING) << "Failed to close L2CAP server socket: " << std::strerror(errno); - } - if (stop_read_fd != -1 && close(stop_read_fd) != 0) { - LOG(WARNING) << "Failed to close stop pipe read fd: " << std::strerror(errno); - } - if (stop_write_fd != -1 && close(stop_write_fd) != 0) { - LOG(WARNING) << "Failed to close stop pipe write fd: " + LOG(WARNING) << "Failed to close L2CAP server socket: " << std::strerror(errno); } - - if (notifier) { - notifier(); - } - return {Exception::kSuccess}; } -void BleL2capServerSocket::SetCloseNotifier( - absl::AnyInvocable notifier) { - absl::AnyInvocable notifier_to_run; - { - absl::MutexLock lock(&mutex_); - if (!closed_) { - close_notifier_ = std::move(notifier); - return; - } - notifier_to_run = std::move(notifier); - } - if (notifier_to_run) { - notifier_to_run(); - } -} - } // namespace linux } // namespace nearby diff --git a/internal/platform/implementation/linux/ble_l2cap_server_socket.h b/internal/platform/implementation/linux/ble_l2cap_server_socket.h index 62956af7..dcd619be 100644 --- a/internal/platform/implementation/linux/ble_l2cap_server_socket.h +++ b/internal/platform/implementation/linux/ble_l2cap_server_socket.h @@ -44,21 +44,14 @@ class BleL2capServerSocket final : public api::ble::BleL2capServerSocket { ABSL_LOCKS_EXCLUDED(mutex_); Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_); - void SetCloseNotifier(absl::AnyInvocable notifier) - ABSL_LOCKS_EXCLUDED(mutex_); private: - bool InitializeServerSocketLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - void AcceptPoll(int server_fd, int stop_fd, int& client_fd, - sockaddr_l2& client_addr, socklen_t& client_len); absl::Mutex mutex_; bool closed_ ABSL_GUARDED_BY(mutex_) = false; - absl::AnyInvocable close_notifier_ ABSL_GUARDED_BY(mutex_); int psm_ = 0; std::string service_id_ ABSL_GUARDED_BY(mutex_); int server_fd_ ABSL_GUARDED_BY(mutex_) = -1; - int stop_pipe_[2] ABSL_GUARDED_BY(mutex_) = {-1, -1}; // read [0], write [1] }; } // namespace linux diff --git a/internal/platform/implementation/linux/ble_l2cap_socket.cc b/internal/platform/implementation/linux/ble_l2cap_socket.cc index eec4a152..e8f26c0c 100644 --- a/internal/platform/implementation/linux/ble_l2cap_socket.cc +++ b/internal/platform/implementation/linux/ble_l2cap_socket.cc @@ -33,332 +33,97 @@ namespace nearby { namespace linux { -namespace { - -constexpr int kHeaderLength = 4; -constexpr int kServiceIdHashLength = 3; -constexpr int kMaxFrameLength = 1024 * 1024; -constexpr uint8_t kControlPacketPrefix[kServiceIdHashLength] = {0x00, 0x00, 0x00}; - -using ::location::nearby::mediums::SocketControlFrame; -using ::location::nearby::mediums::SocketVersion; - -struct ParsedSocketControlFrame { - SocketControlFrame::ControlFrameType type; - ByteArray service_id_hash; - int received_size = 0; -}; - -std::optional ParseSocketControlFramePayload( - absl::string_view payload) { - if (payload.size() <= kServiceIdHashLength) return std::nullopt; - if (!std::equal(std::begin(kControlPacketPrefix), std::end(kControlPacketPrefix), - payload.begin())) { - return std::nullopt; - } - - SocketControlFrame frame; - if (!frame.ParseFromArray(payload.data() + kServiceIdHashLength, - payload.size() - kServiceIdHashLength)) { - return std::nullopt; - } - - ParsedSocketControlFrame parsed{ - .type = frame.type(), - .service_id_hash = ByteArray(), - .received_size = 0, - }; - - switch (frame.type()) { - case SocketControlFrame::INTRODUCTION: - if (!frame.has_introduction() || !frame.introduction().has_service_id_hash() || - frame.introduction().socket_version() != SocketVersion::V2) { - return std::nullopt; - } - parsed.service_id_hash = - ByteArray(frame.introduction().service_id_hash().data(), - frame.introduction().service_id_hash().size()); - return parsed; - case SocketControlFrame::DISCONNECTION: - if (!frame.has_disconnection() || !frame.disconnection().has_service_id_hash()) { - return std::nullopt; - } - parsed.service_id_hash = - ByteArray(frame.disconnection().service_id_hash().data(), - frame.disconnection().service_id_hash().size()); - return parsed; - case SocketControlFrame::PACKET_ACKNOWLEDGEMENT: - if (!frame.has_packet_acknowledgement() || - !frame.packet_acknowledgement().has_service_id_hash()) { - return std::nullopt; - } - parsed.service_id_hash = - ByteArray(frame.packet_acknowledgement().service_id_hash().data(), - frame.packet_acknowledgement().service_id_hash().size()); - parsed.received_size = frame.packet_acknowledgement().has_received_size() - ? frame.packet_acknowledgement().received_size() - : 0; - return parsed; - case SocketControlFrame::UNKNOWN_CONTROL_FRAME_TYPE: - return std::nullopt; - } - return std::nullopt; -} - -std::optional ReadBigEndianUint32(absl::string_view bytes) { - if (bytes.size() != kHeaderLength) return std::nullopt; - const auto* ptr = reinterpret_cast(bytes.data()); - return (static_cast(ptr[0]) << 24) | - (static_cast(ptr[1]) << 16) | - (static_cast(ptr[2]) << 8) | - static_cast(ptr[3]); -} - -std::string WriteBigEndianUint32(uint32_t value) { - std::string out(kHeaderLength, '\0'); - out[0] = static_cast((value >> 24) & 0xFF); - out[1] = static_cast((value >> 16) & 0xFF); - out[2] = static_cast((value >> 8) & 0xFF); - out[3] = static_cast(value & 0xFF); - return out; -} - -} // namespace - -BleL2capInputStream::BleL2capInputStream(BleL2capSocket* owner) : owner_(owner) {} BleL2capInputStream::~BleL2capInputStream() { Close(); } ExceptionOr BleL2capInputStream::Read(std::int64_t size) { - if (owner_ == nullptr) { - return ExceptionOr(Exception::kIo); + std::vector buffer(size); + + 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) { + auto r = recv(fd_raw_->get(), buffer.data() + rcvd, size - rcvd, 0); + if (r < 0){ return Exception{Exception::kIo};} + rcvd += r; + } } - return owner_->ReadFromSocket(size); + + + return ExceptionOr{ByteArray(std::string(buffer.begin(), buffer.end()))}; } Exception BleL2capInputStream::Close() { - if (owner_ == nullptr) return {Exception::kSuccess}; - return owner_->CloseIo(); + if (!fd_raw_->isValid()) return {Exception::kSuccess}; + fd_raw_ -> reset(); + return {Exception::kSuccess}; + } - -BleL2capOutputStream::BleL2capOutputStream(BleL2capSocket* owner) : owner_(owner) {} - BleL2capOutputStream::~BleL2capOutputStream() { Close(); } Exception BleL2capOutputStream::Write(absl::string_view data) { - if (owner_ == nullptr) { - return {Exception::kIo}; + 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) { + auto r = send(fd_raw_->get(), data.data() + sent, data.size(), 0); + if (r < 0){ return Exception{Exception::kIo};} + sent += r; + } } - return owner_->WriteToSocket(data); + return {Exception::kSuccess}; } Exception BleL2capOutputStream::Close() { - if (owner_ == nullptr) return {Exception::kSuccess}; - return owner_->CloseIo(); + if (!fd_raw_->isValid()) return {Exception::kSuccess}; + fd_raw_ -> reset(); + return {Exception::kSuccess}; } BleL2capSocket::BleL2capSocket(int fd, api::ble::BlePeripheral::UniqueId peripheral_id, - std::string service_id, - bool incoming_connection) - : input_stream_(std::make_unique(this)), - output_stream_(std::make_unique(this)), - peripheral_id_(peripheral_id), - fd_(fd), - incoming_connection_(incoming_connection), - intro_packet_validated_(!incoming_connection) { - ByteArray hash = Crypto::Sha256(service_id); - service_id_hash_ = ByteArray(hash.data(), kServiceIdHashLength); -} + std::string service_id + ) + : fd_(std::make_shared(fd)), input_stream_(std::make_unique(fd_)), + output_stream_(std::make_unique(fd_)), + peripheral_id_(peripheral_id) + {} BleL2capSocket::~BleL2capSocket() { Close(); } -bool BleL2capSocket::PollReady(short events, std::optional timeout) const { - int fd = fd_.load(); - if (fd < 0) return false; - - int timeout_ms = -1; - if (timeout.has_value()) { - timeout_ms = std::max(0, absl::ToInt64Milliseconds(*timeout)); - } - - struct pollfd pfd; - pfd.fd = fd; - pfd.events = events; - pfd.revents = 0; - - while (true) { - int ret = poll(&pfd, 1, timeout_ms); - if (ret < 0) { - if (errno == EINTR) continue; - return false; - } - if (ret == 0) return false; - - if ((pfd.revents & events) != 0) return true; - if ((pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) != 0) { - return false; - } - } -} - -bool BleL2capSocket::SendFrame(absl::string_view payload) { - if (payload.size() > kMaxFrameLength) { - return false; - } - std::string framed = WriteBigEndianUint32(static_cast(payload.size())); - framed.append(payload.data(), payload.size()); - - absl::MutexLock lock(&io_mutex_); - int fd = fd_.load(); - if (fd < 0) return false; - - size_t offset = 0; - while (offset < framed.size()) { - if (!PollReady(POLLOUT, std::nullopt)) { - return false; - } - fd = fd_.load(); - if (fd < 0) return false; - - ssize_t sent = send( - fd, framed.data() + offset, framed.size() - offset, -#ifdef MSG_NOSIGNAL - MSG_NOSIGNAL -#else - 0 -#endif - ); - if (sent < 0) { - if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) continue; - return false; - } - if (sent == 0) return false; - offset += static_cast(sent); - } - return true; -} - -bool BleL2capSocket::ReadNextFrame(std::string& payload, - std::optional timeout) { - const std::optional deadline = - timeout.has_value() ? std::make_optional(absl::Now() + *timeout) : std::nullopt; - - while (true) { - if (wire_buffer_.size() >= kHeaderLength) { - auto frame_length = ReadBigEndianUint32( - absl::string_view(wire_buffer_.data(), kHeaderLength)); - if (!frame_length.has_value() || *frame_length > kMaxFrameLength) { - return false; - } - const size_t total = kHeaderLength + *frame_length; - if (wire_buffer_.size() >= total) { - payload = wire_buffer_.substr(kHeaderLength, *frame_length); - wire_buffer_.erase(0, total); - return true; - } - } - - std::optional remaining = std::nullopt; - if (deadline.has_value()) { - remaining = *deadline - absl::Now(); - if (*remaining <= absl::ZeroDuration()) { - return false; - } - } - - if (!PollReady(POLLIN, remaining)) return false; - - int fd = fd_.load(); - if (fd < 0) return false; - - char buffer[1024]; - ssize_t read_count = recv(fd, buffer, sizeof(buffer), 0); - if (read_count < 0) { - if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) continue; - return false; - } - if (read_count == 0) return false; - wire_buffer_.append(buffer, static_cast(read_count)); - } -} - - -ExceptionOr BleL2capSocket::ReadFromSocket(std::int64_t size) { - if (size <= 0) return ExceptionOr(ByteArray()); - - while (read_buffer_.empty()) { - std::string payload; - if (!ReadNextFrame(payload, std::nullopt)) - return ExceptionOr(Exception::kIo); - - read_buffer_.append(payload); - } - - const size_t read_size = - std::min(static_cast(size), read_buffer_.size()); - ByteArray result(read_buffer_.substr(0, read_size)); - read_buffer_.erase(0, read_size); - return ExceptionOr(result); -} - -Exception BleL2capSocket::WriteToSocket(absl::string_view data) { - - if (service_id_hash_.size() != kServiceIdHashLength) return {Exception::kIo}; - - std::string payload; - payload.reserve(service_id_hash_.size() + data.size()); - payload.append(service_id_hash_.AsStringView()); - payload.append(data.data(), data.size()); - - return SendFrame(payload) ? Exception{Exception::kSuccess} - : Exception{Exception::kIo}; -} - -Exception BleL2capSocket::CloseIo() { - int fd = fd_.exchange(-1); - if (fd < 0) return {Exception::kSuccess}; - - shutdown(fd, SHUT_RDWR); - return {Exception::kSuccess}; -} - Exception BleL2capSocket::Close() { - absl::MutexLock lock(&mutex_); - if (closed_) { + if (!fd_->isValid()) return {Exception::kIo}; + fd_ -> reset(); return {Exception::kSuccess}; - } - DoClose(); - return {Exception::kSuccess}; } -void BleL2capSocket::DoClose() { - closed_ = true; - - if (input_stream_) { - input_stream_->Close(); - } - if (output_stream_) { - output_stream_->Close(); - } - - if (close_notifier_) { - auto notifier = std::move(close_notifier_); - mutex_.Unlock(); - notifier(); - mutex_.Lock(); - } -} void BleL2capSocket::SetCloseNotifier(absl::AnyInvocable notifier) { - absl::MutexLock lock(&mutex_); - close_notifier_ = std::move(notifier); } bool BleL2capSocket::IsClosed() const { - absl::MutexLock lock(&mutex_); return closed_; } diff --git a/internal/platform/implementation/linux/ble_l2cap_socket.h b/internal/platform/implementation/linux/ble_l2cap_socket.h index bf4c283a..cdea0bf1 100644 --- a/internal/platform/implementation/linux/ble_l2cap_socket.h +++ b/internal/platform/implementation/linux/ble_l2cap_socket.h @@ -15,6 +15,9 @@ #ifndef PLATFORM_IMPL_LINUX_BLE_L2CAP_SOCKET_H_ #define PLATFORM_IMPL_LINUX_BLE_L2CAP_SOCKET_H_ +#include "dbus.h" + + #include #include #include @@ -36,34 +39,35 @@ class BleL2capSocket; class BleL2capInputStream final : public InputStream { public: - explicit BleL2capInputStream(BleL2capSocket* owner); + explicit BleL2capInputStream(std::shared_ptr fd_raw_): fd_raw_(std::move(fd_raw_)) {}; ~BleL2capInputStream() override; ExceptionOr Read(std::int64_t size) override; Exception Close() override; - private: - BleL2capSocket* owner_ = nullptr; +private: + std::shared_ptr fd_raw_; }; class BleL2capOutputStream final : public OutputStream { - public: - explicit BleL2capOutputStream(BleL2capSocket* owner); +public: + explicit BleL2capOutputStream(std::shared_ptr fd_raw_): fd_raw_(std::move(fd_raw_)) {}; ~BleL2capOutputStream() override; Exception Write(absl::string_view data) override; Exception Flush() override { return {Exception::kSuccess}; } Exception Close() override; - private: - BleL2capSocket* owner_ = nullptr; +private: + std::shared_ptr fd_raw_; + }; class BleL2capSocket final : public api::ble::BleL2capSocket { public: BleL2capSocket(int fd, api::ble::BlePeripheral::UniqueId peripheral_id, - std::string service_id = "", bool incoming_connection = false); + std::string service_id = ""); ~BleL2capSocket() override; InputStream& GetInputStream() override { return *input_stream_; } @@ -81,35 +85,13 @@ class BleL2capSocket final : public api::ble::BleL2capSocket { friend class BleL2capInputStream; friend class BleL2capOutputStream; - - ExceptionOr ReadFromSocket(std::int64_t size) - ABSL_LOCKS_EXCLUDED(io_mutex_); - Exception WriteToSocket(absl::string_view data) ABSL_LOCKS_EXCLUDED(io_mutex_); - Exception CloseIo() ABSL_LOCKS_EXCLUDED(io_mutex_); - - bool ReadNextFrame(std::string& payload, std::optional timeout) - ABSL_LOCKS_EXCLUDED(io_mutex_); - bool SendFrame(absl::string_view payload) ABSL_LOCKS_EXCLUDED(io_mutex_); - - bool PollReady(short events, std::optional timeout) const; - void DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - mutable absl::Mutex mutex_; mutable absl::Mutex io_mutex_; bool closed_ ABSL_GUARDED_BY(mutex_) = false; + std::shared_ptr fd_ ; std::unique_ptr input_stream_; std::unique_ptr output_stream_; api::ble::BlePeripheral::UniqueId peripheral_id_; - absl::AnyInvocable close_notifier_ ABSL_GUARDED_BY(mutex_); - std::atomic fd_{-1}; - - const bool incoming_connection_; - ByteArray service_id_hash_; - - bool intro_packet_validated_ = false; - bool request_data_connection_handled_ = false; - std::string wire_buffer_; - std::string read_buffer_; }; } // namespace linux diff --git a/internal/platform/implementation/linux/ble_l2cap_socket_test.cc b/internal/platform/implementation/linux/ble_l2cap_socket_test.cc index dfdf845e..5bee9a27 100644 --- a/internal/platform/implementation/linux/ble_l2cap_socket_test.cc +++ b/internal/platform/implementation/linux/ble_l2cap_socket_test.cc @@ -27,236 +27,93 @@ #include "absl/time/time.h" #include "internal/platform/implementation/crypto.h" #include "gtest/gtest.h" -#include "proto/mediums/ble_frames.pb.h" namespace nearby { namespace linux { namespace { +class BleL2capSocketTest : public ::testing::Test { +protected: +void SetUp() override { + ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, fds_), 0); -using ::location::nearby::mediums::SocketControlFrame; -using ::location::nearby::mediums::SocketVersion; + socket_fd_ = fds_[0]; + peer_fd_ = fds_[1]; -constexpr uint8_t kRequestDataConnection = 3; -constexpr uint8_t kResponseDataConnectionReady = 23; - -class SocketPair final { - public: - SocketPair() { - int fds[2] = {-1, -1}; - ASSERT_EQ(0, socketpair(AF_UNIX, SOCK_STREAM, 0, fds)); - left_ = fds[0]; - right_ = fds[1]; + socket_ = std::make_unique(socket_fd_, 1); } - ~SocketPair() { - if (left_ >= 0) close(left_); - if (right_ >= 0) close(right_); - } + void TearDown() override { + if (socket_) { + socket_->Close(); + } - int left() const { return left_; } - int right() const { return right_; } - - private: - int left_ = -1; - int right_ = -1; -}; - -bool ReadExact(int fd, char* data, size_t length) { - size_t offset = 0; - while (offset < length) { - ssize_t read_count = recv(fd, data + offset, length - offset, 0); - if (read_count <= 0) { - return false; + if (peer_fd_ >= 0) { + close(peer_fd_); } - offset += static_cast(read_count); } - return true; + int fds_[2]{-1, -1}; + int socket_fd_{-1}; + int peer_fd_{-1}; + std::unique_ptr socket_; +}; +TEST_F(BleL2capSocketTest, ReturnsInputAndOutputStreams) { + InputStream& input = socket_->GetInputStream(); + OutputStream& output = socket_->GetOutputStream(); + + EXPECT_NE(&input, nullptr); + EXPECT_NE(&output, nullptr); } +TEST_F(BleL2capSocketTest, ReturnsSameStreamInstancesAcrossCalls) { + InputStream& input1 = socket_->GetInputStream(); + InputStream& input2 = socket_->GetInputStream(); -std::optional ReceiveFrame(int fd) { - std::array header; - if (!ReadExact(fd, header.data(), header.size())) { - return std::nullopt; - } - const auto* bytes = reinterpret_cast(header.data()); - uint32_t payload_size = (static_cast(bytes[0]) << 24) | - (static_cast(bytes[1]) << 16) | - (static_cast(bytes[2]) << 8) | - static_cast(bytes[3]); - std::string payload(payload_size, '\0'); - if (payload_size > 0 && !ReadExact(fd, payload.data(), payload.size())) { - return std::nullopt; - } - return payload; + OutputStream& output1 = socket_->GetOutputStream(); + OutputStream& output2 = socket_->GetOutputStream(); + + EXPECT_EQ(&input1, &input2); + EXPECT_EQ(&output1, &output2); } +TEST_F(BleL2capSocketTest, ReadReceivesExactBytesFromPeer) { + std::string message = "hello into l2cap socket"; -bool SendFrame(int fd, const std::string& payload) { - std::array header = { - static_cast((payload.size() >> 24) & 0xFF), - static_cast((payload.size() >> 16) & 0xFF), - static_cast((payload.size() >> 8) & 0xFF), - static_cast(payload.size() & 0xFF), - }; - if (send(fd, header.data(), header.size(), 0) != static_cast(header.size())) { - return false; - } - if (!payload.empty() && - send(fd, payload.data(), payload.size(), 0) != static_cast(payload.size())) { - return false; - } - return true; + ASSERT_EQ( + write(peer_fd_, message.data(), message.size()), + static_cast(message.size()) + ); + + InputStream& input = socket_->GetInputStream(); + + std::vector buffer(message.size()); + auto out = input.Read(buffer.size()).GetResult(); + + EXPECT_EQ( + out, + ByteArray(message) + ); + + EXPECT_EQ( + out.AsStringView(), + message + ); } +TEST_F(BleL2capSocketTest, WriteSendsExactBytesToPeer) { + std::string message = "hello from l2cap socket"; -ByteArray ServiceHash(absl::string_view service_id) { - ByteArray full_hash = Crypto::Sha256(service_id); - EXPECT_GE(full_hash.size(), 3); - return ByteArray(full_hash.data(), 3); + OutputStream& output = socket_->GetOutputStream(); + + EXPECT_EQ(output.Write(message).value, Exception::kSuccess); + + std::vector received(message.size()); + + ssize_t n = read(peer_fd_, received.data(), received.size()); + + ASSERT_EQ(n, static_cast(message.size())); + + EXPECT_EQ( + std::string(received.begin(), received.end()), + message + ); } - -std::string BuildLegacyControlPacket(uint8_t command) { - return std::string(1, static_cast(command)); -} - -std::optional ParseLegacyControlCommand(absl::string_view payload) { - if (payload.empty()) return std::nullopt; - if (payload.size() != 1 && payload.size() < 3) return std::nullopt; - return static_cast(payload[0]); -} - -std::string BuildLegacyIntroPacket(const ByteArray& service_hash) { - SocketControlFrame frame; - frame.set_type(SocketControlFrame::INTRODUCTION); - auto* intro = frame.mutable_introduction(); - intro->set_service_id_hash(service_hash.AsStringView()); - intro->set_socket_version(SocketVersion::V2); - - std::string serialized(frame.ByteSizeLong(), '\0'); - EXPECT_TRUE(frame.SerializeToArray(serialized.data(), serialized.size())); - - std::string packet("\x00\x00\x00", 3); - packet.append(serialized); - return packet; -} - -bool IsLegacyIntroPacketForServiceHash(absl::string_view payload, - const ByteArray& service_hash) { - if (payload.size() <= 3 || payload.substr(0, 3) != "\x00\x00\x00") return false; - - SocketControlFrame frame; - if (!frame.ParseFromArray(payload.data() + 3, payload.size() - 3)) return false; - return frame.type() == SocketControlFrame::INTRODUCTION && - frame.has_introduction() && - frame.introduction().socket_version() == SocketVersion::V2 && - frame.introduction().service_id_hash() == service_hash.AsStringView(); -} - -TEST(BleL2capSocketTest, RefactoredOutputStreamWritesFramedPayload) { - SocketPair pair; - BleL2capSocket socket(pair.left(), /*peripheral_id=*/1); - - ASSERT_TRUE(socket.GetOutputStream().Write(ByteArray("hello")).Ok()); - auto payload = ReceiveFrame(pair.right()); - ASSERT_TRUE(payload.has_value()); - EXPECT_EQ(*payload, "hello"); -} - -TEST(BleL2capSocketTest, RefactoredInputStreamReadsFramedPayload) { - SocketPair pair; - BleL2capSocket socket(pair.left(), /*peripheral_id=*/1); - - ASSERT_TRUE(SendFrame(pair.right(), "world")); - ExceptionOr read_result = socket.GetInputStream().Read(5); - ASSERT_TRUE(read_result.ok()); - EXPECT_EQ(read_result.result().string_data(), "world"); -} - -TEST(BleL2capSocketTest, LegacyWritePrefixesServiceHash) { - SocketPair pair; - ByteArray service_hash = ServiceHash("service"); - BleL2capSocket socket(pair.left(), /*peripheral_id=*/1, - BleL2capSocket::ProtocolMode::kLegacy, - /*service_id=*/"service", /*incoming_connection=*/false); - - ASSERT_TRUE(socket.GetOutputStream().Write(ByteArray("abc")).Ok()); - auto payload = ReceiveFrame(pair.right()); - ASSERT_TRUE(payload.has_value()); - ASSERT_EQ(payload->size(), service_hash.size() + 3); - EXPECT_EQ(payload->substr(0, service_hash.size()), service_hash.AsStringView()); - EXPECT_EQ(payload->substr(service_hash.size()), "abc"); -} - -TEST(BleL2capSocketTest, LegacyOutgoingHandshakeSucceeds) { - SocketPair pair; - ByteArray service_hash = ServiceHash("service"); - BleL2capSocket socket(pair.left(), /*peripheral_id=*/1, - BleL2capSocket::ProtocolMode::kLegacy, - /*service_id=*/"service", /*incoming_connection=*/false); - - std::thread peer([&]() { - auto request = ReceiveFrame(pair.right()); - ASSERT_TRUE(request.has_value()); - auto request_command = ParseLegacyControlCommand(*request); - ASSERT_TRUE(request_command.has_value()); - EXPECT_EQ(*request_command, kRequestDataConnection); - - ASSERT_TRUE(SendFrame(pair.right(), BuildLegacyControlPacket(kResponseDataConnectionReady))); - - auto intro = ReceiveFrame(pair.right()); - ASSERT_TRUE(intro.has_value()); - EXPECT_TRUE(IsLegacyIntroPacketForServiceHash(*intro, service_hash)); - }); - - EXPECT_TRUE(socket.PerformLegacyOutgoingHandshake(absl::Seconds(2))); - peer.join(); -} - -TEST(BleL2capSocketTest, LegacyOutgoingHandshakeTimesOutWithoutResponse) { - SocketPair pair; - BleL2capSocket socket(pair.left(), /*peripheral_id=*/1, - BleL2capSocket::ProtocolMode::kLegacy, - /*service_id=*/"service", /*incoming_connection=*/false); - - std::thread peer([&]() { - auto request = ReceiveFrame(pair.right()); - ASSERT_TRUE(request.has_value()); - auto request_command = ParseLegacyControlCommand(*request); - ASSERT_TRUE(request_command.has_value()); - EXPECT_EQ(*request_command, kRequestDataConnection); - }); - - EXPECT_FALSE(socket.PerformLegacyOutgoingHandshake(absl::Milliseconds(200))); - peer.join(); -} - -TEST(BleL2capSocketTest, LegacyIncomingConnectionHandlesHandshakeAndData) { - SocketPair pair; - ByteArray service_hash = ServiceHash("service"); - BleL2capSocket socket(pair.left(), /*peripheral_id=*/1, - BleL2capSocket::ProtocolMode::kLegacy, - /*service_id=*/"service", /*incoming_connection=*/true); - - std::thread peer([&]() { - ASSERT_TRUE(SendFrame(pair.right(), BuildLegacyControlPacket(kRequestDataConnection))); - - auto response = ReceiveFrame(pair.right()); - ASSERT_TRUE(response.has_value()); - auto response_command = ParseLegacyControlCommand(*response); - ASSERT_TRUE(response_command.has_value()); - EXPECT_EQ(*response_command, kResponseDataConnectionReady); - - ASSERT_TRUE(SendFrame(pair.right(), BuildLegacyIntroPacket(service_hash))); - - std::string data_payload(service_hash.AsStringView()); - data_payload.append("hello"); - ASSERT_TRUE(SendFrame(pair.right(), data_payload)); - }); - - ExceptionOr read_result = socket.GetInputStream().Read(5); - ASSERT_TRUE(read_result.ok()); - EXPECT_EQ(read_result.result().string_data(), "hello"); - peer.join(); -} - } // namespace } // namespace linux } // namespace nearby diff --git a/internal/platform/implementation/linux/ble_v2_medium.cc b/internal/platform/implementation/linux/ble_v2_medium.cc index ca255dc2..88710c50 100644 --- a/internal/platform/implementation/linux/ble_v2_medium.cc +++ b/internal/platform/implementation/linux/ble_v2_medium.cc @@ -76,9 +76,6 @@ BleV2Medium::BleV2Medium(BluetoothAdapter &adapter) << ": Failed to initialize known GATT services cache."; } - // generating psm value for l2cap socket - Prng prng; - psm_ = 0x80 + (prng.NextUint32() % 0x80); if (adv_monitor_manager_) { LOG(INFO) @@ -634,8 +631,7 @@ std::unique_ptr BleV2Medium::ConnectOverL2cap( LOG(INFO) << __func__ << ": Successfully connected to L2CAP socket"; auto socket = std::make_unique( - fd, peripheral_id, service_id, - /*incoming_connection=*/false); + fd, peripheral_id, service_id); return socket; } diff --git a/internal/platform/implementation/linux/bluetooth_classic_socket.cc b/internal/platform/implementation/linux/bluetooth_classic_socket.cc index 1ae65cd9..9dd48b98 100644 --- a/internal/platform/implementation/linux/bluetooth_classic_socket.cc +++ b/internal/platform/implementation/linux/bluetooth_classic_socket.cc @@ -31,300 +31,75 @@ #include #include -struct SocketWriteCaps { - int so_type = 0; // SOCK_STREAM / SOCK_SEQPACKET / SOCK_DGRAM - size_t max_chunk = 0; // 0 => unknown/unlimited - bool packet_based = false; -}; - -static SocketWriteCaps DetectCapsNoBtHeaders(int fd) { - SocketWriteCaps caps{}; - - socklen_t len = sizeof(caps.so_type); - if (::getsockopt(fd, SOL_SOCKET, SO_TYPE, &caps.so_type, &len) != 0) { - // If we can't detect, behave conservatively like stream. - caps.so_type = SOCK_STREAM; - } - - caps.packet_based = (caps.so_type == SOCK_SEQPACKET || caps.so_type == SOCK_DGRAM); - - // Initial guess for packet-based sockets. This will be refined on EMSGSIZE. - if (caps.packet_based) caps.max_chunk = 1024; // start guess - else caps.max_chunk = 0; // unlimited/stream - - return caps; -} - namespace nearby { namespace linux { -Exception Poller::Ready() { - while (true) { - auto ret = poll(fds_, 1, -1); - if (ret < 0) { - if (errno == EAGAIN) continue; - LOG(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) { - LOG(ERROR) << __func__ << ": socket disconnected"; - return {Exception::kIo}; - } - if ((fds_[0].revents & (POLLERR | POLLNVAL)) != 0) { - LOG(ERROR) << __func__ << ": an error occured on the socket"; - return {Exception::kIo}; - } - } -} - +// This method blocks until input data is available, end of file is detected, or an exception is thrown. ExceptionOr BluetoothInputStream::Read(std::int64_t size) { - int fd = fd_raw_.load(); - if (fd < 0) return Exception{Exception::kIo}; - - auto poller = Poller::CreateInputPoller(fd); - - int so_type = 0; - socklen_t sl = sizeof(so_type); - if (::getsockopt(fd, SOL_SOCKET, SO_TYPE, &so_type, &sl) != 0) { - // If unknown, default to stream-ish behavior. - so_type = SOCK_STREAM; - } - const bool packet_based = (so_type == SOCK_SEQPACKET || so_type == SOCK_DGRAM); - // Sanity: avoid negative / zero sizes if (size <= 0) return ExceptionOr{ByteArray(std::string())}; - // ---- Packet-based: read ONE message (recommended) ---- - if (packet_based) { - // Wait for readability - while (true) { - if (fd_raw_.load() != fd) return {Exception::kIo}; - auto result = poller.Ready(); - if (result.Raised()) return result; - if (fd_raw_.load() != fd) return {Exception::kIo}; - // Peek the next message length without consuming it. - // For seqpacket/dgram, MSG_TRUNC makes recv() return the *full* message length - // even if the buffer is smaller. - ssize_t msg_len = ::recv(fd, nullptr, 0, MSG_PEEK | MSG_TRUNC); - if (msg_len < 0) { - if (errno == EINTR) continue; - if (errno == EAGAIN || errno == EWOULDBLOCK) continue; - if (errno == EBADF) { - LOG(INFO) << __func__ << ": socket was closed during read"; - return {Exception::kIo}; - } - LOG(ERROR) << __func__ << ": error peeking message length: " - << std::strerror(errno); - return {Exception::kIo}; + std::vector 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; } - if (msg_len == 0) { - LOG(INFO) << __func__ << ": socket closed (EOF)"; - return {Exception::kIo}; - } - - // Decide how much we will actually read/return. - // If caller asked for 'size', cap to that. - size_t want = static_cast(msg_len); - size_t cap = static_cast(size); - size_t to_read = std::min(want, cap); - - std::string buffer; - buffer.resize(to_read); - - // Now read/consume the message. If the message is larger than to_read, - // the remainder will be discarded by the kernel for seqpacket/dgram. - // We can detect that and treat it as an error (or choose a different policy). - if (fd_raw_.load() != fd) return {Exception::kIo}; - ssize_t n = ::recv(fd, buffer.data(), to_read, 0); - if (n < 0) { - if (errno == EINTR) continue; - if (errno == EAGAIN || errno == EWOULDBLOCK) continue; - if (errno == EBADF) { - LOG(INFO) << __func__ << ": socket was closed during read"; - return {Exception::kIo}; - } - LOG(ERROR) << __func__ << ": error reading data on bluetooth socket: " - << std::strerror(errno); - return {Exception::kIo}; - } - if (n == 0) { - LOG(INFO) << __func__ << ": socket closed (EOF)"; - return {Exception::kIo}; - } - - buffer.resize(static_cast(n)); - - // Detect truncation: if msg_len > size, we truncated/discarded remainder. - if (want > cap) { - LOG(ERROR) << __func__ - << ": incoming packet (" << want - << " bytes) exceeds requested size (" << cap - << "). Packet truncated."; - return {Exception::kIo}; - } - - return ExceptionOr{ByteArray(std::move(buffer))}; + return Exception{Exception::kIo}; + } + if (pfds[0].revents & POLLIN) { + auto r = recv(fd_raw_->get(), buffer.data() + rcvd, size - rcvd, 0); + if (r < 0){ return Exception{Exception::kIo};} + rcvd += r; } } - // ---- Stream-based: read exactly 'size' bytes (your original behavior) ---- - std::string buffer; - buffer.resize(static_cast(size)); - char* data = buffer.data(); - size_t total_read = 0; - while (total_read < static_cast(size)) { - if (fd_raw_.load() != fd) return {Exception::kIo}; - auto result = poller.Ready(); - if (result.Raised()) return result; - if (fd_raw_.load() != fd) return {Exception::kIo}; - - ssize_t bytes_read = ::read(fd, - data + total_read, - static_cast(size) - total_read); - if (bytes_read < 0) { - if (errno == EINTR) continue; - if (errno == EAGAIN || errno == EWOULDBLOCK) continue; - if (errno == EBADF) { - LOG(INFO) << __func__ << ": socket was closed during read"; - return {Exception::kIo}; - } - LOG(ERROR) << __func__ << ": error reading data on bluetooth socket: " - << std::strerror(errno); - return {Exception::kIo}; - } - if (bytes_read == 0) { - LOG(INFO) << __func__ << ": socket closed (EOF)"; - return {Exception::kIo}; - } - total_read += static_cast(bytes_read); - } - - return ExceptionOr{ByteArray(std::move(buffer))}; + return ExceptionOr{ByteArray(std::string(buffer.begin(), buffer.end()))}; } Exception BluetoothInputStream::Close() { - int fd = fd_raw_.exchange(-1); - if (fd < 0) return {Exception::kSuccess}; // Already closed - ::shutdown(fd, SHUT_RDWR); - fd_.reset(); - return {Exception::kSuccess}; + if (!fd_raw_->isValid()) return {Exception::kSuccess}; + fd_raw_ -> reset(); + return {Exception::kSuccess}; } Exception BluetoothOutputStream::Write(absl::string_view data) { - int fd = fd_raw_.load(); - if (fd < 0) return Exception{Exception::kIo}; + pollfd pfds[1]; + pfds[0].fd = fd_raw_->get(); + pfds[0].events = POLLOUT; + ssize_t sent = 0; - auto poller = Poller::CreateOutputPoller(fd); - - size_t total_wrote = 0; - - int so_type = 0; - socklen_t sl = sizeof(so_type); - if (::getsockopt(fd, SOL_SOCKET, SO_TYPE, &so_type, &sl) != 0) { - // If we can’t detect, assume stream semantics (no per-message MTU). - so_type = SOCK_STREAM; - } - - const bool packet_based = (so_type == SOCK_SEQPACKET || so_type == SOCK_DGRAM); - - // Initialize a reasonable starting guess for packet-based sockets. - // This will be refined down on EMSGSIZE. - if (packet_based) { - absl::MutexLock lock(&fd_mutex_); - if (max_chunk_ == 0) max_chunk_ = 1024; - } - - size_t max_chunk = 0; - if (packet_based) { - absl::MutexLock lock(&fd_mutex_); - max_chunk = max_chunk_; - } - - - while (total_wrote < data.size()) { - if (fd_raw_.load() != fd) return {Exception::kIo}; - auto result = poller.Ready(); // should wait for POLLOUT/EPOLLOUT - if (result.Raised()) return result; - if (fd_raw_.load() != fd) return {Exception::kIo}; - - const char *buf = data.data(); - size_t remaining = data.size() - total_wrote; - - size_t to_write = remaining; - if (packet_based) { - // For SEQPACKET/DGRAM, one send() == one packet. - // Cap to discovered “MTU-like” limit to avoid EMSGSIZE. - absl::MutexLock lock(&fd_mutex_); - to_write = std::min(to_write, max_chunk_); - } - - // Prefer send() to avoid SIGPIPE (MSG_NOSIGNAL is Linux). - ssize_t wrote = ::send(fd, - buf + total_wrote, - to_write, -#ifdef MSG_NOSIGNAL - MSG_NOSIGNAL -#else - 0 -#endif - ); - - // If send() isn’t appropriate in your environment, you can swap back to write(). - // ssize_t wrote = ::write(fd_.get(), buf + total_wrote, to_write); - - if (wrote < 0) { - if (errno == EINTR) continue; - if (errno == EAGAIN || errno == EWOULDBLOCK) continue; - - if (errno == EMSGSIZE && packet_based) { - // Our packet is too large; shrink max_chunk_ and retry. - { - absl::MutexLock lock(&fd_mutex_); - if (max_chunk_ > 1) { - max_chunk_ = std::max(1, max_chunk_ / 2); - LOG(INFO) << __func__ << ": EMSGSIZE; reducing max_chunk_ to " - << max_chunk_; - continue; // retry with smaller chunk - } - } - LOG(ERROR) << __func__ << ": EMSGSIZE even at 1 byte"; - return {Exception::kIo}; + while (sent < data.size()) { + int r = poll(pfds, 1, -1); + if (r < 0) { + if (errno == EINTR) { + continue; } - - if (errno == EBADF || errno == EPIPE) { - LOG(INFO) << __func__ << ": socket was closed during write"; - return {Exception::kIo}; - } - - LOG(ERROR) << __func__ - << ": error writing data on bluetooth socket: " - << std::strerror(errno); - return {Exception::kIo}; + return Exception{Exception::kIo}; } - - if (wrote == 0) { - // For sockets, 0 usually means peer closed. - LOG(INFO) << __func__ << ": peer closed during write"; - return {Exception::kIo}; + if (pfds[0].revents & POLLOUT) { + auto r = send(fd_raw_->get(), data.data() + sent, data.size(), 0); + if (r < 0){ return Exception{Exception::kIo};} + sent += r; } - - total_wrote += static_cast(wrote); } - return {Exception::kSuccess}; } Exception BluetoothOutputStream::Close() { - int fd = fd_raw_.exchange(-1); - if (fd < 0) return {Exception::kSuccess}; // Already closed - ::shutdown(fd, SHUT_RDWR); - fd_.reset(); + if (!fd_raw_->isValid()) return {Exception::kSuccess}; + fd_raw_ -> reset(); return {Exception::kSuccess}; } diff --git a/internal/platform/implementation/linux/bluetooth_classic_socket.h b/internal/platform/implementation/linux/bluetooth_classic_socket.h index df700046..b9fe89e6 100644 --- a/internal/platform/implementation/linux/bluetooth_classic_socket.h +++ b/internal/platform/implementation/linux/bluetooth_classic_socket.h @@ -32,56 +32,23 @@ 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); - } - - static Poller CreateInputPoller(int fd) { return Poller(fd, POLLIN); } - - static Poller CreateOutputPoller(int 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; - } - - Poller(int fd, short event) : poll_event_(event) { - fds_[0].fd = fd; - 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)), fd_raw_(fd_.get()) {} + explicit BluetoothInputStream(std::shared_ptr fd) + : fd_raw_(std::move(fd)){} ExceptionOr Read(std::int64_t size) override; Exception Close() override; private: - sdbus::UnixFd fd_; - std::atomic fd_raw_{-1}; + std::shared_ptr fd_raw_; }; class BluetoothOutputStream : public nearby::OutputStream { public: - explicit BluetoothOutputStream(sdbus::UnixFd fd) - : fd_(std::move(fd)), fd_raw_(fd_.get()) {} + explicit BluetoothOutputStream(std::shared_ptr fd) + : fd_raw_(std::move(fd)) {} Exception Write(absl::string_view data) override; Exception Flush() override { return {Exception::kSuccess}; } @@ -89,19 +56,14 @@ class BluetoothOutputStream : public nearby::OutputStream { private: mutable absl::Mutex fd_mutex_; - sdbus::UnixFd fd_; - std::atomic fd_raw_{-1}; - - // For packet sockets, discovered max payload per send/write. - // 0 means "unknown", we’ll initialize on first packet write. - size_t max_chunk_ ABSL_GUARDED_BY(fd_mutex_) = 0; + std::shared_ptr fd_raw_; }; class BluetoothSocket final : public api::BluetoothSocket { public: BluetoothSocket(std::shared_ptr device, - const sdbus::UnixFd &fd) - : device_(std::move(device)), output_stream_(fd), input_stream_(fd) {} + sdbus::UnixFd fd) + :fd_(std::make_shared(fd)), device_(std::move(device)), output_stream_(fd_), input_stream_(fd_) {} nearby::InputStream &GetInputStream() override { return input_stream_; } nearby::OutputStream &GetOutputStream() override { return output_stream_; } @@ -114,6 +76,7 @@ class BluetoothSocket final : public api::BluetoothSocket { api::BluetoothDevice *GetRemoteDevice() override { return device_.get(); }; private: + std::shared_ptr fd_; std::shared_ptr device_; BluetoothOutputStream output_stream_; BluetoothInputStream input_stream_;