Implement L2CAP stream wrapper.

PiperOrigin-RevId: 741161066

internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPStream.h

@@ -0,0 +1,60 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#import <Foundation/Foundation.h>
+
+NS_ASSUME_NONNULL_BEGIN
+
+@class GNCBLEL2CAPStream;
+
+/// Block invoked when the stream is closed.
+typedef void (^GNCBLEL2CAPStreamClosedBlock)(void);
+
+/// Block invoked when |data| is received from the remote device on the L2CAP connection.
+typedef void (^GNCBLEL2CAPControllerReceivedDataBlock)(NSData *data);
+
+/**
+ * Abstraction to take in two streams returned from L2CAP controller and provide a simplified
+ * interface to send and receive data from the streams.
+ * This class is thread-safe.
+ */
+@interface GNCBLEL2CAPStream : NSObject
+
+/// Returns an instance of @c GNCBLEL2CAPStream which sends and receives data on |inputStream|
+/// and |outputStream|.
+/// Invokes |closedBlock| if stream closed signal is received when reading data.
+/// The stream should be torn down when this block is called.
+/// Invokes |receivedDataBlock| with data received |inputStream|.
+/// The blocks are invoked on an arbitrary queue with DISPATCH_QUEUE_PRIORITY_HIGH.
+- (instancetype)initWithClosedBlock:(GNCBLEL2CAPStreamClosedBlock)closedBlock
+                  receivedDataBlock:(GNCBLEL2CAPControllerReceivedDataBlock)receivedDataBlock
+                        inputStream:(NSInputStream *)inputStream
+                       outputStream:(NSOutputStream *)outputStream NS_DESIGNATED_INITIALIZER;
+
+- (instancetype)init NS_UNAVAILABLE;
+
+/// Queues up |data| to be sent to the device. Asserts if |data| is empty.
+///
+/// |completionBlock| is invoked on an arbitrary queue after sending finishes. If the parameter is
+/// YES, data is completely written to the local L2CAP socket. Note that this does not necessarily
+/// mean the device received it. NO indicates writing failed, which typically happens when the
+/// stream is torn down.
+- (void)sendData:(NSData *)data completionBlock:(void (^)(BOOL))completionBlock;
+
+/// Tears down the stream.
+- (void)tearDown;
+
+@end
+
+NS_ASSUME_NONNULL_END

internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPStream.m

