fix: Subscription failed event should be terminal event

AppSyncRealTimeClient.xcodeproj/project.pbxproj

@@ -10,6 +10,7 @@
 		0B59161BB37D32073E4FD61B /* Pods_AppSyncRTCSample.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = 7CF486070B34EFD15B4DB8FC /* Pods_AppSyncRTCSample.framework */; };
 		2143D46727B5D9B40066B2F7 /* RealTimeConnectionProviderResponseTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 2143D46627B5D9B40066B2F7 /* RealTimeConnectionProviderResponseTests.swift */; };
 		2143D4B027BC49BE0066B2F7 /* AWSAppSyncRealTimeClient.swift in Sources */ = {isa = PBXBuildFile; fileRef = 2143D4AF27BC49BE0066B2F7 /* AWSAppSyncRealTimeClient.swift */; };
+		2143D4B227BE40B30066B2F7 /* AppSyncRealTimeClientFailureTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 2143D4B127BE40B30066B2F7 /* AppSyncRealTimeClientFailureTests.swift */; };
 		2164E65D2639AD5600385027 /* StarscreamAdapterTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 2164E65C2639AD5600385027 /* StarscreamAdapterTests.swift */; };
 		2164E674263C58CE00385027 /* AppSyncRealTimeClientTestBase.swift in Sources */ = {isa = PBXBuildFile; fileRef = 2164E673263C58CD00385027 /* AppSyncRealTimeClientTestBase.swift */; };
 		217F39992405D9D500F1A0B3 /* AppSyncRealTimeClient.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = 217F398F2405D9D500F1A0B3 /* AppSyncRealTimeClient.framework */; };
@@ -125,6 +126,7 @@
 		18D6E56CE03BAC33493CC19B /* Pods-HostApp.debug.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-HostApp.debug.xcconfig"; path = "Target Support Files/Pods-HostApp/Pods-HostApp.debug.xcconfig"; sourceTree = "<group>"; };
 		2143D46627B5D9B40066B2F7 /* RealTimeConnectionProviderResponseTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = RealTimeConnectionProviderResponseTests.swift; sourceTree = "<group>"; };
 		2143D4AF27BC49BE0066B2F7 /* AWSAppSyncRealTimeClient.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = AWSAppSyncRealTimeClient.swift; sourceTree = "<group>"; };
+		2143D4B127BE40B30066B2F7 /* AppSyncRealTimeClientFailureTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = AppSyncRealTimeClientFailureTests.swift; sourceTree = "<group>"; };
 		2164E65C2639AD5600385027 /* StarscreamAdapterTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = StarscreamAdapterTests.swift; sourceTree = "<group>"; };
 		2164E673263C58CD00385027 /* AppSyncRealTimeClientTestBase.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = AppSyncRealTimeClientTestBase.swift; sourceTree = "<group>"; };
 		217F398F2405D9D500F1A0B3 /* AppSyncRealTimeClient.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; includeInIndex = 0; path = AppSyncRealTimeClient.framework; sourceTree = BUILT_PRODUCTS_DIR; };
@@ -455,6 +457,7 @@
 			isa = PBXGroup;
 			children = (
 				21D38B4B2409B6C000EC2A8D /* amplifyconfiguration.json */,
+				2143D4B127BE40B30066B2F7 /* AppSyncRealTimeClientFailureTests.swift */,
 				21D38B402409AFBD00EC2A8D /* AppSyncRealTimeClientIntegrationTests.swift */,
 				2164E673263C58CD00385027 /* AppSyncRealTimeClientTestBase.swift */,
 				21D38B422409AFBD00EC2A8D /* Info.plist */,
@@ -1176,6 +1179,7 @@
 				21D38B99240C4E1C00EC2A8D /* ConfigurationHelper.swift in Sources */,
 				2164E674263C58CE00385027 /* AppSyncRealTimeClientTestBase.swift in Sources */,
 				21D38B97240C4DCF00EC2A8D /* Error+Extension.swift in Sources */,
+				2143D4B227BE40B30066B2F7 /* AppSyncRealTimeClientFailureTests.swift in Sources */,
 				21D38B9D240C540D00EC2A8D /* TestCommonConstants.swift in Sources */,
 				21D38B412409AFBD00EC2A8D /* AppSyncRealTimeClientIntegrationTests.swift in Sources */,
 				2164E65D2639AD5600385027 /* StarscreamAdapterTests.swift in Sources */,

