TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-16 15:36:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

rewrote bluetooth classic and ble l2cap sockets. added test for l2cap socket

Browse Source
This commit is contained in:
lasan
2026-06-05 01:10:47 +05:30
parent 275f31edb9
commit 6de5147e35
9 changed files with 254 additions and 1089 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/platform/implementation/linux/BUILD
+1 -2
View File
@@ -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",
internal/platform/implementation/linux/ble_l2cap_server_socket.cc
+63 -228
View File
@@ -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<sockaddr*>(&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<sockaddr*>(&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<sockaddr*>(&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<api::ble::BleL2capSocket> 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<sockaddr*>(&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<sockaddr*>(&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<api::ble::BleL2capSocket> BleL2capServerSocket::Accept() {
api::ble::BlePeripheral::UniqueId peripheral_id = 0;
for (int i = 0; i < 6; ++i) {
peripheral_id =
(peripheral_id << 8) |
static_cast<uint8_t>(client_addr.l2_bdaddr.b[i]);
(peripheral_id << 8) | static_cast<uint8_t>(client_addr.l2_bdaddr.b[i]);
}
std::string service_id;
@@ -258,67 +135,25 @@ std::unique_ptr<api::ble::BleL2capSocket> BleL2capServerSocket::Accept() {
absl::MutexLock lock(&mutex_);
service_id = service_id_;
}
return std::make_unique<BleL2capSocket>(
client_fd, peripheral_id, service_id,
/*incoming_connection=*/true);
return std::make_unique<BleL2capSocket>(client_fd, peripheral_id, service_id );
}
Exception BleL2capServerSocket::Close() {
absl::AnyInvocable<void()> 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<void()> notifier) {
absl::AnyInvocable<void()> 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
internal/platform/implementation/linux/ble_l2cap_server_socket.h
-7
View File
@@ -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<void()> 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<void()> 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
internal/platform/implementation/linux/ble_l2cap_socket.cc
+57 -292
View File
@@ -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<ParsedSocketControlFrame> 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<uint32_t> ReadBigEndianUint32(absl::string_view bytes) {
if (bytes.size() != kHeaderLength) return std::nullopt;
const auto* ptr = reinterpret_cast<const uint8_t*>(bytes.data());
return (static_cast<uint32_t>(ptr[0]) << 24) |
(static_cast<uint32_t>(ptr[1]) << 16) |
(static_cast<uint32_t>(ptr[2]) << 8) |
static_cast<uint32_t>(ptr[3]);
}
std::string WriteBigEndianUint32(uint32_t value) {
std::string out(kHeaderLength, '\0');
out[0] = static_cast<char>((value >> 24) & 0xFF);
out[1] = static_cast<char>((value >> 16) & 0xFF);
out[2] = static_cast<char>((value >> 8) & 0xFF);
out[3] = static_cast<char>(value & 0xFF);
return out;
}
} // namespace
BleL2capInputStream::BleL2capInputStream(BleL2capSocket* owner) : owner_(owner) {}
BleL2capInputStream::~BleL2capInputStream() { Close(); }
ExceptionOr<ByteArray> BleL2capInputStream::Read(std::int64_t size) {
if (owner_ == nullptr) {
return ExceptionOr<ByteArray>(Exception::kIo);
std::vector<char> 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<BleL2capInputStream>(this)),
output_stream_(std::make_unique<BleL2capOutputStream>(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<sdbus::UnixFd>(fd)), input_stream_(std::make_unique<BleL2capInputStream>(fd_)),
output_stream_(std::make_unique<BleL2capOutputStream>(fd_)),
peripheral_id_(peripheral_id)
{}
BleL2capSocket::~BleL2capSocket() { Close(); }
bool BleL2capSocket::PollReady(short events, std::optional<absl::Duration> timeout) const {
int fd = fd_.load();
if (fd < 0) return false;
int timeout_ms = -1;
if (timeout.has_value()) {
timeout_ms = std::max<int64_t>(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<uint32_t>(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<size_t>(sent);
}
return true;
}
bool BleL2capSocket::ReadNextFrame(std::string& payload,
std::optional<absl::Duration> timeout) {
const std::optional<absl::Time> 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<absl::Duration> 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<size_t>(read_count));
}
}
ExceptionOr<ByteArray> BleL2capSocket::ReadFromSocket(std::int64_t size) {
if (size <= 0) return ExceptionOr<ByteArray>(ByteArray());
while (read_buffer_.empty()) {
std::string payload;
if (!ReadNextFrame(payload, std::nullopt))
return ExceptionOr<ByteArray>(Exception::kIo);
read_buffer_.append(payload);
}
const size_t read_size =
std::min<size_t>(static_cast<size_t>(size), read_buffer_.size());
ByteArray result(read_buffer_.substr(0, read_size));
read_buffer_.erase(0, read_size);
return ExceptionOr<ByteArray>(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<void()> notifier) {
absl::MutexLock lock(&mutex_);
close_notifier_ = std::move(notifier);
}
bool BleL2capSocket::IsClosed() const {
absl::MutexLock lock(&mutex_);
return closed_;
}
internal/platform/implementation/linux/ble_l2cap_socket.h
+13 -31
View File
@@ -15,6 +15,9 @@
#ifndef PLATFORM_IMPL_LINUX_BLE_L2CAP_SOCKET_H_
#define PLATFORM_IMPL_LINUX_BLE_L2CAP_SOCKET_H_
#include "dbus.h"
#include <atomic>
#include <memory>
#include <optional>
@@ -36,34 +39,35 @@ class BleL2capSocket;
class BleL2capInputStream final : public InputStream {
public:
explicit BleL2capInputStream(BleL2capSocket* owner);
explicit BleL2capInputStream(std::shared_ptr<sdbus::UnixFd> fd_raw_): fd_raw_(std::move(fd_raw_)) {};
~BleL2capInputStream() override;
ExceptionOr<ByteArray> Read(std::int64_t size) override;
Exception Close() override;
private:
BleL2capSocket* owner_ = nullptr;
private:
std::shared_ptr<sdbus::UnixFd> fd_raw_;
};
class BleL2capOutputStream final : public OutputStream {
public:
explicit BleL2capOutputStream(BleL2capSocket* owner);
public:
explicit BleL2capOutputStream(std::shared_ptr<sdbus::UnixFd> 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<sdbus::UnixFd> 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<ByteArray> 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<absl::Duration> timeout)
ABSL_LOCKS_EXCLUDED(io_mutex_);
bool SendFrame(absl::string_view payload) ABSL_LOCKS_EXCLUDED(io_mutex_);
bool PollReady(short events, std::optional<absl::Duration> 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<sdbus::UnixFd > fd_ ;
std::unique_ptr<BleL2capInputStream> input_stream_;
std::unique_ptr<BleL2capOutputStream> output_stream_;
api::ble::BlePeripheral::UniqueId peripheral_id_;
absl::AnyInvocable<void()> close_notifier_ ABSL_GUARDED_BY(mutex_);
std::atomic<int> 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
internal/platform/implementation/linux/ble_l2cap_socket_test.cc
+69 -212
View File
@@ -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<BleL2capSocket>(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<size_t>(read_count);
}
return true;
int fds_[2]{-1, -1};
int socket_fd_{-1};
int peer_fd_{-1};
std::unique_ptr<BleL2capSocket> 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<std::string> ReceiveFrame(int fd) {
std::array<char, 4> header;
if (!ReadExact(fd, header.data(), header.size())) {
return std::nullopt;
}
const auto* bytes = reinterpret_cast<const uint8_t*>(header.data());
uint32_t payload_size = (static_cast<uint32_t>(bytes[0]) << 24) |
(static_cast<uint32_t>(bytes[1]) << 16) |
(static_cast<uint32_t>(bytes[2]) << 8) |
static_cast<uint32_t>(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<char, 4> header = {
static_cast<char>((payload.size() >> 24) & 0xFF),
static_cast<char>((payload.size() >> 16) & 0xFF),
static_cast<char>((payload.size() >> 8) & 0xFF),
static_cast<char>(payload.size() & 0xFF),
};
if (send(fd, header.data(), header.size(), 0) != static_cast<ssize_t>(header.size())) {
return false;
}
if (!payload.empty() &&
send(fd, payload.data(), payload.size(), 0) != static_cast<ssize_t>(payload.size())) {
return false;
}
return true;
ASSERT_EQ(
write(peer_fd_, message.data(), message.size()),
static_cast<ssize_t>(message.size())
);
InputStream& input = socket_->GetInputStream();
std::vector<char> 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<char> received(message.size());
ssize_t n = read(peer_fd_, received.data(), received.size());
ASSERT_EQ(n, static_cast<ssize_t>(message.size()));
EXPECT_EQ(
std::string(received.begin(), received.end()),
message
);
}
std::string BuildLegacyControlPacket(uint8_t command) {
return std::string(1, static_cast<char>(command));
}
std::optional<uint8_t> ParseLegacyControlCommand(absl::string_view payload) {
if (payload.empty()) return std::nullopt;
if (payload.size() != 1 && payload.size() < 3) return std::nullopt;
return static_cast<uint8_t>(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<ByteArray> 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<ByteArray> 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
internal/platform/implementation/linux/ble_v2_medium.cc
+1 -5
View File
@@ -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<api::ble::BleL2capSocket> BleV2Medium::ConnectOverL2cap(
LOG(INFO) << __func__ << ": Successfully connected to L2CAP socket";
auto socket = std::make_unique<BleL2capSocket>(
fd, peripheral_id, service_id,
/*incoming_connection=*/false);
fd, peripheral_id, service_id);
return socket;
}
internal/platform/implementation/linux/bluetooth_classic_socket.cc
+41 -266
View File
@@ -31,300 +31,75 @@
#include <cstddef>
#include <algorithm>
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<ByteArray> 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<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;
}
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<size_t>(msg_len);
size_t cap = static_cast<size_t>(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<size_t>(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_t>(size));
char* data = buffer.data();
size_t total_read = 0;
while (total_read < static_cast<size_t>(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_t>(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<size_t>(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<size_t>(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<size_t>(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};
}
internal/platform/implementation/linux/bluetooth_classic_socket.h
+9 -46
View File
@@ -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<sdbus::UnixFd> fd)
: fd_raw_(std::move(fd)){}
ExceptionOr<ByteArray> Read(std::int64_t size) override;
Exception Close() override;
private:
sdbus::UnixFd fd_;
std::atomic<int> fd_raw_{-1};
std::shared_ptr<sdbus::UnixFd> 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<sdbus::UnixFd> 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<int> 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<sdbus::UnixFd> fd_raw_;
};
class BluetoothSocket final : public api::BluetoothSocket {
public:
BluetoothSocket(std::shared_ptr<BluetoothDevice> device,
const sdbus::UnixFd &fd)
: device_(std::move(device)), output_stream_(fd), input_stream_(fd) {}
sdbus::UnixFd fd)
:fd_(std::make_shared<sdbus::UnixFd>(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<sdbus::UnixFd> fd_;
std::shared_ptr<BluetoothDevice> device_;
BluetoothOutputStream output_stream_;
BluetoothInputStream input_stream_;