converted StarscreamAdapterAsync into an actor

.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

@@ -23,7 +23,7 @@
 		217F39D02406E98400F1A0B3 /* AppSyncMessage.swift in Sources */ = {isa = PBXBuildFile; fileRef = 217F39AE2406E98300F1A0B3 /* AppSyncMessage.swift */; };
 		217F39D12406E98400F1A0B3 /* AppSyncConnectionRequest.swift in Sources */ = {isa = PBXBuildFile; fileRef = 217F39AF2406E98300F1A0B3 /* AppSyncConnectionRequest.swift */; };
 		217F39D32406E98400F1A0B3 /* RealtimeConnectionProvider.swift in Sources */ = {isa = PBXBuildFile; fileRef = 217F39B22406E98300F1A0B3 /* RealtimeConnectionProvider.swift */; };
-		217F39D42406E98400F1A0B3 /* RealtimeConnectionProvider+Websocket.swift in Sources */ = {isa = PBXBuildFile; fileRef = 217F39B32406E98300F1A0B3 /* RealtimeConnectionProvider+Websocket.swift */; };
+		217F39D42406E98400F1A0B3 /* RealtimeConnectionProviderBase+Websocket.swift in Sources */ = {isa = PBXBuildFile; fileRef = 217F39B32406E98300F1A0B3 /* RealtimeConnectionProviderBase+Websocket.swift */; };
 		217F39D52406E98400F1A0B3 /* RealtimeConnectionProviderResponse.swift in Sources */ = {isa = PBXBuildFile; fileRef = 217F39B42406E98300F1A0B3 /* RealtimeConnectionProviderResponse.swift */; };
 		217F39D62406E98400F1A0B3 /* RealtimeConnectionProvider+MessageInterceptable.swift in Sources */ = {isa = PBXBuildFile; fileRef = 217F39B52406E98300F1A0B3 /* RealtimeConnectionProvider+MessageInterceptable.swift */; };
 		217F39D72406E98400F1A0B3 /* RealtimeConnectionProvider+ConnectionInterceptable.swift in Sources */ = {isa = PBXBuildFile; fileRef = 217F39B62406E98300F1A0B3 /* RealtimeConnectionProvider+ConnectionInterceptable.swift */; };
@@ -76,6 +76,22 @@
 		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 */; };
+		5C5BE6432831797400B7B58C /* ConcurrencyTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5C5BE6422831797400B7B58C /* ConcurrencyTests.swift */; };
+		5C5BE6442831797400B7B58C /* AppSyncRealTimeClient.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = 217F398F2405D9D500F1A0B3 /* AppSyncRealTimeClient.framework */; };
+		5C89F782282AD0CC00912619 /* RealtimeConnectionProviderAsyncTestBase.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5C89F781282AD0CC00912619 /* RealtimeConnectionProviderAsyncTestBase.swift */; };
+		5C89F784282AD20A00912619 /* ConnectionProviderAsyncTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5C89F783282AD20A00912619 /* ConnectionProviderAsyncTests.swift */; };
+		5CC10129283017E00093AC44 /* RealtimeConnectionProviderAsync.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CC10121283017E00093AC44 /* RealtimeConnectionProviderAsync.swift */; };
+		5CC1012A283017E00093AC44 /* RealtimeConnectionProviderAsync+StaleConnection.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CC10122283017E00093AC44 /* RealtimeConnectionProviderAsync+StaleConnection.swift */; };
+		5CC1012C283017E00093AC44 /* RealtimeConnectionProviderAsync+Websocket.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CC10124283017E00093AC44 /* RealtimeConnectionProviderAsync+Websocket.swift */; };
+		5CC1012E283017E00093AC44 /* RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CC10126283017E00093AC44 /* RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift */; };
+		5CC10130283017E00093AC44 /* RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CC10128283017E00093AC44 /* RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift */; };
+		5CC1013228301BDB0093AC44 /* ConnectionProviderAsync.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CC1013128301BDB0093AC44 /* ConnectionProviderAsync.swift */; };
+		5CC10134283025000093AC44 /* AppSyncWebsocketProviderAsync.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CC10133283025000093AC44 /* AppSyncWebsocketProviderAsync.swift */; };
+		5CC101372830288B0093AC44 /* StarscreamAdapterAsync.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CC101352830288B0093AC44 /* StarscreamAdapterAsync.swift */; };
+		5CC101382830288B0093AC44 /* StarscreamAdapterAsync+Delegate.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CC101362830288B0093AC44 /* StarscreamAdapterAsync+Delegate.swift */; };
+		5CC1013A28304A470093AC44 /* MockWebsocketProviderAsync.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CC1013928304A470093AC44 /* MockWebsocketProviderAsync.swift */; };
+		5CC1013C28315E520093AC44 /* Test.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5CC1013B28315E520093AC44 /* Test.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 */; };
@@ -95,7 +111,7 @@
 		FA91B21024D306730017404D /* AppSyncRTCProvider.swift in Sources */ = {isa = PBXBuildFile; fileRef = FAB5AA1D24D1CD31001F370F /* AppSyncRTCProvider.swift */; };
 		FAAEF946255229DA009DA4D5 /* AtomicValue.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA2EFABB2550CAC6007698C7 /* AtomicValue.swift */; };
 		FAB5AA1F24D1CD84001F370F /* amplifyconfiguration.json in Resources */ = {isa = PBXBuildFile; fileRef = 21D38B4B2409B6C000EC2A8D /* amplifyconfiguration.json */; };
-		FAB7E91224D2644E00DF1EA1 /* RealtimeConnectionProvider+StaleConnection.swift in Sources */ = {isa = PBXBuildFile; fileRef = 217F39B12406E98300F1A0B3 /* RealtimeConnectionProvider+StaleConnection.swift */; };
+		FAB7E91224D2644E00DF1EA1 /* RealtimeConnectionProviderBase+StaleConnection.swift in Sources */ = {isa = PBXBuildFile; fileRef = 217F39B12406E98300F1A0B3 /* RealtimeConnectionProviderBase+StaleConnection.swift */; };
 /* End PBXBuildFile section */
 
 /* Begin PBXContainerItemProxy section */
@@ -120,6 +136,13 @@
 			remoteGlobalIDString = 21D38B522409B93F00EC2A8D;
 			remoteInfo = HostApp;
 		};
