Merge pull request #597 from davidswi/feature/issue_566

IAP transport read/writes on background thread

7 files changed, 370 insertions, 61 deletions

diff --git a/SmartDeviceLink-iOS.xcodeproj/project.pbxproj b/SmartDeviceLink-iOS.xcodeproj/project.pbxproj
index 46cc8683f..e9422ea73 100644
--- a/SmartDeviceLink-iOS.xcodeproj/project.pbxproj
+++ b/SmartDeviceLink-iOS.xcodeproj/project.pbxproj
@@ -885,6 +885,8 @@
 		5DEE55C01B8509CB004F0D0F /* SDLURLRequestTaskSpec.m in Sources */ = {isa = PBXBuildFile; fileRef = 5DEE55BF1B8509CB004F0D0F /* SDLURLRequestTaskSpec.m */; };
 		5DF2BB9D1B94E38A00CE5994 /* SDLURLSessionSpec.m in Sources */ = {isa = PBXBuildFile; fileRef = 5DF2BB9C1B94E38A00CE5994 /* SDLURLSessionSpec.m */; };
 		5DFFB9151BD7C89700DB3F04 /* SDLConnectionManagerType.h in Headers */ = {isa = PBXBuildFile; fileRef = 5DFFB9141BD7C89700DB3F04 /* SDLConnectionManagerType.h */; };
+		97E26DEC1E807AD70074A3C7 /* SDLMutableDataQueue.h in Headers */ = {isa = PBXBuildFile; fileRef = 97E26DEA1E807AD70074A3C7 /* SDLMutableDataQueue.h */; };
+		97E26DED1E807AD70074A3C7 /* SDLMutableDataQueue.m in Sources */ = {isa = PBXBuildFile; fileRef = 97E26DEB1E807AD70074A3C7 /* SDLMutableDataQueue.m */; };
 		DA4353DF1D271FD10099B8C4 /* CGPointUtilSpec.m in Sources */ = {isa = PBXBuildFile; fileRef = DA4353DE1D271FD10099B8C4 /* CGPointUtilSpec.m */; };
 		DA4353E31D2720A30099B8C4 /* SDLPinchGestureSpec.m in Sources */ = {isa = PBXBuildFile; fileRef = DA4353E21D2720A30099B8C4 /* SDLPinchGestureSpec.m */; };
 		DA4353E91D2721680099B8C4 /* DispatchTimerSpec.m in Sources */ = {isa = PBXBuildFile; fileRef = DA4353E61D2721680099B8C4 /* DispatchTimerSpec.m */; };
@@ -1924,6 +1926,8 @@
 		5DEE55BF1B8509CB004F0D0F /* SDLURLRequestTaskSpec.m */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.objc; name = SDLURLRequestTaskSpec.m; path = "UtilitiesSpecs/HTTP Connection/SDLURLRequestTaskSpec.m"; sourceTree = "<group>"; };
 		5DF2BB9C1B94E38A00CE5994 /* SDLURLSessionSpec.m */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.objc; name = SDLURLSessionSpec.m; path = "UtilitiesSpecs/HTTP Connection/SDLURLSessionSpec.m"; sourceTree = "<group>"; };
 		5DFFB9141BD7C89700DB3F04 /* SDLConnectionManagerType.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; path = SDLConnectionManagerType.h; sourceTree = "<group>"; };
+		97E26DEA1E807AD70074A3C7 /* SDLMutableDataQueue.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; path = SDLMutableDataQueue.h; sourceTree = "<group>"; };
+		97E26DEB1E807AD70074A3C7 /* SDLMutableDataQueue.m */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.objc; path = SDLMutableDataQueue.m; sourceTree = "<group>"; };
 		DA4353DE1D271FD10099B8C4 /* CGPointUtilSpec.m */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.objc; name = CGPointUtilSpec.m; path = UtilitiesSpecs/Touches/CGPointUtilSpec.m; sourceTree = "<group>"; };
 		DA4353E21D2720A30099B8C4 /* SDLPinchGestureSpec.m */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.objc; name = SDLPinchGestureSpec.m; path = UtilitiesSpecs/Touches/SDLPinchGestureSpec.m; sourceTree = "<group>"; };
 		DA4353E61D2721680099B8C4 /* DispatchTimerSpec.m */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.objc; name = DispatchTimerSpec.m; path = UtilitiesSpecs/Touches/DispatchTimerSpec.m; sourceTree = "<group>"; };
