feat: add swift concurrency (async/await) support for async interceptors (#85)

Co-authored-by: Ian Saultz <52051793+atierian@users.noreply.github.com>

.swiftformat

@@ -13,10 +13,8 @@
   --header "//\n// Copyright Amazon.com Inc. or its affiliates.\n// All Rights Reserved.\n//\n// SPDX-License-Identifier: Apache-2.0\n//"
 
 --disable hoistPatternLet
-  --patternlet inline
 
 --disable indent
-  --ifdef outdent
   --indent 4
   --indentcase false
   --xcodeindentation disabled
@@ -45,7 +43,6 @@
   --semicolons never
 
 --disable sortedImports
-  --importgrouping testable-bottom
 
 --enable spaceAroundOperators
   --operatorfunc spaced
@@ -54,13 +51,11 @@
   --trailingclosures 
 
 --disable trailingCommas
-  --commas inline
 
 --enable trailingSpace
   --trimwhitespace always
 
 --disable unusedArguments
-  --stripunusedargs closure-only
 
 --enable void
   --empty void

AppSyncRealTimeClient.xcodeproj/project.pbxproj

@@ -76,6 +76,17 @@
 		259F9EB83B9F67D0413A6D4C /* Pods_AppSyncRealTimeClient.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = 81AFCBED5A97E398A4BD9D06 /* Pods_AppSyncRealTimeClient.framework */; };
 		450019BB151D701382536BD0 /* Pods_HostApp.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = 29CDD85F44666C7241232D29 /* Pods_HostApp.framework */; };
 		4A3EAC9B20D96D81CC3A7EF4 /* Pods_HostApp_AppSyncRealTimeClientIntegrationTests.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = 43D112B03D6D38F84995D6FA /* Pods_HostApp_AppSyncRealTimeClientIntegrationTests.framework */; };
+		5C5609EE2821E6FC0002ACF5 /* InterceptableConnectionAsync.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5C5609ED2821E6FC0002ACF5 /* InterceptableConnectionAsync.swift */; };
+		5CF3E96B283C33D40036EAD2 /* AppSyncRealTimeClientAsyncFailureTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CF3E968283C33D40036EAD2 /* AppSyncRealTimeClientAsyncFailureTests.swift */; };
+		5CF3E96D283C33D40036EAD2 /* AppSyncRealTimeClientAsyncIntegrationTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CF3E96A283C33D40036EAD2 /* AppSyncRealTimeClientAsyncIntegrationTests.swift */; };
+		5CFF7233283C1971001D5471 /* RealtimeConnectionProviderAsync.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CFF722E283C1971001D5471 /* RealtimeConnectionProviderAsync.swift */; };
+		5CFF7234283C1971001D5471 /* RealtimeConnectionProviderAsync+StaleConnection.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CFF722F283C1971001D5471 /* RealtimeConnectionProviderAsync+StaleConnection.swift */; };
+		5CFF7235283C1971001D5471 /* RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CFF7230283C1971001D5471 /* RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift */; };
+		5CFF7236283C1971001D5471 /* RealtimeConnectionProviderAsync+Websocket.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CFF7231283C1971001D5471 /* RealtimeConnectionProviderAsync+Websocket.swift */; };
+		5CFF7237283C1971001D5471 /* RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CFF7232283C1971001D5471 /* RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift */; };
+		5CFF7239283C19CF001D5471 /* TaskQueue.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CFF7238283C19CF001D5471 /* TaskQueue.swift */; };
+		5CFF723D283C1AF5001D5471 /* ConnectionProviderAsyncTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CFF723B283C1AF5001D5471 /* ConnectionProviderAsyncTests.swift */; };
+		5CFF723E283C1AF5001D5471 /* RealtimeConnectionProviderAsyncTestBase.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CFF723C283C1AF5001D5471 /* RealtimeConnectionProviderAsyncTestBase.swift */; };
 		978409B82739C7BE002362A7 /* AppSyncURLHelper.swift in Sources */ = {isa = PBXBuildFile; fileRef = 978409B72739C7BE002362A7 /* AppSyncURLHelper.swift */; };
 		978409BA2739C7E1002362A7 /* AppSyncURLHelperTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 978409B92739C7E1002362A7 /* AppSyncURLHelperTests.swift */; };
 		DCFE701B5D1380E566694A48 /* Pods_AppSyncRealTimeClient_AppSyncRealTimeClientTests.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = 3E662A46AB2C93EE316F784C /* Pods_AppSyncRealTimeClient_AppSyncRealTimeClientTests.framework */; };
@@ -207,6 +218,17 @@
 		356411189EAD2C776D250FB7 /* Pods-HostApp.release.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-HostApp.release.xcconfig"; path = "Target Support Files/Pods-HostApp/Pods-HostApp.release.xcconfig"; sourceTree = "<group>"; };
 		3E662A46AB2C93EE316F784C /* Pods_AppSyncRealTimeClient_AppSyncRealTimeClientTests.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; includeInIndex = 0; path = Pods_AppSyncRealTimeClient_AppSyncRealTimeClientTests.framework; sourceTree = BUILT_PRODUCTS_DIR; };
 		43D112B03D6D38F84995D6FA /* Pods_HostApp_AppSyncRealTimeClientIntegrationTests.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; includeInIndex = 0; path = Pods_HostApp_AppSyncRealTimeClientIntegrationTests.framework; sourceTree = BUILT_PRODUCTS_DIR; };
+		5C5609ED2821E6FC0002ACF5 /* InterceptableConnectionAsync.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = InterceptableConnectionAsync.swift; sourceTree = "<group>"; };
+		5CF3E968283C33D40036EAD2 /* AppSyncRealTimeClientAsyncFailureTests.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = AppSyncRealTimeClientAsyncFailureTests.swift; sourceTree = "<group>"; };
+		5CF3E96A283C33D40036EAD2 /* AppSyncRealTimeClientAsyncIntegrationTests.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = AppSyncRealTimeClientAsyncIntegrationTests.swift; sourceTree = "<group>"; };
+		5CFF722E283C1971001D5471 /* RealtimeConnectionProviderAsync.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = RealtimeConnectionProviderAsync.swift; sourceTree = "<group>"; };
+		5CFF722F283C1971001D5471 /* RealtimeConnectionProviderAsync+StaleConnection.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProviderAsync+StaleConnection.swift"; sourceTree = "<group>"; };
+		5CFF7230283C1971001D5471 /* RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift"; sourceTree = "<group>"; };
+		5CFF7231283C1971001D5471 /* RealtimeConnectionProviderAsync+Websocket.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProviderAsync+Websocket.swift"; sourceTree = "<group>"; };
+		5CFF7232283C1971001D5471 /* RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift"; sourceTree = "<group>"; };
+		5CFF7238283C19CF001D5471 /* TaskQueue.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = TaskQueue.swift; sourceTree = "<group>"; };
+		5CFF723B283C1AF5001D5471 /* ConnectionProviderAsyncTests.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = ConnectionProviderAsyncTests.swift; sourceTree = "<group>"; };
+		5CFF723C283C1AF5001D5471 /* RealtimeConnectionProviderAsyncTestBase.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = RealtimeConnectionProviderAsyncTestBase.swift; sourceTree = "<group>"; };
 		7CF486070B34EFD15B4DB8FC /* Pods_AppSyncRTCSample.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; includeInIndex = 0; path = Pods_AppSyncRTCSample.framework; sourceTree = BUILT_PRODUCTS_DIR; };
 		7D4F451B830A5837CE5274A3 /* Pods-AppSyncRealTimeClient-AppSyncRealTimeClientTests.release.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-AppSyncRealTimeClient-AppSyncRealTimeClientTests.release.xcconfig"; path = "Target Support Files/Pods-AppSyncRealTimeClient-AppSyncRealTimeClientTests/Pods-AppSyncRealTimeClient-AppSyncRealTimeClientTests.release.xcconfig"; sourceTree = "<group>"; };
 		81AFCBED5A97E398A4BD9D06 /* Pods_AppSyncRealTimeClient.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; includeInIndex = 0; path = Pods_AppSyncRealTimeClient.framework; sourceTree = BUILT_PRODUCTS_DIR; };
@@ -353,12 +375,14 @@
 				217F39AF2406E98300F1A0B3 /* AppSyncConnectionRequest.swift */,
 				217F39AE2406E98300F1A0B3 /* AppSyncMessage.swift */,
 				217F39AD2406E98300F1A0B3 /* AppSyncMessage+Encodable.swift */,
+				217F39B02406E98300F1A0B3 /* AppsyncRealtimeConnection */,
+				5CFF722D283C1971001D5471 /* AppsyncRealtimeConnectionAsync */,
 				217F39AA2406E98300F1A0B3 /* AppSyncResponse.swift */,
 				217F39B72406E98300F1A0B3 /* ConnectionProvider.swift */,
 				217F39AC2406E98300F1A0B3 /* ConnectionProviderError.swift */,
 				21D38B8D240A3C2300EC2A8D /* ConnectionProviderFactory.swift */,
 				217F39AB2406E98300F1A0B3 /* InterceptableConnection.swift */,
-				217F39B02406E98300F1A0B3 /* AppsyncRealtimeConnection */,
+				5C5609ED2821E6FC0002ACF5 /* InterceptableConnectionAsync.swift */,
 			);
 			path = ConnectionProvider;
 			sourceTree = "<group>";
