fix(iot): fix race conditions and crashes (#5511)

* fix(iot): fix race conditions and crashes

* fix syntax error

* add tests

* work on review comments

Author: harsh62

AWSIoT/AWSIoTDataManager.h

@@ -359,7 +359,7 @@ NS_ASSUME_NONNULL_BEGIN
 
  *Objective-C*
 
- AWSIoTDataManager *IoTDataManager = [AWSIoTDataManager IoTDataManagerForKey:@"USWest2IoTDataManager"];
+     AWSIoTDataManager *IoTDataManager = [AWSIoTDataManager IoTDataManagerForKey:@"USWest2IoTDataManager"];
 
  @warning After calling this method, do not modify the configuration object. It may cause unspecified behaviors.
 
@@ -538,9 +538,10 @@ DEPRECATED_MSG_ATTRIBUTE("Use `updateUserMetaData` for updating the user meta da
                            statusCallback:(void (^)(AWSIoTMQTTStatus status))callback;
 
 /**
- Disconnect from a mqtt client (close current mqtt session)
+ Disconnect from the AWS IoT Service. This will also cancel any active 
+ subscriptions to topics.
  */
-- (void)disconnect;
+- (void) disconnect;
 
 /**
  Get the current connection status

AWSIoT/AWSIoTDataManager.m

@@ -356,7 +356,10 @@ - (void)addUserMetaData:(NSDictionary<NSString *, NSString *> *)userMetaDataMap
     if (userMetaDataMap) {
         for (id key in userMetaDataMap) {
             if (!([key isEqualToString:@"SDK"] || [key isEqualToString:@"Version"])) {
-                [userMetaDataString appendFormat:@"&%@=%@", key, [userMetaDataMap objectForKey:key]];
+                [userMetaDataString appendFormat:@"&%@", key];
+                if (!([(NSString *)[userMetaDataMap objectForKey:key] isEqualToString:@""] || [userMetaDataMap objectForKey:key] == nil)){
+                    [userMetaDataString appendFormat:@"=%@", [userMetaDataMap objectForKey:key]];
+                }
             } else {
                 AWSDDLogWarn(@"Keynames 'SDK' and 'Version' are reserved and will be skipped");
             }
@@ -560,14 +563,37 @@ - (BOOL)connectUsingWebSocketWithClientId:(NSString *)clientId
                                  statusCallback:callback];
 }
 
-- (void)disconnect{
-    if ( !_userDidIssueConnect || _userDidIssueDisconnect ) {
-        //Have to be connected to make this call. noop this call by returning
-        return ;
+- (void)disconnect {
+    // Check if already disconnected to prevent multiple disconnect calls
+    if (!_userDidIssueConnect || _userDidIssueDisconnect) {
+        // Already disconnected, no-op
+        return;
     }
+    
+    // Update state flags
     _userDidIssueConnect = NO;
     _userDidIssueDisconnect = YES;
-    [self.mqttClient disconnect];
+    
+    // Execute the disconnect on the main thread for stability
+    // The AWS IoT MQTT client uses NSRunLoop which works best on the main thread
+    if ([NSThread isMainThread]) {
+        // Already on main thread, call directly
+        [self.mqttClient disconnect];
+        
+        // Force a runloop cycle to allow disconnect processing to start
+        // This doesn't add a delay but ensures that disconnect processing has begun
+        CFRunLoopRunInMode(kCFRunLoopDefaultMode, 0, false);
+    } else {
+        // Not on main thread, dispatch to main thread and wait for it to complete
+        // This ensures the disconnect is processed before any potential credential clearing
+        dispatch_sync(dispatch_get_main_queue(), ^{
+            [self.mqttClient disconnect];
+            
+            // Force a runloop cycle to allow disconnect processing to start
+            // This doesn't add a delay but ensures that disconnect processing has begun
+            CFRunLoopRunInMode(kCFRunLoopDefaultMode, 0, false);
+        });
+    }
 }
 
 - (AWSIoTMQTTStatus) getConnectionStatus {

AWSIoT/Internal/AWSIoTMQTTClient.m

@@ -620,7 +620,6 @@ - (void)initWebSocketConnectionForURL:(NSString *)urlString {
 }
 
 - (void)disconnect {
-
     if (self.userDidIssueDisconnect ) {
         //Issuing disconnect multiple times. Turn this function into a noop by returning here.
         return;
@@ -637,8 +636,13 @@ - (void)disconnect {
     [self.session disconnect];
     self.connectionAgeInSeconds = 0;
 
-    //Cancel the current streams thread
-    [self.streamsThread cancelAndDisconnect:YES];
+    //Cancel the current streams thread - synchronize this to prevent race conditions
+    @synchronized(self) {
+        if (self.streamsThread && !self.streamsThread.isCancelled) {
+            AWSDDLogVerbose(@"Cancelling stream thread during disconnect: [%@]", self.streamsThread);
+            [self.streamsThread cancelAndDisconnect:YES];
+        }
+    }
 
     __weak AWSIoTMQTTClient *weakSelf = self;
     self.streamsThread.onStop = ^{

AWSIoT/Internal/AWSIoTStreamThread.m

@@ -143,16 +143,21 @@ - (void)cleanUp {
         }
 
         if (self.shouldDisconnect) {
+            // Properly handle session closure first
             if (self.session) {
                 [self.session close];
                 self.session = nil;
             }
 
+            // Make sure we handle the streams in a thread-safe way
             if (self.outputStream) {
+                // Remove from runLoop first before closing
+                if (self.runLoopForStreamsThread) {
+                    [self.outputStream removeFromRunLoop:self.runLoopForStreamsThread
+                                                 forMode:NSDefaultRunLoopMode];
+                }
                 self.outputStream.delegate = nil;
                 [self.outputStream close];
-                [self.outputStream removeFromRunLoop:self.runLoopForStreamsThread
-                                             forMode:NSDefaultRunLoopMode];
                 self.outputStream = nil;
             }
 
@@ -169,9 +174,13 @@ - (void)cleanUp {
             AWSDDLogVerbose(@"Skipping disconnect for thread: [%@]", (NSThread *)self);
         }
 
-        if (self.onStop) {
-            self.onStop();
+        // Make sure onStop is called on the main thread to avoid UI issues
+        void (^stopBlock)(void) = self.onStop;
+        if (stopBlock) {
             self.onStop = nil;
+            dispatch_async(dispatch_get_main_queue(), ^{
+                stopBlock();
+            });
         }
     });
 }