+		5C5BE6452831797400B7B58C /* PBXContainerItemProxy */ = {
+			isa = PBXContainerItemProxy;
+			containerPortal = 217F39862405D9D500F1A0B3 /* Project object */;
+			proxyType = 1;
+			remoteGlobalIDString = 217F398E2405D9D500F1A0B3;
+			remoteInfo = AppSyncRealTimeClient;
+		};
 		FA2B2C3424D4B77600C6B108 /* PBXContainerItemProxy */ = {
 			isa = PBXContainerItemProxy;
 			containerPortal = 217F39862405D9D500F1A0B3 /* Project object */;
@@ -148,9 +171,9 @@
 		217F39AD2406E98300F1A0B3 /* AppSyncMessage+Encodable.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "AppSyncMessage+Encodable.swift"; sourceTree = "<group>"; };
 		217F39AE2406E98300F1A0B3 /* AppSyncMessage.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = AppSyncMessage.swift; sourceTree = "<group>"; };
 		217F39AF2406E98300F1A0B3 /* AppSyncConnectionRequest.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = AppSyncConnectionRequest.swift; sourceTree = "<group>"; };
-		217F39B12406E98300F1A0B3 /* RealtimeConnectionProvider+StaleConnection.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProvider+StaleConnection.swift"; sourceTree = "<group>"; };
+		217F39B12406E98300F1A0B3 /* RealtimeConnectionProviderBase+StaleConnection.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProviderBase+StaleConnection.swift"; sourceTree = "<group>"; };
 		217F39B22406E98300F1A0B3 /* RealtimeConnectionProvider.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = RealtimeConnectionProvider.swift; sourceTree = "<group>"; };
-		217F39B32406E98300F1A0B3 /* RealtimeConnectionProvider+Websocket.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProvider+Websocket.swift"; sourceTree = "<group>"; };
+		217F39B32406E98300F1A0B3 /* RealtimeConnectionProviderBase+Websocket.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProviderBase+Websocket.swift"; sourceTree = "<group>"; };
 		217F39B42406E98300F1A0B3 /* RealtimeConnectionProviderResponse.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = RealtimeConnectionProviderResponse.swift; sourceTree = "<group>"; };
 		217F39B52406E98300F1A0B3 /* RealtimeConnectionProvider+MessageInterceptable.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProvider+MessageInterceptable.swift"; sourceTree = "<group>"; };
 		217F39B62406E98300F1A0B3 /* RealtimeConnectionProvider+ConnectionInterceptable.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProvider+ConnectionInterceptable.swift"; sourceTree = "<group>"; };
@@ -207,6 +230,22 @@
 		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>"; };
+		5C5BE6402831797400B7B58C /* ConcurrencyTests.xctest */ = {isa = PBXFileReference; explicitFileType = wrapper.cfbundle; includeInIndex = 0; path = ConcurrencyTests.xctest; sourceTree = BUILT_PRODUCTS_DIR; };
+		5C5BE6422831797400B7B58C /* ConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = ConcurrencyTests.swift; sourceTree = "<group>"; };
+		5C89F781282AD0CC00912619 /* RealtimeConnectionProviderAsyncTestBase.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = RealtimeConnectionProviderAsyncTestBase.swift; sourceTree = "<group>"; };
+		5C89F783282AD20A00912619 /* ConnectionProviderAsyncTests.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = ConnectionProviderAsyncTests.swift; sourceTree = "<group>"; };
+		5CC10121283017E00093AC44 /* RealtimeConnectionProviderAsync.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = RealtimeConnectionProviderAsync.swift; sourceTree = "<group>"; };
+		5CC10122283017E00093AC44 /* RealtimeConnectionProviderAsync+StaleConnection.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProviderAsync+StaleConnection.swift"; sourceTree = "<group>"; };
+		5CC10124283017E00093AC44 /* RealtimeConnectionProviderAsync+Websocket.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProviderAsync+Websocket.swift"; sourceTree = "<group>"; };
+		5CC10126283017E00093AC44 /* RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift"; sourceTree = "<group>"; };
+		5CC10128283017E00093AC44 /* RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift"; sourceTree = "<group>"; };
+		5CC1013128301BDB0093AC44 /* ConnectionProviderAsync.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = ConnectionProviderAsync.swift; sourceTree = "<group>"; };
+		5CC10133283025000093AC44 /* AppSyncWebsocketProviderAsync.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = AppSyncWebsocketProviderAsync.swift; sourceTree = "<group>"; };
+		5CC101352830288B0093AC44 /* StarscreamAdapterAsync.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = StarscreamAdapterAsync.swift; sourceTree = "<group>"; };
+		5CC101362830288B0093AC44 /* StarscreamAdapterAsync+Delegate.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "StarscreamAdapterAsync+Delegate.swift"; sourceTree = "<group>"; };
+		5CC1013928304A470093AC44 /* MockWebsocketProviderAsync.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = MockWebsocketProviderAsync.swift; sourceTree = "<group>"; };
+		5CC1013B28315E520093AC44 /* Test.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; name = Test.swift; path = AppSyncRealTimeClientTests/Test.swift; sourceTree = SOURCE_ROOT; };
 		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; };
@@ -272,6 +311,14 @@
 			);
 			runOnlyForDeploymentPostprocessing = 0;
 		};
+		5C5BE63D2831797400B7B58C /* Frameworks */ = {
+			isa = PBXFrameworksBuildPhase;
+			buildActionMask = 2147483647;
+			files = (
+				5C5BE6442831797400B7B58C /* AppSyncRealTimeClient.framework in Frameworks */,
+			);
+			runOnlyForDeploymentPostprocessing = 0;
+		};
 		FA91B1F824D3063E0017404D /* Frameworks */ = {
 			isa = PBXFrameworksBuildPhase;
 			buildActionMask = 2147483647;
@@ -299,6 +346,7 @@
 				21D38B3F2409AFBD00EC2A8D /* AppSyncRealTimeClientIntegrationTests */,
 				FA91B1FC24D3063E0017404D /* AppSyncRTCSample */,
 				21D38B542409B93F00EC2A8D /* HostApp */,
+				5C5BE6412831797400B7B58C /* ConcurrencyTests */,
 				217F39902405D9D500F1A0B3 /* Products */,
 				D146D6DDE2251CCA7A4ECD7F /* Pods */,
 				CC47BA7DC6033B1626BAD959 /* Frameworks */,
@@ -313,6 +361,7 @@
 				21D38B3E2409AFBD00EC2A8D /* AppSyncRealTimeClientIntegrationTests.xctest */,
 				21D38B532409B93F00EC2A8D /* HostApp.app */,
 				FA91B1FB24D3063E0017404D /* AppSyncRTCSample.app */,
+				5C5BE6402831797400B7B58C /* ConcurrencyTests.xctest */,
 			);
 			name = Products;
 			sourceTree = "<group>";
@@ -353,12 +402,15 @@
 				217F39AF2406E98300F1A0B3 /* AppSyncConnectionRequest.swift */,
 				217F39AE2406E98300F1A0B3 /* AppSyncMessage.swift */,
 				217F39AD2406E98300F1A0B3 /* AppSyncMessage+Encodable.swift */,
+				217F39B02406E98300F1A0B3 /* AppsyncRealtimeConnection */,
+				5CC10120283017E00093AC44 /* AppsyncRealtimeConnectionAsync */,
 				217F39AA2406E98300F1A0B3 /* AppSyncResponse.swift */,
 				217F39B72406E98300F1A0B3 /* ConnectionProvider.swift */,
+				5CC1013128301BDB0093AC44 /* ConnectionProviderAsync.swift */,
 				217F39AC2406E98300F1A0B3 /* ConnectionProviderError.swift */,
 				21D38B8D240A3C2300EC2A8D /* ConnectionProviderFactory.swift */,
 				217F39AB2406E98300F1A0B3 /* InterceptableConnection.swift */,
-				217F39B02406E98300F1A0B3 /* AppsyncRealtimeConnection */,
+				5C5609ED2821E6FC0002ACF5 /* InterceptableConnectionAsync.swift */,
 			);
 			path = ConnectionProvider;
 			sourceTree = "<group>";
@@ -369,8 +421,8 @@
 				217F39B22406E98300F1A0B3 /* RealtimeConnectionProvider.swift */,
 				217F39B62406E98300F1A0B3 /* RealtimeConnectionProvider+ConnectionInterceptable.swift */,
 				217F39B52406E98300F1A0B3 /* RealtimeConnectionProvider+MessageInterceptable.swift */,
-				217F39B12406E98300F1A0B3 /* RealtimeConnectionProvider+StaleConnection.swift */,
-				217F39B32406E98300F1A0B3 /* RealtimeConnectionProvider+Websocket.swift */,
+				217F39B12406E98300F1A0B3 /* RealtimeConnectionProviderBase+StaleConnection.swift */,
+				217F39B32406E98300F1A0B3 /* RealtimeConnectionProviderBase+Websocket.swift */,
 				217F39B42406E98300F1A0B3 /* RealtimeConnectionProviderResponse.swift */,
 			);
 			path = AppsyncRealtimeConnection;
@@ -402,6 +454,7 @@
 			isa = PBXGroup;
 			children = (
 				217F39C22406E98400F1A0B3 /* AppSyncWebsocketProvider.swift */,
+				5CC10133283025000093AC44 /* AppSyncWebsocketProviderAsync.swift */,
 				217F39C32406E98400F1A0B3 /* Starscream */,
 			);
 			path = Websocket;
@@ -412,6 +465,8 @@
 			children = (
 				217F39C42406E98400F1A0B3 /* StarscreamAdapter.swift */,
 				217F39C52406E98400F1A0B3 /* StarscreamAdapter+Delegate.swift */,
+				5CC101352830288B0093AC44 /* StarscreamAdapterAsync.swift */,
+				5CC101362830288B0093AC44 /* StarscreamAdapterAsync+Delegate.swift */,
 			);
 			path = Starscream;
 			sourceTree = "<group>";
@@ -445,8 +500,9 @@
 			isa = PBXGroup;
 			children = (
 				217F39EB2406EA3F00F1A0B3 /* MockConnectionProvider.swift */,
-				FA67507724D3244A005A1345 /* MockWebsocketProvider.swift */,
 				219BFF6B27A9AF85000FC148 /* MockConnectivityMonitor.swift */,
+				FA67507724D3244A005A1345 /* MockWebsocketProvider.swift */,
+				5CC1013928304A470093AC44 /* MockWebsocketProviderAsync.swift */,
 			);
 			path = Mocks;
 			sourceTree = "<group>";
@@ -455,10 +511,13 @@
 			isa = PBXGroup;
 			children = (
 				2143D46527B5D9730066B2F7 /* AppSyncRealTimeConnection */,
+				5C89F783282AD20A00912619 /* ConnectionProviderAsyncTests.swift */,
 				2151D5C927C68C3C00F3C866 /* ConnectionProviderHandleErrorTests.swift */,
 				FA67507F24D339B0005A1345 /* ConnectionProviderStaleConnectionTests.swift */,
 				217F39ED2406EA4000F1A0B3 /* ConnectionProviderTests.swift */,
+				5C89F781282AD0CC00912619 /* RealtimeConnectionProviderAsyncTestBase.swift */,
 				FA67507C24D338FA005A1345 /* RealtimeConnectionProviderTestBase.swift */,
+				5CC1013B28315E520093AC44 /* Test.swift */,
 			);
 			path = ConnectionProvider;
 			sourceTree = "<group>";
@@ -558,6 +617,26 @@
 			path = Support;
 			sourceTree = "<group>";
 		};