@@ -419,15 +443,16 @@
 		217F39C62406E98400F1A0B3 /* Support */ = {
 			isa = PBXGroup;
 			children = (
-				978409B72739C7BE002362A7 /* AppSyncURLHelper.swift */,
 				21D38B6C240A262800EC2A8D /* AppSyncJSONHelper.swift */,
 				217F39C92406E98400F1A0B3 /* AppSyncJSONValue.swift */,
 				217F39C72406E98400F1A0B3 /* AppSyncLogger.swift */,
+				978409B72739C7BE002362A7 /* AppSyncURLHelper.swift */,
 				FA2EFABB2550CAC6007698C7 /* AtomicValue.swift */,
 				FA67508124D33A7A005A1345 /* CountdownTimer.swift */,
 				21D38B93240C4A2A00EC2A8D /* OIDCAuthProvider.swift */,
 				217F39CB2406E98400F1A0B3 /* SubscriptionConnectionType.swift */,
 				217F39CA2406E98400F1A0B3 /* SubscriptionConstants.swift */,
+				5CFF7238283C19CF001D5471 /* TaskQueue.swift */,
 			);
 			path = Support;
 			sourceTree = "<group>";
@@ -454,6 +479,7 @@
 		217F39EC2406EA4000F1A0B3 /* ConnectionProvider */ = {
 			isa = PBXGroup;
 			children = (
+				5CFF723A283C1AA0001D5471 /* ConnectionProviderAsync */,
 				2143D46527B5D9730066B2F7 /* AppSyncRealTimeConnection */,
 				2151D5C927C68C3C00F3C866 /* ConnectionProviderHandleErrorTests.swift */,
 				FA67507F24D339B0005A1345 /* ConnectionProviderStaleConnectionTests.swift */,
@@ -492,6 +518,7 @@
 		21D38B3F2409AFBD00EC2A8D /* AppSyncRealTimeClientIntegrationTests */ = {
 			isa = PBXGroup;
 			children = (
+				5CF3E967283C33D40036EAD2 /* AppSyncRealTimeClientAsync */,
 				21D38B4B2409B6C000EC2A8D /* amplifyconfiguration.json */,
 				2143D4B127BE40B30066B2F7 /* AppSyncRealTimeClientFailureTests.swift */,
 				21D38B402409AFBD00EC2A8D /* AppSyncRealTimeClientIntegrationTests.swift */,
@@ -558,6 +585,36 @@
 			path = Support;
 			sourceTree = "<group>";
 		};
+		5CF3E967283C33D40036EAD2 /* AppSyncRealTimeClientAsync */ = {
+			isa = PBXGroup;
+			children = (
+				5CF3E968283C33D40036EAD2 /* AppSyncRealTimeClientAsyncFailureTests.swift */,
+				5CF3E96A283C33D40036EAD2 /* AppSyncRealTimeClientAsyncIntegrationTests.swift */,
+			);
+			path = AppSyncRealTimeClientAsync;
+			sourceTree = "<group>";
+		};
+		5CFF722D283C1971001D5471 /* AppsyncRealtimeConnectionAsync */ = {
+			isa = PBXGroup;
+			children = (
+				5CFF722E283C1971001D5471 /* RealtimeConnectionProviderAsync.swift */,
+				5CFF7230283C1971001D5471 /* RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift */,
+				5CFF7232283C1971001D5471 /* RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift */,
+				5CFF722F283C1971001D5471 /* RealtimeConnectionProviderAsync+StaleConnection.swift */,
+				5CFF7231283C1971001D5471 /* RealtimeConnectionProviderAsync+Websocket.swift */,
+			);
+			path = AppsyncRealtimeConnectionAsync;
+			sourceTree = "<group>";
+		};
+		5CFF723A283C1AA0001D5471 /* ConnectionProviderAsync */ = {
+			isa = PBXGroup;
+			children = (
+				5CFF723B283C1AF5001D5471 /* ConnectionProviderAsyncTests.swift */,
+				5CFF723C283C1AF5001D5471 /* RealtimeConnectionProviderAsyncTestBase.swift */,
+			);
+			path = ConnectionProviderAsync;
+			sourceTree = "<group>";
+		};
 		CC47BA7DC6033B1626BAD959 /* Frameworks */ = {
 			isa = PBXGroup;
 			children = (
@@ -1150,6 +1207,7 @@
 			buildActionMask = 2147483647;
 			files = (
 				21D38B7C240A2A1300EC2A8D /* OIDCAuthInterceptor.swift in Sources */,
+				5CFF7235283C1971001D5471 /* RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift in Sources */,
 				217F39DB2406E98400F1A0B3 /* AppSyncSubscriptionConnection+DataHandler.swift in Sources */,
 				21D38B94240C4A2A00EC2A8D /* OIDCAuthProvider.swift in Sources */,
 				217F39DF2406E98400F1A0B3 /* SubscriptionItem.swift in Sources */,
@@ -1162,6 +1220,7 @@
 				FAB7E91224D2644E00DF1EA1 /* RealtimeConnectionProvider+StaleConnection.swift in Sources */,
 				217F39D32406E98400F1A0B3 /* RealtimeConnectionProvider.swift in Sources */,
 				217F39D12406E98400F1A0B3 /* AppSyncConnectionRequest.swift in Sources */,
+				5C5609EE2821E6FC0002ACF5 /* InterceptableConnectionAsync.swift in Sources */,
 				217F39DD2406E98400F1A0B3 /* AppSyncSubscriptionConnection+Connection.swift in Sources */,
 				217F39E52406E98400F1A0B3 /* AppSyncJSONValue.swift in Sources */,
 				217F39E42406E98400F1A0B3 /* RealtimeGatewayURLInterceptor.swift in Sources */,
@@ -1169,8 +1228,10 @@
 				217F39E22406E98400F1A0B3 /* StarscreamAdapter+Delegate.swift in Sources */,
 				217F39D52406E98400F1A0B3 /* RealtimeConnectionProviderResponse.swift in Sources */,
 				217F39D62406E98400F1A0B3 /* RealtimeConnectionProvider+MessageInterceptable.swift in Sources */,
+				5CFF7233283C1971001D5471 /* RealtimeConnectionProviderAsync.swift in Sources */,
 				FA67508224D33A7A005A1345 /* CountdownTimer.swift in Sources */,
 				217F39E62406E98400F1A0B3 /* SubscriptionConstants.swift in Sources */,
+				5CFF7234283C1971001D5471 /* RealtimeConnectionProviderAsync+StaleConnection.swift in Sources */,
 				217F39D02406E98400F1A0B3 /* AppSyncMessage.swift in Sources */,
 				217F39E32406E98400F1A0B3 /* AppSyncLogger.swift in Sources */,
 				217F39D82406E98400F1A0B3 /* ConnectionProvider.swift in Sources */,
@@ -1179,15 +1240,18 @@
 				217F39DA2406E98400F1A0B3 /* RetryableConnection.swift in Sources */,
 				978409B82739C7BE002362A7 /* AppSyncURLHelper.swift in Sources */,
 				217F39CC2406E98400F1A0B3 /* AppSyncResponse.swift in Sources */,
+				5CFF7239283C19CF001D5471 /* TaskQueue.swift in Sources */,
 				217F39E72406E98400F1A0B3 /* SubscriptionConnectionType.swift in Sources */,
 				217F39D42406E98400F1A0B3 /* RealtimeConnectionProvider+Websocket.swift in Sources */,
 				217F39DC2406E98400F1A0B3 /* AppSyncSubscriptionConnection.swift in Sources */,
+				5CFF7236283C1971001D5471 /* RealtimeConnectionProviderAsync+Websocket.swift in Sources */,
 				21D38B6D240A262800EC2A8D /* AppSyncJSONHelper.swift in Sources */,
 				219BFF4B27A3B24F000FC148 /* ConnectivityPath.swift in Sources */,
 				217F39CE2406E98400F1A0B3 /* ConnectionProviderError.swift in Sources */,
 				217F39E12406E98400F1A0B3 /* StarscreamAdapter.swift in Sources */,
 				217F39D72406E98400F1A0B3 /* RealtimeConnectionProvider+ConnectionInterceptable.swift in Sources */,
 				217F39DE2406E98400F1A0B3 /* AppSyncSubscriptionConnection+ErrorHandler.swift in Sources */,
+				5CFF7237283C1971001D5471 /* RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift in Sources */,
 			);
 			runOnlyForDeploymentPostprocessing = 0;
 		};
@@ -1203,6 +1267,8 @@
 				217F39F02406EA4000F1A0B3 /* AppSyncSubscriptionConnectionTests.swift in Sources */,
 				FA67507B24D338CB005A1345 /* Error+Extension.swift in Sources */,
 				21D38B8C240A39E400EC2A8D /* APIKeyAuthInterceptorTests.swift in Sources */,
+				5CFF723E283C1AF5001D5471 /* RealtimeConnectionProviderAsyncTestBase.swift in Sources */,
+				5CFF723D283C1AF5001D5471 /* ConnectionProviderAsyncTests.swift in Sources */,
 				21D38B8B240A39E400EC2A8D /* AppSyncJSONHelperTests.swift in Sources */,
 				217F39F32406EA4000F1A0B3 /* RealtimeGatewayURLInterceptorTests.swift in Sources */,
 				217C74D227C45B6600AE054F /* AppSyncSubscriptionConnectionErrorHandlerTests.swift in Sources */,
@@ -1221,10 +1287,12 @@
 			buildActionMask = 2147483647;
 			files = (
 				21D38B99240C4E1C00EC2A8D /* ConfigurationHelper.swift in Sources */,
+				5CF3E96D283C33D40036EAD2 /* AppSyncRealTimeClientAsyncIntegrationTests.swift in Sources */,
 				2164E674263C58CE00385027 /* AppSyncRealTimeClientTestBase.swift in Sources */,
 				21D38B97240C4DCF00EC2A8D /* Error+Extension.swift in Sources */,
 				2143D4B227BE40B30066B2F7 /* AppSyncRealTimeClientFailureTests.swift in Sources */,
 				21D38B9D240C540D00EC2A8D /* TestCommonConstants.swift in Sources */,
+				5CF3E96B283C33D40036EAD2 /* AppSyncRealTimeClientAsyncFailureTests.swift in Sources */,
 				21D38B412409AFBD00EC2A8D /* AppSyncRealTimeClientIntegrationTests.swift in Sources */,
 				2164E65D2639AD5600385027 /* StarscreamAdapterTests.swift in Sources */,
 			);
@@ -1347,7 +1415,7 @@
 				GCC_WARN_UNINITIALIZED_AUTOS = YES_AGGRESSIVE;
 				GCC_WARN_UNUSED_FUNCTION = YES;
 				GCC_WARN_UNUSED_VARIABLE = YES;
-				IPHONEOS_DEPLOYMENT_TARGET = 13.2;
+				IPHONEOS_DEPLOYMENT_TARGET = 9.0;
 				MTL_ENABLE_DEBUG_INFO = INCLUDE_SOURCE;
 				MTL_FAST_MATH = YES;
 				ONLY_ACTIVE_ARCH = YES;
@@ -1404,7 +1472,7 @@
 				GCC_WARN_UNINITIALIZED_AUTOS = YES_AGGRESSIVE;
 				GCC_WARN_UNUSED_FUNCTION = YES;
 				GCC_WARN_UNUSED_VARIABLE = YES;
-				IPHONEOS_DEPLOYMENT_TARGET = 13.2;
+				IPHONEOS_DEPLOYMENT_TARGET = 9.0;
 				MTL_ENABLE_DEBUG_INFO = NO;
 				MTL_FAST_MATH = YES;
 				SDKROOT = iphoneos;

AppSyncRealTimeClient/Connection/AppSyncConnection/AppSyncSubscriptionConnection+Connection.swift

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

AppSyncRealTimeClient/Connection/AppSyncConnection/AppSyncSubscriptionConnection+ErrorHandler.swift

@@ -57,6 +57,7 @@ extension AppSyncSubscriptionConnection {
 
         let retryAdvice = retryHandler.shouldRetryRequest(for: connectionError)
         if retryAdvice.shouldRetry, let retryInterval = retryAdvice.retryInterval {
+            // swiftlint:disable:next line_length
             AppSyncLogger.debug("[AppSyncSubscriptionConnection] Retrying subscription \(subscriptionItem.identifier) after \(retryInterval)")
             DispatchQueue.global().asyncAfter(deadline: .now() + retryInterval) {
                 self.connectionProvider?.connect()

AppSyncRealTimeClient/Connection/AppSyncConnection/AppSyncSubscriptionConnection.swift

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

AppSyncRealTimeClient/ConnectionProvider/AppsyncRealtimeConnection/RealtimeConnectionProvider+StaleConnection.swift

@@ -12,7 +12,9 @@ extension RealtimeConnectionProvider {
 
     /// Start a stale connection timer, first invalidating and destroying any existing timer
     func startStaleConnectionTimer() {
-        AppSyncLogger.debug("[RealtimeConnectionProvider] Starting stale connection timer for \(staleConnectionTimer.interval)s")
+        AppSyncLogger.debug(
+            "[RealtimeConnectionProvider] Starting stale connection timer for \(staleConnectionTimer.interval)s"
+        )
 
         staleConnectionTimer.start(interval: RealtimeConnectionProvider.staleConnectionTimeout) {
             self.disconnectStaleConnection()
@@ -36,13 +38,19 @@ extension RealtimeConnectionProvider {
             guard let self = self else {
                 return
             }
-            AppSyncLogger.debug("[RealtimeConnectionProvider] Status: \(self.status). Connectivity status: \(connectivity.status)")
+            AppSyncLogger.debug(
+                "[RealtimeConnectionProvider] Status: \(self.status). Connectivity status: \(connectivity.status)"
+            )
             if self.status == .connected && connectivity.status == .unsatisfied && !self.isStaleConnection {
-                AppSyncLogger.debug("[RealtimeConnectionProvider] Connetion is stale. Pending reconnect on connectivity.")
+                AppSyncLogger.debug(
+                    "[RealtimeConnectionProvider] Connetion is stale. Pending reconnect on connectivity."
+                )
                 self.isStaleConnection = true
 
             } else if self.status == .connected && self.isStaleConnection && connectivity.status == .satisfied {
-                AppSyncLogger.debug("[RealtimeConnectionProvider] Connetion is stale. Disconnecting to begin reconnect.")
+                AppSyncLogger.debug(
+                    "[RealtimeConnectionProvider] Connetion is stale. Disconnecting to begin reconnect."
+                )
                 self.staleConnectionTimer.invalidate()
                 self.disconnectStaleConnection()
             }

AppSyncRealTimeClient/ConnectionProvider/AppsyncRealtimeConnection/RealtimeConnectionProvider.swift

@@ -55,7 +55,7 @@ public class RealtimeConnectionProvider: ConnectionProvider {
         if iLimitExceededSubject == nil {
             iLimitExceededSubject = PassthroughSubject<ConnectionProviderError, Never>()
         }
-        return iLimitExceededSubject as! PassthroughSubject<ConnectionProviderError, Never> // swiftlint:disable:this force_cast
+        return iLimitExceededSubject as! PassthroughSubject<ConnectionProviderError, Never> // swiftlint:disable:this force_cast line_length
     }
 
     public convenience init(for url: URL, websocket: AppSyncWebsocketProvider) {
@@ -169,7 +169,9 @@ public class RealtimeConnectionProvider: ConnectionProvider {
             self.listeners.removeValue(forKey: identifier)
 
             if self.listeners.isEmpty {
-                AppSyncLogger.debug("[RealtimeConnectionProvider] all subscriptions removed, disconnecting websocket connection.")
+                AppSyncLogger.debug(
+                    "[RealtimeConnectionProvider] all subscriptions removed, disconnecting websocket connection."
+                )
                 self.status = .notConnected
                 self.websocket.disconnect()
                 self.invalidateStaleConnectionTimer()

AppSyncRealTimeClient/ConnectionProvider/AppsyncRealtimeConnectionAsync/RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift

@@ -0,0 +1,32 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+#if swift(>=5.5.2)
+
+import Foundation
+
+@available(iOS 13.0.0, *)
+extension RealtimeConnectionProviderAsync: ConnectionInterceptableAsync {
+
+    public func addInterceptor(_ interceptor: ConnectionInterceptorAsync) {
+        connectionInterceptors.append(interceptor)
+    }
+
+    public func interceptConnection(
+        _ request: AppSyncConnectionRequest,
+        for endpoint: URL
+    ) async -> AppSyncConnectionRequest {
+        var finalRequest = request
+        for interceptor in connectionInterceptors {
+            finalRequest = await interceptor.interceptConnection(finalRequest, for: endpoint)
+        }
+
+        return finalRequest
+    }
+}
+
+#endif

AppSyncRealTimeClient/ConnectionProvider/AppsyncRealtimeConnectionAsync/RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift

@@ -0,0 +1,28 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+#if swift(>=5.5.2)
+
+import Foundation
+
+@available(iOS 13.0.0, *)
+extension RealtimeConnectionProviderAsync: MessageInterceptableAsync {
+    public func addInterceptor(_ interceptor: MessageInterceptorAsync) {
+        messageInterceptors.append(interceptor)
+    }
+
+    public func interceptMessage(_ message: AppSyncMessage, for endpoint: URL) async -> AppSyncMessage {
+        var finalMessage = message
+        for interceptor in messageInterceptors {
+            finalMessage = await interceptor.interceptMessage(finalMessage, for: endpoint)
+        }
+
+        return finalMessage
+    }
+}
+
+#endif

AppSyncRealTimeClient/ConnectionProvider/AppsyncRealtimeConnectionAsync/RealtimeConnectionProviderAsync+StaleConnection.swift

@@ -0,0 +1,77 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+#if swift(>=5.5.2)
+
+import Foundation
+
+/// Consolidates usage and parameters passed to the `staleConnectionTimer` methods.
+@available(iOS 13.0, *)
+extension RealtimeConnectionProviderAsync {
+
+    /// Start a stale connection timer, first invalidating and destroying any existing timer
+    func startStaleConnectionTimer() {
+        AppSyncLogger.debug(
+            "[RealtimeConnectionProvider] Starting stale connection timer for \(staleConnectionTimer.interval)s"
+        )
+
+        staleConnectionTimer.start(interval: RealtimeConnectionProviderAsync.staleConnectionTimeout) {
+            self.disconnectStaleConnection()
+        }
+    }
+
+    /// Reset the stale connection timer in response to receiving a message from the websocket
+    func resetStaleConnectionTimer(interval: TimeInterval? = nil) {
+        AppSyncLogger.verbose("[RealtimeConnectionProvider] Resetting stale connection timer")
+        staleConnectionTimer.reset(interval: interval)
+    }
+
+    /// Stops the timer when disconnecting the websocket.
+    func invalidateStaleConnectionTimer() {
+        staleConnectionTimer.invalidate()
+    }
+
+    /// Handle updates from the ConnectivityMonitor
+    func handleConnectivityUpdates(connectivity: ConnectivityPath) {
+        taskQueue.async { [weak self] in
+            guard let self = self else {
+                return
+            }
+            AppSyncLogger.debug(
+                "[RealtimeConnectionProvider] Status: \(self.status). Connectivity status: \(connectivity.status)"
+            )
+            if self.status == .connected && connectivity.status == .unsatisfied && !self.isStaleConnection {
+                AppSyncLogger.debug(
+                    "[RealtimeConnectionProvider] Connetion is stale. Pending reconnect on connectivity."
+                )
+                self.isStaleConnection = true
+
+            } else if self.status == .connected && self.isStaleConnection && connectivity.status == .satisfied {
+                AppSyncLogger.debug(
+                    "[RealtimeConnectionProvider] Connetion is stale. Disconnecting to begin reconnect."
+                )
+                self.staleConnectionTimer.invalidate()
+                self.disconnectStaleConnection()
+            }
+        }
+    }
+
+    /// Fired when the stale connection timer expires
+    private func disconnectStaleConnection() {
+        taskQueue.async { [weak self] in
+            guard let self = self else {
+                return
+            }
+            AppSyncLogger.error("[RealtimeConnectionProvider] Realtime connection is stale, disconnecting.")
+            self.status = .notConnected
+            self.isStaleConnection = false
+            self.websocket.disconnect()
+            self.updateCallback(event: .error(ConnectionProviderError.connection))
+        }
+    }
+}
+#endif

AppSyncRealTimeClient/ConnectionProvider/AppsyncRealtimeConnectionAsync/RealtimeConnectionProviderAsync+Websocket.swift

@@ -0,0 +1,149 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+#if swift(>=5.5.2)
+
+import Foundation
+
+@available(iOS 13.0, *)
+extension RealtimeConnectionProviderAsync: AppSyncWebsocketDelegate {
+
+    public func websocketDidConnect(provider: AppSyncWebsocketProvider) {
+        // Call the ack to finish the connection handshake
+        // Inform the callback when ack gives back a response.
+        AppSyncLogger.debug("[RealtimeConnectionProvider] WebsocketDidConnect, sending init message")
+        sendConnectionInitMessage()
+        startStaleConnectionTimer()
+    }
+
+    public func websocketDidDisconnect(provider: AppSyncWebsocketProvider, error: Error?) {
+        taskQueue.async { [weak self] in
+            guard let self = self else {
+                return
+            }
+            self.status = .notConnected
+            guard error != nil else {
+                self.updateCallback(event: .connection(self.status))
+                return
+            }
+            self.updateCallback(event: .error(ConnectionProviderError.connection))
+        }
+    }
+
+    public func websocketDidReceiveData(provider: AppSyncWebsocketProvider, data: Data) {
+        do {
+            let response = try JSONDecoder().decode(RealtimeConnectionProviderResponse.self, from: data)
+            handleResponse(response)
+        } catch {
+            AppSyncLogger.error(error)
+            updateCallback(event: .error(ConnectionProviderError.jsonParse(nil, error)))
+        }
+    }
+
+    // MARK: - Handle websocket response
+
+    private func handleResponse(_ response: RealtimeConnectionProviderResponse) {
+        resetStaleConnectionTimer()
+
+        switch response.responseType {
+        case .connectionAck:
+            AppSyncLogger.debug("[RealtimeConnectionProvider] received connectionAck")
+            taskQueue.async { [weak self] in
+                self?.handleConnectionAck(response: response)
+            }
+        case .error:
+            AppSyncLogger.verbose("[RealtimeConnectionProvider] received error")
+            taskQueue.async { [weak self] in
+                self?.handleError(response: response)
+            }
+        case .subscriptionAck, .unsubscriptionAck, .data:
+            if let appSyncResponse = response.toAppSyncResponse() {
+                updateCallback(event: .data(appSyncResponse))
+            }
+        case .keepAlive:
+            AppSyncLogger.verbose("[RealtimeConnectionProvider] received keepAlive")
+        }
+    }
+
+    /// Updates connection status callbacks and sets stale connection timeout
+    ///
+    /// - Warning: This method must be invoked on the `connectionQueue`
+    private func handleConnectionAck(response: RealtimeConnectionProviderResponse) {
+        // Only from in progress state, the connection can transition to connected state.
+        // The below guard statement make sure that. If we get connectionAck in other
+        // state means that we have initiated a disconnect parallely.
+        guard status == .inProgress else {
+            return
+        }
+
+        status = .connected
+        updateCallback(event: .connection(status))
+
+        // If the service returns a connection timeout, use that instead of the default
+        guard case let .number(value) = response.payload?["connectionTimeoutMs"] else {
+            return
+        }
+
+        let interval = value / 1_000
+
+        guard interval != staleConnectionTimer.interval else {
+            return
+        }
+
+        AppSyncLogger.debug(
+            """
+            Resetting keep alive timer in response to service timeout \
+            instructions: \(interval)s
+            """
+        )
+        resetStaleConnectionTimer(interval: interval)
+    }
+
+    /// Resolves & dispatches errors from `response`.
+    ///
+    /// - Warning: This method must be invoked on the `connectionQueue`
+    func handleError(response: RealtimeConnectionProviderResponse) {
+        // If we get an error in connection inprogress state, return back as connection error.
+        guard status != .inProgress else {
+            status = .notConnected
+            updateCallback(event: .error(ConnectionProviderError.connection))
+            return
+        }
+
+        if response.isLimitExceededError() {
+            let limitExceedError = ConnectionProviderError.limitExceeded(response.id)
+
+            guard response.id == nil else {
+                updateCallback(event: .error(limitExceedError))
+                return
+            }
+
+            limitExceededSubject.send(limitExceedError)
+            return
+        }
+
+        if response.isMaxSubscriptionReachedError() {
+            let limitExceedError = ConnectionProviderError.limitExceeded(response.id)
+            updateCallback(event: .error(limitExceedError))
+            return
+        }
+
+        // If the type of error is not handled (by checking `isLimitExceededError`, `isMaxSubscriptionReachedError`,
+        // etc), and is not for a specific subscription, then return a generic error
+        guard let identifier = response.id else {
+            let genericError = ConnectionProviderError.other
+            updateCallback(event: .error(genericError))
+            return
+        }
+
+        // Default scenario - return the error with subscription id and error payload.
+        let subscriptionError = ConnectionProviderError.subscription(identifier, response.payload)
+        updateCallback(event: .error(subscriptionError))
+    }
+
+}
+#endif

AppSyncRealTimeClient/ConnectionProvider/AppsyncRealtimeConnectionAsync/RealtimeConnectionProviderAsync.swift

@@ -0,0 +1,204 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+#if swift(>=5.5.2)
+
+import Foundation
+import Combine
+
+/// Appsync Real time connection that connects to subscriptions
+/// through websocket.
+@available(iOS 13.0, *)
+public class RealtimeConnectionProviderAsync: ConnectionProvider {
+    /// Maximum number of seconds a connection may go without receiving a keep alive
+    /// message before we consider it stale and force a disconnect
+    static let staleConnectionTimeout: TimeInterval = 5 * 60
+
+    let url: URL
+    var listeners: [String: ConnectionProviderCallback]
+
+    let websocket: AppSyncWebsocketProvider
+
+    var status: ConnectionState
+
+    var messageInterceptors = [MessageInterceptorAsync]()
+    var connectionInterceptors = [ConnectionInterceptorAsync]()
+
+    /// A timer that automatically disconnects the current connection if it goes longer
+    /// than `staleConnectionTimeout` without activity. Receiving any data or "keep
+    /// alive" message will cause the timer to be reset to the full interval.
+    var staleConnectionTimer: CountdownTimer
+
+    /// Intermediate state when the connection is connected and connectivity updates to unsatisfied (offline)
+    var isStaleConnection: Bool
+
+    /// Manages concurrency for socket connections, disconnections, writes, and status reports.
+    ///
+    /// Each connection request will be sent to this queue. Connection request are
+    /// handled one at a time.
+    let taskQueue = TaskQueue<Void>()
+
+    /// Monitor for connectivity updates
+    let connectivityMonitor: ConnectivityMonitor
+
+    /// The serial queue on which status & message callbacks from the web socket are invoked.
+    private let serialCallbackQueue: DispatchQueue
+
+    /// Throttle when AppSync sends LimitExceeded error. High rate of subscriptions requests will cause AppSync to send
+    /// connection level LimitExceeded errors for each subscribe made. A connection level error means that there is no
+    /// subscription id associated with the error. When handling these errors, all subscriptions will receive a message
+    /// for the error. Use this subject to send and throttle the errors on the client side.
+    var limitExceededThrottleSink: Any?
+    var iLimitExceededSubject: Any?
+    var limitExceededSubject: PassthroughSubject<ConnectionProviderError, Never> {
+        if iLimitExceededSubject == nil {
+            iLimitExceededSubject = PassthroughSubject<ConnectionProviderError, Never>()
+        }
+        // swiftlint:disable:next force_cast
+        return iLimitExceededSubject as! PassthroughSubject<ConnectionProviderError, Never>
+    }
+
+    init(
+        url: URL,
+        websocket: AppSyncWebsocketProvider,
+
+        serialCallbackQueue: DispatchQueue = DispatchQueue(
+            label: "com.amazonaws.AppSyncRealTimeConnectionProvider.callbackQueue"
+        ),
+        connectivityMonitor: ConnectivityMonitor = ConnectivityMonitor()
+    ) {
+        self.url = url
+        self.websocket = websocket
+        self.listeners = [:]
+        self.status = .notConnected
+        self.staleConnectionTimer = CountdownTimer()
+        self.isStaleConnection = false
+        self.serialCallbackQueue = serialCallbackQueue
+        self.connectivityMonitor = connectivityMonitor
+
+        connectivityMonitor.start(onUpdates: handleConnectivityUpdates(connectivity:))
+    }
+
+    public convenience init(for url: URL, websocket: AppSyncWebsocketProvider) {
+        self.init(url: url, websocket: websocket)
+    }
+
+    // MARK: - ConnectionProvider methods
+
+    public func connect() {
+        taskQueue.async { [weak self] in
+            guard let self = self else {
+                return
+            }
+            guard self.status == .notConnected else {
+                self.updateCallback(event: .connection(self.status))
+                return
+            }
+            self.status = .inProgress
+            self.updateCallback(event: .connection(self.status))
+            let request = AppSyncConnectionRequest(url: self.url)
+
+            let signedRequest = await self.interceptConnection(request, for: self.url)
+            self.websocket.connect(
+                url: signedRequest.url,
+                protocols: ["graphql-ws"],
+                delegate: self
+            )
+        }
+    }
+
+    public func write(_ message: AppSyncMessage) {
+        taskQueue.async { [weak self] in
+            guard let self = self else {
+                return
+            }
+
+            let signedMessage = await self.interceptMessage(message, for: self.url)
+
+            let jsonEncoder = JSONEncoder()
+            do {
+                let jsonData = try jsonEncoder.encode(signedMessage)
+                guard let jsonString = String(data: jsonData, encoding: .utf8) else {
+                    let jsonError = ConnectionProviderError.jsonParse(signedMessage.id, nil)
+                    self.updateCallback(event: .error(jsonError))
+                    return
+                }
+                self.websocket.write(message: jsonString)
+            } catch {
+                AppSyncLogger.error(error)
+                switch signedMessage.messageType {
+                case .connectionInit:
+                    self.receivedConnectionInit()
+                default:
+                    self.updateCallback(event: .error(ConnectionProviderError.jsonParse(signedMessage.id, error)))
+                }
+            }
+        }
+    }
+
+    public func disconnect() {
+        taskQueue.async {
+            self.websocket.disconnect()
+            self.invalidateStaleConnectionTimer()
+        }
+    }
+
+    public func addListener(identifier: String, callback: @escaping ConnectionProviderCallback) {
+        taskQueue.async { [weak self] in
+            self?.listeners[identifier] = callback
+        }
+    }
+
+    public func removeListener(identifier: String) {
+        taskQueue.async { [weak self] in
+            guard let self = self else {
+                return
+            }
+
+            self.listeners.removeValue(forKey: identifier)
+
+            if self.listeners.isEmpty {
+                AppSyncLogger.debug(
+                    "[RealtimeConnectionProvider] all subscriptions removed, disconnecting websocket connection."
+                )
+                self.status = .notConnected
+                self.websocket.disconnect()
+                self.invalidateStaleConnectionTimer()
+            }
+        }
+    }
+
+    // MARK: -
+
+    func sendConnectionInitMessage() {
+        let message = AppSyncMessage(type: .connectionInit("connection_init"))
+        write(message)
+    }
+
+    /// Invokes all registered listeners with `event`. The event is dispatched on `serialCallbackQueue`,
+    /// but internally this method uses the connectionQueue to get the currently registered listeners.
+    ///
+    /// - Parameter event: The connection event to dispatch
+    func updateCallback(event: ConnectionProviderEvent) {
+        taskQueue.async { [weak self] in
+            guard let self = self else {
+                return
+            }
+            let allListeners = Array(self.listeners.values)
+            self.serialCallbackQueue.async {
+                allListeners.forEach { $0(event) }
+            }
+        }
+    }
+
+    /// - Warning: This must be invoked from the `taskQueue`
+    private func receivedConnectionInit() {
+        status = .notConnected
+        updateCallback(event: .error(ConnectionProviderError.connection))
+    }
+}
+#endif

AppSyncRealTimeClient/ConnectionProvider/ConnectionProviderFactory.swift

@@ -15,7 +15,13 @@ public enum ConnectionProviderFactory {
         authInterceptor: AuthInterceptor,
         connectionType: SubscriptionConnectionType
     ) -> ConnectionProvider {
-        let provider = ConnectionProviderFactory.createConnectionProvider(for: url, connectionType: connectionType)
+        let provider: ConnectionProvider
+
+        switch connectionType {
+        case .appSyncRealtime:
+            let websocketProvider = StarscreamAdapter()
+            provider = RealtimeConnectionProvider(for: url, websocket: websocketProvider)
+        }
 
         if let messageInterceptable = provider as? MessageInterceptable {
             messageInterceptable.addInterceptor(authInterceptor)
@@ -28,15 +34,30 @@ public enum ConnectionProviderFactory {
         return provider
     }
 
-    static func createConnectionProvider(
+    #if swift(>=5.5.2)
+    @available(iOS 13.0.0, *)
+    public static func createConnectionProviderAsync(
         for url: URL,
+        authInterceptor: AuthInterceptorAsync,
         connectionType: SubscriptionConnectionType
     ) -> ConnectionProvider {
+        let provider: ConnectionProvider
+
         switch connectionType {
         case .appSyncRealtime:
             let websocketProvider = StarscreamAdapter()
-            let connectionProvider = RealtimeConnectionProvider(for: url, websocket: websocketProvider)
+            provider = RealtimeConnectionProviderAsync(for: url, websocket: websocketProvider)
-            return connectionProvider
         }
+
+        if let messageInterceptable = provider as? MessageInterceptableAsync {
+            messageInterceptable.addInterceptor(authInterceptor)
         }
+        if let connectionInterceptable = provider as? ConnectionInterceptableAsync {
+            connectionInterceptable.addInterceptor(RealtimeGatewayURLInterceptor())
+            connectionInterceptable.addInterceptor(authInterceptor)
+        }
+
+        return provider
+    }
+    #endif
 }

AppSyncRealTimeClient/ConnectionProvider/InterceptableConnectionAsync.swift

@@ -0,0 +1,46 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Foundation
+
+@available(iOS 13.0.0, *)
+public protocol ConnectionInterceptableAsync {
+    #if swift(>=5.5.2)
+    /// Add a new interceptor to the object.
+    ///
+    /// - Parameter interceptor: interceptor to be added
+    func addInterceptor(_ interceptor: ConnectionInterceptorAsync)
+
+    func interceptConnection(_ request: AppSyncConnectionRequest, for endpoint: URL) async -> AppSyncConnectionRequest
+    #endif
+}
+
+@available(iOS 13.0.0, *)
+public protocol MessageInterceptableAsync {
+    #if swift(>=5.5.2)
+    func addInterceptor(_ interceptor: MessageInterceptorAsync)
+
+    func interceptMessage(_ message: AppSyncMessage, for endpoint: URL) async -> AppSyncMessage
+    #endif
+}
+
+@available(iOS 13.0.0, *)
+public protocol ConnectionInterceptorAsync {
+    #if swift(>=5.5.2)
+    func interceptConnection(_ request: AppSyncConnectionRequest, for endpoint: URL) async -> AppSyncConnectionRequest
+    #endif
+}
+
+@available(iOS 13.0.0, *)
+public protocol MessageInterceptorAsync {
+    #if swift(>=5.5.2)
+    func interceptMessage(_ message: AppSyncMessage, for endpoint: URL) async -> AppSyncMessage
+    #endif
+}
+
+@available(iOS 13.0.0, *)
+public protocol AuthInterceptorAsync: MessageInterceptorAsync, ConnectionInterceptorAsync {}

AppSyncRealTimeClient/Interceptor/APIKeyAuthInterceptor.swift

@@ -8,7 +8,7 @@
 import Foundation
 
 /// Auth interceptor for API Key based authentication
-public class APIKeyAuthInterceptor: AuthInterceptor {
+public class APIKeyAuthInterceptor: AuthInterceptor, AuthInterceptorAsync {
 
     let apiKey: String
 

AppSyncRealTimeClient/Interceptor/OIDCAuthInterceptor.swift

@@ -7,7 +7,7 @@
 
 import Foundation
 
-public class OIDCAuthInterceptor: AuthInterceptor {
+public class OIDCAuthInterceptor: AuthInterceptor, AuthInterceptorAsync {
 
     let authProvider: OIDCAuthProvider
 

AppSyncRealTimeClient/Interceptor/RealtimeGatewayURLInterceptor.swift

@@ -8,7 +8,7 @@
 import Foundation
 
 /// Connection interceptor for real time connection provider
-public class RealtimeGatewayURLInterceptor: ConnectionInterceptor {
+public class RealtimeGatewayURLInterceptor: ConnectionInterceptor, ConnectionInterceptorAsync {
     public init() {
         // Do nothing
     }

AppSyncRealTimeClient/Support/AppSyncJSONHelper.swift

@@ -13,7 +13,9 @@ public enum AppSyncJSONHelper {
         let jsonEncoder = JSONEncoder()
         do {
             let jsonHeader = try jsonEncoder.encode(header)
-            AppSyncLogger.verbose("Generated Header for request - \(String(describing: String(data: jsonHeader, encoding: .utf8)))")
+            AppSyncLogger.verbose(
+                "Generated Header for request - \(String(describing: String(data: jsonHeader, encoding: .utf8)))"
+            )
             return jsonHeader.base64EncodedString()
         } catch {
             AppSyncLogger.error(error.localizedDescription)

AppSyncRealTimeClient/Support/TaskQueue.swift

@@ -0,0 +1,30 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+#if swift(>=5.5.2)
+
+import Foundation
+
+@available(iOS 13.0, *)
+actor TaskQueue<Success> {
+    private var previousTask: Task<Success, Error>?
+
+    func sync(block: @Sendable @escaping () async throws -> Success) rethrows {
+        previousTask = Task { [previousTask] in
+            _ = await previousTask?.result
+            return try await block()
+        }
+    }
+
+    nonisolated func async(block: @Sendable @escaping () async throws -> Success) rethrows {
+        Task {
+            try await sync(block: block)
+        }
+    }
+}
+
+#endif

AppSyncRealTimeClient/Websocket/Starscream/StarscreamAdapter+Delegate.swift

@@ -45,7 +45,9 @@ extension StarscreamAdapter: Starscream.WebSocketDelegate {
     }
 
     private func websocketDidDisconnect(socket: WebSocketClient, error: Error?) {
-        AppSyncLogger.verbose("[StarscreamAdapter] websocketDidDisconnect: \(error?.localizedDescription ?? "No error")")
+        AppSyncLogger.verbose(
+            "[StarscreamAdapter] websocketDidDisconnect: \(error?.localizedDescription ?? "No error")"
+        )
         serialQueue.async {
             self._isConnected = false
             self.delegate?.websocketDidDisconnect(provider: self, error: error)

AppSyncRealTimeClientIntegrationTests/AppSyncRealTimeClientAsync/AppSyncRealTimeClientAsyncFailureTests.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 AppSyncRealTimeClientAsyncFailureTests: AppSyncRealTimeClientTestBase {
+
+    /// Test the current AppSync limit of 100 subscriptions per connection
+    func testMaxSubscriptionReached() { // swiftlint:disable:this cyclomatic_complexity
+        let subscribeSuccess = expectation(description: "subscribe successfully")
+        subscribeSuccess.expectedFulfillmentCount = 100
+        let authInterceptor = APIKeyAuthInterceptor(apiKey)
+        let connectionProvider = ConnectionProviderFactory.createConnectionProviderAsync(
+            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() { // swiftlint:disable:this cyclomatic_complexity
+        let subscribeSuccess = expectation(description: "subscribe successfully")
+        subscribeSuccess.expectedFulfillmentCount = 100
+        let authInterceptor = APIKeyAuthInterceptor(apiKey)
+        let connectionProvider = ConnectionProviderFactory.createConnectionProviderAsync(
+            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/AppSyncRealTimeClientAsync/AppSyncRealTimeClientAsyncIntegrationTests.swift

@@ -0,0 +1,262 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import XCTest
+@testable import AppSyncRealTimeClient
+
+class AppSyncRealTimeClientAsyncIntegrationTests: AppSyncRealTimeClientTestBase { // swiftlint:disable:this type_name
+
+    /// Simple integration test against an AppSync service provisioned with a simple
+    /// Todo model generated by the GraphQL Transform on the `model` directive.
+    ///
+    /// - Given: A subscription connection on an AppSync endpoint with Todo model provisioned
+    /// - When:
+    ///    - Subscribe to the `onCreateTodo`
+    /// - Then:
+    ///    - Webosocket connection and subscription connection is established.
+    ///
+    func testSubscribeWithSubscriptionConnection() {
+        let subscribeSuccess = expectation(description: "subscribe successfully")
+        let authInterceptor = APIKeyAuthInterceptor(apiKey)
+        let connectionProvider = ConnectionProviderFactory.createConnectionProviderAsync(
+            for: url,
+            authInterceptor: authInterceptor,
+            connectionType: .appSyncRealtime
+        )
+        let subscriptionConnection = AppSyncSubscriptionConnection(provider: connectionProvider)
+        _ = subscriptionConnection.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)")
+            }
+        }
+
+        wait(for: [subscribeSuccess], timeout: TestCommonConstants.networkTimeout)
+    }
+
+    /// The purpose of this test is to ensure that all websockets can be successfully
+    /// created, exercised and terminated while keeping a single connection provider in
+    /// memory.
+    ///
+    /// Specifically, the following test exercises the following:
+    /// 1. Create a new connection provider
+    /// 2. Create multiple subscriptions
+    /// 3. Unsubscribe the subscriptions
+    /// 4. Ensure the socket is disconnected
+    /// 5. Repeat Steps 2-4 with the existing connection provider.
+    ///
+    /// - Given: Connected subscriptions
+    /// - When:
+    ///    - All subscription items are unsubscribed
+    /// - Then:
+    ///    - Underlying websocket is disconnected
+    func testAllSubscriptionsCancelledShouldDisconnectTheWebsocket2() {
+        let connectedInvoked = expectation(description: "Connection established")
+        connectedInvoked.expectedFulfillmentCount = 3
+
+        let authInterceptor = APIKeyAuthInterceptor(apiKey)
+        let connectionProvider = ConnectionProviderFactory.createConnectionProviderAsync(
+            for: url,
+            authInterceptor: authInterceptor,
+            connectionType: .appSyncRealtime
+        )
+        let subscriptionConnection1 = AppSyncSubscriptionConnection(provider: connectionProvider)
+        let item1 = subscriptionConnection1.subscribe(
+            requestString: requestString,
+            variables: nil
+        ) { event, _ in
+            if case let .connection(state) = event {
+                if case .connected = state {
+                    connectedInvoked.fulfill()
+                }
+            }
+        }
+        let subscriptionConnection2 = AppSyncSubscriptionConnection(provider: connectionProvider)
+        let item2 = subscriptionConnection2.subscribe(
+            requestString: requestString,
+            variables: nil
+        ) { event, _ in
+            if case let .connection(state) = event {
+                if case .connected = state {
+                    connectedInvoked.fulfill()
+                }
+            }
+        }
+        let subscriptionConnection3 = AppSyncSubscriptionConnection(provider: connectionProvider)
+        let item3 = subscriptionConnection3.subscribe(
+            requestString: requestString,
+            variables: nil
+        ) { event, _ in
+            if case let .connection(state) = event {
+                if case .connected = state {
+                    connectedInvoked.fulfill()
+                }
+            }
+        }
+
+        XCTAssertNotNil(item1)
+        XCTAssertNotNil(item2)
+        XCTAssertNotNil(item3)
+        wait(for: [connectedInvoked], timeout: TestCommonConstants.networkTimeout)
+
+        guard let realTimeConnectionProvider = connectionProvider as? RealtimeConnectionProviderAsync else {
+            XCTFail("Could not retrieve concrete connection provider")
+            return
+        }
+
+        assertStatus(of: realTimeConnectionProvider, equals: .connected)
+
+        subscriptionConnection1.unsubscribe(item: item1)
+        assertStatus(of: realTimeConnectionProvider, equals: .connected)
+        subscriptionConnection2.unsubscribe(item: item2)
+        assertStatus(of: realTimeConnectionProvider, equals: .connected)
+        subscriptionConnection3.unsubscribe(item: item3)
+
+        assertStatus(of: realTimeConnectionProvider, equals: .notConnected)
+
+        let newConnectedInvoked = expectation(description: "Connection established")
+        let subscriptionConnection4 = AppSyncSubscriptionConnection(provider: connectionProvider)
+        let newItem = subscriptionConnection4.subscribe(
+            requestString: requestString,
+            variables: nil
+        ) { event, _ in
+            if case let .connection(state) = event {
+                if case .connected = state {
+                    newConnectedInvoked.fulfill()
+                }
+            }
+        }
+        wait(for: [newConnectedInvoked], timeout: TestCommonConstants.networkTimeout)
+        assertStatus(of: realTimeConnectionProvider, equals: .connected)
+        subscriptionConnection4.unsubscribe(item: newItem)
+        sleep(5)
+        assertStatus(of: realTimeConnectionProvider, equals: .notConnected)
+    }
+
+    /// The purpose of this test is to ensure that a signifcant number of subscriptions
+    /// can be created on a websocket, then unsubscribed, and repeated.
+    ///
+    /// Specifically, the following test exercises the following:
+    /// 1. Create a new connection provider
+    /// 2. Create multiple subscriptions
+    /// 3. Unsubscribe the subscriptions
+    /// 4. Ensure the socket is disconnected
+    /// 5. Repeat Steps 2-4 with the existing connection provider.
+    ///
+    /// - Given: Connected subscriptions
+    /// - When:
+    ///    - All subscription items are unsubscribed
+    /// - Then:
+    ///    - Underlying websocket is disconnected
+    func testSubscribeUnsubscribeRepeat() {
+        let authInterceptor = APIKeyAuthInterceptor(apiKey)
+        let connectionProvider = ConnectionProviderFactory.createConnectionProviderAsync(
+            for: url,
+            authInterceptor: authInterceptor,
+            connectionType: .appSyncRealtime
+        )
+        guard let realTimeConnectionProvider = connectionProvider as? RealtimeConnectionProviderAsync else {
+            XCTFail("Could not retrieve concrete connection provider")
+            return
+        }
+
+        let count = 30
+        let subscriptions = subscribe(connectionProvider, count: count)
+        assertStatus(of: realTimeConnectionProvider, equals: .connected)
+        for index in 0 ..< count {
+            subscriptions[index].1.unsubscribe(item: subscriptions[index].0)
+        }
+        assertStatus(of: realTimeConnectionProvider, equals: .notConnected)
+        let subscriptions2 = subscribe(connectionProvider, count: count)
+        assertStatus(of: realTimeConnectionProvider, equals: .connected)
+        for index in 0 ..< count {
+            subscriptions2[index].1.unsubscribe(item: subscriptions2[index].0)
+        }
+        assertStatus(of: realTimeConnectionProvider, equals: .notConnected)
+    }
+
+    func testMultipleThreadsSubscribeUnsubscribe() {
+        let authInterceptor = APIKeyAuthInterceptor(apiKey)
+        let connectionProvider = ConnectionProviderFactory.createConnectionProviderAsync(
+            for: url,
+            authInterceptor: authInterceptor,
+            connectionType: .appSyncRealtime
+        )
+        guard let realTimeConnectionProvider = connectionProvider as? RealtimeConnectionProviderAsync else {
+            XCTFail("Could not retrieve concrete connection provider")
+            return
+        }
+        let expectedPerforms = expectation(description: "total performs")
+        expectedPerforms.expectedFulfillmentCount = 1_000
+        DispatchQueue.concurrentPerform(iterations: 1_000) { _ in
+            let subscriptionConnection = AppSyncSubscriptionConnection(provider: connectionProvider)
+            let item = subscriptionConnection.subscribe(
+                requestString: requestString,
+                variables: nil
+            ) { _, _ in }
+            subscriptionConnection.unsubscribe(item: item)
+            expectedPerforms.fulfill()
+        }
+        wait(for: [expectedPerforms], timeout: 1)
+        assertStatus(of: realTimeConnectionProvider, equals: .notConnected)
+    }
+
+    // MARK: - Helpers
+
+    private func subscribe(
+        _ connectionProvider: ConnectionProvider,
+        count: Int
+    ) -> [(SubscriptionItem, AppSyncSubscriptionConnection)] {
+        let connectedInvoked = expectation(description: "Connection established")
+        connectedInvoked.expectedFulfillmentCount = count
+        var subscriptions = [(SubscriptionItem, AppSyncSubscriptionConnection)]()
+
+        for _ in 1 ... count {
+            let subscriptionConnection = AppSyncSubscriptionConnection(provider: connectionProvider)
+            let item = subscriptionConnection.subscribe(
+                requestString: requestString,
+                variables: nil
+            ) { event, _ in
+                if case let .connection(state) = event {
+                    if case .connected = state {
+                        connectedInvoked.fulfill()
+                    }
+                }
+            }
+            subscriptions.append((item, subscriptionConnection))
+        }
+
+        wait(for: [connectedInvoked], timeout: TestCommonConstants.networkTimeout)
+        return subscriptions
+    }
+
+    /// Checks the status of the provider in a thread-safe way. This is only needed for tests; real-world
+    /// call sites wouldn't be able to access the `status` as it has `internal` access.
+    private func assertStatus(
+        of provider: RealtimeConnectionProviderAsync,
+        equals status: ConnectionState
+    ) {
+        provider.taskQueue.async {
+            XCTAssertEqual(provider.status, status)
+        }
+    }
+}

AppSyncRealTimeClientIntegrationTests/AppSyncRealTimeClientFailureTests.swift

@@ -11,7 +11,7 @@ import XCTest
 class AppSyncRealTimeClientFailureTests: AppSyncRealTimeClientTestBase {
 
     /// Test the current AppSync limit of 100 subscriptions per connection
-    func testMaxSubscriptionReached() {
+    func testMaxSubscriptionReached() { // swiftlint:disable:this cyclomatic_complexity
         let subscribeSuccess = expectation(description: "subscribe successfully")
         subscribeSuccess.expectedFulfillmentCount = 100
         let authInterceptor = APIKeyAuthInterceptor(apiKey)
@@ -86,7 +86,7 @@ class AppSyncRealTimeClientFailureTests: AppSyncRealTimeClientTestBase {
     }
 
     /// Subscriptions receiving a failed event should only receive it once.
-    func testMaxSubscriptionReachedWithRetry() {
+    func testMaxSubscriptionReachedWithRetry() { // swiftlint:disable:this cyclomatic_complexity
         let subscribeSuccess = expectation(description: "subscribe successfully")
         subscribeSuccess.expectedFulfillmentCount = 100
         let authInterceptor = APIKeyAuthInterceptor(apiKey)

AppSyncRealTimeClientIntegrationTests/AppSyncRealTimeClientIntegrationTests.swift

@@ -152,7 +152,8 @@ class AppSyncRealTimeClientIntegrationTests: AppSyncRealTimeClientTestBase {
         assertStatus(of: realTimeConnectionProvider, equals: .notConnected)
     }
 
-    /// The purpose of this test is to ensure that a signifcant number of subscriptions can be created on a websocket, then unsubscribed, and repeated.
+    /// The purpose of this test is to ensure that a signifcant number of subscriptions
+    /// can be created on a websocket, then unsubscribed, and repeated.
     ///
     /// Specifically, the following test exercises the following:
     /// 1. Create a new connection provider
@@ -211,7 +212,7 @@ class AppSyncRealTimeClientIntegrationTests: AppSyncRealTimeClientTestBase {
             let item = subscriptionConnection.subscribe(
                 requestString: requestString,
                 variables: nil
-            ) { event, _ in }
+            ) { _, _ in }
             subscriptionConnection.unsubscribe(item: item)
             expectedPerforms.fulfill()
         }

AppSyncRealTimeClientTests/Connection/AppSyncSubscriptionConnectionErrorHandlerTests.swift

@@ -8,6 +8,7 @@
 import XCTest
 @testable import AppSyncRealTimeClient
 
+// swiftlint:disable:next type_name type_body_length
 class AppSyncSubscriptionConnectionErrorHandlerTests: XCTestCase {
 
     let connectionProvider = MockConnectionProvider()
@@ -67,7 +68,7 @@ class AppSyncSubscriptionConnectionErrorHandlerTests: XCTestCase {
             case .failed(let error):
                 guard let connection = error as? ConnectionProviderError,
                       case .subscription(let id, _) = connection,
-                      id != nil else {
+                      !id.isEmpty else {
                           XCTFail("Should be .subscription(item.identifier)")
                           return
                 }

AppSyncRealTimeClientTests/ConnectionProvider/ConnectionProviderAsync/ConnectionProviderAsyncTests.swift

@@ -0,0 +1,228 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import XCTest
+@testable import AppSyncRealTimeClient
+
+class ConnectionProviderAsyncTests: RealtimeConnectionProviderTestBase {
+
+    /// Provider test
+    ///
+    /// Given:
+    /// - A configured subscriber -> provider -> websocket chain
+    /// When:
+    /// - I invoke `provider.connect()`
+    /// - And the websocket properly connects
+    /// Then:
+    /// - The subscriber is notified of the successful connection
+    func testSuccessfulConnection() {
+        receivedNotConnected.isInverted = true
+        receivedError.isInverted = true
+
+        let onConnect: MockWebsocketProvider.OnConnect = { _, _, delegate in
+            self.websocketDelegate = delegate
+            DispatchQueue.global().async {
+                delegate?.websocketDidConnect(provider: self.websocket)
+            }
+        }
+
+        let onDisconnect: MockWebsocketProvider.OnDisconnect = { }
+
+        let onWrite: MockWebsocketProvider.OnWrite = { message in
+            guard RealtimeConnectionProviderTestBase.messageType(of: message, equals: "connection_init") else {
+                XCTFail("Incoming message did not have 'connection_init' type")
+                return
+            }
+
+            self.websocketDelegate.websocketDidReceiveData(
+                provider: self.websocket,
+                data: RealtimeConnectionProviderTestBase.makeConnectionAckMessage()
+            )
+        }
+
+        websocket = MockWebsocketProvider(
+            onConnect: onConnect,
+            onDisconnect: onDisconnect,
+            onWrite: onWrite
+        )
+
+        // Retain the provider so it doesn't release prior to executing callbacks
+        let provider = createProviderAndConnect()
+
+        // Get rid of "written to, but never read" compiler warnings
+        print(provider)
+
+        waitForExpectations(timeout: 0.05)
+    }
+
+    /// Provider add and remove listeners tests
+    ///
+    /// Given:
+    /// - A connected websocket with a listener
+    /// When:
+    /// - remove all listeners
+    /// Then:
+    /// - The listeners are removed and the connection is disconnected
+    func testAddRemoveListeners() {
+        receivedNotConnected.isInverted = true
+        receivedError.isInverted = true
+
+        let onConnect: MockWebsocketProvider.OnConnect = { _, _, delegate in
+            self.websocketDelegate = delegate
+            DispatchQueue.global().async {
+                delegate?.websocketDidConnect(provider: self.websocket)
+            }
+        }
+
+        let receivedDisconnect = expectation(description: "receivedDisconnect")
+        let onDisconnect: MockWebsocketProvider.OnDisconnect = {
+            receivedDisconnect.fulfill()
+        }
+
+        let onWrite: MockWebsocketProvider.OnWrite = { message in
+            guard RealtimeConnectionProviderTestBase.messageType(of: message, equals: "connection_init") else {
+                XCTFail("Incoming message did not have 'connection_init' type")
+                return
+            }
+
+            self.websocketDelegate.websocketDidReceiveData(
+                provider: self.websocket,
+                data: RealtimeConnectionProviderTestBase.makeConnectionAckMessage()
+            )
+        }
+
+        websocket = MockWebsocketProvider(
+            onConnect: onConnect,
+            onDisconnect: onDisconnect,
+            onWrite: onWrite
+        )
+
+        // Retain the provider so it doesn't release prior to executing callbacks
+        let provider = createProviderAndConnect(listeners: ["1", "2", "3", "4"])
+
+        wait(
+            for: [receivedInProgress, receivedConnected, receivedNotConnected, receivedError],
+            timeout: 1
+        )
+
+        XCTAssertFalse(provider.listeners.isEmpty)
+
+        let listenersToRemove = provider.listeners.map { $0.key }
+
+        // Removing all the listeners will disconnect the websocket connection
+        for identifier in listenersToRemove {
+            provider.removeListener(identifier: identifier)
+        }
+
+        // Since removing listeners is asynchronous, we have to wait for the disconnect
+        wait(for: [receivedDisconnect], timeout: 1)
+        XCTAssertTrue(provider.listeners.isEmpty)
+        XCTAssertEqual(provider.status, .notConnected)
+    }
+
+    /// Provider test
+    ///
+    /// Given:
+    /// - A configured subscriber -> provider -> websocket chain
+    /// When:
+    /// - I invoke `provider.connect()`
+    /// - And the websocket reports a connection error
+    /// Then:
+    /// - The subscriber is notified of the unsuccessful connection
+    func testConnectionError() {
+        receivedConnected.isInverted = true
+        receivedNotConnected.isInverted = true
+
+        let onConnect: MockWebsocketProvider.OnConnect = { _, _, delegate in
+            self.websocketDelegate = delegate
+            DispatchQueue.global().async {
+                delegate?.websocketDidConnect(provider: self.websocket)
+            }
+        }
+
+        let onDisconnect: MockWebsocketProvider.OnDisconnect = { }
+
+        let onWrite: MockWebsocketProvider.OnWrite = { message in
+            guard RealtimeConnectionProviderTestBase.messageType(of: message, equals: "connection_init") else {
+                XCTFail("Incoming message did not have 'connection_init' type")
+                return
+            }
+
+            self.websocketDelegate.websocketDidDisconnect(
+                provider: self.websocket,
+                error: "test error"
+            )
+        }
+
+        websocket = MockWebsocketProvider(
+            onConnect: onConnect,
+            onDisconnect: onDisconnect,
+            onWrite: onWrite
+        )
+
+        // Retain the provider so it doesn't release prior to executing callbacks
+        let provider = createProviderAndConnect()
+
+        // Get rid of "written to, but never read" compiler warnings
+        print(provider)
+
+        waitForExpectations(timeout: 0.05)
+    }
+
+    /// Stale connection test
+    ///
+    /// Given:
+    /// - A provider configured with a default stale connection timeout
+    /// When:
+    /// - The service sends a message containing an override timeout value
+    /// Then:
+    /// - The provider updates its stale connection timeout to the service-provided value
+    func testServiceOverridesStaleConnectionTimeout() {
+        receivedNotConnected.isInverted = true
+        receivedError.isInverted = true
+
+        let expectedTimeoutInSeconds = 60.0
+        let timeoutInMilliseconds = Int(expectedTimeoutInSeconds) * 1_000
+
+        let onConnect: MockWebsocketProvider.OnConnect = { _, _, delegate in
+            self.websocketDelegate = delegate
+            DispatchQueue.global().async {
+                delegate?.websocketDidConnect(provider: self.websocket)
+            }
+        }
+
+        let onDisconnect: MockWebsocketProvider.OnDisconnect = { }
+
+        let connectionAckMessage = RealtimeConnectionProviderTestBase
+            .makeConnectionAckMessage(withTimeout: timeoutInMilliseconds)
+        let onWrite: MockWebsocketProvider.OnWrite = { message in
+            guard RealtimeConnectionProviderTestBase.messageType(of: message, equals: "connection_init") else {
+                XCTFail("Incoming message did not have 'connection_init' type")
+                return
+            }
+
+            self.websocketDelegate.websocketDidReceiveData(
+                provider: self.websocket,
+                data: connectionAckMessage
+            )
+        }
+
+        websocket = MockWebsocketProvider(
+            onConnect: onConnect,
+            onDisconnect: onDisconnect,
+            onWrite: onWrite
+        )
+
+        let provider = createProviderAndConnect()
+
+        wait(for: [receivedConnected], timeout: 0.05)
+        XCTAssertEqual(provider.staleConnectionTimer.interval, expectedTimeoutInSeconds)
+
+        waitForExpectations(timeout: 0.05)
+    }
+
+}

AppSyncRealTimeClientTests/ConnectionProvider/ConnectionProviderAsync/RealtimeConnectionProviderAsyncTestBase.swift

@@ -0,0 +1,107 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import XCTest
+@testable import AppSyncRealTimeClient
+
+class RealtimeConnectionProviderAsyncTestBase: XCTestCase {
+
+    let url = URL(string: "https://www.appsyncrealtimeclient.test/")!
+
+    var websocket: MockWebsocketProvider!
+
+    // swiftlint:disable:next weak_delegate
+    var websocketDelegate: AppSyncWebsocketDelegate!
+
+    // Shared test expectations. Set expected fulfillment counts and inversions as
+    // needed in the body of the test.
+    var receivedInProgress: XCTestExpectation!
+    var receivedConnected: XCTestExpectation!
+    var receivedNotConnected: XCTestExpectation!
+    var receivedError: XCTestExpectation!
+
+    override func setUp() {
+        receivedInProgress = expectation(description: "receivedInProgress")
+        receivedConnected = expectation(description: "receivedConnected")
+        receivedNotConnected = expectation(description: "receivedNotConnected")
+        receivedError = expectation(description: "receivedError")
+    }
+
+    // MARK: - Utilities
+
+    /// Creates a RealtimeConnectionProvider, adds a listener that fulfills the shared
+    /// expectations as appropriate, and invokes `connect()`. Returns the provider for
+    /// subsequent testing.
+    ///
+    /// Preconditions:
+    /// - `self.websocket` must be initialized in the mock provider's `onConnect`
+    func createProviderAndConnect(
+        listeners: [String]? = nil,
+        serialCallbackQueue: DispatchQueue = DispatchQueue(
+            label: "com.amazonaws.RealtimeConnectionProviderTestBase.serialCallbackQueue"
+        ),
+        connectivityMonitor: ConnectivityMonitor = ConnectivityMonitor()
+    ) -> RealtimeConnectionProvider {
+        let provider = RealtimeConnectionProvider(
+            url: url,
+            websocket: websocket,
+            serialCallbackQueue: serialCallbackQueue,
+            connectivityMonitor: connectivityMonitor
+        )
+        provider.addListener(identifier: "testListener") { event in
+            switch event {
+            case .connection(let connectionState):
+                switch connectionState {
+                case .inProgress:
+                    self.receivedInProgress.fulfill()
+                case .connected:
+                    self.receivedConnected.fulfill()
+                case .notConnected:
+                    self.receivedNotConnected.fulfill()
+                }
+            case .error:
+                self.receivedError.fulfill()
+            default:
+                break
+            }
+        }
+        if let listeners = listeners {
+            listeners.forEach { identifier in
+                provider.addListener(identifier: identifier) { _ in }
+            }
+        }
+        provider.connect()
+        return provider
+    }
+
+    /// Given a Stringified AppSyncMessage, validates the `type` is equal to `expectedType`
+    /// - Parameter message: a string representation of a websocket message
+    /// - Parameter expectedType: the expected value of the type
+    /// - Returns: type `type` field of the message, if present
+    static func messageType(of message: String, equals expectedType: String) -> Bool {
+        guard
+            let messageData = message.data(using: .utf8),
+            let dict = try? JSONSerialization.jsonObject(with: messageData) as? [String: String]
+            else {
+                return false
+        }
+
+        guard let type = dict["type"] else {
+            return false
+        }
+
+        return type == expectedType
+    }
+
+    /// Creates a connection acknowledgement message with the specified timeout
+    /// - Parameter timeout: stale connection timeout, in milliseconds (defaults to 300,000)
+    static func makeConnectionAckMessage(withTimeout timeout: Int = 300_000) -> Data {
+        #"{"type":"connection_ack","payload":{"connectionTimeoutMs":\#(timeout)}}"#
+            .data(using: .utf8)!
+    }
+
+}

AppSyncRealTimeClientTests/ConnectionProvider/RealtimeConnectionProviderTestBase.swift

@@ -41,8 +41,12 @@ class RealtimeConnectionProviderTestBase: XCTestCase {
     /// - `self.websocket` must be initialized in the mock provider's `onConnect`
     func createProviderAndConnect(
         listeners: [String]? = nil,
-        connectionQueue: DispatchQueue = DispatchQueue(label: "com.amazonaws.RealtimeConnectionProviderTestBase.connectionQueue"),
+        connectionQueue: DispatchQueue = DispatchQueue(
-        serialCallbackQueue: DispatchQueue = DispatchQueue(label: "com.amazonaws.RealtimeConnectionProviderTestBase.serialCallbackQueue"),
+            label: "com.amazonaws.RealtimeConnectionProviderTestBase.connectionQueue"
+        ),
+        serialCallbackQueue: DispatchQueue = DispatchQueue(
+            label: "com.amazonaws.RealtimeConnectionProviderTestBase.serialCallbackQueue"
+        ),
         connectivityMonitor: ConnectivityMonitor = ConnectivityMonitor()
     ) -> RealtimeConnectionProvider {
         let provider = RealtimeConnectionProvider(

Podfile.lock

@@ -21,4 +21,4 @@ SPEC CHECKSUMS:
 
 PODFILE CHECKSUM: faccccbac411bb105d282d128939114db76875b9
 
-COCOAPODS: 1.11.2
+COCOAPODS: 1.11.3

Pods/Manifest.lock

@@ -21,4 +21,4 @@ SPEC CHECKSUMS:
 
 PODFILE CHECKSUM: faccccbac411bb105d282d128939114db76875b9
 
-COCOAPODS: 1.11.2
+COCOAPODS: 1.11.3

Pods/Pods.xcodeproj/project.pbxproj

@@ -15,6 +15,7 @@
 			dependencies = (
 			);
 			name = SwiftFormat;
+			productName = SwiftFormat;
 		};
 		52B60EC2A583F24ACBB69C113F5488B9 /* SwiftLint */ = {
 			isa = PBXAggregateTarget;
@@ -24,6 +25,7 @@
 			dependencies = (
 			);
 			name = SwiftLint;
+			productName = SwiftLint;
 		};
 /* End PBXAggregateTarget section */
 
@@ -203,7 +205,7 @@
 		999D7A0732B0168D9EB631C456DEC8A3 /* Pods-AppSyncRealTimeClient-AppSyncRealTimeClientTests.release.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; path = "Pods-AppSyncRealTimeClient-AppSyncRealTimeClientTests.release.xcconfig"; sourceTree = "<group>"; };
 		9B273836AFE7E07EE713B29F94B54285 /* Starscream.release.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; path = Starscream.release.xcconfig; sourceTree = "<group>"; };
 		9D2BCC3D81009251CF149E4E6A0C224E /* StringHTTPHandler.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; name = StringHTTPHandler.swift; path = Sources/Framer/StringHTTPHandler.swift; sourceTree = "<group>"; };
-		9D940727FF8FB9C785EB98E56350EF41 /* Podfile */ = {isa = PBXFileReference; explicitFileType = text.script.ruby; includeInIndex = 1; indentWidth = 2; lastKnownFileType = text; name = Podfile; path = ../Podfile; sourceTree = SOURCE_ROOT; tabWidth = 2; xcLanguageSpecificationIdentifier = xcode.lang.ruby; };
+		9D940727FF8FB9C785EB98E56350EF41 /* Podfile */ = {isa = PBXFileReference; explicitFileType = text.script.ruby; includeInIndex = 1; indentWidth = 2; name = Podfile; path = ../Podfile; sourceTree = SOURCE_ROOT; tabWidth = 2; };
 		9FF7D529C51D3F1B0D444B08C40EE3AD /* Pods-AppSyncRTCSample-frameworks.sh */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.script.sh; path = "Pods-AppSyncRTCSample-frameworks.sh"; sourceTree = "<group>"; };
 		A26BD722B33E83A41BD4A44DD2BDC4E3 /* Pods-AppSyncRTCSample-umbrella.h */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.c.h; path = "Pods-AppSyncRTCSample-umbrella.h"; sourceTree = "<group>"; };
 		A31447240FB4AA22B5D3E1287DAF4DCB /* Pods-HostApp-AppSyncRealTimeClientIntegrationTests-acknowledgements.markdown */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text; path = "Pods-HostApp-AppSyncRealTimeClientIntegrationTests-acknowledgements.markdown"; sourceTree = "<group>"; };
@@ -327,7 +329,6 @@
 				F07B207CD9F5DF97C2D649C58C619BB7 /* WSEngine.swift */,
 				83681F15B53C4D30045511A41B8A97FD /* Support Files */,
 			);
-			name = Starscream;
 			path = Starscream;
 			sourceTree = "<group>";
 		};
@@ -443,7 +444,6 @@
 			children = (
 				37E4C763E3D1AD54AEBD3E13FA588594 /* Support Files */,
 			);
-			name = SwiftLint;
 			path = SwiftLint;
 			sourceTree = "<group>";
 		};
@@ -469,7 +469,6 @@
 			children = (
 				120316207F6661A29A01BB8F9BBEF8F1 /* Support Files */,
 			);
-			name = SwiftFormat;
 			path = SwiftFormat;
 			sourceTree = "<group>";
 		};
@@ -1173,7 +1172,7 @@
 			buildSettings = {
 				ASSETCATALOG_COMPILER_APPICON_NAME = AppIcon;
 				ASSETCATALOG_COMPILER_GLOBAL_ACCENT_COLOR_NAME = AccentColor;
-				IPHONEOS_DEPLOYMENT_TARGET = 8.0;
+				IPHONEOS_DEPLOYMENT_TARGET = 9.0;
 				LD_RUNPATH_SEARCH_PATHS = (
 					"$(inherited)",
 					"@executable_path/Frameworks",
@@ -1296,7 +1295,7 @@
 			buildSettings = {
 				ASSETCATALOG_COMPILER_APPICON_NAME = AppIcon;
 				ASSETCATALOG_COMPILER_GLOBAL_ACCENT_COLOR_NAME = AccentColor;
-				IPHONEOS_DEPLOYMENT_TARGET = 8.0;
+				IPHONEOS_DEPLOYMENT_TARGET = 9.0;
 				LD_RUNPATH_SEARCH_PATHS = (
 					"$(inherited)",
 					"@executable_path/Frameworks",