AppSyncRealTimeClient/Connection/AppSyncConnection/AppSyncSubscriptionConnection+Connection.swift

@@ -31,7 +31,7 @@ extension AppSyncSubscriptionConnection {
         else {
             return
         }
-        AppSyncLogger.debug("[AppSyncSubscriptionConnection] \(#function): connection is connected, start subscription.")
+        AppSyncLogger.debug("[AppSyncSubscriptionConnection]: Connection connected, start subscription \(subscriptionItem.identifier).")
         subscriptionState = .inProgress
 
         guard let payload = convertToPayload(

AppSyncRealTimeClient/Connection/AppSyncConnection/AppSyncSubscriptionConnection+ErrorHandler.swift

@@ -32,6 +32,7 @@ extension AppSyncSubscriptionConnection {
             let connectionError = error as? ConnectionProviderError
         else {
             subscriptionItem.subscriptionEventHandler(.failed(error), subscriptionItem)
+            connectionProvider?.removeListener(identifier: subscriptionItem.identifier)
             return
         }
 
@@ -43,6 +44,7 @@ extension AppSyncSubscriptionConnection {
             }
         } else {
             subscriptionItem.subscriptionEventHandler(.failed(error), subscriptionItem)
+            connectionProvider?.removeListener(identifier: subscriptionItem.identifier)
         }
     }
 

AppSyncRealTimeClient/Connection/AppSyncConnection/AppSyncSubscriptionConnection.swift

@@ -81,7 +81,7 @@ public class AppSyncSubscriptionConnection: SubscriptionConnection, RetryableCon
 
         connectionProvider.addListener(identifier: subscriptionItem.identifier) { [weak self] event in
             guard let self = self else {
-                AppSyncLogger.debug("[AppSyncSubscriptionConnection] \(#function): Self is nil, listener is not called.")
+                AppSyncLogger.debug("[AppSyncSubscriptionConnection]: Subscription (Self) is nil, connection event is not handled.")
                 return
             }
             switch event {

AppSyncRealTimeClient/ConnectionProvider/AppsyncRealtimeConnection/RealtimeConnectionProvider+StaleConnection.swift

@@ -21,7 +21,7 @@ extension RealtimeConnectionProvider {
 
     /// Reset the stale connection timer in response to receiving a message from the websocket
     func resetStaleConnectionTimer(interval: TimeInterval? = nil) {
-        AppSyncLogger.debug("[RealtimeConnectionProvider] Resetting stale connection timer")
+        AppSyncLogger.verbose("[RealtimeConnectionProvider] Resetting stale connection timer")
         staleConnectionTimer.reset(interval: interval)
     }
 

AppSyncRealTimeClient/ConnectionProvider/AppsyncRealtimeConnection/RealtimeConnectionProvider+Websocket.swift

@@ -53,7 +53,7 @@ extension RealtimeConnectionProvider: AppSyncWebsocketDelegate {
                 self?.handleConnectionAck(response: response)
             }
         case .error:
-            AppSyncLogger.debug("[RealtimeConnectionProvider] received error")
+            AppSyncLogger.verbose("[RealtimeConnectionProvider] received error")
             connectionQueue.async { [weak self] in
                 self?.handleError(response: response)
             }
@@ -62,7 +62,7 @@ extension RealtimeConnectionProvider: AppSyncWebsocketDelegate {
                 updateCallback(event: .data(appSyncResponse))
             }
         case .keepAlive:
-            AppSyncLogger.debug("[RealtimeConnectionProvider] received keepAlive")
+            AppSyncLogger.verbose("[RealtimeConnectionProvider] received keepAlive")
         }
     }
 

AppSyncRealTimeClientIntegrationTests/AppSyncRealTimeClientFailureTests.swift

@@ -0,0 +1,190 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import XCTest
+@testable import AppSyncRealTimeClient
+
+class AppSyncRealTimeClientFailureTests: AppSyncRealTimeClientTestBase {
+
+    /// Test the current AppSync limit of 100 subscriptions per connection
+    func testMaxSubscriptionReached() {
+        let subscribeSuccess = expectation(description: "subscribe successfully")
+        subscribeSuccess.expectedFulfillmentCount = 100
+        let authInterceptor = APIKeyAuthInterceptor(apiKey)
+        let connectionProvider = ConnectionProviderFactory.createConnectionProvider(
+            for: url,
+            authInterceptor: authInterceptor,
+            connectionType: .appSyncRealtime
+        )
+        var subscriptions = [AppSyncSubscriptionConnection]()
+        for _ in 1 ... 100 {
+            let subscription = AppSyncSubscriptionConnection(provider: connectionProvider)
+            _ = subscription.subscribe(
+                requestString: requestString,
+                variables: nil
+            ) { event, _ in
+                switch event {
+                case .connection(let subscriptionConnectionEvent):
+                    switch subscriptionConnectionEvent {
+                    case .connecting:
+                        break
+                    case .connected:
+                        subscribeSuccess.fulfill()
+                    case .disconnected:
+                        break
+                    }
+                case .data(let data):
+                    print("Got data back \(data)")
+                case .failed(let error):
+                    XCTFail("Got error \(error)")
+                }
+            }
+            subscriptions.append(subscription)
+        }
+
+        wait(for: [subscribeSuccess], timeout: TestCommonConstants.networkTimeout)
+        XCTAssertEqual(subscriptions.count, 100)
+        let limitExceeded = expectation(description: "Received Limit Exceeded error")
+        let subscription = AppSyncSubscriptionConnection(provider: connectionProvider)
+        _ = subscription.subscribe(
+            requestString: requestString,
+            variables: nil
+        ) { event, _ in
+            switch event {
+            case .connection(let subscriptionConnectionEvent):
+                switch subscriptionConnectionEvent {
+                case .connecting:
+                    break
+                case .connected:
+                    XCTFail("Got connected successfully - Should have been limit exceeded")
+                case .disconnected:
+                    break
+                }
+            case .data(let data):
+                print("Got data back \(data)")
+            case .failed(let error):
+                guard let connectionError = error as? ConnectionProviderError,
+                      case .limitExceeded = connectionError else {
+                          XCTFail("Should Be Limited Exceeded error")
+                          return
+                }
+
+                limitExceeded.fulfill()
+            }
+        }
+        wait(for: [limitExceeded], timeout: TestCommonConstants.networkTimeout)
+
+        for subscription in subscriptions {
+            if let item = subscription.subscriptionItem {
+                subscription.unsubscribe(item: item)
+            }
+        }
+    }
+
+    /// Subscriptions receiving a failed event should only receive it once.
+    func testMaxSubscriptionReachedWithRetry() {
+        let subscribeSuccess = expectation(description: "subscribe successfully")
+        subscribeSuccess.expectedFulfillmentCount = 100
+        let authInterceptor = APIKeyAuthInterceptor(apiKey)
+        let connectionProvider = ConnectionProviderFactory.createConnectionProvider(
+            for: url,
+            authInterceptor: authInterceptor,
+            connectionType: .appSyncRealtime
+        )
+        var subscriptions = [AppSyncSubscriptionConnection]()
+        for _ in 1 ... 100 {
+            let subscription = AppSyncSubscriptionConnection(provider: connectionProvider)
+            _ = subscription.subscribe(
+                requestString: requestString,
+                variables: nil
+            ) { event, _ in
+                switch event {
+                case .connection(let subscriptionConnectionEvent):
+                    switch subscriptionConnectionEvent {
+                    case .connecting:
+                        break
+                    case .connected:
+                        subscribeSuccess.fulfill()
+                    case .disconnected:
+                        break
+                    }
+                case .data(let data):
+                    print("Got data back \(data)")
+                case .failed(let error):
+                    XCTFail("Got error \(error)")
+                }
+            }
+            subscriptions.append(subscription)
+        }
+
+        wait(for: [subscribeSuccess], timeout: TestCommonConstants.networkTimeout)
+        XCTAssertEqual(subscriptions.count, 100)
+        let limitExceeded = expectation(description: "Received Limit Exceeded error")
+        limitExceeded.expectedFulfillmentCount = 2
+        for _ in 1 ... 2 {
+            let subscription = AppSyncSubscriptionConnection(provider: connectionProvider)
+            subscription.addRetryHandler(handler: TestConnectionRetryHandler())
+            _ = subscription.subscribe(
+                requestString: requestString,
+                variables: nil
+            ) { event, _ in
+                switch event {
+                case .connection(let subscriptionConnectionEvent):
+                    switch subscriptionConnectionEvent {
+                    case .connecting:
+                        break
+                    case .connected:
+                        XCTFail("Got connected successfully - Should have been limit exceeded")
+                    case .disconnected:
+                        break
+                    }
+                case .data(let data):
+                    print("Got data back \(data)")
+                case .failed(let error):
+                    guard let connectionError = error as? ConnectionProviderError,
+                          case .limitExceeded = connectionError else {
+                              XCTFail("Should Be Limited Exceeded error")
+                              return
+                    }
+
+                    limitExceeded.fulfill()
+                }
+            }
+            subscriptions.append(subscription)
+        }
+
+        wait(for: [limitExceeded], timeout: TestCommonConstants.networkTimeout)
+
+        for subscription in subscriptions {
+            if let item = subscription.subscriptionItem {
+                subscription.unsubscribe(item: item)
+            }
+        }
+    }
+
+    class TestConnectionRetryHandler: ConnectionRetryHandler {
+        var count: Int = 0
+        func shouldRetryRequest(for error: ConnectionProviderError) -> RetryAdvice {
+            if count > 10 {
+                return TestRetryAdvice(shouldRetry: false)
+            }
+
+            if case .limitExceeded = error {
+                self.count += 1
+                return TestRetryAdvice(shouldRetry: true, retryInterval: .seconds(1))
+            }
+            return TestRetryAdvice(shouldRetry: false)
+
+        }
+    }
+
+    struct TestRetryAdvice: RetryAdvice {
+        var shouldRetry: Bool
+
+        var retryInterval: DispatchTimeInterval?
+    }
+}

AppSyncRealTimeClientIntegrationTests/AppSyncRealTimeClientTestBase.swift

@@ -23,7 +23,7 @@ class AppSyncRealTimeClientTestBase: XCTestCase {
         """
 
     override func setUp() {
-        AppSyncRealTimeClient.logLevel = .verbose
+        AppSyncRealTimeClient.logLevel = .debug
         do {
             let json = try ConfigurationHelper.retrieve(forResource: "amplifyconfiguration")
             if let data = json as? [String: Any],

AppSyncRealTimeClientTests/Connection/AppSyncSubscriptionConnectionTests.swift

@@ -59,6 +59,7 @@ class AppSyncSubscriptionConnectionTests: XCTestCase {
         }
         XCTAssertNotNil(item, "Subscription item should not be nil")
         wait(for: [connectingMessageExpectation, connectedMessageExpectation], timeout: 5, enforceOrder: true)
+        XCTAssertNotNil(connectionProvider.listener)
     }
 
     /// Test unsubscribe subscription gives us back the right events
@@ -101,8 +102,10 @@ class AppSyncSubscriptionConnectionTests: XCTestCase {
         XCTAssertNotNil(item, "Subscription item should not be nil")
         wait(for: [connectingMessageExpectation, connectedMessageExpectation], timeout: 5, enforceOrder: true)
 
+        XCTAssertNotNil(connectionProvider.listener)
         connection.unsubscribe(item: item)
         wait(for: [unsubscribeAckExpectation], timeout: 2)
+        XCTAssertNil(connectionProvider.listener)
     }
 
     /// Test subscription with invalid connection
@@ -141,6 +144,7 @@ class AppSyncSubscriptionConnectionTests: XCTestCase {
         }
         XCTAssertNotNil(item, "Subscription item should not be nil")
         wait(for: [connectingMessageExpectation, errorEventExpectation], timeout: 5, enforceOrder: true)
+        XCTAssertNil(connectionProvider.listener)
     }
 
     /// Test if trying to subscribe with a 'not connected' connection gives error
@@ -181,6 +185,7 @@ class AppSyncSubscriptionConnectionTests: XCTestCase {
         }
         XCTAssertNotNil(item, "Subscription item should not be nil")
         wait(for: [connectingMessageExpectation, errorEventExpectation], timeout: 5, enforceOrder: true)
+        XCTAssertNil(connectionProvider.listener)
     }
 
     /// Test if valid data is returned
@@ -230,6 +235,7 @@ class AppSyncSubscriptionConnectionTests: XCTestCase {
         )
         connectionProvider.sendDataResponse(mockResponse)
         wait(for: [dataEventExpectation], timeout: 2)
+        XCTAssertNotNil(connectionProvider.listener)
     }
 
     func testNilDataInVariables() {
@@ -261,6 +267,7 @@ class AppSyncSubscriptionConnectionTests: XCTestCase {
         }
         XCTAssertNotNil(item, "Subscription item should not be nil")
         wait(for: [connectingMessageExpectation, connectedMessageExpectation], timeout: 5, enforceOrder: true)
+        XCTAssertNotNil(connectionProvider.listener)
     }
 
 }