+		5C5BE6412831797400B7B58C /* ConcurrencyTests */ = {
+			isa = PBXGroup;
+			children = (
+				5C5BE6422831797400B7B58C /* ConcurrencyTests.swift */,
+			);
+			path = ConcurrencyTests;
+			sourceTree = "<group>";
+		};
+		5CC10120283017E00093AC44 /* AppsyncRealtimeConnectionAsync */ = {
+			isa = PBXGroup;
+			children = (
+				5CC10121283017E00093AC44 /* RealtimeConnectionProviderAsync.swift */,
+				5CC10122283017E00093AC44 /* RealtimeConnectionProviderAsync+StaleConnection.swift */,
+				5CC10124283017E00093AC44 /* RealtimeConnectionProviderAsync+Websocket.swift */,
+				5CC10126283017E00093AC44 /* RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift */,
+				5CC10128283017E00093AC44 /* RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift */,
+			);
+			path = AppsyncRealtimeConnectionAsync;
+			sourceTree = "<group>";
+		};
 		CC47BA7DC6033B1626BAD959 /* Frameworks */ = {
 			isa = PBXGroup;
 			children = (
@@ -715,6 +794,24 @@
 			productReference = 21D38B532409B93F00EC2A8D /* HostApp.app */;
 			productType = "com.apple.product-type.application";
 		};
+		5C5BE63F2831797400B7B58C /* ConcurrencyTests */ = {
+			isa = PBXNativeTarget;
+			buildConfigurationList = 5C5BE6492831797400B7B58C /* Build configuration list for PBXNativeTarget "ConcurrencyTests" */;
+			buildPhases = (
+				5C5BE63C2831797400B7B58C /* Sources */,
+				5C5BE63D2831797400B7B58C /* Frameworks */,
+				5C5BE63E2831797400B7B58C /* Resources */,
+			);
+			buildRules = (
+			);
+			dependencies = (
+				5C5BE6462831797400B7B58C /* PBXTargetDependency */,
+			);
+			name = ConcurrencyTests;
+			productName = ConcurrencyTests;
+			productReference = 5C5BE6402831797400B7B58C /* ConcurrencyTests.xctest */;
+			productType = "com.apple.product-type.bundle.unit-test";
+		};
 		FA91B1FA24D3063E0017404D /* AppSyncRTCSample */ = {
 			isa = PBXNativeTarget;
 			buildConfigurationList = FA91B20E24D306430017404D /* Build configuration list for PBXNativeTarget "AppSyncRTCSample" */;
@@ -743,7 +840,7 @@
 		217F39862405D9D500F1A0B3 /* Project object */ = {
 			isa = PBXProject;
 			attributes = {
-				LastSwiftUpdateCheck = 1160;
+				LastSwiftUpdateCheck = 1330;
 				LastUpgradeCheck = 1130;
 				ORGANIZATIONNAME = amazonaws;
 				TargetAttributes = {
@@ -760,6 +857,9 @@
 					21D38B522409B93F00EC2A8D = {
 						CreatedOnToolsVersion = 11.3.1;
 					};
+					5C5BE63F2831797400B7B58C = {
+						CreatedOnToolsVersion = 13.3.1;
+					};
 					FA91B1FA24D3063E0017404D = {
 						CreatedOnToolsVersion = 11.6;
 					};
@@ -783,6 +883,7 @@
 				21D38B3D2409AFBD00EC2A8D /* AppSyncRealTimeClientIntegrationTests */,
 				FA91B1FA24D3063E0017404D /* AppSyncRTCSample */,
 				21D38B522409B93F00EC2A8D /* HostApp */,
+				5C5BE63F2831797400B7B58C /* ConcurrencyTests */,
 			);
 		};
 /* End PBXProject section */
@@ -823,6 +924,13 @@
 			);
 			runOnlyForDeploymentPostprocessing = 0;
 		};
+		5C5BE63E2831797400B7B58C /* Resources */ = {
+			isa = PBXResourcesBuildPhase;
+			buildActionMask = 2147483647;
+			files = (
+			);
+			runOnlyForDeploymentPostprocessing = 0;
+		};
 		FA91B1F924D3063E0017404D /* Resources */ = {
 			isa = PBXResourcesBuildPhase;
 			buildActionMask = 2147483647;
@@ -1153,16 +1261,22 @@
 				217F39DB2406E98400F1A0B3 /* AppSyncSubscriptionConnection+DataHandler.swift in Sources */,
 				21D38B94240C4A2A00EC2A8D /* OIDCAuthProvider.swift in Sources */,
 				217F39DF2406E98400F1A0B3 /* SubscriptionItem.swift in Sources */,
+				5CC1012E283017E00093AC44 /* RealtimeConnectionProviderAsync+ConnectionInterceptableAsync.swift in Sources */,
+				5CC1012C283017E00093AC44 /* RealtimeConnectionProviderAsync+Websocket.swift in Sources */,
 				FAAEF946255229DA009DA4D5 /* AtomicValue.swift in Sources */,
 				217F39CF2406E98400F1A0B3 /* AppSyncMessage+Encodable.swift in Sources */,
 				217F39CD2406E98400F1A0B3 /* InterceptableConnection.swift in Sources */,
 				21D38B8E240A3C2300EC2A8D /* ConnectionProviderFactory.swift in Sources */,
 				217F39E02406E98400F1A0B3 /* AppSyncWebsocketProvider.swift in Sources */,
+				5CC101372830288B0093AC44 /* StarscreamAdapterAsync.swift in Sources */,
 				219BFF4927A3B238000FC148 /* ConnectivityMonitor.swift in Sources */,
-				FAB7E91224D2644E00DF1EA1 /* RealtimeConnectionProvider+StaleConnection.swift in Sources */,
+				FAB7E91224D2644E00DF1EA1 /* RealtimeConnectionProviderBase+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 */,
+				5CC10130283017E00093AC44 /* RealtimeConnectionProviderAsync+MessageInterceptableAsync.swift in Sources */,
+				5CC101382830288B0093AC44 /* StarscreamAdapterAsync+Delegate.swift in Sources */,
 				217F39E52406E98400F1A0B3 /* AppSyncJSONValue.swift in Sources */,
 				217F39E42406E98400F1A0B3 /* RealtimeGatewayURLInterceptor.swift in Sources */,
 				2143D4B027BC49BE0066B2F7 /* AWSAppSyncRealTimeClient.swift in Sources */,
@@ -1172,6 +1286,8 @@
 				FA67508224D33A7A005A1345 /* CountdownTimer.swift in Sources */,
 				217F39E62406E98400F1A0B3 /* SubscriptionConstants.swift in Sources */,
 				217F39D02406E98400F1A0B3 /* AppSyncMessage.swift in Sources */,
+				5CC1013228301BDB0093AC44 /* ConnectionProviderAsync.swift in Sources */,
+				5CC1012A283017E00093AC44 /* RealtimeConnectionProviderAsync+StaleConnection.swift in Sources */,
 				217F39E32406E98400F1A0B3 /* AppSyncLogger.swift in Sources */,
 				217F39D82406E98400F1A0B3 /* ConnectionProvider.swift in Sources */,
 				21D38B83240A392B00EC2A8D /* APIKeyAuthInterceptor.swift in Sources */,
@@ -1180,11 +1296,13 @@
 				978409B82739C7BE002362A7 /* AppSyncURLHelper.swift in Sources */,
 				217F39CC2406E98400F1A0B3 /* AppSyncResponse.swift in Sources */,
 				217F39E72406E98400F1A0B3 /* SubscriptionConnectionType.swift in Sources */,
-				217F39D42406E98400F1A0B3 /* RealtimeConnectionProvider+Websocket.swift in Sources */,
+				217F39D42406E98400F1A0B3 /* RealtimeConnectionProviderBase+Websocket.swift in Sources */,
 				217F39DC2406E98400F1A0B3 /* AppSyncSubscriptionConnection.swift in Sources */,
 				21D38B6D240A262800EC2A8D /* AppSyncJSONHelper.swift in Sources */,
 				219BFF4B27A3B24F000FC148 /* ConnectivityPath.swift in Sources */,
+				5CC10129283017E00093AC44 /* RealtimeConnectionProviderAsync.swift in Sources */,
 				217F39CE2406E98400F1A0B3 /* ConnectionProviderError.swift in Sources */,
+				5CC10134283025000093AC44 /* AppSyncWebsocketProviderAsync.swift in Sources */,
 				217F39E12406E98400F1A0B3 /* StarscreamAdapter.swift in Sources */,
 				217F39D72406E98400F1A0B3 /* RealtimeConnectionProvider+ConnectionInterceptable.swift in Sources */,
 				217F39DE2406E98400F1A0B3 /* AppSyncSubscriptionConnection+ErrorHandler.swift in Sources */,
@@ -1200,18 +1318,22 @@
 				217F39F22406EA4000F1A0B3 /* ConnectionProviderTests.swift in Sources */,
 				2143D46727B5D9B40066B2F7 /* RealTimeConnectionProviderResponseTests.swift in Sources */,
 				FA67507824D3244A005A1345 /* MockWebsocketProvider.swift in Sources */,
+				5CC1013C28315E520093AC44 /* Test.swift in Sources */,
 				217F39F02406EA4000F1A0B3 /* AppSyncSubscriptionConnectionTests.swift in Sources */,
 				FA67507B24D338CB005A1345 /* Error+Extension.swift in Sources */,
+				5C89F782282AD0CC00912619 /* RealtimeConnectionProviderAsyncTestBase.swift in Sources */,
 				21D38B8C240A39E400EC2A8D /* APIKeyAuthInterceptorTests.swift in Sources */,
 				21D38B8B240A39E400EC2A8D /* AppSyncJSONHelperTests.swift in Sources */,
 				217F39F32406EA4000F1A0B3 /* RealtimeGatewayURLInterceptorTests.swift in Sources */,
 				217C74D227C45B6600AE054F /* AppSyncSubscriptionConnectionErrorHandlerTests.swift in Sources */,
 				FA67507E24D33976005A1345 /* RealtimeConnectionProviderTestBase.swift in Sources */,
+				5C89F784282AD20A00912619 /* ConnectionProviderAsyncTests.swift in Sources */,
 				219BFF6A27A9AC6E000FC148 /* ConnectivityMonitorTests.swift in Sources */,
 				2151D5CA27C68C3C00F3C866 /* ConnectionProviderHandleErrorTests.swift in Sources */,
 				978409BA2739C7E1002362A7 /* AppSyncURLHelperTests.swift in Sources */,
 				FA67508424D33ACC005A1345 /* CountdownTimerTests.swift in Sources */,
 				FA67508024D339B0005A1345 /* ConnectionProviderStaleConnectionTests.swift in Sources */,
+				5CC1013A28304A470093AC44 /* MockWebsocketProviderAsync.swift in Sources */,
 				217F39F12406EA4000F1A0B3 /* MockConnectionProvider.swift in Sources */,
 			);
 			runOnlyForDeploymentPostprocessing = 0;
@@ -1240,6 +1362,14 @@
 			);
 			runOnlyForDeploymentPostprocessing = 0;
 		};
