From e2256ca83df31abd8e225cb8a5b69adca6455e81 Mon Sep 17 00:00:00 2001 From: kidfromjupiter Date: Sun, 4 Jan 2026 17:14:27 +0000 Subject: [PATCH] endpoints advertised on bluetooth are immediately detected now. bluetooth socket underlying fd type auto detection added. Fixed segfault with download paths --- .../linux/bluetooth_classic_medium.cc | 4 +- .../linux/bluetooth_classic_socket.cc | 226 +++++++++++++++--- .../linux/bluetooth_classic_socket.h | 4 + .../implementation/linux/bluetooth_devices.cc | 60 +++-- .../implementation/linux/bluetooth_devices.h | 8 +- .../platform/implementation/linux/platform.cc | 59 ++++- 6 files changed, 303 insertions(+), 58 deletions(-) diff --git a/internal/platform/implementation/linux/bluetooth_classic_medium.cc b/internal/platform/implementation/linux/bluetooth_classic_medium.cc index 00449b77..117518e4 100644 --- a/internal/platform/implementation/linux/bluetooth_classic_medium.cc +++ b/internal/platform/implementation/linux/bluetooth_classic_medium.cc @@ -48,9 +48,9 @@ BluetoothClassicMedium::BluetoothClassicMedium(BluetoothAdapter &adapter) bool BluetoothClassicMedium::StartDiscovery( DiscoveryCallback discovery_callback) { device_watcher_ = std::make_unique( - *system_bus_, adapter_.GetObjectPath(), devices_, // BUG: this is getting called with devices_ being a nullptr + *system_bus_, adapter_.GetObjectPath(), adapter_, devices_, std::make_unique(std::move(discovery_callback)), - observers_); // BUG: observers_ is a nullptr + observers_); std::map filter; filter["Transport"] = "auto"; diff --git a/internal/platform/implementation/linux/bluetooth_classic_socket.cc b/internal/platform/implementation/linux/bluetooth_classic_socket.cc index 2bc6a03c..6991b89c 100644 --- a/internal/platform/implementation/linux/bluetooth_classic_socket.cc +++ b/internal/platform/implementation/linux/bluetooth_classic_socket.cc @@ -25,6 +25,36 @@ #include "internal/platform/implementation/linux/bluetooth_classic_socket.h" #include "internal/platform/logging.h" +#include +#include +#include +#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() { @@ -50,45 +80,124 @@ Exception Poller::Ready() { } } -ExceptionOr BluetoothInputStream::Read(std::int64_t size) { - // Check if FD is valid before proceeding - { - absl::MutexLock lock(&fd_mutex_); - if (!fd_.isValid()) return Exception{Exception::kIo}; - } - // Create poller while we still have the FD (copy the fd value) +ExceptionOr BluetoothInputStream::Read(std::int64_t size) { + absl::MutexLock lock(&fd_mutex_); + if (!fd_.isValid()) return Exception{Exception::kIo}; + auto poller = Poller::CreateInputPoller(fd_); + int so_type = 0; + socklen_t sl = sizeof(so_type); + if (::getsockopt(fd_.get(), 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) { + auto result = poller.Ready(); + if (result.Raised()) return result; + + // 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_.get(), 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}; + } + 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). + ssize_t n = ::recv(fd_.get(), 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))}; + } + } + + // ---- Stream-based: read exactly 'size' bytes (your original behavior) ---- std::string buffer; - buffer.resize(size); - char *data = buffer.data(); + buffer.resize(static_cast(size)); + char* data = buffer.data(); size_t total_read = 0; - - while (total_read < size) { + while (total_read < static_cast(size)) { auto result = poller.Ready(); if (result.Raised()) return result; - auto bytes_read = read(fd_.get(), &data[total_read], (size - total_read)); + ssize_t bytes_read = ::read(fd_.get(), + 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) { - // FD was closed by another thread LOG(INFO) << __func__ << ": socket was closed during read"; return {Exception::kIo}; } - LOG(ERROR) << __func__ - << ": error reading data on bluetooth socket: " - << std::strerror(errno); + LOG(ERROR) << __func__ << ": error reading data on bluetooth socket: " + << std::strerror(errno); return {Exception::kIo}; } if (bytes_read == 0) { - // EOF - socket closed LOG(INFO) << __func__ << ": socket closed (EOF)"; return {Exception::kIo}; } - total_read += bytes_read; + total_read += static_cast(bytes_read); } return ExceptionOr{ByteArray(std::move(buffer))}; @@ -102,38 +211,93 @@ Exception BluetoothInputStream::Close() { } Exception BluetoothOutputStream::Write(const ByteArray &data) { - // Check if FD is valid before proceeding - { - absl::MutexLock lock(&fd_mutex_); - if (!fd_.isValid()) return Exception{Exception::kIo}; - } + absl::MutexLock lock(&fd_mutex_); + if (!fd_.isValid()) return Exception{Exception::kIo}; - // Create poller while we still have the FD (copy the fd value) auto poller = Poller::CreateOutputPoller(fd_); size_t total_wrote = 0; + int so_type = 0; + socklen_t sl = sizeof(so_type); + if (::getsockopt(fd_.get(), 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 && max_chunk_ == 0) max_chunk_ = 1024; + + LOG(INFO) << "SO_TYPE=" << so_type + << (so_type == SOCK_SEQPACKET ? " (SEQPACKET)" : + so_type == SOCK_STREAM ? " (STREAM)" : + so_type == SOCK_DGRAM ? " (DGRAM)" : " (other)") + << (packet_based ? absl::StrFormat(" max_chunk=%zu", max_chunk_) : ""); + while (total_wrote < data.size()) { - auto result = poller.Ready(); + auto result = poller.Ready(); // should wait for POLLOUT/EPOLLOUT if (result.Raised()) return result; const char *buf = data.data(); - auto wrote = - write(fd_.get(), &buf[total_wrote], (data.size() - total_wrote)); + 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. + to_write = std::min(to_write, max_chunk_); + } + + // Prefer send() to avoid SIGPIPE (MSG_NOSIGNAL is Linux). + ssize_t wrote = ::send(fd_.get(), + 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. + 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}; + } + if (errno == EBADF || errno == EPIPE) { - // FD was closed by another thread LOG(INFO) << __func__ << ": socket was closed during write"; return {Exception::kIo}; } + LOG(ERROR) << __func__ - << ": error writing data on bluetooth socket: " - << std::strerror(errno); + << ": error writing data on bluetooth socket: " + << std::strerror(errno); return {Exception::kIo}; } - total_wrote += wrote; + if (wrote == 0) { + // For sockets, 0 usually means peer closed. + LOG(INFO) << __func__ << ": peer closed during write"; + return {Exception::kIo}; + } + + total_wrote += static_cast(wrote); } return {Exception::kSuccess}; diff --git a/internal/platform/implementation/linux/bluetooth_classic_socket.h b/internal/platform/implementation/linux/bluetooth_classic_socket.h index 8aac4053..7b3b93db 100644 --- a/internal/platform/implementation/linux/bluetooth_classic_socket.h +++ b/internal/platform/implementation/linux/bluetooth_classic_socket.h @@ -78,6 +78,10 @@ class BluetoothOutputStream : public nearby::OutputStream { private: mutable absl::Mutex fd_mutex_; sdbus::UnixFd fd_ ABSL_GUARDED_BY(fd_mutex_); + + // 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; }; class BluetoothSocket final : public api::BluetoothSocket { diff --git a/internal/platform/implementation/linux/bluetooth_devices.cc b/internal/platform/implementation/linux/bluetooth_devices.cc index 822abe9f..259a24ab 100644 --- a/internal/platform/implementation/linux/bluetooth_devices.cc +++ b/internal/platform/implementation/linux/bluetooth_devices.cc @@ -21,6 +21,7 @@ #include "absl/strings/substitute.h" #include "absl/synchronization/mutex.h" +#include "internal/platform/implementation/linux/bluetooth_adapter.h" #include "internal/platform/implementation/linux/bluetooth_classic_device.h" #include "internal/platform/implementation/linux/bluetooth_devices.h" #include "internal/platform/implementation/linux/bluez.h" @@ -159,7 +160,6 @@ void DeviceWatcher::onInterfacesRemoved( } void DeviceWatcher::notifyExistingDevices() { - // NOTE: Existing devices don't get identified as endpoints. They only std::map>> objects; @@ -169,23 +169,53 @@ void DeviceWatcher::notifyExistingDevices() { DBUS_LOG_METHOD_CALL_ERROR(this, "GetManagedObjects", e); return; } - auto device_it = - std::find_if(objects.begin(), objects.end(), [&](auto entry) { - auto &[device_path, interfaces] = entry; + std::vector existing_device_paths; - return device_path.find( - absl::Substitute("$0/dev_", adapter_object_path_)) == 0 && - interfaces.count(org::bluez::Device1_proxy::INTERFACE_NAME) == 1; - }); + for (const auto& [device_path, interfaces] : objects) { + if (device_path.find(absl::Substitute("$0/dev_", adapter_object_path_)) == 0 && + interfaces.count(org::bluez::Device1_proxy::INTERFACE_NAME) == 1) { + + // Don't remove bonded, paired, connected, or trusted devices + bool should_skip = false; + auto device_interface_it = interfaces.find(org::bluez::Device1_proxy::INTERFACE_NAME); + if (device_interface_it != interfaces.end()) { + const auto& properties = device_interface_it->second; + + auto check_bool_property = [&properties](const std::string& prop_name) -> bool { + auto it = properties.find(prop_name); + if (it != properties.end()) { + try { + return it->second.get(); + } catch (...) { + return false; + } + } + return false; + }; + + if (check_bool_property("Bonded") || + check_bool_property("Paired") || + check_bool_property("Connected") || + check_bool_property("Trusted")) { + should_skip = true; + LOG(INFO) << __func__ << ": Skipping device " << device_path + << " (bonded/paired/connected/trusted)"; + } + } + + if (!should_skip) { + existing_device_paths.push_back(device_path); + } + } + } - - for (; device_it != objects.end(); device_it++) { - LOG(INFO) << __func__ << ": Adding existing device " - << device_it->first; - auto device = devices_->add_new_device(device_it->first); - if (discovery_cb_ != nullptr) { - device->SetDiscoveryCallback(discovery_cb_); + // Remove existing devices - they will be immediately re-discovered + // This triggers InterfacesAdded signals which properly invoke discovery callbacks + for (const auto& device_path : existing_device_paths) { + LOG(INFO) << __func__ << ": Refreshing existing device " << device_path; + if (!adapter_.RemoveDeviceByObjectPath(device_path)) { + LOG(WARNING) << __func__ << ": Failed to remove device " << device_path; } } } diff --git a/internal/platform/implementation/linux/bluetooth_devices.h b/internal/platform/implementation/linux/bluetooth_devices.h index 19966b92..a18ac126 100644 --- a/internal/platform/implementation/linux/bluetooth_devices.h +++ b/internal/platform/implementation/linux/bluetooth_devices.h @@ -34,6 +34,8 @@ namespace nearby { namespace linux { +class BluetoothAdapter; + class BluetoothDevices final { public: BluetoothDevices( @@ -97,6 +99,7 @@ class DeviceWatcher final : sdbus::ProxyInterfaces { DeviceWatcher( sdbus::IConnection &system_bus, const sdbus::ObjectPath &adapter_object_path, + BluetoothAdapter &adapter, std::shared_ptr devices, std::unique_ptr discovery_callback, @@ -104,6 +107,7 @@ class DeviceWatcher final : sdbus::ProxyInterfaces { observers) : ProxyInterfaces(system_bus, "org.bluez", "/"), adapter_object_path_(adapter_object_path), + adapter_(adapter), devices_(std::move(devices)), discovery_cb_(std::move(discovery_callback)), observers_(std::move(observers)) { @@ -112,8 +116,9 @@ class DeviceWatcher final : sdbus::ProxyInterfaces { } DeviceWatcher(sdbus::IConnection &system_bus, const sdbus::ObjectPath &adapter_object_path, + BluetoothAdapter &adapter, std::shared_ptr devices) - : DeviceWatcher(system_bus, adapter_object_path, std::move(devices), + : DeviceWatcher(system_bus, adapter_object_path, adapter, std::move(devices), nullptr, nullptr) {} ~DeviceWatcher() { unregisterProxy(); } @@ -128,6 +133,7 @@ class DeviceWatcher final : sdbus::ProxyInterfaces { void notifyExistingDevices(); sdbus::ObjectPath adapter_object_path_; + BluetoothAdapter &adapter_; std::shared_ptr devices_; std::shared_ptr discovery_cb_; std::shared_ptr> diff --git a/internal/platform/implementation/linux/platform.cc b/internal/platform/implementation/linux/platform.cc index 4ade4f66..e95107b0 100644 --- a/internal/platform/implementation/linux/platform.cc +++ b/internal/platform/implementation/linux/platform.cc @@ -61,27 +61,68 @@ namespace api { std::string ImplementationPlatform::GetCustomSavePath( const std::string &parent_folder, const std::string &file_name) { auto fs = std::filesystem::path(parent_folder); - return fs / file_name; + return (fs / file_name).string(); } std::string ImplementationPlatform::GetDownloadPath( const std::string &parent_folder, const std::string &file_name) { - auto downloads = std::filesystem::path(getenv("XDG_DOWNLOAD_DIR")); - - return downloads / std::filesystem::path(parent_folder).filename() / - std::filesystem::path(file_name).filename(); + std::filesystem::path downloads; + const char* download_dir = getenv("XDG_DOWNLOAD_DIR"); + + if (download_dir != nullptr) { + downloads = std::filesystem::path(download_dir); + } else { + // Fallback to ~/Downloads if XDG_DOWNLOAD_DIR is not set + const char* home = getenv("HOME"); + if (home != nullptr) { + downloads = std::filesystem::path(home) / "Downloads"; + } else { + downloads = "/tmp/Downloads"; + } + } + + return (downloads / std::filesystem::path(parent_folder).filename() / + std::filesystem::path(file_name).filename()).string(); } std::string ImplementationPlatform::GetDownloadPath( const std::string &file_name) { - auto downloads = std::filesystem::path(getenv("XDG_DOWNLOAD_DIR")); - return downloads / std::filesystem::path(file_name).filename(); + std::filesystem::path downloads; + const char* download_dir = getenv("XDG_DOWNLOAD_DIR"); + + if (download_dir != nullptr) { + downloads = std::filesystem::path(download_dir); + } else { + // Fallback to ~/Downloads if XDG_DOWNLOAD_DIR is not set + const char* home = getenv("HOME"); + if (home != nullptr) { + downloads = std::filesystem::path(home) / "Downloads"; + } else { + downloads = "/tmp/Downloads"; + } + } + + return (downloads / std::filesystem::path(file_name).filename()).string(); } std::string ImplementationPlatform::GetAppDataPath( const std::string &file_name) { - auto state = std::filesystem::path(getenv("XDG_STATE_HOME")); - return state / std::filesystem::path(file_name).filename(); + std::filesystem::path state; + const char* state_home = getenv("XDG_STATE_HOME"); + + if (state_home != nullptr) { + state = std::filesystem::path(state_home); + } else { + // Fallback to ~/.local/state if XDG_STATE_HOME is not set + const char* home = getenv("HOME"); + if (home != nullptr) { + state = std::filesystem::path(home) / ".local" / "state"; + } else { + state = "/tmp/state"; + } + } + + return (state / std::filesystem::path(file_name).filename()).string(); } OSName ImplementationPlatform::GetCurrentOS() { return OSName::kWindows; }