@@ -0,0 +1,349 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPStream.h"
+
+#import "GoogleToolboxForMac/GTMLogger.h"
+
+#define READ_BUFFER_SIZE 409600
+
+/** A pending packet that will be written to the L2CAP socket. */
+@interface GNCBLEL2CAPStreamWriteOperation : NSObject
+
+/// Initializes a write with given data and completion block.
+- (instancetype)initWithData:(NSData *)data
+             completionBlock:(void (^)(BOOL))completionBlock NS_DESIGNATED_INITIALIZER;
+- (instancetype)init NS_UNAVAILABLE;
+
+/// The remaining data that should be written to the L2CAP socket.
+@property(nonatomic, readonly) NSData *remainingData;
+
+/// Invoked when this packet is completely written.
+@property(nonatomic, readonly) void (^completionBlock)(BOOL);
+
+/// Removes given number of bytes from the beginning of the remaining data.
+- (void)consumeBytes:(NSUInteger)consumedByteCount;
+
+@end
+
+@interface GNCBLEL2CAPStream () <NSStreamDelegate>
+
+/// Input stream from the watch. Operations to this stream are synchronized by |_streamQueue|
+/// dispatch queue.
+@property(nonatomic, nullable) NSInputStream *inputStream;
+
+/// Output stream to the watch. Operations to this stream are synchronized by synchronized access on
+/// |_writeBufferArray|.
+@property(nonatomic, nullable) NSOutputStream *outputStream;
+
+@end
+
+@implementation GNCBLEL2CAPStream {
+  GNCBLEL2CAPStreamClosedBlock _closedBlock;
+  GNCBLEL2CAPControllerReceivedDataBlock _receivedDataBlock;
+
+  /// Serial queue used when invoking |_receivedDataBlock|.
+  dispatch_queue_t _receivedDataQueue;
+
+  /// Queue used exclusively from events on |toWatchStream| and |fromWatchStream|.
+  dispatch_queue_t _streamQueue;
+
+  /// Pending data to be written to the remote device, synchronized access on itself.
+  /// @synchronized used since 3x speedup in benchmark over dispatch_async.
+  NSMutableArray<GNCBLEL2CAPStreamWriteOperation *> *_writeBufferArray;
+
+  /// NSOutputStream has notified that space is available to write data.
+  /// synchronized access on |_writeBufferArray|.
+  BOOL _writeBufferReadyForData;
+
+  /// Verbose logging for some statements which are only useful when debugging but produce far too
+  /// much log-spam to enable on Dev.
+  BOOL _verboseLoggingEnabled;
+}
+
+#pragma mark Public
+
+- (instancetype)initWithClosedBlock:(GNCBLEL2CAPStreamClosedBlock)closedBlock
+                  receivedDataBlock:(GNCBLEL2CAPControllerReceivedDataBlock)receivedDataBlock
+                        inputStream:(NSInputStream *)inputStream
+                       outputStream:(NSOutputStream *)outputStream {
+  self = [super init];
+
+  if (self) {
+    _streamQueue = dispatch_queue_create("com.google.nearby.GNCBLEL2CAPStream",
+                                         dispatch_queue_attr_make_with_qos_class(
+                                             DISPATCH_QUEUE_SERIAL, QOS_CLASS_USER_INITIATED, -1));
+    _receivedDataQueue =
+        dispatch_queue_create("com.google.nearby.GNCBLEL2CAPStream.receivedData",
+                              dispatch_queue_attr_make_with_qos_class(
+                                  DISPATCH_QUEUE_SERIAL, QOS_CLASS_USER_INITIATED, -1));
+
+    _closedBlock = closedBlock;
+    _receivedDataBlock = receivedDataBlock;
+
+    _writeBufferArray = [NSMutableArray array];
+
+    [self configureStreamsWithInputStream:inputStream outputStream:outputStream];
+  }
+  return self;
+}
+
+- (void)tearDown {
+  dispatch_async(_streamQueue, ^{
+    GTMLoggerDebug(@"[NEARBY] Closing inputStream %@ by tearDown", self.inputStream);
+    [self.inputStream close];
+    self.inputStream.delegate = nil;
+    self.inputStream = nil;
+
+    @synchronized(self->_writeBufferArray) {
+      GTMLoggerDebug(@"[NEARBY] Closing outputStream %@ by tearDown", self.outputStream);
+      [self.outputStream close];
+      self.outputStream.delegate = nil;
+      self.outputStream = nil;
+
+      for (GNCBLEL2CAPStreamWriteOperation *pendingWrite in [self->_writeBufferArray copy]) {
+        pendingWrite.completionBlock(NO);
+      }
+
+      [self->_writeBufferArray removeAllObjects];
+      self->_writeBufferReadyForData = NO;
+    }
+  });
+}
+
+- (void)dealloc {
+  NSStream *inputStream = _inputStream;
+  NSStream *outputStream = _outputStream;
+  dispatch_async(_streamQueue, ^{
+    GTMLoggerDebug(@"[NEARBY] Closing streams inputStream %@ outputStream %@ by deallocation",
+                   inputStream, outputStream);
+    [inputStream close];
+    inputStream.delegate = nil;
+
+    [outputStream close];
+    outputStream.delegate = nil;
+  });
+}
+
+- (void)sendData:(NSData *)data completionBlock:(void (^)(BOOL))completionBlock {
+  if (!data || data.length == 0) {
+    GTMLoggerError(@"[NEARBY] Sending data cannot be nil or empty");
+  }
+
+  GNCBLEL2CAPStreamWriteOperation *write =
+      [[GNCBLEL2CAPStreamWriteOperation alloc] initWithData:data completionBlock:completionBlock];
+  @synchronized(_writeBufferArray) {
+    [_writeBufferArray addObject:write];
+
+    if (_writeBufferReadyForData) {
+      _writeBufferReadyForData = NO;
+      [self sendWriteBufferData];
+    }
+  }
+}
+
+#pragma mark NSStreamDelegate
+
+- (void)stream:(NSStream *)stream handleEvent:(NSStreamEvent)eventCode {
+  if (_verboseLoggingEnabled) {
+    GTMLoggerDebug(@"[NEARBY] Stream event %@ for stream %@",
+                   [[self class] stringFromStreamEventCode:eventCode], stream);
+  }
+  if ([stream isEqual:self.inputStream]) {
+    switch (eventCode) {
+      case NSStreamEventHasBytesAvailable:
+        [self receiveStreamData];
+        break;
+      case NSStreamEventErrorOccurred:
+      case NSStreamEventEndEncountered:
+        [self tearDown];
+        break;
+      case NSStreamEventOpenCompleted:
+      case NSStreamEventHasSpaceAvailable:
+      case NSStreamEventNone:
+      default:
+        break;
+    }
+    return;
+  }
+
+  if ([stream isEqual:self.outputStream]) {
+    switch (eventCode) {
+      case NSStreamEventHasSpaceAvailable: {
+        @synchronized(_writeBufferArray) {
+          if (!_writeBufferArray.count) {
+            GTMLoggerInfo(@"[NEARBY] No data to write in buffer, setting flag");
+            _writeBufferReadyForData = YES;
+            return;
+          }
+          [self sendWriteBufferData];
+        }
+        break;
+      }
+      case NSStreamEventHasBytesAvailable:
+      case NSStreamEventOpenCompleted:
+      case NSStreamEventErrorOccurred:
+      case NSStreamEventEndEncountered:
+      case NSStreamEventNone:
+      default:
+        break;
+    }
+    return;
+  }
+}
+
+#pragma mark Private
+
+/// Returns a string representation of @c NSStreamEvent.
++ (NSString *)stringFromStreamEventCode:(NSStreamEvent)eventCode {
+  switch (eventCode) {
+    case NSStreamEventOpenCompleted:
+      return @"NSStreamEventOpenCompleted";
+    case NSStreamEventHasSpaceAvailable:
+      return @"NSStreamEventHasSpaceAvailable";
+    case NSStreamEventHasBytesAvailable:
+      return @"NSStreamEventHasBytesAvailable";
+    case NSStreamEventErrorOccurred:
+      return @"NSStreamEventErrorOccurred";
+    case NSStreamEventEndEncountered:
+      return @"NSStreamEventEndEncountered";
+    case NSStreamEventNone:
+      return @"NSStreamEventNone";
+    default:
+      return [NSString stringWithFormat:@"Unknown NSStreamEvent %@", @(eventCode)];
+  }
+}
+
+/// Sets up |inputStream| and |outputStream| on |_streamQueue|.
+- (void)configureStreamsWithInputStream:(NSInputStream *)inputStream
+                           outputStream:(NSOutputStream *)outputStream {
+  self.inputStream = inputStream;
+  self.outputStream = outputStream;
+
+  if (inputStream.delegate) {
+    GTMLoggerError(@"[NEARBY] Should not have a delegate.");
+    return;
+  }
+  if (outputStream.delegate) {
+    GTMLoggerError(@"[NEARBY] Should not have a delegate.");
+    return;
+  }
+
+  inputStream.delegate = self;
+  outputStream.delegate = self;
+
+  GTMLoggerDebug(@"[NEARBY] streams inputStream %@ outputStream %@", inputStream, outputStream);
+
+  if (!_streamQueue) {
+    GTMLoggerError(@"[NEARBY] Stream queue must be initialized.");
+    return;
+  }
+
+  CFReadStreamSetDispatchQueue((__bridge CFReadStreamRef)inputStream, _streamQueue);
+  CFWriteStreamSetDispatchQueue((__bridge CFWriteStreamRef)outputStream, _streamQueue);
+  // Need to open streams on |_streamQueue|.
+  dispatch_async(_streamQueue, ^{
+    [inputStream open];
+    [outputStream open];
+  });
+}
+
+/// Sends data in the write buffer to the remote device.
+/// Must be called after synchronizing on |_writeBufferArray|.
+- (void)sendWriteBufferData {
+  if (!_writeBufferArray.count) {
+    GTMLoggerError(@"[NEARBY] sendWriteBufferData should not be called with empty buffer");
+    return;
+  }
+
+  NSUInteger totalBytesToWrite = _writeBufferArray.firstObject.remainingData.length;
+  NSInteger result =
+      [self.outputStream write:(const uint8_t *)_writeBufferArray.firstObject.remainingData.bytes
+                     maxLength:totalBytesToWrite];
+  if (result < 0) {
+    GTMLoggerError(@"[NEARBY] Stream write error %@", self.outputStream.streamError);
+    return;
+  }
+
+  if (result < 0) {
+    GTMLoggerError(@"[NEARBY] Write result should not be negative.");
+    return;
+  }
+  NSUInteger totalBytesWritten = (NSUInteger)result;
+
+  if (_verboseLoggingEnabled) {
+    GTMLoggerInfo(@"[NEARBY] Wrote %@/%@ bytes to stream", @(totalBytesWritten),
+                  @(totalBytesToWrite));
+  }
+
+  if (totalBytesWritten == totalBytesToWrite) {
+    GNCBLEL2CAPStreamWriteOperation *finishedWrite = _writeBufferArray.firstObject;
+    [_writeBufferArray removeObjectAtIndex:0];
+    finishedWrite.completionBlock(YES);
+  } else {
+    [_writeBufferArray.firstObject consumeBytes:totalBytesWritten];
+    if (_writeBufferArray.firstObject.remainingData.length == 0) {
+      GTMLoggerError(@"[NEARBY] Remaining data cannot be empty.");
+      return;
+    }
+  }
+}
+
+/// Receives data from watch and invokes |_receivedDataBlock|.
+- (void)receiveStreamData {
+  dispatch_assert_queue_debug(_streamQueue);
+
+  uint8_t readBuffer[READ_BUFFER_SIZE];
+  NSInteger bytesRead = [self.inputStream read:readBuffer maxLength:READ_BUFFER_SIZE];
+
+  if (bytesRead > 0) {
+    NSMutableData *data = [NSMutableData data];
+    [data appendBytes:readBuffer length:(NSUInteger)bytesRead];
+
+    if (_verboseLoggingEnabled) {
+      GTMLoggerDebug(@"[NEARBY] Stream data from watch of length %@", @(data.length));
+    }
+
+    dispatch_async(_receivedDataQueue, ^{
+      self->_receivedDataBlock(data);
+    });
+  } else if (bytesRead < 0) {
+    GTMLoggerError(@"[NEARBY] Stream read error: %@", self.inputStream.streamError);
+  } else if (bytesRead == 0) {
+    GTMLoggerDebug(@"[NEARBY] End of stream reached. Disconnecting");
+    // This indicates the L2CAP socket is closed. Notifying the owner so that it can tear down this
+    // stream and update its own state.
+    _closedBlock();
+  }
+}
+
+@end
+
+@implementation GNCBLEL2CAPStreamWriteOperation
+
+- (instancetype)initWithData:(NSData *)data completionBlock:(void (^)(BOOL))completionBlock {
+  if (self = [super init]) {
+    // Create copy of data to prevent client modification.
+    _remainingData = [data copy];
+    _completionBlock = completionBlock;
+  }
+  return self;
+}
+
+- (void)consumeBytes:(NSUInteger)consumedByteCount {
+  _remainingData = [_remainingData
+      subdataWithRange:NSMakeRange(consumedByteCount, _remainingData.length - consumedByteCount)];
+}
+
+@end

