Fix RTN7c: notify publish callbacks of broken connection.

ably-ios/ARTClientOptions.m

@@ -84,6 +84,11 @@ - (id)copyWithZone:(NSZone *)zone {
     options.tls = self.tls;
     options.logLevel = self.logLevel;
     options.suspendedRetryTimeout = self.suspendedRetryTimeout;
+    options.disconnectedRetryTimeout = self.disconnectedRetryTimeout;
+    options.httpMaxRetryCount = self.httpMaxRetryCount;
+    options.httpMaxRetryDuration = self.httpMaxRetryDuration;
+    options.httpOpenTimeout = self.httpOpenTimeout;
+    options.httpRequestTimeout = self.httpRequestTimeout;
 
     return options;
 }

ably-ios/ARTRealtime.m

@@ -425,7 +425,6 @@ - (void)onDisconnected {
     [self.logger info:@"ARTRealtime disconnected"];
     switch (self.connection.state) {
         case ARTRealtimeConnected:
-            [self.connection setId:nil];
             [self transition:ARTRealtimeDisconnected];
             break;
         default:

ably-ios/ARTRealtimeChannel.m

@@ -140,14 +140,36 @@ - (void)publishProtocolMessage:(ARTProtocolMessage *)pm cb:(ARTStatusCallback)cb
         }
         case ARTRealtimeChannelAttached:
         {
-            [self.realtime send:pm cb:cb];
+            [self sendMessage:pm cb:cb];
             break;
         }
         default:
             NSAssert(NO, @"Invalid State");
     }
 }
 
+
+- (void)sendMessage:(ARTProtocolMessage *)pm cb:(ARTStatusCallback)cb {
+    __block BOOL gotFailure = false;
+    NSString *oldConnectionId = self.realtime.connection.id;
+    __block ARTEventListener *listener = [self.realtime.connection on:^(ARTConnectionStateChange *stateChange) {
+        if (!(stateChange.current == ARTRealtimeClosed || stateChange.current == ARTRealtimeFailed
+              || (stateChange.current == ARTRealtimeConnected && ![oldConnectionId isEqual:self.realtime.connection.id] /* connection state lost */))) {
+            return;
+        }
+        gotFailure = true;
+        [self.realtime.connection off:listener];
+        if (!cb) return;
+        ARTErrorInfo *reason = stateChange.reason ? stateChange.reason : [ARTErrorInfo createWithCode:0 message:@"connection broken before receiving publishing acknowledgement."];
+        cb([ARTStatus state:ARTStateError info:reason]);
+    }];
+
+    [self.realtime send:pm cb:^(ARTStatus *status) {
+        [self.realtime.connection off:listener];
+        if (cb && !gotFailure) cb(status);
+    }];
+}
+
 - (ARTPresenceMap *) presenceMap {
     return _presenceMap;
 }
@@ -446,7 +468,7 @@ - (void)sendQueuedMessages {
     NSArray *qms = self.queuedMessages;
     self.queuedMessages = [NSMutableArray array];
     for (ARTQueuedMessage *qm in qms) {
-        [self.realtime send:qm.msg cb:qm.cb];
+        [self sendMessage:qm.msg cb:qm.cb];
     }
 }
 

ably-ios/ARTWebSocketTransport.h

@@ -26,6 +26,8 @@ ART_ASSUME_NONNULL_BEGIN
 
 @property (readonly, getter=getIsConnected) BOOL isConnected;
 
+@property (readwrite, assign, nonatomic) BOOL closing;
+
 - (NSURL *)setupWebSocket:(__GENERIC(NSArray, NSURLQueryItem *) *)params withOptions:(ARTClientOptions *)options resumeKey:(NSString *__art_nullable)resumeKey connectionSerial:(NSNumber *__art_nullable)connectionSerial;
 
 - (BOOL)getIsConnected;

ably-ios/ARTWebSocketTransport.m

@@ -41,7 +41,6 @@ @interface ARTWebSocketTransport () <WebSocketDelegate>
 @property (readonly, assign, nonatomic) CFRunLoopRef rl;
 @property (readonly, strong, nonatomic) dispatch_queue_t queue;
 @property (readwrite, strong, nonatomic) WebSocket *websocket;
-@property (readwrite, assign, nonatomic) BOOL closing;
 
 // From RestClient
 @property (readwrite, strong, nonatomic) id<ARTEncoder> encoder;

ablySpec/RealtimeClientConnection.swift

@@ -687,7 +687,7 @@ class RealtimeClientConnection: QuickSpec {
                 }
 
                 // RTN7c