+		5C5BE63C2831797400B7B58C /* Sources */ = {
+			isa = PBXSourcesBuildPhase;
+			buildActionMask = 2147483647;
+			files = (
+				5C5BE6432831797400B7B58C /* ConcurrencyTests.swift in Sources */,
+			);
+			runOnlyForDeploymentPostprocessing = 0;
+		};
 		FA91B1F724D3063E0017404D /* Sources */ = {
 			isa = PBXSourcesBuildPhase;
 			buildActionMask = 2147483647;
@@ -1269,6 +1399,11 @@
 			target = 21D38B522409B93F00EC2A8D /* HostApp */;
 			targetProxy = 21D38B672409B95000EC2A8D /* PBXContainerItemProxy */;
 		};
+		5C5BE6462831797400B7B58C /* PBXTargetDependency */ = {
+			isa = PBXTargetDependency;
+			target = 217F398E2405D9D500F1A0B3 /* AppSyncRealTimeClient */;
+			targetProxy = 5C5BE6452831797400B7B58C /* PBXContainerItemProxy */;
+		};
 		FA2B2C3524D4B77600C6B108 /* PBXTargetDependency */ = {
 			isa = PBXTargetDependency;
 			target = 217F398E2405D9D500F1A0B3 /* AppSyncRealTimeClient */;
@@ -1347,7 +1482,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 +1539,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;
@@ -1594,6 +1729,44 @@
 			};
 			name = Release;
 		};
+		5C5BE6472831797400B7B58C /* Debug */ = {
+			isa = XCBuildConfiguration;
+			buildSettings = {
+				CLANG_CXX_LANGUAGE_STANDARD = "gnu++17";
+				CLANG_WARN_QUOTED_INCLUDE_IN_FRAMEWORK_HEADER = YES;
+				CODE_SIGN_STYLE = Automatic;
+				CURRENT_PROJECT_VERSION = 1;
+				DEVELOPMENT_TEAM = W3DRXD72QU;
+				GENERATE_INFOPLIST_FILE = YES;
+				IPHONEOS_DEPLOYMENT_TARGET = 15.4;
+				MARKETING_VERSION = 1.0;
+				PRODUCT_BUNDLE_IDENTIFIER = com.amazonaws.ConcurrencyTests;
+				PRODUCT_NAME = "$(TARGET_NAME)";
+				SWIFT_EMIT_LOC_STRINGS = NO;
+				SWIFT_VERSION = 5.0;
+				TARGETED_DEVICE_FAMILY = "1,2";
+			};
+			name = Debug;
+		};
+		5C5BE6482831797400B7B58C /* Release */ = {
+			isa = XCBuildConfiguration;
+			buildSettings = {
+				CLANG_CXX_LANGUAGE_STANDARD = "gnu++17";
+				CLANG_WARN_QUOTED_INCLUDE_IN_FRAMEWORK_HEADER = YES;
+				CODE_SIGN_STYLE = Automatic;
+				CURRENT_PROJECT_VERSION = 1;
+				DEVELOPMENT_TEAM = W3DRXD72QU;
+				GENERATE_INFOPLIST_FILE = YES;
+				IPHONEOS_DEPLOYMENT_TARGET = 15.4;
+				MARKETING_VERSION = 1.0;
+				PRODUCT_BUNDLE_IDENTIFIER = com.amazonaws.ConcurrencyTests;
+				PRODUCT_NAME = "$(TARGET_NAME)";
+				SWIFT_EMIT_LOC_STRINGS = NO;
+				SWIFT_VERSION = 5.0;
+				TARGETED_DEVICE_FAMILY = "1,2";
+			};
+			name = Release;
+		};
 		FA91B20C24D306430017404D /* Debug */ = {
 			isa = XCBuildConfiguration;
 			baseConfigurationReference = BBF14AA5E9A4D036CA87B952 /* Pods-AppSyncRTCSample.debug.xcconfig */;
@@ -1684,6 +1857,15 @@
 			defaultConfigurationIsVisible = 0;
 			defaultConfigurationName = Release;
 		};