@@ -3091,6 +3095,8 @@
 		5D5934F61A85189500687FB9 /* Utilities */ = {
 			isa = PBXGroup;
 			children = (
+				97E26DEA1E807AD70074A3C7 /* SDLMutableDataQueue.h */,
+				97E26DEB1E807AD70074A3C7 /* SDLMutableDataQueue.m */,
 				DAC5724C1D0FE3B60004288B /* Touches */,
 				5DCC199D1B8221D2004FFAD9 /* HTTP Connection */,
 				E9C32B831AB20B2900F283AF /* @categories */,
@@ -4201,6 +4207,7 @@
 				5D61FD4A1A84238C00846EE7 /* SDLProtocolMessageAssembler.h in Headers */,
 				5D61FD4C1A84238C00846EE7 /* SDLProtocolMessageDisassembler.h in Headers */,
 				5D61FD4E1A84238C00846EE7 /* SDLProtocolReceivedMessageRouter.h in Headers */,
+				97E26DEC1E807AD70074A3C7 /* SDLMutableDataQueue.h in Headers */,
 				5D61FDF71A84238C00846EE7 /* SDLV1ProtocolMessage.h in Headers */,
 				5D61FDFB1A84238C00846EE7 /* SDLV2ProtocolMessage.h in Headers */,
 				5D7F87EB1CE3C1A1002DD7C4 /* SDLDeleteFileOperation.h in Headers */,
@@ -4668,6 +4675,7 @@
 				5D61FD7A1A84238C00846EE7 /* SDLScreenParams.m in Sources */,
 				5D61FC831A84238C00846EE7 /* SDLDeviceInfo.m in Sources */,
 				5D7F87EC1CE3C1A1002DD7C4 /* SDLDeleteFileOperation.m in Sources */,
+				97E26DED1E807AD70074A3C7 /* SDLMutableDataQueue.m in Sources */,
 				5D61FD641A84238C00846EE7 /* SDLResetGlobalProperties.m in Sources */,
 				5D60088B1BE3ED540094A505 /* SDLStateMachine.m in Sources */,
 				5D61FD181A84238C00846EE7 /* SDLOnPermissionsChange.m in Sources */,
diff --git a/SmartDeviceLink/SDLIAPSession.h b/SmartDeviceLink/SDLIAPSession.h
index d21343894..751d10086 100644
--- a/SmartDeviceLink/SDLIAPSession.h
+++ b/SmartDeviceLink/SDLIAPSession.h
@@ -17,11 +17,12 @@ typedef void (^SessionCompletionHandler)(BOOL success);
 @property (strong, atomic) EASession *easession;
 @property (weak) id<SDLIAPSessionDelegate> delegate;
 @property (strong, atomic) SDLStreamDelegate *streamDelegate;
+@property (assign, readonly, getter=isStopped) BOOL stopped;
 
 - (instancetype)initWithAccessory:(EAAccessory *)accessory
                       forProtocol:(NSString *)protocol;
 
 - (BOOL)start;
 - (void)stop;
-
+- (void)sendData:(NSData *)data;
 @end
diff --git a/SmartDeviceLink/SDLIAPSession.m b/SmartDeviceLink/SDLIAPSession.m
index 5020b2081..2048f88de 100644
--- a/SmartDeviceLink/SDLIAPSession.m
+++ b/SmartDeviceLink/SDLIAPSession.m
@@ -4,14 +4,21 @@
 
 #import "SDLIAPSession.h"
 #import "SDLDebugTool.h"
+#import "SDLMutableDataQueue.h"
 #import "SDLStreamDelegate.h"
 #import "SDLTimer.h"
 
+NSString *const IOStreamThreadName  = @"com.smartdevicelink.iostream";
+NSTimeInterval const streamThreadWaitSecs = 1.0;
 
 @interface SDLIAPSession ()
 
 @property (assign) BOOL isInputStreamOpen;
 @property (assign) BOOL isOutputStreamOpen;
+@property (assign, nonatomic) BOOL isDataSession;
+@property (nonatomic, strong) NSThread *ioStreamThread;
+@property (nonatomic, strong) SDLMutableDataQueue *sendDataQueue;
+@property (nonatomic, strong) dispatch_semaphore_t canceledSemaphore;
 
 @end
 
@@ -27,8 +34,11 @@
     self = [super init];
     if (self) {
         _delegate = nil;
+        _isDataSession = [protocol isEqualToString:@"com.smartdevicelink.prot0"] ? NO : YES;
         _accessory = accessory;
         _protocol = protocol;
+        _canceledSemaphore = dispatch_semaphore_create(0);
+        _sendDataQueue = [[SDLMutableDataQueue alloc] init];
         _streamDelegate = nil;
         _easession = nil;
         _isInputStreamOpen = NO;
@@ -43,20 +53,28 @@
 - (BOOL)start {
     __weak typeof(self) weakSelf = self;
 
-    NSString *logMessage = [NSString stringWithFormat:@"Opening EASession withAccessory:%@ forProtocol:%@", _accessory.name, _protocol];
+    NSString *logMessage = [NSString stringWithFormat:@"Opening EASession withAccessory:%@ forProtocol:%@", self.accessory.name, self.protocol];
     [SDLDebugTool logInfo:logMessage];
 
-    if ((self.easession = [[EASession alloc] initWithAccessory:_accessory forProtocol:_protocol])) {
+    // TODO: This assignment should be broken out of the if and the if / else should be flipped.
+    if ((self.easession = [[EASession alloc] initWithAccessory:self.accessory forProtocol:self.protocol])) {
         __strong typeof(self) strongSelf = weakSelf;
 
         [SDLDebugTool logInfo:@"Created Session Object"];
 
         strongSelf.streamDelegate.streamErrorHandler = [self streamErroredHandler];
         strongSelf.streamDelegate.streamOpenHandler = [self streamOpenedHandler];
-
-        [strongSelf startStream:weakSelf.easession.outputStream];
-        [strongSelf startStream:weakSelf.easession.inputStream];
-
+        if (self.isDataSession) {
+            self.streamDelegate.streamHasSpaceHandler = [self sdl_streamHasSpaceHandler];
+            // Start I/O event loop processing events in iAP channel
+            self.ioStreamThread = [[NSThread alloc] initWithTarget:self selector:@selector(sdl_accessoryEventLoop) object:nil];
+            [self.ioStreamThread setName:IOStreamThreadName];
+            [self.ioStreamThread start];
+        } else {
+            // Set up control session -- no need for its own thread
+            [self startStream:self.easession.outputStream];
+            [self startStream:self.easession.inputStream];
+        }
         return YES;
 
     } else {
@@ -66,17 +84,124 @@
 }
 
 - (void)stop {
-    [self stopStream:self.easession.outputStream];
-    [self stopStream:self.easession.inputStream];
+    if (self.isDataSession) {
+        [self.ioStreamThread cancel];
+        
+        long lWait = dispatch_semaphore_wait(self.canceledSemaphore, dispatch_time(DISPATCH_TIME_NOW, streamThreadWaitSecs * NSEC_PER_SEC));
+        if (lWait == 0) {
+            [SDLDebugTool logInfo:@"Stream thread canceled"];
+        } else {
+            [SDLDebugTool logInfo:@"Error: failed to cancel stream thread"];
+        }
+        self.ioStreamThread = nil;
+        self.isDataSession = NO;
+    } else {
+        // Stop control session
+        [self stopStream:self.easession.outputStream];
+        [self stopStream:self.easession.inputStream];
+    }
     self.easession = nil;
 }
 
+- (BOOL)isStopped {
+    return !self.isOutputStreamOpen && !self.isInputStreamOpen;
+}
+
+# pragma mark - data send methods
+
+- (void)sendData:(NSData *)data {
+    // Enqueue the data for transmission on the IO thread
+    [self.sendDataQueue enqueueBuffer:data.mutableCopy];
+}
+
+- (BOOL)sdl_dequeueAndWriteToOutputStream:(NSError **)error {
+    NSOutputStream *ostream = self.easession.outputStream;
+    NSMutableData *remainder = [self.sendDataQueue frontBuffer];
+    BOOL allDataWritten = NO;
+    
+    if (error != nil && remainder != nil && ostream.streamStatus == NSStreamStatusOpen) {
+        NSInteger bytesRemaining = remainder.length;
+        NSInteger bytesWritten = [ostream write:remainder.bytes maxLength:bytesRemaining];
+        if (bytesWritten < 0) {
+            *error = ostream.streamError;
+        } else if (bytesWritten == bytesRemaining) {
+            // Remove the data from the queue
+            [self.sendDataQueue popBuffer];
+            allDataWritten = YES;
+        } else {
+            // Cleave the sent bytes from the data, the remainder will sit at the head of the queue
+            [remainder replaceBytesInRange:NSMakeRange(0, bytesWritten) withBytes:NULL length:0];
+        }
+    }
+
+    return allDataWritten;
+}
+
+- (void)sdl_handleOutputStreamWriteError:(NSError *)error {
+    NSString *errString = [NSString stringWithFormat:@"Output stream error: %@", error];
+    [SDLDebugTool logInfo:errString];
+    // TODO: We should look at the domain and the code as a tuple and decide how to handle the error based on both values. For now, if the stream is closed, we will flush the send queue and leave it as-is otherwise so that temporary error conditions can be dealt with by retrying
+    if (self.easession == nil ||
+        self.easession.outputStream == nil ||
+        self.easession.outputStream.streamStatus != NSStreamStatusOpen) {
+        [self.sendDataQueue removeAllObjects];
+    }
+}
+
+# pragma mark - background I/O for data session
+
+// Data session I/O thread
+- (void)sdl_accessoryEventLoop {
+    @autoreleasepool {
+        NSAssert(self.easession, @"_session must be assigned before calling");
+
+        if (!self.easession) {
+            return;
+        }
+        
+        [self startStream:self.easession.inputStream];
+        [self startStream:self.easession.outputStream];
+
+        [SDLDebugTool logInfo:@"starting the event loop for accessory"];
+        do {
+            if (self.sendDataQueue.count > 0 && !self.sendDataQueue.frontDequeued) {
+                NSError *sendErr = nil;
+                if (![self sdl_dequeueAndWriteToOutputStream:&sendErr] && sendErr != nil) {
+                    [self sdl_handleOutputStreamWriteError:sendErr];
+                }
+            }
+            // The principle here is to give the event loop enough time to process stream events while also allowing it to handle new enqueued data buffers in a timely manner. We're capping the run loop CPU time at 0.25s maximum before it will return control to the rest of the loop.
+            [[NSRunLoop currentRunLoop] runMode:NSDefaultRunLoopMode beforeDate:[NSDate dateWithTimeIntervalSinceNow:0.25f]];
+        } while (![NSThread currentThread].cancelled);
+
+        [SDLDebugTool logInfo:@"closing accessory session"];
+
+        // Close I/O streams of the iAP session
+        [self sdl_closeSession];
+        dispatch_semaphore_signal(self.canceledSemaphore);
+    }
+}
+
+// Must be called on accessoryEventLoop.
+- (void)sdl_closeSession {
+    if (!self.easession) {
+        return;
+    }
+    
+    NSString *closeSessionString = [NSString stringWithFormat:@"Close EASession: %tu", self.easession.accessory.connectionID];
+    [SDLDebugTool logInfo:closeSessionString];
+
+    [self stopStream:[self.easession inputStream]];
+    [self stopStream:[self.easession outputStream]];
+}
+
 
 #pragma mark - Private Stream Lifecycle Helpers
 
 - (void)startStream:(NSStream *)stream {
     stream.delegate = self.streamDelegate;
-    [stream scheduleInRunLoop:[NSRunLoop mainRunLoop] forMode:NSDefaultRunLoopMode];
+    NSAssert((self.isDataSession && [[NSThread currentThread] isEqual:self.ioStreamThread]) || [NSThread isMainThread], @"startStream is being called on the wrong thread!!!");
+    [stream scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
     [stream open];
 }
 
@@ -86,6 +211,8 @@
 
     // When you disconect the cable you get a stream end event and come here but stream is already in closed state.
     // Still need to remove from run loop.
+    
+    NSAssert((self.isDataSession && [[NSThread currentThread] isEqual:self.ioStreamThread]) || [NSThread isMainThread], @"stopStream is being called on the wrong thread!!!");
 
     NSUInteger status1 = stream.streamStatus;
     if (status1 != NSStreamStatusNotOpen &&
@@ -93,15 +220,17 @@
         [stream close];
     }
 
-    [stream removeFromRunLoop:[NSRunLoop mainRunLoop] forMode:NSDefaultRunLoopMode];
+    [stream removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
     [stream setDelegate:nil];
 
     NSUInteger status2 = stream.streamStatus;
     if (status2 == NSStreamStatusClosed) {
         if (stream == [self.easession inputStream]) {
             [SDLDebugTool logInfo:@"Input Stream Closed"];
+            self.isInputStreamOpen = NO;
         } else if (stream == [self.easession outputStream]) {
             [SDLDebugTool logInfo:@"Output Stream Closed"];
+            self.isOutputStreamOpen = NO;
         }
     }
 }
@@ -141,16 +270,22 @@
     };
 }
 
-
-#pragma mark - Lifecycle Destruction
-
-- (void)dealloc {
-    self.delegate = nil;
-    self.accessory = nil;
-    self.protocol = nil;
-    self.streamDelegate = nil;
-    self.easession = nil;
-    [SDLDebugTool logInfo:@"SDLIAPSession Dealloc"];
+- (SDLStreamHasSpaceHandler)sdl_streamHasSpaceHandler {
+    __weak typeof(self) weakSelf = self;
+    
+    return ^(NSStream *stream) {
+        __strong typeof(weakSelf) strongSelf = weakSelf;
+        
+        if (!strongSelf.isDataSession){
+            return;
+        }
+        
+        NSError *sendErr = nil;
+        
+        if (![strongSelf sdl_dequeueAndWriteToOutputStream:&sendErr] && sendErr != nil) {
+            [strongSelf sdl_handleOutputStreamWriteError:sendErr];
+        }
+    };
 }
 
 @end
diff --git a/SmartDeviceLink/SDLIAPTransport.m b/SmartDeviceLink/SDLIAPTransport.m
index b90296d07..651df5fbb 100644
--- a/SmartDeviceLink/SDLIAPTransport.m
+++ b/SmartDeviceLink/SDLIAPTransport.m
@@ -6,6 +6,7 @@
 #import <UIKit/UIKit.h>
 
 #import "EAAccessoryManager+SDLProtocols.h"
+#import "EAAccessory+SDLProtocols.h"
 #import "SDLDebugTool.h"
 #import "SDLGlobals.h"
 #import "SDLIAPSession.h"
@@ -27,7 +28,6 @@ int const streamOpenTimeoutSeconds = 2;
 
 
 @interface SDLIAPTransport () {
-    dispatch_queue_t _transmit_queue;
     BOOL _alreadyDestructed;
 }
 
@@ -48,7 +48,6 @@ int const streamOpenTimeoutSeconds = 2;
         _retryCounter = 0;
         _sessionSetupInProgress = NO;
         _protocolIndexTimer = nil;
-        _transmit_queue = dispatch_queue_create("com.sdl.transport.iap.transmit", DISPATCH_QUEUE_SERIAL);
 
         [self sdl_startEventListening];
         [SDLSiphonServer init];
@@ -78,6 +77,8 @@ int const streamOpenTimeoutSeconds = 2;
                                              selector:@selector(sdl_applicationWillEnterForeground:)
                                                  name:UIApplicationWillEnterForegroundNotification
                                                object:nil];
+    
+    [[EAAccessoryManager sharedAccessoryManager] registerForLocalNotifications];
 }
 
 - (void)sdl_stopEventListening {
@@ -88,11 +89,13 @@ int const streamOpenTimeoutSeconds = 2;
 #pragma mark - EAAccessory Notifications
 
 - (void)sdl_accessoryConnected:(NSNotification *)notification {
+    EAAccessory *accessory = notification.userInfo[EAAccessoryKey];
     NSMutableString *logMessage = [NSMutableString stringWithFormat:@"Accessory Connected, Opening in %0.03fs", self.retryDelay];
     [SDLDebugTool logInfo:logMessage withType:SDLDebugType_Transport_iAP toOutput:SDLDebugOutput_All toGroup:self.debugConsoleGroupName];
 
     self.retryCounter = 0;
-    [self performSelector:@selector(connect) withObject:nil afterDelay:self.retryDelay];
+
+    [self performSelector:@selector(sdl_connect:) withObject:accessory afterDelay:self.retryDelay];
 }
 
 - (void)sdl_accessoryDisconnected:(NSNotification *)notification {
@@ -100,7 +103,10 @@ int const streamOpenTimeoutSeconds = 2;
 
     // Only check for the data session, the control session is handled separately
     EAAccessory *accessory = [notification.userInfo objectForKey:EAAccessoryKey];
-    if (accessory.connectionID == self.session.accessory.connectionID) {
+    if (accessory.connectionID != self.session.accessory.connectionID) {
+         [SDLDebugTool logInfo:@"Accessory connection ID mismatch!!!" withType:SDLDebugType_Transport_iAP toOutput:SDLDebugOutput_All toGroup:self.debugConsoleGroupName];
+    }
+    if ([accessory.serialNumber isEqualToString:self.session.accessory.serialNumber]){
         self.sessionSetupInProgress = NO;
         [self disconnect];
         [self.delegate onTransportDisconnected];
@@ -117,9 +123,13 @@ int const streamOpenTimeoutSeconds = 2;
 #pragma mark - Stream Lifecycle
 
 - (void)connect {
+    [self sdl_connect:nil];
+}
+
+- (void)sdl_connect:(EAAccessory *)accessory {
     if (!self.session && !self.sessionSetupInProgress) {
         self.sessionSetupInProgress = YES;
-        [self sdl_establishSession];
+        [self sdl_establishSessionWithAccessory:accessory];
     } else if (self.session) {
         [SDLDebugTool logInfo:@"Session already established."];
     } else {
@@ -129,7 +139,9 @@ int const streamOpenTimeoutSeconds = 2;
 
 - (void)disconnect {
     [SDLDebugTool logInfo:@"IAP Disconnecting" withType:SDLDebugType_Transport_iAP toOutput:SDLDebugOutput_All toGroup:self.debugConsoleGroupName];
-
+    // Stop event listening here so that even if the transport is disconnected by the proxy
+    // we unregister for accessory local notifications
+    [self sdl_stopEventListening];
     // Only disconnect the data session, the control session does not stay open and is handled separately
     if (self.session != nil) {
         [self.session stop];
@@ -140,23 +152,43 @@ int const streamOpenTimeoutSeconds = 2;
 
 #pragma mark - Creating Session Streams
 
-- (void)sdl_establishSession {
+- (BOOL)sdl_connectAccessory:(EAAccessory *)accessory {
+    BOOL connecting = NO;
+    
+    if ([accessory supportsProtocol:controlProtocolString]) {
+        [self sdl_createIAPControlSessionWithAccessory:accessory];
+        connecting = YES;
+    } else if ([accessory supportsProtocol:legacyProtocolString]) {
+        [self sdl_createIAPDataSessionWithAccessory:accessory forProtocol:legacyProtocolString];
+        connecting = YES;
+    }
+    
+    return connecting;
+}
+
+- (void)sdl_establishSessionWithAccessory:(EAAccessory *)accessory {
     [SDLDebugTool logInfo:@"Attempting To Connect"];
     if (self.retryCounter < createSessionRetries) {
         // We should be attempting to connect
         self.retryCounter++;
-        EAAccessory *accessory = nil;
-
+        EAAccessory *sdlAccessory = accessory;
+        // If we are being called from sdl_connectAccessory, the EAAccessoryDidConnectNotification will contain the SDL accessory to connect to and we can connect without searching the accessory manager's connected accessory list. Otherwise, we fall through to a search.
+        if (sdlAccessory != nil && [self sdl_connectAccessory:sdlAccessory]){
+            // Connection underway, exit
+            return;
+        }
+        
         // Determine if we can start a multi-app session or a legacy (single-app) session
-        if ((accessory = [EAAccessoryManager findAccessoryForProtocol:controlProtocolString])) {
-            [self sdl_createIAPControlSessionWithAccessory:accessory];
-        } else if ((accessory = [EAAccessoryManager findAccessoryForProtocol:legacyProtocolString])) {
-            [self sdl_createIAPDataSessionWithAccessory:accessory forProtocol:legacyProtocolString];
+        if ((sdlAccessory = [EAAccessoryManager findAccessoryForProtocol:controlProtocolString])) {
+            [self sdl_createIAPControlSessionWithAccessory:sdlAccessory];
+        } else if ((sdlAccessory = [EAAccessoryManager findAccessoryForProtocol:legacyProtocolString])) {
+            [self sdl_createIAPDataSessionWithAccessory:sdlAccessory forProtocol:legacyProtocolString];
         } else {
             // No compatible accessory
             [SDLDebugTool logInfo:@"No accessory supporting a required sync protocol was found."];
             self.sessionSetupInProgress = NO;
         }
+        
     } else {
         // We are beyond the number of retries allowed
         [SDLDebugTool logInfo:@"Create session retries exhausted."];
@@ -234,6 +266,13 @@ int const streamOpenTimeoutSeconds = 2;
 - (void)sdl_retryEstablishSession {
     // Current strategy disallows automatic retries.
     self.sessionSetupInProgress = NO;
+    if (self.session != nil){
+        [self.session stop];
+        self.session.delegate = nil;
+        self.session = nil;
+    }
+    // No accessory to use this time, search connected accessories
+    [self sdl_connect:nil];
 }
 
 // This gets called after both I/O streams of the session have opened.
@@ -268,23 +307,11 @@ int const streamOpenTimeoutSeconds = 2;
 #pragma mark - Data Transmission
 
 - (void)sendData:(NSData *)data {
-    dispatch_async(_transmit_queue, ^{
-        NSOutputStream *ostream = self.session.easession.outputStream;
-        NSMutableData *remainder = data.mutableCopy;
-
-        while (ostream.streamStatus == NSStreamStatusOpen && remainder.length != 0) {
-            if (ostream.hasSpaceAvailable){
-                NSInteger bytesWritten = [ostream write:remainder.bytes maxLength:remainder.length];
-
-                if (bytesWritten == -1) {
-                    [SDLDebugTool logInfo:[NSString stringWithFormat:@"Error: %@", [ostream streamError]] withType:SDLDebugType_Transport_iAP toOutput:SDLDebugOutput_All];
-                    break;
-                }
-
-                [remainder replaceBytesInRange:NSMakeRange(0, bytesWritten) withBytes:NULL length:0];
-            }
-        }
-    });
+    if (self.session == nil || !self.session.accessory.connected) {
+        return;
+    }
+    
+    [self.session sendData:data];
 }
 
 
@@ -327,15 +354,19 @@ int const streamOpenTimeoutSeconds = 2;
 
             // Destroy the control session
             [strongSelf.protocolIndexTimer cancel];
-            [strongSelf.controlSession stop];
-            strongSelf.controlSession.streamDelegate = nil;
-            strongSelf.controlSession = nil;
+            dispatch_sync(dispatch_get_main_queue(), ^{
+                [strongSelf.controlSession stop];
+                strongSelf.controlSession.streamDelegate = nil;
+                strongSelf.controlSession = nil;
+            });
 
             // Determine protocol string of the data session, then create that data session
             NSString *indexedProtocolString = [NSString stringWithFormat:@"%@%@", indexedProtocolStringPrefix, @(buf[0])];
-            dispatch_sync(dispatch_get_main_queue(), ^{
-                [strongSelf sdl_createIAPDataSessionWithAccessory:accessory forProtocol:indexedProtocolString];
-            });
+            if (accessory.isConnected){
+                dispatch_async(dispatch_get_main_queue(), ^{
+                    [strongSelf sdl_createIAPDataSessionWithAccessory:accessory forProtocol:indexedProtocolString];
+                });
+            }
         }
     };
 }
@@ -383,7 +414,9 @@ int const streamOpenTimeoutSeconds = 2;
         __strong typeof(weakSelf) strongSelf = weakSelf;
 
         uint8_t buf[[SDLGlobals globals].maxMTUSize];
-        while ([istream hasBytesAvailable]) {
+        while (istream.streamStatus == NSStreamStatusOpen && istream.hasBytesAvailable) {
+            // It is necessary to check the stream status and whether there are bytes available because the dataStreamHasBytesHandler is executed on the IO thread and the accessory disconnect notification arrives on the main thread, causing data to be passed to the delegate while the main thread is tearing down the transport.
+            
             NSInteger bytesRead = [istream read:buf maxLength:[SDLGlobals globals].maxMTUSize];
             NSData *dataIn = [NSData dataWithBytes:buf length:bytesRead];
 
@@ -458,7 +491,6 @@ int const streamOpenTimeoutSeconds = 2;
 - (void)sdl_destructObjects {
     if (!_alreadyDestructed) {
         _alreadyDestructed = YES;
-        [self sdl_stopEventListening];
         self.controlSession = nil;
         self.session = nil;
         self.delegate = nil;
diff --git a/SmartDeviceLink/SDLMutableDataQueue.h b/SmartDeviceLink/SDLMutableDataQueue.h
new file mode 100755
index 000000000..1632acf5a
--- /dev/null
+++ b/SmartDeviceLink/SDLMutableDataQueue.h
@@ -0,0 +1,60 @@
+//
+//  SDLMutableDataQueue.h
+//  
+//
+//  Created by David Switzer on 3/2/17.
+//
+//
+
+#import <Foundation/Foundation.h>
+
+/**
+ This queue class is used by the IAP data session to provide safe, async writing of RPC messages to the connected SDL compliant accessory. It uses NSMutableData objects because the NSOutputStream has a variable amount of space available based on iOS and accessory buffering. We want to be able to pop the buffer at the head multiiple times and consume as many bytes as there are available in the stream.
+ */
+
+@interface SDLMutableDataQueue : NSObject
+
+NS_ASSUME_NONNULL_BEGIN
+
+/**
+ frontBuffer returns the NSMutableData * buffer at the front of the queue, but does not remove it - modeled after the STL C++ queue front method
+
+ @return NSMutableData containing the buffer at the head of the queue or nil if
+ the queue is empty
+ */
+- (NSMutableData  * _Nullable)frontBuffer;
+
+
+/**
+ popBuffer removes the buffer at the head of the queue -- modeled after the
+ STL C++ queue pop method. Use to remove the buffer once all of the data has
+ been consumed or the data is no longer valid/needed.
+ */
+- (void)popBuffer;
+
+/**
+ Enqueues a new buffer at the back of the queue
+
+ @param data NSMutableData containing the buffer to enqueue
+ */
+- (void)enqueueBuffer:(NSMutableData *)data;
+
+/**
+ Empty the queue
+ */
+- (void)removeAllObjects;
+
+/**
+ Gets the number of buffers currently enqueued
+ */
+@property(nonatomic, assign, readonly) NSUInteger count;
+
+/**
+ A flag indicating whether the buffer at the head of the queue has been dequeued at least once.
+ */
+@property(nonatomic, assign, readonly, getter=isFrontBufferDequeued) BOOL frontDequeued;
+
+NS_ASSUME_NONNULL_END
+
+
+@end
diff --git a/SmartDeviceLink/SDLMutableDataQueue.m b/SmartDeviceLink/SDLMutableDataQueue.m
new file mode 100755
index 000000000..c91b25d85
--- /dev/null
+++ b/SmartDeviceLink/SDLMutableDataQueue.m
@@ -0,0 +1,75 @@
+//
+//  SDLMutableDataQueue.m
+//  
+//
+//  Created by David Switzer on 3/2/17.
+//
+//
+
+#import "SDLMutableDataQueue.h"
+
+NS_ASSUME_NONNULL_BEGIN
+
+@interface SDLMutableDataQueue()
+
+@property(nonatomic, strong) NSMutableArray *elements;
+@property(nonatomic, assign, readwrite) BOOL frontDequeued;
+
+@end
+
+@implementation SDLMutableDataQueue
+
+- (instancetype)init {
+    self = [super init];
+    if (!self) {
+        return nil;
+    }
+    
+    _elements = [NSMutableArray array];
+    
+    return self;
+}
+
+- (void)enqueueBuffer:(NSMutableData *)data {
+    // Since this method is being called from the main thread and the dequeue methods are being called from the data session stream thread, we need to put critical sections around the queue members. Use the @synchronized object level lock to do this.
+    @synchronized (self) {
+        [self.elements addObject:data];
+        self.frontDequeued = NO;
+    }
+}
+
+- (NSMutableData  * _Nullable )frontBuffer {
+    NSMutableData *dataAtFront = nil;
+    
+    @synchronized (self) {
+        if (self.elements.count) {
+            // The front of the queue is always at index 0
+            dataAtFront = self.elements[0];
+            self.frontDequeued = YES;
+        }
+    }
+    
+    return dataAtFront;
+}
+
+- (void)popBuffer {
+    @synchronized (self) {
+        [self.elements removeObjectAtIndex:0];
+    }
+}
+
+- (void)removeAllObjects {
+    @synchronized (self) {
+        [self.elements removeAllObjects];
+    }
+}
+
+- (NSUInteger)count {
+    @synchronized (self) {
+        return self.elements.count;
+    }
+}
+
+NS_ASSUME_NONNULL_END
+
+@end
diff --git a/SmartDeviceLink/SDLProxy.m b/SmartDeviceLink/SDLProxy.m
index 7a9088c3e..68eb8f6fc 100644
--- a/SmartDeviceLink/SDLProxy.m
+++ b/SmartDeviceLink/SDLProxy.m
@@ -88,7 +88,6 @@ const int POLICIES_CORRELATION_ID = 65535;
         [self.transport connect];
 
         [SDLDebugTool logInfo:@"SDLProxy initWithTransport"];
-        [[EAAccessoryManager sharedAccessoryManager] registerForLocalNotifications];
     }
 
     return self;
@@ -99,7 +98,6 @@ const int POLICIES_CORRELATION_ID = 65535;
         _alreadyDestructed = YES;
 
         [[NSNotificationCenter defaultCenter] removeObserver:self];
-        [[EAAccessoryManager sharedAccessoryManager] unregisterForLocalNotifications];
 
         [[SDLURLSession defaultSession] cancelAllTasks];