-                pending("should trigger the failure callback for the remaining pending messages if") {
+                context("should trigger the failure callback for the remaining pending messages if") {
 
                     it("connection is closed") {
                         let options = AblyTests.commonAppSetup()
@@ -752,6 +752,7 @@ class RealtimeClientConnection: QuickSpec {
                     it("lost connection state") {
                         let options = AblyTests.commonAppSetup()
                         options.autoConnect = false
+                        options.disconnectedRetryTimeout = 0.1
                         let client = ARTRealtime(options: options)
                         client.setTransportClass(TestProxyTransport.self)
                         client.connect()
@@ -765,22 +766,26 @@ class RealtimeClientConnection: QuickSpec {
                         let transport = client.transport as! TestProxyTransport
                         transport.actionsIgnored += [.Ack, .Nack]
 
-                        waitUntil(timeout: testTimeout + options.disconnectedRetryTimeout) { done in
-                            channel.on { errorInfo in
-                                if channel.state == .Attached {
-                                    channel.publish(nil, data: "message", cb: { errorInfo in
-                                        expect(errorInfo).toNot(beNil())
-                                        done()
-                                    })
-                                    // Wait until the message is pushed to Ably first
-                                    delay(1.0) {
-                                        client.simulateLostConnection()
-                                        expect(client.connection.state).toEventually(equal(ARTRealtimeConnectionState.Connecting), timeout: options.disconnectedRetryTimeout)
-                                    }
-                                }
-                            }
-                            channel.attach()
+                        channel.attach()
+                        expect(channel.state).toEventually(equal(ARTRealtimeChannelState.Attached), timeout: testTimeout)
+
+                        var gotPublishedCallback = false
+                        channel.publish(nil, data: "message", cb: { errorInfo in
+                            expect(errorInfo).toNot(beNil())
+                            gotPublishedCallback = true
+                        })
+
+                        let oldConnectionId = client.connection.id!
+                        // Wait until the message is pushed to Ably first
+                        waitUntil(timeout: testTimeout) { done in
+                            delay(1.0) { done() }
                         }
+
+                        client.simulateLostConnection()
+                        expect(gotPublishedCallback).to(beFalse())
+                        expect(client.connection.state).toEventually(equal(ARTRealtimeConnectionState.Connected), timeout: testTimeout)
+                        expect(client.connection.id).toNot(equal(oldConnectionId))
+                        expect(gotPublishedCallback).to(beTrue())
                     }
 
                 }
@@ -1357,12 +1362,12 @@ class RealtimeClientConnection: QuickSpec {
                                 done()
                             })
                         }
-
+                        
                         expect(client.connection.state).toEventually(equal(ARTRealtimeConnectionState.Connecting), timeout: options.suspendedRetryTimeout)
-
+                        
                         channel.attach()
                         expect(channel.state).toEventually(equal(ARTRealtimeChannelState.Attached), timeout: testTimeout)
-
+                        
                         waitUntil(timeout: testTimeout) { done in
                             // Accept publishing of messages
                             channel.publish(nil, data: "message", cb: { errorInfo in
@@ -1371,7 +1376,7 @@ class RealtimeClientConnection: QuickSpec {
                             })
                         }
                     }
-
+                    
                     it("when a connection enters CLOSED state") {
                         let options = AblyTests.commonAppSetup()
                         let client = ARTRealtime(options: options)
@@ -1381,14 +1386,14 @@ class RealtimeClientConnection: QuickSpec {
                         }
                         let channel = client.channels.get("test")
                         channel.attach()
-
+                        
                         expect(channel.state).toEventually(equal(ARTRealtimeChannelState.Attached), timeout: testTimeout)
-
+                        
                         client.close()
-
+                        
                         expect(client.connection.state).to(equal(ARTRealtimeConnectionState.Closed))
                         expect(channel.state).toEventually(equal(ARTRealtimeChannelState.Detached), timeout: testTimeout)
-
+                        
                         waitUntil(timeout: testTimeout) { done in
                             // Reject publishing of messages
                             channel.publish(nil, data: "message", cb: { errorInfo in
@@ -1398,9 +1403,9 @@ class RealtimeClientConnection: QuickSpec {
                             })
                         }
                     }
-
+                    
                 }
-
+                
                 // RTN18c
                 it("when a connection enters FAILED state, all channels will move to FAILED state") {
                     let options = AblyTests.commonAppSetup()
@@ -1411,14 +1416,14 @@ class RealtimeClientConnection: QuickSpec {
                     }
                     let channel = client.channels.get("test")
                     channel.attach()
-
+                    
                     expect(channel.state).toEventually(equal(ARTRealtimeChannelState.Attached), timeout: testTimeout)
-
+                    
                     client.onError(AblyTests.newErrorProtocolMessage())
-
+                    
                     expect(client.connection.state).to(equal(ARTRealtimeConnectionState.Failed))
                     expect(channel.state).toEventually(equal(ARTRealtimeChannelState.Failed), timeout: testTimeout)
-
+                    
                     waitUntil(timeout: testTimeout) { done in
                         // Reject publishing of messages
                         channel.publish(nil, data: "message", cb: { errorInfo in
@@ -1428,9 +1433,9 @@ class RealtimeClientConnection: QuickSpec {
                         })
                     }
                 }
-
+                
             }
-
+            
         }
     }
 }

ablySpec/TestUtilities.swift

@@ -442,7 +442,7 @@ extension ARTRealtime {
         //2. Change the `Connection#id` and `Connection#key` before the client 
         //   library attempts to reconnect and resume the connection
         self.connection.setId("lost")
-        self.connection.setKey("lost")
+        self.connection.setKey("xxxxx!xxxxxxx-xxxxxxxx-xxxxxxxx")
         self.onDisconnected()
     }
 
@@ -467,6 +467,7 @@ extension ARTWebSocketTransport {
 
     func simulateIncomingNormalClose() {
         let CLOSE_NORMAL = 1000
+        self.closing = true
         let webSocketDelegate = self as! WebSocketDelegate
         webSocketDelegate.webSocketClose(CLOSE_NORMAL, reason: "", wasClean: true)
     }