+		5C5BE6492831797400B7B58C /* Build configuration list for PBXNativeTarget "ConcurrencyTests" */ = {
+			isa = XCConfigurationList;
+			buildConfigurations = (
+				5C5BE6472831797400B7B58C /* Debug */,
+				5C5BE6482831797400B7B58C /* Release */,
+			);
+			defaultConfigurationIsVisible = 0;
+			defaultConfigurationName = Release;
+		};
 		FA91B20E24D306430017404D /* Build configuration list for PBXNativeTarget "AppSyncRTCSample" */ = {
 			isa = XCConfigurationList;
 			buildConfigurations = (

AppSyncRealTimeClient.xcodeproj/xcshareddata/xcschemes/AppSyncRealTimeClientIntegrationTests.xcscheme

@@ -10,7 +10,7 @@
       buildConfiguration = "Debug"
       selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
       selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
-      shouldUseLaunchSchemeArgsEnv = "YES"
+      shouldUseLaunchSchemeArgsEnv = "NO"
       enableThreadSanitizer = "YES">
       <Testables>
          <TestableReference

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.swift

@@ -10,19 +10,17 @@ import Combine
 
 /// Appsync Real time connection that connects to subscriptions
 /// through websocket.
-public class RealtimeConnectionProvider: ConnectionProvider {
+public class RealtimeConnectionProviderBase {
     /// 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
 
-    private let url: URL
+    let url: URL
     var listeners: [String: ConnectionProviderCallback]
 
     let websocket: AppSyncWebsocketProvider
 
     var status: ConnectionState
-    var messageInterceptors: [MessageInterceptor]
-    var connectionInterceptors: [ConnectionInterceptor]
 
     /// A timer that automatically disconnects the current connection if it goes longer
     /// than `staleConnectionTimeout` without activity. Receiving any data or "keep
@@ -55,14 +53,13 @@ public class RealtimeConnectionProvider: ConnectionProvider {
         if iLimitExceededSubject == nil {
             iLimitExceededSubject = PassthroughSubject<ConnectionProviderError, Never>()
         }
-        return iLimitExceededSubject as! PassthroughSubject<ConnectionProviderError, Never> // swiftlint:disable:this force_cast
+        // swiftlint:disable:next force_cast
+        return iLimitExceededSubject as! PassthroughSubject<ConnectionProviderError, Never>
     }
 
-    public convenience init(for url: URL, websocket: AppSyncWebsocketProvider) {
-        self.init(url: url, websocket: websocket)
-    }
-
-    init(
+    // Prevent this class from being instantiated directly.
+    // Use RealtimeConnectionProvider or RealtimeConnectionProviderAsync instead
+    fileprivate init(
         url: URL,
         websocket: AppSyncWebsocketProvider,
         connectionQueue: DispatchQueue = DispatchQueue(
@@ -77,8 +74,6 @@ public class RealtimeConnectionProvider: ConnectionProvider {
         self.websocket = websocket
         self.listeners = [:]
         self.status = .notConnected
-        self.messageInterceptors = []
-        self.connectionInterceptors = []
         self.staleConnectionTimer = CountdownTimer()
         self.isStaleConnection = false
         self.connectionQueue = connectionQueue
@@ -94,57 +89,35 @@ public class RealtimeConnectionProvider: ConnectionProvider {
 
     // MARK: - ConnectionProvider methods
 
-    public func connect() {
-        connectionQueue.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 = self.interceptConnection(request, for: self.url)
-            DispatchQueue.global().async {
-                self.websocket.connect(
-                    url: signedRequest.url,
-                    protocols: ["graphql-ws"],
-                    delegate: self
-                )
-            }
-        }
+    // The following two methods are overridden in
+    // RealtimeConnectionProvider and RealtimeConnectionProviderAsync below.
+    public func connect() {}
+    public func write(_ message: AppSyncMessage) {}
+
+    func sendConnectionInitMessage() {
+        let message = AppSyncMessage(type: .connectionInit("connection_init"))
+        write(message)
     }
 
-    public func write(_ message: AppSyncMessage) {
-
-        connectionQueue.async { [weak self] in
-            guard let self = self else {
+    func finishWrite(_ signedMessage: AppSyncMessage) {
+        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)
+                updateCallback(event: .error(jsonError))
                 return
             }
-
-            let signedMessage = 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(message.id, nil)
-                    self.updateCallback(event: .error(jsonError))
-                    return
-                }
-                self.websocket.write(message: jsonString)
-            } catch {
-                AppSyncLogger.error(error)
-                switch message.messageType {
-                case .connectionInit:
-                    self.receivedConnectionInit()
-                default:
-                    self.updateCallback(event: .error(ConnectionProviderError.jsonParse(message.id, error)))
-                }
+            websocket.write(message: jsonString)
+        } catch {
+            AppSyncLogger.error(error)
+            switch signedMessage.messageType {
+            case .connectionInit:
+                receivedConnectionInit()
+            default:
+                updateCallback(event: .error(ConnectionProviderError.jsonParse(signedMessage.id, error)))
             }
         }
-
     }
 
     public func disconnect() {
@@ -169,7 +142,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()
@@ -178,10 +153,6 @@ public class RealtimeConnectionProvider: ConnectionProvider {
     }
 
     // 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.
@@ -229,3 +200,69 @@ public class RealtimeConnectionProvider: ConnectionProvider {
         updateCallback(event: .error(ConnectionProviderError.connection))
     }
 }
+
+// MARK: - RealtimeConnectionProvider
+
+public class RealtimeConnectionProvider: RealtimeConnectionProviderBase, ConnectionProvider {
+
+    var messageInterceptors = [MessageInterceptor]()
+    var connectionInterceptors = [ConnectionInterceptor]()
+
+    override internal init(
+        url: URL,
+        websocket: AppSyncWebsocketProvider,
+        connectionQueue: DispatchQueue = DispatchQueue(
+            label: "com.amazonaws.AppSyncRealTimeConnectionProvider.serialQueue"
+        ),
+        serialCallbackQueue: DispatchQueue = DispatchQueue(
+            label: "com.amazonaws.AppSyncRealTimeConnectionProvider.callbackQueue"
+        ),
+        connectivityMonitor: ConnectivityMonitor = ConnectivityMonitor()
+    ) {
+        super.init(
+            url: url,
+            websocket: websocket,
+            connectionQueue: connectionQueue,
+            serialCallbackQueue: serialCallbackQueue,
+            connectivityMonitor: connectivityMonitor
+        )
+    }
+
+    public convenience init(for url: URL, websocket: AppSyncWebsocketProvider) {
+        self.init(url: url, websocket: websocket)
+    }
+
+    override public func connect() {
+        connectionQueue.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 = self.interceptConnection(request, for: self.url)
+            DispatchQueue.global().async {
+                self.websocket.connect(
+                    url: signedRequest.url,
+                    protocols: ["graphql-ws"],
+                    delegate: self
+                )
+            }
+        }
+    }
+
+    override public func write(_ message: AppSyncMessage) {
+        connectionQueue.async { [weak self] in
+            guard let self = self else {
+                return
+            }
+
+            let signedMessage = self.interceptMessage(message, for: self.url)
+            self.finishWrite(signedMessage)
+        }
+    }
+}

AppSyncRealTimeClient/ConnectionProvider/AppsyncRealtimeConnection/RealtimeConnectionProviderBase+StaleConnection.swift

@@ -8,13 +8,15 @@
 import Foundation
 
 /// Consolidates usage and parameters passed to the `staleConnectionTimer` methods.
-extension RealtimeConnectionProvider {
+extension RealtimeConnectionProviderBase {
 
     /// 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) {
+        staleConnectionTimer.start(interval: RealtimeConnectionProviderBase.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/RealtimeConnectionProviderBase+Websocket.swift

@@ -7,7 +7,7 @@
 
 import Foundation
 
-extension RealtimeConnectionProvider: AppSyncWebsocketDelegate {
+extension RealtimeConnectionProviderBase: AppSyncWebsocketDelegate {
 
     public func websocketDidConnect(provider: AppSyncWebsocketProvider) {
         // Call the ack to finish the connection handshake

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) async {
+        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) async {
+        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,66 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Foundation
+
+/// Consolidates usage and parameters passed to the `staleConnectionTimer` methods.
+@available(iOS 13.0.0, *)
+extension RealtimeConnectionProviderAsync {
+
+    /// Start a stale connection timer, first invalidating and destroying any existing timer
+    func startStaleConnectionTimer() async {
+        AppSyncLogger.debug(
+            "[RealtimeConnectionProvider] Starting stale connection timer for \(staleConnectionTimer.interval)s"
+        )
+
+        staleConnectionTimer.start(interval: RealtimeConnectionProviderBase.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) {
+            AppSyncLogger.debug(
+                "[RealtimeConnectionProvider] Status: \(status). Connectivity status: \(connectivity.status)"
+            )
+            if status == .connected && connectivity.status == .unsatisfied && !isStaleConnection {
+                AppSyncLogger.debug(
+                    "[RealtimeConnectionProvider] Connetion is stale. Pending reconnect on connectivity."
+                )
+                isStaleConnection = true
+
+            } else if status == .connected && isStaleConnection && connectivity.status == .satisfied {
+                AppSyncLogger.debug(
+                    "[RealtimeConnectionProvider] Connetion is stale. Disconnecting to begin reconnect."
+                )
+                staleConnectionTimer.invalidate()
+                disconnectStaleConnection()
+            }
+    }
+
+    /// Fired when the stale connection timer expires
+    private func disconnectStaleConnection() {
+        Task {
+            AppSyncLogger.error("[RealtimeConnectionProvider] Realtime connection is stale, disconnecting.")
+            status = .notConnected
+            isStaleConnection = false
+            await websocket.disconnect()
+            updateCallback(event: .error(ConnectionProviderError.connection))
+        }
+    }
+}

AppSyncRealTimeClient/ConnectionProvider/AppsyncRealtimeConnectionAsync/RealtimeConnectionProviderAsync+Websocket.swift

@@ -0,0 +1,136 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Foundation
+
+@available(iOS 13.0.0, *)
+extension RealtimeConnectionProviderAsync: AppSyncWebsocketDelegateAsync {
+    public func websocketDidConnect(provider: AppSyncWebsocketProviderAsync) async {
+        // 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()
+        await startStaleConnectionTimer()
+    }
+
+    public func websocketDidDisconnect(provider: AppSyncWebsocketProviderAsync, error: Error?) async {
+            status = .notConnected
+            guard error != nil else {
+                updateCallback(event: .connection(status))
+                return
+            }
+            updateCallback(event: .error(ConnectionProviderError.connection))
+    }
+
+    public func websocketDidReceiveData(provider: AppSyncWebsocketProviderAsync, data: Data) async {
+        do {
+            let response = try JSONDecoder().decode(RealtimeConnectionProviderResponse.self, from: data)
+            await handleResponse(response)
+        } catch {
+            AppSyncLogger.error(error)
+            updateCallback(event: .error(ConnectionProviderError.jsonParse(nil, error)))
+        }
+    }
+
+    // MARK: - Handle websocket response
+
+    private func handleResponse(_ response: RealtimeConnectionProviderResponse) async {
+        resetStaleConnectionTimer()
+
+        switch response.responseType {
+        case .connectionAck:
+            AppSyncLogger.debug("[RealtimeConnectionProvider] received connectionAck")
+                handleConnectionAck(response: response)
+        case .error:
+            AppSyncLogger.verbose("[RealtimeConnectionProvider] received error")
+                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))
+    }
+
+}

AppSyncRealTimeClient/ConnectionProvider/AppsyncRealtimeConnectionAsync/RealtimeConnectionProviderAsync.swift

@@ -0,0 +1,217 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Foundation
+import Combine
+
+/// Appsync Real time connection that connects to subscriptions
+/// through websocket.
+@available(iOS 13.0.0, *)
+public actor RealtimeConnectionProviderAsync: ConnectionProviderAsync {
+    /// 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: ConnectionProviderCallbackAsync]
+
+    let websocket: AppSyncWebsocketProviderAsync
+
+    var status: ConnectionState
+
+    /// 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 connectionQueue: DispatchQueue
+
+    /// 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?
+    @available(iOS 13.0, *)
+    var limitExceededSubject: PassthroughSubject<ConnectionProviderError, Never> {
+        if iLimitExceededSubject == nil {
+            iLimitExceededSubject = PassthroughSubject<ConnectionProviderError, Never>()
+        }
+        // swiftlint:disable:next force_cast
+        return iLimitExceededSubject as! PassthroughSubject<ConnectionProviderError, Never>
+    }
+
+    // Prevent this class from being instantiated directly.
+    // Use RealtimeConnectionProvider or RealtimeConnectionProviderAsync instead
+    init(
+        url: URL,
+        websocket: AppSyncWebsocketProviderAsync,
+        connectionQueue: DispatchQueue = DispatchQueue(
+            label: "com.amazonaws.AppSyncRealTimeConnectionProvider.serialQueue"
+        ),
+        serialCallbackQueue: DispatchQueue = DispatchQueue(
+            label: "com.amazonaws.AppSyncRealTimeConnectionProvider.callbackQueue"
+        ),
+        connectivityMonitor: ConnectivityMonitor = ConnectivityMonitor()
+    ) async {
+        self.url = url
+        self.websocket = websocket
+        self.listeners = [:]
+        self.status = .notConnected
+        self.staleConnectionTimer = CountdownTimer()
+        self.isStaleConnection = false
+        self.connectionQueue = connectionQueue
+        self.serialCallbackQueue = serialCallbackQueue
+        self.connectivityMonitor = connectivityMonitor
+
+        connectivityMonitor.start(onUpdates: handleConnectivityUpdates(connectivity:))
+
+        subscribeToLimitExceededThrottle()
+    }
+
+    public convenience init(for url: URL, websocket: AppSyncWebsocketProviderAsync) async {
+        await self.init(url: url, websocket: websocket)
+    }
+
+    // MARK: - ConnectionProvider methods
+
+    func sendConnectionInitMessage() {
+        let message = AppSyncMessage(type: .connectionInit("connection_init"))
+        write(message)
+    }
+
+    func finishWrite(_ signedMessage: AppSyncMessage) async {
+        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)
+                updateCallback(event: .error(jsonError))
+                return
+            }
+            await websocket.write(message: jsonString)
+        } catch {
+            AppSyncLogger.error(error)
+            switch signedMessage.messageType {
+            case .connectionInit:
+                receivedConnectionInit()
+            default:
+                updateCallback(event: .error(ConnectionProviderError.jsonParse(signedMessage.id, error)))
+            }
+        }
+    }
+
+    public func disconnect() async {
+         await websocket.disconnect()
+         invalidateStaleConnectionTimer()
+    }
+
+    public func addListener(identifier: String, callback: @escaping ConnectionProviderCallbackAsync) {
+          listeners[identifier] = callback
+    }
+
+    public func removeListener(identifier: String) async {
+            listeners.removeValue(forKey: identifier)
+
+            if listeners.isEmpty {
+                AppSyncLogger.debug(
+                    "[RealtimeConnectionProvider] all subscriptions removed, disconnecting websocket connection."
+                )
+               status = .notConnected
+            await websocket.disconnect()
+             invalidateStaleConnectionTimer()
+            }
+
+    }
+
+    // MARK: -
+
+    /// 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) {
+            let allListeners = Array(listeners.values)
+            Task {
+                for listener in allListeners {
+                    await listener(event)
+                }
+            }
+    }
+
+    @available(iOS 13.0, *)
+    func subscribeToLimitExceededThrottle() {
+        limitExceededThrottleSink = limitExceededSubject
+            .filter {
+                // Make sure the limitExceeded error is a connection level error (no subscription id present).
+                // When id is present, it is passed back directly subscription via `updateCallback`.
+                if case .limitExceeded(let id) = $0, id == nil {
+                    return true
+                }
+                return false
+            }
+            .throttle(for: .milliseconds(150), scheduler: connectionQueue, latest: true)
+            .sink { completion in
+                switch completion {
+                case .failure(let error):
+                    AppSyncLogger.verbose("limitExceededThrottleSink failed \(error)")
+                case .finished:
+                    AppSyncLogger.verbose("limitExceededThrottleSink finished")
+                }
+        } receiveValue: { result in
+            self.updateCallback(event: .error(result))
+        }
+    }
+
+    /// - Warning: This must be invoked from the `connectionQueue`
+    private func receivedConnectionInit() {
+        status = .notConnected
+        updateCallback(event: .error(ConnectionProviderError.connection))
+    }
+
+    var messageInterceptors = [MessageInterceptorAsync]()
+    var connectionInterceptors = [ConnectionInterceptorAsync]()
+
+    public func connect() {
+            guard status == .notConnected else {
+                updateCallback(event: .connection(status))
+                return
+            }
+            status = .inProgress
+            updateCallback(event: .connection(status))
+            let request = AppSyncConnectionRequest(url: url)
+
+            Task {
+                let signedRequest = await self.interceptConnection(request, for: self.url)
+                await self.websocket.connect(
+                    url: signedRequest.url,
+                    protocols: ["graphql-ws"],
+                    delegate: self
+                )
+            }
+    }
+
+    public func write(_ message: AppSyncMessage) {
+            Task {
+                let signedMessage = await self.interceptMessage(message, for: self.url)
+                await self.finishWrite(signedMessage)
+            }
+    }
+}

AppSyncRealTimeClient/ConnectionProvider/ConnectionProviderAsync.swift

@@ -0,0 +1,24 @@
+//
+// 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 ConnectionProviderAsync: AnyObject {
+
+    func connect() async
+
+    func write(_ message: AppSyncMessage) async
+
+    func disconnect() async
+
+    func addListener(identifier: String, callback: @escaping ConnectionProviderCallbackAsync) async
+
+    func removeListener(identifier: String) async
+}
+
+public typealias ConnectionProviderCallbackAsync = (ConnectionProviderEvent) async -> Void

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 {
+    ) async -> ConnectionProviderAsync {
+        let provider: ConnectionProviderAsync
+
         switch connectionType {
         case .appSyncRealtime:
-            let websocketProvider = StarscreamAdapter()
-            let connectionProvider = RealtimeConnectionProvider(for: url, websocket: websocketProvider)
-            return connectionProvider
+            let websocketProvider = StarscreamAdapterAsync()
+            provider = await RealtimeConnectionProviderAsync(for: url, websocket: websocketProvider)
         }
+
+        if let messageInterceptable = provider as? MessageInterceptableAsync {
+            await messageInterceptable.addInterceptor(authInterceptor)
+        }
+        if let connectionInterceptable = provider as? ConnectionInterceptableAsync {
+            await connectionInterceptable.addInterceptor(RealtimeGatewayURLInterceptor())
+            await 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) async
+
+    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) async
+
+    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/Websocket/AppSyncWebsocketProviderAsync.swift

@@ -0,0 +1,40 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Foundation
+
+/// Protocol to be implemented by different websocket providers
+@available(iOS 13.0.0, *)
+public protocol AppSyncWebsocketProviderAsync {
+
+    /// Initiates a connection to the given url.
+    ///
+    /// This is an async call. After the connection is succesfully established, the delegate
+    /// will receive the callback on `websocketDidConnect(:)`
+    func connect(url: URL, protocols: [String], delegate: AppSyncWebsocketDelegateAsync?) async
+
+    /// Disconnects the websocket.
+    func disconnect() async
+
+    /// Write message to the websocket provider
+    /// - Parameter message: Message to write
+    func write(message: String) async
+
+    /// Returns `true` if the websocket is connected
+    func isConnected() async -> Bool
+}
+
+/// Delegate method to get callbacks on websocket provider connection
+@available(iOS 13.0.0, *)
+public protocol AppSyncWebsocketDelegateAsync: AnyObject {
+
+    func websocketDidConnect(provider: AppSyncWebsocketProviderAsync) async
+
+    func websocketDidDisconnect(provider: AppSyncWebsocketProviderAsync, error: Error?) async
+
+    func websocketDidReceiveData(provider: AppSyncWebsocketProviderAsync, data: Data) async
+}

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)

AppSyncRealTimeClient/Websocket/Starscream/StarscreamAdapterAsync+Delegate.swift

@@ -0,0 +1,72 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Foundation
+import Starscream
+
+/// Extension to handle delegate callback from Starscream
+@available(iOS 13.0.0, *)
+extension StarscreamAdapterAsync: Starscream.WebSocketDelegate {
+    public nonisolated func didReceive(event: WebSocketEvent, client: WebSocket) {
+        switch event {
+        case .connected:
+            websocketDidConnect(socket: client)
+        case .disconnected(let reason, let code):
+            AppSyncLogger.verbose("[StarscreamAdapter] disconnected: reason=\(reason); code=\(code)")
+            websocketDidDisconnect(socket: client, error: nil)
+        case .text(let string):
+            websocketDidReceiveMessage(socket: client, text: string)
+        case .binary(let data):
+            websocketDidReceiveData(socket: client, data: data)
+        case .ping:
+            AppSyncLogger.verbose("[StarscreamAdapter] ping")
+        case .pong:
+            AppSyncLogger.verbose("[StarscreamAdapter] pong")
+        case .viabilityChanged(let viability):
+            AppSyncLogger.verbose("[StarscreamAdapter] viabilityChanged: \(viability)")
+        case .reconnectSuggested(let suggestion):
+            AppSyncLogger.verbose("[StarscreamAdapter] reconnectSuggested: \(suggestion)")
+        case .cancelled:
+            websocketDidDisconnect(socket: client, error: nil)
+        case .error(let error):
+            websocketDidDisconnect(socket: client, error: error)
+        }
+    }
+
+    private nonisolated func websocketDidConnect(socket: WebSocketClient) {
+        AppSyncLogger.verbose("[StarscreamAdapter] websocketDidConnect: websocket has been connected.")
+        Task {
+            await setIsConnected(true)
+            await delegate?.websocketDidConnect(provider: self)
+        }
+    }
+
+    private nonisolated func websocketDidDisconnect(socket: WebSocketClient, error: Error?) {
+        AppSyncLogger.verbose(
+            "[StarscreamAdapter] websocketDidDisconnect: \(error?.localizedDescription ?? "No error")"
+        )
+        Task {
+            await setIsConnected(false)
+            await delegate?.websocketDidDisconnect(provider: self, error: error)
+        }
+    }
+
+    private nonisolated func websocketDidReceiveMessage(socket: WebSocketClient, text: String) {
+        AppSyncLogger.verbose("[StarscreamAdapter] websocketDidReceiveMessage: - \(text)")
+        let data = text.data(using: .utf8) ?? Data()
+        Task {
+            await delegate?.websocketDidReceiveData(provider: self, data: data)
+        }
+    }
+
+    private nonisolated func websocketDidReceiveData(socket: WebSocketClient, data: Data) {
+        AppSyncLogger.verbose("[StarscreamAdapter] WebsocketDidReceiveData - \(data)")
+        Task {
+            await delegate?.websocketDidReceiveData(provider: self, data: data)
+        }
+    }
+}

AppSyncRealTimeClient/Websocket/Starscream/StarscreamAdapterAsync.swift

@@ -0,0 +1,58 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Foundation
+import Starscream
+
+@available(iOS 13.0.0, *)
+public actor StarscreamAdapterAsync: AppSyncWebsocketProviderAsync {
+    var socket: WebSocket?
+    weak var delegate: AppSyncWebsocketDelegateAsync?
+
+    // swiftlint:disable:next identifier_name
+    var _isConnected: Bool
+
+    public func isConnected() -> Bool {
+        _isConnected
+    }
+
+    func setIsConnected(_ isConnected: Bool) {
+        _isConnected = isConnected
+    }
+
+    public init() {
+        self._isConnected = false
+    }
+
+    public func connect(url: URL, protocols: [String], delegate: AppSyncWebsocketDelegateAsync?) async {
+        Task {
+            AppSyncLogger.verbose("[StarscreamAdapter] connect. Connecting to url")
+            var urlRequest = URLRequest(url: url)
+            let protocolHeaderValue = protocols.joined(separator: ", ")
+            urlRequest.setValue(protocolHeaderValue, forHTTPHeaderField: "Sec-WebSocket-Protocol")
+            socket = WebSocket(request: urlRequest)
+            self.delegate = delegate
+            socket?.delegate = self
+            socket?.connect()
+        }
+    }
+
+    public func disconnect() async {
+        Task {
+            AppSyncLogger.verbose("[StarscreamAdapter] socket.disconnect")
+            self.socket?.disconnect()
+            self.socket = nil
+        }
+    }
+
+    public func write(message: String) async {
+        Task {
+            AppSyncLogger.verbose("[StarscreamAdapter] socket.write - \(message)")
+            self.socket?.write(string: message)
+        }
+    }
+}

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/ConnectionProviderAsyncTests.swift

@@ -0,0 +1,257 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import XCTest
+@testable import AppSyncRealTimeClient
+
+#if swift(>=5.5.2)
+@available(iOS 13.0, *)
+class ConnectionProviderAsyncTests: XCTestCase { // }: RealtimeConnectionProviderAsyncTestBase {
+
+    func test() async {
+        print("ok")
+
+        await waitForExpectations(timeout: 3)
+    }
+
+}
+
+    /// 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() async {
+//        await expectations.invertReceivedNotConnected()
+//        await expectations.invertReceivedError()
+
+//        let receivedInProgress = expectations.receivedInProgress
+//        let receivedConnected = expectations.receivedConnected
+
+//        let expectation = expectation(description: "blah")
+//        await expectations.setExpectations(
+//            receivedInProgress: expectation,
+//            receivedConnected: expectation,
+//            receivedNotConnected: expectation,
+//            receivedError: expectation
+//        )
+
+//        let onConnect: MockWebsocketProviderAsync.OnConnect = { _, _, delegate in
+//            self.websocketDelegate = delegate
+////            Task {
+//                await delegate?.websocketDidConnect(provider: self.websocket)
+////            }
+//        }
+//
+//        let onDisconnect: MockWebsocketProviderAsync.OnDisconnect = { }
+//
+//        let onWrite: MockWebsocketProviderAsync.OnWrite = { message in
+//            guard RealtimeConnectionProviderTestBase.messageType(of: message, equals: "connection_init") else {
+//                XCTFail("Incoming message did not have 'connection_init' type")
+//                return
+//            }
+//
+//            await self.websocketDelegate.websocketDidReceiveData(
+//                provider: self.websocket,
+//                data: RealtimeConnectionProviderTestBase.makeConnectionAckMessage()
+//            )
+//        }
+//
+//        websocket = MockWebsocketProviderAsync(
+//            onConnect: onConnect,
+//            onDisconnect: onDisconnect,
+//            onWrite: onWrite
+//        )
+
+        // Retain the provider so it doesn't release prior to executing callbacks
+//        let provider = await createProviderAndConnect()
+
+        // Get rid of "written to, but never read" compiler warnings
+//        print(provider)
+
+        //await waitForExpectations(timeout: 20.05)
+//        wait(for: [expectations.receivedConnected], timeout: 5.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() async {
+//        await expectations.invertReceivedNotConnected()
+//        await expectations.invertReceivedError()
+//
+//        let onConnect: MockWebsocketProviderAsync.OnConnect = { _, _, delegate in
+//            self.websocketDelegate = delegate
+//            Task {
+//                await delegate?.websocketDidConnect(provider: self.websocket)
+//            }
+//        }
+//
+//        let receivedDisconnect = expectation(description: "receivedDisconnect")
+//        let onDisconnect: MockWebsocketProviderAsync.OnDisconnect = {
+//            receivedDisconnect.fulfill()
+//        }
+//
+//        let onWrite: MockWebsocketProviderAsync.OnWrite = { message in
+//            guard RealtimeConnectionProviderTestBase.messageType(of: message, equals: "connection_init") else {
+//                XCTFail("Incoming message did not have 'connection_init' type")
+//                return
+//            }
+//
+//            await self.websocketDelegate.websocketDidReceiveData(
+//                provider: self.websocket,
+//                data: RealtimeConnectionProviderTestBase.makeConnectionAckMessage()
+//            )
+//        }
+//
+//        websocket = MockWebsocketProviderAsync(
+//            onConnect: onConnect,
+//            onDisconnect: onDisconnect,
+//            onWrite: onWrite
+//        )
+//
+//        // Retain the provider so it doesn't release prior to executing callbacks
+//        let provider = await createProviderAndConnect(listeners: ["1", "2", "3", "4"])
+//
+//        wait(
+//            for: [receivedInProgress, receivedConnected, receivedNotConnected, receivedError],
+//            timeout: 1
+//        )
+//
+//        var listenersEmpty = await provider.listeners.isEmpty
+//        XCTAssertFalse(listenersEmpty)
+//
+//        let listenersToRemove = await provider.listeners.map { $0.key }
+//
+//        // Removing all the listeners will disconnect the websocket connection
+//        for identifier in listenersToRemove {
+//            await provider.removeListener(identifier: identifier)
+//        }
+//
+//        // Since removing listeners is asynchronous, we have to wait for the disconnect
+//        // wait(for: [receivedDisconnect], timeout: 1)
+//        listenersEmpty = await provider.listeners.isEmpty
+//        XCTAssertTrue(listenersEmpty)
+//
+//        let status = await provider.status
+//        XCTAssertEqual(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() async {
+//        receivedConnected.isInverted = true
+//        receivedNotConnected.isInverted = true
+//
+//        let onConnect: MockWebsocketProviderAsync.OnConnect = { _, _, delegate in
+//            self.websocketDelegate = delegate
+//            // DispatchQueue.global().async {
+//            await delegate?.websocketDidConnect(provider: self.websocket)
+//            // }
+//        }
+//
+//        let onDisconnect: MockWebsocketProviderAsync.OnDisconnect = { }
+//
+//        let onWrite: MockWebsocketProviderAsync.OnWrite = { message in
+//            guard RealtimeConnectionProviderTestBase.messageType(of: message, equals: "connection_init") else {
+//                XCTFail("Incoming message did not have 'connection_init' type")
+//                return
+//            }
+//
+//            await self.websocketDelegate.websocketDidDisconnect(
+//                provider: self.websocket,
+//                error: "test error"
+//            )
+//        }
+//
+//        websocket = MockWebsocketProviderAsync(
+//            onConnect: onConnect,
+//            onDisconnect: onDisconnect,
+//            onWrite: onWrite
+//        )
+//
+//        // Retain the provider so it doesn't release prior to executing callbacks
+//        let provider = await createProviderAndConnect()
+//
+//        // Get rid of "written to, but never read" compiler warnings
+//        print(provider)
+//
+//        await 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() async {
+//        receivedNotConnected.isInverted = true
+//        receivedError.isInverted = true
+//
+//        let expectedTimeoutInSeconds = 60.0
+//        let timeoutInMilliseconds = Int(expectedTimeoutInSeconds) * 1_000
+//
+//        let onConnect: MockWebsocketProviderAsync.OnConnect = { _, _, delegate in
+//            self.websocketDelegate = delegate
+//            Task {
+//                await delegate?.websocketDidConnect(provider: self.websocket)
+//            }
+//        }
+//
+//        let onDisconnect: MockWebsocketProviderAsync.OnDisconnect = { }
+//
+//        let connectionAckMessage = RealtimeConnectionProviderTestBase
+//            .makeConnectionAckMessage(withTimeout: timeoutInMilliseconds)
+//        let onWrite: MockWebsocketProviderAsync.OnWrite = { message in
+//            guard RealtimeConnectionProviderTestBase.messageType(of: message, equals: "connection_init") else {
+//                XCTFail("Incoming message did not have 'connection_init' type")
+//                return
+//            }
+//
+//            await self.websocketDelegate.websocketDidReceiveData(
+//                provider: self.websocket,
+//                data: connectionAckMessage
+//            )
+//        }
+//
+//        websocket = MockWebsocketProviderAsync(
+//            onConnect: onConnect,
+//            onDisconnect: onDisconnect,
+//            onWrite: onWrite
+//        )
+//
+//        let provider = await createProviderAndConnect()
+//
+//        wait(for: [receivedConnected], timeout: 0.05)
+//
+//        let staleConnectionTimerInterval = await provider.staleConnectionTimer.interval
+//        XCTAssertEqual(staleConnectionTimerInterval, expectedTimeoutInSeconds)
+//
+//        await waitForExpectations(timeout: 0.05)
+//    }
+
+//}
+#endif

AppSyncRealTimeClientTests/ConnectionProvider/RealtimeConnectionProviderAsyncTestBase.swift

@@ -0,0 +1,158 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+////
+//// Copyright Amazon.com Inc. or its affiliates.
+//// All Rights Reserved.
+////
+//// SPDX-License-Identifier: Apache-2.0
+////
+//
+//import XCTest
+//@testable import AppSyncRealTimeClient
+//
+//#if swift(>=5.5.2)
+//
+//@available(iOS 13.0.0, *)
+//actor AsyncExpectations {
+////    var receivedInProgress = XCTestExpectation(description: "receivedInProgress")
+//    var receivedConnected = XCTestExpectation(description: "receivedConnected")
+////    var receivedNotConnected = XCTestExpectation(description: "receivedNotConnected")
+////    var receivedError = XCTestExpectation(description: "receivedError")
+//
+//    func setExpectations(
+//        //        receivedInProgress: XCTestExpectation,
+//        receivedConnected: XCTestExpectation
+////        receivedNotConnected: XCTestExpectation,
+////        receivedError: XCTestExpectation
+//    ) {
+////        self.receivedInProgress = receivedInProgress
+//        self.receivedConnected = receivedConnected
+////        self.receivedNotConnected = receivedNotConnected
+////        self.receivedError = receivedError
+//    }
+//
+////    func invertReceivedInProgress() {
+////        receivedInProgress.isInverted = true
+////    }
+//
+//    func invertReceivedConnected() {
+//        receivedConnected.isInverted = true
+//    }
+//
+////    func invertReceivedNotConnected() {
+////        receivedNotConnected.isInverted = true
+////    }
+//
+////    func invertReceivedError() {
+////        receivedError.isInverted = true
+////    }
+//}
+//
+//@available(iOS 13.0, *)
+//class RealtimeConnectionProviderAsyncTestBase: XCTestCase {
+//
+//    let url = URL(string: "https://www.appsyncrealtimeclient.test/")!
+//
+//    var websocket: MockWebsocketProviderAsync!
+//
+//    // swiftlint:disable:next weak_delegate
+//    var websocketDelegate: AppSyncWebsocketDelegateAsync!
+//
+//    // Shared test expectations. Set expected fulfillment counts and inversions as
+//    // needed in the body of the test.
+//    let expectations = AsyncExpectations()
+//
+//    override func setUp() async throws {
+////        let receivedInProgress = expectation(description: "receivedInProgress")
+//     //   let receivedConnected = expectation(description: "receivedConnected")
+////        let receivedNotConnected = expectation(description: "receivedNotConnected")
+////        let receivedError = expectation(description: "receivedError")
+//
+//       // await expectations.setExpectations(
+//            //            receivedInProgress: receivedInProgress,
+//      //      receivedConnected: receivedConnected
+////            receivedNotConnected: receivedNotConnected,
+////            receivedError: 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,
+//        connectivityMonitor: ConnectivityMonitor = ConnectivityMonitor()
+//    ) async -> RealtimeConnectionProviderAsync {
+//        let provider = await RealtimeConnectionProviderAsync(
+//            url: url,
+//            websocket: websocket,
+//            connectivityMonitor: connectivityMonitor
+//        )
+//        await provider.addListener(identifier: "testListener") { event in
+//            switch event {
+//            case .connection(let connectionState):
+//                switch connectionState {
+//                case .inProgress:
+//                    print("ok")
+////                    await self.expectations.receivedInProgress.fulfill()
+//                case .connected:
+//                    print("ok")
+//                    //await self.expectations.receivedConnected.fulfill()
+//                case .notConnected:
+//                    print("ok")
+////                    await self.expectations.receivedNotConnected.fulfill()
+//                }
+//            case .error:
+//                print("ok")
+////                await self.expectations.receivedError.fulfill()
+//            default:
+//                break
+//            }
+//        }
+////        if let listeners = listeners {
+////            for listener in listeners {
+////                await provider.addListener(identifier: listener)  { _ in }
+////            }
+////        }
+//        await 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)!
+////    }
+//
+//}
+//#endif

AppSyncRealTimeClientTests/ConnectionProvider/RealtimeConnectionProviderTestBase.swift

@@ -14,7 +14,7 @@ class RealtimeConnectionProviderTestBase: XCTestCase {
 
     var websocket: MockWebsocketProvider!
 
-    //swiftlint:disable:next weak_delegate
+    // swiftlint:disable:next weak_delegate
     var websocketDelegate: AppSyncWebsocketDelegate!
 
     // Shared test expectations. Set expected fulfillment counts and inversions as
@@ -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"),
-        serialCallbackQueue: DispatchQueue = DispatchQueue(label: "com.amazonaws.RealtimeConnectionProviderTestBase.serialCallbackQueue"),
+        connectionQueue: DispatchQueue = DispatchQueue(
+            label: "com.amazonaws.RealtimeConnectionProviderTestBase.connectionQueue"
+        ),
+        serialCallbackQueue: DispatchQueue = DispatchQueue(
+            label: "com.amazonaws.RealtimeConnectionProviderTestBase.serialCallbackQueue"
+        ),
         connectivityMonitor: ConnectivityMonitor = ConnectivityMonitor()
     ) -> RealtimeConnectionProvider {
         let provider = RealtimeConnectionProvider(

AppSyncRealTimeClientTests/Mocks/MockWebsocketProviderAsync.swift

@@ -0,0 +1,46 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Foundation
+import AppSyncRealTimeClient
+
+@available(iOS 13.0.0, *)
+class MockWebsocketProviderAsync: AppSyncWebsocketProviderAsync {
+    typealias OnConnect = (URL, [String], AppSyncWebsocketDelegateAsync?) async -> Void
+    typealias OnDisconnect = () async -> Void
+    typealias OnWrite = (String) async -> Void
+
+    var isConnected: Bool
+
+    let onConnect: OnConnect?
+    let onDisconnect: OnDisconnect?
+    let onWrite: OnWrite?
+
+    init(
+        onConnect: OnConnect? = nil,
+        onDisconnect: OnDisconnect? = nil,
+        onWrite: OnWrite? = nil
+    ) {
+        self.isConnected = false
+        self.onConnect = onConnect
+        self.onDisconnect = onDisconnect
+        self.onWrite = onWrite
+    }
+
+    func connect(url: URL, protocols: [String], delegate: AppSyncWebsocketDelegateAsync?) async {
+        await onConnect?(url, protocols, delegate)
+    }
+
+    func disconnect() async {
+        await onDisconnect?()
+    }
+
+    func write(message: String) async {
+        await onWrite?(message)
+    }
+
+}

ConcurrencyTests/ConcurrencyTests.swift

@@ -0,0 +1,36 @@
+//
+//  ConcurrencyTests.swift
+//  ConcurrencyTests
+//
+//  Created by Ameter, Chris on 5/15/22.
+//  Copyright © 2022 amazonaws. All rights reserved.
+//
+
+import XCTest
+
+class ConcurrencyTests: XCTestCase {
+
+    override func setUpWithError() throws {
+        // Put setup code here. This method is called before the invocation of each test method in the class.
+    }
+
+    override func tearDownWithError() throws {
+        // Put teardown code here. This method is called after the invocation of each test method in the class.
+    }
+
+    func testExample() throws {
+        // This is an example of a functional test case.
+        // Use XCTAssert and related functions to verify your tests produce the correct results.
+        // Any test you write for XCTest can be annotated as throws and async.
+        // Mark your test throws to produce an unexpected failure when your test encounters an uncaught error.
+        // Mark your test async to allow awaiting for asynchronous code to complete. Check the results with assertions afterwards.
+    }
+
+    func testPerformanceExample() throws {
+        // This is an example of a performance test case.
+        measure {
+            // Put the code you want to measure the time of here.
+        }
+    }
+
+}

Podfile

@@ -21,6 +21,8 @@ target 'AppSyncRealTimeClient' do
     # Pods for testing
   end
 
+  target 'ConcurrencyTests' do
+  end
 end
 
 target "AppSyncRTCSample" do 
@@ -50,3 +52,15 @@ target "HostApp" do
   end
 end
 
+# target 'ConcurrencyTests' do
+#   # Comment the next line if you don't want to use dynamic frameworks
+#   use_frameworks!
+
+#   # Pods for AppSyncRealTimeClient
+  
+#   # If you update this dependency version, be sure to update the Cartfile also
+#   pod "Starscream", "~> 4.0.4"
+  
+#   include_build_tools!
+
+# end

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",