diff --git a/internal/platform/implementation/apple/ble_l2cap_server_socket.h b/internal/platform/implementation/apple/ble_l2cap_server_socket.h index 5a6c3453..87a148ba 100644 --- a/internal/platform/implementation/apple/ble_l2cap_server_socket.h +++ b/internal/platform/implementation/apple/ble_l2cap_server_socket.h @@ -12,18 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Note: File language is detected using heuristics. Many Objective-C++ headers -// are incorrectly classified as C++ resulting in invalid linter errors. The use -// of "NSArray" and other Foundation classes like "NSData", "NSDictionary" and -// "NSUUID" are highly weighted for Objective-C and Objective-C++ scores. Oddly, -// "#import " does not contribute any points. This -// comment alone should be enough to trick the IDE in to believing this is -// actually some sort of Objective-C file. See: -// cs/google3/devtools/search/lang/recognize_language_classifiers_data - #include -#import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPServer.h" +#include "absl/functional/any_invocable.h" +#include "internal/platform/exception.h" +#include "internal/platform/implementation/ble_v2.h" #import "internal/platform/implementation/apple/ble_l2cap_socket.h" namespace nearby { @@ -33,14 +26,16 @@ namespace apple { class BleL2capServerSocket : public api::ble_v2::BleL2capServerSocket { public: BleL2capServerSocket() = default; - ~BleL2capServerSocket() override = default; + ~BleL2capServerSocket() override; // Gets PSM value has been published by the server. int GetPSM() const override; - + // Sets PSM value has been published by the server. - void SetPSM(int PSM); + void SetPSM(int psm); + // Wait for an available socket. + // // Blocks until either: // - at least one incoming connection request is available, or // - ServerSocket is closed. @@ -48,17 +43,25 @@ class BleL2capServerSocket : public api::ble_v2::BleL2capServerSocket { // Returns nullptr on error. // Once error is reported, it is permanent, and L2CAP ServerSocket has to be // closed. - std::unique_ptr Accept() override; + std::unique_ptr Accept() override + ABSL_LOCKS_EXCLUDED(mutex_); // Closes the L2CAP server socket. Exception Close() override; - // Connects to the L2CAP server socket. - bool Connect(std::unique_ptr socket); + // Adds a pending socket to the server socket. + bool AddPendingSocket(std::unique_ptr socket); private: - // The PSM value of the L2CAP server socket. - int PSM_ = 0; + Exception DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + mutable absl::Mutex mutex_; + absl::CondVar cond_; + absl::flat_hash_set> pending_sockets_ + ABSL_GUARDED_BY(mutex_); + absl::AnyInvocable close_notifier_ ABSL_GUARDED_BY(mutex_); + bool closed_ ABSL_GUARDED_BY(mutex_) = false; + int psm_ = 0; }; } // namespace apple diff --git a/internal/platform/implementation/apple/ble_l2cap_server_socket.mm b/internal/platform/implementation/apple/ble_l2cap_server_socket.mm index 30d640b1..c03fca06 100644 --- a/internal/platform/implementation/apple/ble_l2cap_server_socket.mm +++ b/internal/platform/implementation/apple/ble_l2cap_server_socket.mm @@ -20,28 +20,62 @@ #include #import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPServer.h" -#import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPStream.h" -#import "internal/platform/implementation/apple/ble_l2cap_socket.h" #import "GoogleToolboxForMac/GTMLogger.h" namespace nearby { namespace apple { -int BleL2capServerSocket::GetPSM() const { return PSM_; } - -void BleL2capServerSocket::SetPSM(int PSM) { PSM_ = PSM; } - -std::unique_ptr BleL2capServerSocket::Accept() { - // TODO: b/399815436 - Implement to accept incoming l2cap connection. - return nullptr; +BleL2capServerSocket::~BleL2capServerSocket() { + absl::MutexLock lock(&mutex_); + DoClose(); } -bool BleL2capServerSocket::Connect(std::unique_ptr socket) { - // TODO: b/399815436 - Implement to connect to l2cap server socket. - return false; +int BleL2capServerSocket::GetPSM() const { return psm_; } + +void BleL2capServerSocket::SetPSM(int psm) { psm_ = psm; } + +// TODO: b/399815436 - Refactor Accept() and AddPendingSocket() for better readability. +std::unique_ptr BleL2capServerSocket::Accept() { + absl::MutexLock lock(&mutex_); + while (!closed_ && pending_sockets_.empty()) { + cond_.Wait(&mutex_); + } + if (closed_) return {}; + + std::unique_ptr remote_socket = + std::move(pending_sockets_.extract(pending_sockets_.begin()).value()); + return std::move(remote_socket); +} + +bool BleL2capServerSocket::AddPendingSocket(std::unique_ptr socket) { + absl::MutexLock lock(&mutex_); + if (closed_) { + return false; + } + pending_sockets_.insert(std::move(socket)); + cond_.SignalAll(); + return !closed_; } Exception BleL2capServerSocket::Close() { + absl::MutexLock lock(&mutex_); + return DoClose(); +} + +Exception BleL2capServerSocket::DoClose() { + bool should_notify = !closed_; + closed_ = true; + if (should_notify) { + cond_.SignalAll(); + if (close_notifier_) { + auto notifier = std::move(close_notifier_); + mutex_.Unlock(); + // Notifier may contain calls to public API, and may cause deadlock, if + // mutex_ is held during the call. + notifier(); + mutex_.Lock(); + } + } return {Exception::kSuccess}; } diff --git a/internal/platform/implementation/apple/ble_l2cap_socket.h b/internal/platform/implementation/apple/ble_l2cap_socket.h index b520e3e4..2f31821a 100644 --- a/internal/platform/implementation/apple/ble_l2cap_socket.h +++ b/internal/platform/implementation/apple/ble_l2cap_socket.h @@ -14,21 +14,21 @@ #include -#import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPStream.h" -#import "internal/platform/implementation/apple/ble_peripheral.h" #include "internal/platform/implementation/ble_v2.h" +#import "internal/platform/implementation/apple/ble_peripheral.h" + +@class GNCMConnectionHandlers; +@class GNCBLEL2CAPConnection; + namespace nearby { namespace apple { /** A readable stream of bytes. */ class BleL2capInputStream : public InputStream { public: - // Creates a BleL2capInputStream. - // - // @param stream The underlying stream to use for reading and writing. - explicit BleL2capInputStream(GNCBLEL2CAPStream* stream); - ~BleL2capInputStream() override = default; + explicit BleL2capInputStream(GNCBLEL2CAPConnection* connection); + ~BleL2capInputStream() override; // Reads at most `size` bytes from the input stream. // @@ -41,22 +41,24 @@ class BleL2capInputStream : public InputStream { Exception Close() override; private: - GNCBLEL2CAPStream* stream_; + GNCMConnectionHandlers *connectionHandlers_; + GNCBLEL2CAPConnection* connection_; + NSMutableArray *newDataPackets_; + NSMutableData *accumulatedData_; + NSCondition *condition_; }; /** A writable stream of bytes. */ class BleL2capOutputStream : public OutputStream { public: - // Creates a BleL2capOutputStream. - // - // @param stream The underlying stream to use for reading and writing. - explicit BleL2capOutputStream(GNCBLEL2CAPStream* stream); - ~BleL2capOutputStream() override = default; + explicit BleL2capOutputStream(GNCBLEL2CAPConnection* connection) + : connection_(connection), condition_([[NSCondition alloc] init]) {} + ~BleL2capOutputStream() override; // Write the provided bytes to the output stream. // // Returns Exception::kIo on error, otherwise Exception::kSuccess. - Exception Write(const ByteArray& data) override; + Exception Write(const ByteArray &data) override; // no-op // @@ -69,7 +71,8 @@ class BleL2capOutputStream : public OutputStream { Exception Close() override; private: - GNCBLEL2CAPStream* stream_; + GNCBLEL2CAPConnection* connection_; + NSCondition *condition_; }; /** @@ -77,46 +80,47 @@ class BleL2capOutputStream : public OutputStream { */ class BleL2capSocket : public api::ble_v2::BleL2capSocket { public: - explicit BleL2capSocket(GNCBLEL2CAPStream* stream); + explicit BleL2capSocket(GNCBLEL2CAPConnection* connection); - // The peripheral used to create the socket must outlive the socket or - // undefined behavior will occur. - BleL2capSocket(GNCBLEL2CAPStream* stream, - api::ble_v2::BlePeripheral* peripheral); - ~BleL2capSocket() override = default; + // The peripheral used to create the socket must outlive the socket or undefined behavior will + // occur. + BleL2capSocket(GNCBLEL2CAPConnection* connection, api::ble_v2::BlePeripheral *peripheral); + ~BleL2capSocket() override; // Returns the InputStream of the BleL2capSocket. // On error, returned stream will report Exception::kIo on any operation. // // The returned object is not owned by the caller, and can be invalidated once // the BleL2capSocket object is destroyed. - InputStream& GetInputStream() override; + BleL2capInputStream &GetInputStream() override { return *input_stream_; } // Returns the OutputStream of the BleL2capSocket. // On error, returned stream will report Exception::kIo on any operation. // // The returned object is not owned by the caller, and can be invalidated once // the BleL2capSocket object is destroyed. - OutputStream& GetOutputStream() override; + BleL2capOutputStream &GetOutputStream() override { return *output_stream_; } // Returns Exception::kIo on error, Exception::kSuccess otherwise. - Exception Close() override; - + Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_); // Sets the close notifier by client side. - void SetCloseNotifier(absl::AnyInvocable notifier) override; + void SetCloseNotifier(absl::AnyInvocable notifier) override {}; // Returns valid BlePeripheral pointer if there is a connection, and // nullptr otherwise. - api::ble_v2::BlePeripheral* GetRemotePeripheral() override { - return peripheral_; - } + api::ble_v2::BlePeripheral *GetRemotePeripheral() override { return peripheral_; } + + bool IsClosed() const ABSL_LOCKS_EXCLUDED(mutex_); private: - GNCBLEL2CAPStream* stream_; + void DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + mutable absl::Mutex mutex_; + bool closed_ ABSL_GUARDED_BY(mutex_) = false; std::unique_ptr input_stream_; std::unique_ptr output_stream_; - api::ble_v2::BlePeripheral* peripheral_; + api::ble_v2::BlePeripheral *peripheral_; }; } // namespace apple -} // namespace nearby \ No newline at end of file +} // namespace nearby diff --git a/internal/platform/implementation/apple/ble_l2cap_socket.mm b/internal/platform/implementation/apple/ble_l2cap_socket.mm index e09e5947..bce1548e 100644 --- a/internal/platform/implementation/apple/ble_l2cap_socket.mm +++ b/internal/platform/implementation/apple/ble_l2cap_socket.mm @@ -14,9 +14,9 @@ #import "internal/platform/implementation/apple/ble_l2cap_socket.h" -#import - -#import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPStream.h" +#include "internal/platform/implementation/ble_v2.h" +#import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPConnection.h" +#import "internal/platform/implementation/apple/utils.h" #import "GoogleToolboxForMac/GTMLogger.h" namespace nearby { @@ -24,55 +24,182 @@ namespace apple { #pragma mark - BleL2capInputStream -BleL2capInputStream::BleL2capInputStream(GNCBLEL2CAPStream* stream) : stream_(stream) { - GTMLoggerInfo(@"BleL2capInputStream::BleL2capInputStream"); +BleL2capInputStream::BleL2capInputStream(GNCBLEL2CAPConnection *connection) + : connection_(connection), + newDataPackets_([NSMutableArray array]), + accumulatedData_([NSMutableData data]), + condition_([[NSCondition alloc] init]) { + // Create the handlers of incoming data from the remote endpoint. + connection.connectionHandlers = [GNCMConnectionHandlers + payloadHandler:^(NSData *data) { + [condition_ lock]; + // Add the incoming data to the data packet array to be processed in read() below. + [newDataPackets_ addObject:data]; + [condition_ broadcast]; + [condition_ unlock]; + } + disconnectedHandler:^{ + [condition_ lock]; + // Release the data packet array, meaning the stream has been closed or severed. + newDataPackets_ = nil; + [condition_ broadcast]; + [condition_ unlock]; + }]; +} + +BleL2capInputStream::~BleL2capInputStream() { + NSCAssert(!newDataPackets_, @"BleInputStream not closed before destruction"); } ExceptionOr BleL2capInputStream::Read(std::int64_t size) { - // TODO: edwinwu - Implement to read data from l2cap channel. - return {Exception::kIo}; + // Block until either (a) the connection has been closed, (b) we have enough data to return. + NSData *dataToReturn; + [condition_ lock]; + while (true) { + // Check if the stream has been closed or severed. + if (!newDataPackets_) break; + + if (newDataPackets_.count > 0) { + // Add the packet data to the accumulated data. + for (NSData *data in newDataPackets_) { + if (data.length > 0) { + [accumulatedData_ appendData:data]; + } + } + [newDataPackets_ removeAllObjects]; + } + + if (accumulatedData_.length > 0) { + // Return up to |size| bytes of the data. + std::int64_t sizeToReturn = (accumulatedData_.length < size) ? accumulatedData_.length : size; + NSRange range = NSMakeRange(0, (NSUInteger)sizeToReturn); + dataToReturn = [accumulatedData_ subdataWithRange:range]; + [accumulatedData_ replaceBytesInRange:range withBytes:nil length:0]; + break; + } + + [condition_ wait]; + } + [condition_ unlock]; + + if (dataToReturn) { + GTMLoggerInfo(@"[NEARBY] BleL2capInputStream: Received data of size: %lu", + (unsigned long)dataToReturn.length); + return ExceptionOr{ByteArray((const char *)dataToReturn.bytes, dataToReturn.length)}; + } else { + return ExceptionOr{Exception::kIo}; + } } Exception BleL2capInputStream::Close() { - // The input stream reads directly from the connection. It can not be closed without closing the - // connection itself. A call to `BleL2capSocket::Close` will close the connection. + // Unblock pending read operation. + [condition_ lock]; + newDataPackets_ = nil; + [condition_ broadcast]; + [condition_ unlock]; return {Exception::kSuccess}; } #pragma mark - BleL2capOutputStream -BleL2capOutputStream::BleL2capOutputStream(GNCBLEL2CAPStream* stream) : stream_(stream) {} - -Exception BleL2capOutputStream::Write(const ByteArray& data) { - // TODO: edwinwu - Implement to write data to l2cap channel. - return {Exception::kIo}; +BleL2capOutputStream::~BleL2capOutputStream() { + NSCAssert(!connection_, @"BleL2capOutputStream not closed before destruction"); } -Exception BleL2capOutputStream::Flush() { return {Exception::kSuccess}; } +Exception BleL2capOutputStream::Write(const ByteArray &data) { + [condition_ lock]; + GTMLoggerInfo(@"[NEARBY] BleL2capOutputStream: Sending data of size: %lu", + NSDataFromByteArray(data).length); -Exception BleL2capOutputStream::Close() { return {Exception::kSuccess}; } + if (!connection_) { + [condition_ unlock]; + return {Exception::kIo}; + } -#pragma mark - BleL2capSocket + NSMutableData *packet = [NSMutableData dataWithBytes:data.data() length:data.size()]; -BleL2capSocket::BleL2capSocket(GNCBLEL2CAPStream* stream) - : BleL2capSocket(stream, new EmptyBlePeripheral()) {} + // Send the data, blocking until the completion handler is called. + __block BOOL isComplete = NO; + __block BOOL sendResult = NO; + NSCondition *condition = condition_; // don't capture |this| in completion -BleL2capSocket::BleL2capSocket(GNCBLEL2CAPStream* stream, api::ble_v2::BlePeripheral* peripheral) - : stream_(stream), - input_stream_(std::make_unique(stream)), - output_stream_(std::make_unique(stream)), - peripheral_(peripheral) {} + [connection_ sendData:packet + completion:^(BOOL result) { + [condition lock]; + if (isComplete) { + [condition unlock]; + return; + } + isComplete = YES; + sendResult = result; + [condition broadcast]; + [condition unlock]; + }]; -InputStream& BleL2capSocket::GetInputStream() { return *input_stream_; } + while (connection_ && !isComplete) { + [condition_ wait]; + } -OutputStream& BleL2capSocket::GetOutputStream() { return *output_stream_; } + if (sendResult == YES) { + [condition_ unlock]; + return {Exception::kSuccess}; + } else { + [condition_ unlock]; + return {Exception::kIo}; + } +} -Exception BleL2capSocket::Close() { - [stream_ close]; +Exception BleL2capOutputStream::Flush() { + // The write() function blocks until the data is received by the remote endpoint, so there's + // nothing to do here. return {Exception::kSuccess}; } -void BleL2capSocket::SetCloseNotifier(absl::AnyInvocable notifier) {} +Exception BleL2capOutputStream::Close() { + GTMLoggerInfo(@"[NEARBY] edwin : BleL2capOutputStream Closing"); + // Unblock pending write operation. + [condition_ lock]; + connection_ = nil; + [condition_ broadcast]; + [condition_ unlock]; + return {Exception::kSuccess}; +} + +#pragma mark - BleL2capSocket + +BleL2capSocket::BleL2capSocket(GNCBLEL2CAPConnection *connection) + : BleL2capSocket(connection, new EmptyBlePeripheral()) {} + +BleL2capSocket::BleL2capSocket(GNCBLEL2CAPConnection *connection, + api::ble_v2::BlePeripheral *peripheral) + : input_stream_(std::make_unique(connection)), + output_stream_(std::make_unique(connection)), + peripheral_(peripheral) {} + +BleL2capSocket::~BleL2capSocket() { + absl::MutexLock lock(&mutex_); + DoClose(); +} + +bool BleL2capSocket::IsClosed() const { + absl::MutexLock lock(&mutex_); + return closed_; +} + +Exception BleL2capSocket::Close() { + absl::MutexLock lock(&mutex_); + DoClose(); + return {Exception::kSuccess}; +} + +void BleL2capSocket::DoClose() { + GTMLoggerInfo(@"[NEARBY] edwin : BleL2capSocket DoClose"); + if (!closed_) { + input_stream_->Close(); + output_stream_->Close(); + closed_ = true; + } +} } // namespace apple } // namespace nearby diff --git a/internal/platform/implementation/apple/ble_medium.mm b/internal/platform/implementation/apple/ble_medium.mm index 9a851656..0baf2963 100644 --- a/internal/platform/implementation/apple/ble_medium.mm +++ b/internal/platform/implementation/apple/ble_medium.mm @@ -406,7 +406,15 @@ std::unique_ptr BleMedium::OpenL2capServerSoc GTMLoggerError(@"Error opening L2CAP channel in L2CAP server: %@", error); return; } - // TODO: b/399815436 - Implement to create socket when stream is ready. + GNCBLEL2CAPConnection *connection = + [GNCBLEL2CAPConnection connectionWithStream:stream + serviceID:@(service_id_str.c_str()) + incomingConnection:YES + callbackQueue:dispatch_get_main_queue()]; + auto socket = std::make_unique(connection); + if (l2cap_server_socket_ptr) { + l2cap_server_socket_ptr->AddPendingSocket(std::move(socket)); + } } peripheralManager:nil]; dispatch_semaphore_wait(semaphore, DISPATCH_TIME_FOREVER);