internal/platform/implementation/apple/Mediums/BUILD

@@ -26,6 +26,7 @@ objc_library(
         "BLEv2/GNCBLEGATTClient.m",
         "BLEv2/GNCBLEGATTServer.m",
         "BLEv2/GNCBLEL2CAPServer.m",
+        "BLEv2/GNCBLEL2CAPStream.m",
         "BLEv2/GNCBLEMedium.m",
         "BLEv2/GNCCentralManager.m",
         "BLEv2/GNCPeripheral.m",
@@ -52,6 +53,7 @@ objc_library(
         "BLEv2/GNCBLEGATTClient.h",
         "BLEv2/GNCBLEGATTServer.h",
         "BLEv2/GNCBLEL2CAPServer.h",
+        "BLEv2/GNCBLEL2CAPStream.h",
         "BLEv2/GNCBLEMedium.h",
         "BLEv2/GNCCentralManager.h",
         "BLEv2/GNCPeripheral.h",

internal/platform/implementation/apple/Tests/BUILD

@@ -28,8 +28,11 @@ objc_library(
         "GNCBLEGATTClientTest.m",
         "GNCBLEGATTServer+Testing.h",
         "GNCBLEGATTServerTest.m",
+        "GNCBLEL2CAPFakeInputOutputStream.h",
+        "GNCBLEL2CAPFakeInputOutputStream.m",
         "GNCBLEL2CAPServer+Testing.h",
         "GNCBLEL2CAPServerTest.m",
+        "GNCBLEL2CAPStreamTest.m",
         "GNCBLEMedium+Testing.h",
         "GNCBLEMediumTest.m",
         "GNCBLEUtilsTest.mm",

internal/platform/implementation/apple/Tests/GNCBLEL2CAPFakeInputOutputStream.h

@@ -0,0 +1,41 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#import <Foundation/Foundation.h>
+
+/**
+ * Test helper to create fake @c NSInputStream and @c NSOutputStream objects to simulate BLE
+ * communication with a watch via L2CAP channels during tests.
+ */
+@interface GNCBLEL2CAPFakeInputOutputStream : NSObject
+
+@property(readonly) NSInputStream *inputStream;
+
+@property(readonly) NSOutputStream *outputStream;
+
+/// Returns an instance of streams initialized with the given |bufferSize|.
+- (instancetype)initWithBufferSize:(NSUInteger)bufferSize NS_DESIGNATED_INITIALIZER;
+
+- (instancetype)init NS_UNAVAILABLE;
+
+/// Should be called in [XCTestCase tearDown] to tear down the streams.
+- (void)tearDown;
+
+/// Returns the data "sent" to the watch with |maxBytes|.
+- (NSData *)dataSentToWatchWithMaxBytes:(NSUInteger)maxBytes;
+
+/// Simulates the device writing |data| to |inputStream|.
+- (void)writeFromDevice:(NSData *)data;
+
+@end

internal/platform/implementation/apple/Tests/GNCBLEL2CAPFakeInputOutputStream.m

@@ -0,0 +1,101 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#import "internal/platform/implementation/apple/Tests/GNCBLEL2CAPFakeInputOutputStream.h"
+
+@implementation GNCBLEL2CAPFakeInputOutputStream {
+  /// Output stream for |_inputStream| to "simulate" data from watch during tests.
+  NSOutputStream* _outputTestStream;
+
+  /// Input stream for |_outputStream| to "simulate" data to watch during tests.
+  NSInputStream* _inputTestStream;
+}
+
+#pragma mark Public
+
+- (instancetype)initWithBufferSize:(NSUInteger)bufferSize {
+  self = [super init];
+  if (self) {
+    [self setupStreamsWithBufferSize:bufferSize];
+  }
+  return self;
+}
+
+- (void)tearDown {
+  [_inputTestStream close];
+  [_outputTestStream close];
+
+  _inputStream = nil;
+  _outputStream = nil;
+  _inputTestStream = nil;
+  _outputTestStream = nil;
+}
+
+- (void)writeFromDevice:(NSData*)data {
+  [_outputTestStream write:data.bytes maxLength:data.length];
+}
+
+- (NSData*)dataSentToWatchWithMaxBytes:(NSUInteger)maxBytes {
+  NSMutableData* readData = [NSMutableData data];
+  uint8_t buf[409600];
+  NSUInteger totalBytesRead = 0u;
+
+  while (totalBytesRead < maxBytes) {
+    NSInteger bytesRead = [_inputTestStream read:buf maxLength:409600];
+    [readData appendBytes:buf length:(NSUInteger)bytesRead];
+    totalBytesRead += (NSUInteger)bytesRead;
+  }
+  return readData;
+}
+
+#pragma mark Private
+
+/// Opens a connection to |stream| on the main run loop.
++ (void)openStream:(NSStream*)stream {
+  [stream scheduleInRunLoop:[NSRunLoop mainRunLoop] forMode:NSDefaultRunLoopMode];
+  [stream open];
+}
+
+/// Sets up the stream instance variables with the given |bufferSize|.
+- (void)setupStreamsWithBufferSize:(NSUInteger)bufferSize {
+  {
+    // Local variables created to avoid error in passing pointer of non-local object.
+    NSInputStream* inputStream;
+    NSOutputStream* outputStream;
+
+    [NSStream getBoundStreamsWithBufferSize:bufferSize
+                                inputStream:&inputStream
+                               outputStream:&outputStream];
+
+    _inputStream = inputStream;
+    _outputTestStream = outputStream;
+  }
+
+  {
+    // Local variables created to avoid error in passing pointer of non-local object.
+    NSInputStream* inputStream;
+    NSOutputStream* outputStream;
+    [NSStream getBoundStreamsWithBufferSize:bufferSize
+                                inputStream:&inputStream
+                               outputStream:&outputStream];
+
+    _inputTestStream = inputStream;
+    _outputStream = outputStream;
+  }
+
+  [[self class] openStream:_inputTestStream];
+  [[self class] openStream:_outputTestStream];
+}
+
+@end

internal/platform/implementation/apple/Tests/GNCBLEL2CAPStreamTest.m

@@ -0,0 +1,324 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#import "internal/platform/implementation/apple/Mediums/BLEv2/GNCBLEL2CAPStream.h"
+
+#import <XCTest/XCTest.h>
+
+#import "internal/platform/implementation/apple/Tests/GNCBLEL2CAPFakeInputOutputStream.h"
+
+@interface GNCBLEL2CAPStreamTest : XCTestCase
+@end
+
+@implementation GNCBLEL2CAPStreamTest {
+  GNCBLEL2CAPFakeInputOutputStream* _fakeInputOutputStream;
+  GNCBLEL2CAPStream* _stream;
+}
+
+- (void)tearDown {
+  [_stream tearDown];
+  [_fakeInputOutputStream tearDown];
+
+  _stream = nil;
+  [super tearDown];
+}
+
+#pragma mark Tests
+
+/// Tests that received data block called twice with two messages (in order) with size of the stream
+/// buffer.
+- (void)testReceivedDataBlockForTwoBufferSizeMessagesFromWatch {
+  // GIVEN
+  NSData* dummyData1 = [@"dummyData1" dataUsingEncoding:NSASCIIStringEncoding];
+  NSData* dummyData2 = [@"dummyData2" dataUsingEncoding:NSASCIIStringEncoding];
+  NSMutableData* expectedData = [NSMutableData dataWithData:dummyData1];
+  [expectedData appendData:dummyData2];
+
+  _fakeInputOutputStream =
+      [[GNCBLEL2CAPFakeInputOutputStream alloc] initWithBufferSize:dummyData1.length];
+
+  XCTestExpectation* expectation = [self expectationWithDescription:@"Received data from watch."];
+  expectation.expectedFulfillmentCount = 2;
+
+  NSMutableData* receivedData = [NSMutableData data];
+  _stream = [[GNCBLEL2CAPStream alloc]
+      initWithClosedBlock:^{
+        XCTFail(@"Should not be invoked.");
+      }
+      receivedDataBlock:^(NSData* data) {
+        [receivedData appendData:data];
+        [expectation fulfill];
+      }
+      inputStream:_fakeInputOutputStream.inputStream
+      outputStream:_fakeInputOutputStream.outputStream];
+
+  // WHEN
+  [_fakeInputOutputStream writeFromDevice:dummyData1];
+  [_fakeInputOutputStream writeFromDevice:dummyData2];
+
+  // THEN
+  [self waitForExpectations:@[ expectation ] timeout:1.0];
+
+  XCTAssertEqualObjects(receivedData, expectedData);
+}
+
+/// Tests data larger that buffer size is sent to watch despite chunking.
+- (void)testSendDataWithSmallerStreamBuffer {
+  // GIVEN
+  NSData* dummyData = [@"dummyData" dataUsingEncoding:NSASCIIStringEncoding];
+  _fakeInputOutputStream =
+      [[GNCBLEL2CAPFakeInputOutputStream alloc] initWithBufferSize:dummyData.length - 1];
+
+  _stream = [[GNCBLEL2CAPStream alloc]
+      initWithClosedBlock:^{
+      }
+      receivedDataBlock:^(NSData* data) {
+        XCTFail(@"Should not call block.");
+      }
+      inputStream:_fakeInputOutputStream.inputStream
+      outputStream:_fakeInputOutputStream.outputStream];
+
+  // WHEN
+  [_stream sendData:dummyData
+      completionBlock:^(BOOL result){
+      }];
+
+  // THEN
+  NSData* sentData = [_fakeInputOutputStream dataSentToWatchWithMaxBytes:dummyData.length];
+  XCTAssertEqualObjects(sentData, dummyData);
+}
+
+- (void)testSendDataCompletionIsCalled {
+  // GIVEN
+  NSData* dummyData = [@"dummyData" dataUsingEncoding:NSASCIIStringEncoding];
+  _fakeInputOutputStream =
+      [[GNCBLEL2CAPFakeInputOutputStream alloc] initWithBufferSize:dummyData.length * 2];
+
+  _stream = [[GNCBLEL2CAPStream alloc]
+      initWithClosedBlock:^{
+      }
+      receivedDataBlock:^(NSData* data) {
+        XCTFail(@"Should not call block.");
+      }
+      inputStream:_fakeInputOutputStream.inputStream
+      outputStream:_fakeInputOutputStream.outputStream];
+
+  // WHEN
+  XCTestExpectation* completionExpectation =
+      [self expectationWithDescription:@"Completion is called"];
+  [_stream sendData:dummyData
+      completionBlock:^(BOOL result) {
+        XCTAssert(result);
+        [completionExpectation fulfill];
+      }];
+
+  // THEN
+  [self waitForExpectations:@[ completionExpectation ] timeout:1.0];
+}
+
+- (void)testSendDataCompletionIsNotCalledIfDataIsNotCompletelyWritten {
+  // GIVEN
+  NSData* dummyData = [@"dummyData" dataUsingEncoding:NSASCIIStringEncoding];
+  // Small buffer size so that the data cannot be completely written.
+  _fakeInputOutputStream =
+      [[GNCBLEL2CAPFakeInputOutputStream alloc] initWithBufferSize:dummyData.length - 1];
+
+  _stream = [[GNCBLEL2CAPStream alloc]
+      initWithClosedBlock:^{
+      }
+      receivedDataBlock:^(NSData* data) {
+        XCTFail(@"Should not call block.");
+      }
+      inputStream:_fakeInputOutputStream.inputStream
+      outputStream:_fakeInputOutputStream.outputStream];
+
+  // WHEN
+  XCTestExpectation* noCompletionExpectation =
+      [self expectationWithDescription:@"Completion is not called"];
+  noCompletionExpectation.inverted = YES;
+  [_stream sendData:dummyData
+      completionBlock:^(BOOL result) {
+        [noCompletionExpectation fulfill];
+      }];
+
+  // THEN
+  [self waitForExpectations:@[ noCompletionExpectation ] timeout:1.0];
+}
+
+- (void)testSendDataCompletionIsCalledAfterDataIsCompletelyWritten {
+  // GIVEN
+  NSData* dummyData = [@"dummyData" dataUsingEncoding:NSASCIIStringEncoding];
+  // Small buffer size so that the data is not completely written until the watch starts reading.
+  _fakeInputOutputStream =
+      [[GNCBLEL2CAPFakeInputOutputStream alloc] initWithBufferSize:dummyData.length - 1];
+
+  _stream = [[GNCBLEL2CAPStream alloc]
+      initWithClosedBlock:^{
+      }
+      receivedDataBlock:^(NSData* data) {
+        XCTFail(@"Should not call block.");
+      }
+      inputStream:_fakeInputOutputStream.inputStream
+      outputStream:_fakeInputOutputStream.outputStream];
+
+  // WHEN
+  XCTestExpectation* completionExpectation =
+      [self expectationWithDescription:@"Completion is called"];
+  [_stream sendData:dummyData
+      completionBlock:^(BOOL result) {
+        XCTAssert(result);
+        [completionExpectation fulfill];
+      }];
+  // Reads the written data on the buffer so that the stream send finish writing the full packet.
+  [_fakeInputOutputStream dataSentToWatchWithMaxBytes:dummyData.length];
+
+  // THEN
+  [self waitForExpectations:@[ completionExpectation ] timeout:1.0];
+}
+
+- (void)testSendDataCompletionIsCalledWhenStreamIsTornDown {
+  // GIVEN
+  NSData* dummyData = [@"dummyData" dataUsingEncoding:NSASCIIStringEncoding];
+  // Small buffer size so that the data is not completely written until the watch starts reading.
+  _fakeInputOutputStream =
+      [[GNCBLEL2CAPFakeInputOutputStream alloc] initWithBufferSize:dummyData.length - 1];
+
+  _stream = [[GNCBLEL2CAPStream alloc]
+      initWithClosedBlock:^{
+      }
+      receivedDataBlock:^(NSData* data) {
+        XCTFail(@"Should not call block.");
+      }
+      inputStream:_fakeInputOutputStream.inputStream
+      outputStream:_fakeInputOutputStream.outputStream];
+
+  XCTestExpectation* completionExpectation =
+      [self expectationWithDescription:@"Completion is called"];
+
+  // WHEN
+  [_stream sendData:dummyData
+      completionBlock:^(BOOL result) {
+        XCTAssertFalse(result);
+        [completionExpectation fulfill];
+      }];
+  [_stream tearDown];
+
+  // THEN
+  [self waitForExpectations:@[ completionExpectation ] timeout:1.0];
+}
+
+/// Tests receiving data is not possible after invoking the tear down method.
+- (void)testDataCannotBeReceivedAfterInvokingTearDown {
+  // GIVEN
+  NSData* dummyData1 = [@"dummyData1" dataUsingEncoding:NSASCIIStringEncoding];
+
+  _fakeInputOutputStream =
+      [[GNCBLEL2CAPFakeInputOutputStream alloc] initWithBufferSize:dummyData1.length];
+
+  XCTestExpectation* expectation = [self expectationWithDescription:@"Received data from watch."];
+  expectation.inverted = YES;
+
+  _stream = [[GNCBLEL2CAPStream alloc]
+      initWithClosedBlock:^{
+      }
+      receivedDataBlock:^(NSData* data) {
+        [expectation fulfill];
+      }
+      inputStream:_fakeInputOutputStream.inputStream
+      outputStream:_fakeInputOutputStream.outputStream];
+
+  // WHEN
+  [_stream tearDown];
+
+  [_fakeInputOutputStream writeFromDevice:dummyData1];
+
+  // THEN
+  [self waitForExpectations:@[ expectation ] timeout:1.0];
+}
+
+/// Tests closing the remote end disconnects the L2CAP stream.
+- (void)testClosedBlockIsCalledWhenStreamIsClosed {
+  // GIVEN
+  // Hard-coded from @c GNCBLEL2CAPStream.
+  NSUInteger kMaxAllowedQueuedDataBytes = 20480;
+  _fakeInputOutputStream =
+      [[GNCBLEL2CAPFakeInputOutputStream alloc] initWithBufferSize:kMaxAllowedQueuedDataBytes];
+
+  XCTestExpectation* disconnectedExpectation =
+      [self expectationWithDescription:@"State disconnected"];
+
+  _stream = [[GNCBLEL2CAPStream alloc]
+      initWithClosedBlock:^{
+        [disconnectedExpectation fulfill];
+      }
+      receivedDataBlock:^(NSData* data) {
+        XCTFail(@"Should not call block.");
+      }
+      inputStream:_fakeInputOutputStream.inputStream
+      outputStream:_fakeInputOutputStream.outputStream];
+
+  // WHEN
+  [_fakeInputOutputStream tearDown];
+
+  // THEN
+  [self waitForExpectations:@[ disconnectedExpectation ] timeout:1.0];
+}
+
+/// Tests that received data block is not invoked concurrently and the data is passed in the order
+/// it was written.
+- (void)testForwardsReceivedDataSeriallyInCorrectOrder {
+  // GIVEN
+  NSMutableArray<NSData*>* testData = [NSMutableArray array];
+  for (int i = 0; i < 10; i++) {
+    [testData addObject:[[NSString stringWithFormat:@"testData%@", @(i)]
+                            dataUsingEncoding:NSASCIIStringEncoding]];
+  }
+  // Buffer size is as small as possible so that the stream does not combine the written packets.
+  _fakeInputOutputStream =
+      [[GNCBLEL2CAPFakeInputOutputStream alloc] initWithBufferSize:testData[0].length];
+
+  XCTestExpectation* receivedDataExpectation =
+      [self expectationWithDescription:@"Received data from watch."];
+  receivedDataExpectation.expectedFulfillmentCount = testData.count;
+
+  __block BOOL processingData = NO;
+  NSMutableArray<NSData*>* receivedData = [NSMutableArray array];
+  _stream = [[GNCBLEL2CAPStream alloc]
+      initWithClosedBlock:^{
+        XCTFail(@"Should not be invoked.");
+      }
+      receivedDataBlock:^(NSData* data) {
+        XCTAssertFalse(processingData, @"Received data block must not be invoked concurrently");
+        processingData = YES;
+        dispatch_sync(dispatch_get_main_queue(), ^{
+          [receivedData addObject:data];
+          [receivedDataExpectation fulfill];
+        });
+        processingData = NO;
+      }
+      inputStream:_fakeInputOutputStream.inputStream
+      outputStream:_fakeInputOutputStream.outputStream];
+
+  // WHEN
+  for (NSData* data in testData) {
+    [_fakeInputOutputStream writeFromDevice:data];
+  }
+
+  // THEN
+  [self waitForExpectations:@[ receivedDataExpectation ] timeout:1];
+
+  XCTAssertEqualObjects(receivedData, testData);
+}
+
+@end