feat(api): GraphQL Subscription Stream (aws-amplify#852)

* chore(datastore): Remove redundant files for the datastore example (aws-amplify#762) * feat(datastore): Populate belongs-to nested models (aws-amplify#658) * feat(datastore): Populate has-one and belongs-to nested models * Nest data under serializedData key and persist modelName * Resolve comment * Update packages/amplify_datastore/example/ios/unit_tests/resources/SchemaData.swift Co-authored-by: Chris F <[email protected]> * chore(amplify_api): add support for apiName to GraphQL requests (aws-amplify#553) * fix(amplify_api): prevent some fatal REST errors in Android Throw ApiException in android when PUT, POST, and PATCH REST requests have no body to prevent fatal error (aws-amplify#661). * chore(release): 0.2.2 (aws-amplify#781) * chore(release): missing dependency (aws-amplify#783) * Bump analytics version * Fix PR # * Bump all versions * Missing bump * Pin collections * chore: skip FlutterURLSessionTests (aws-amplify#795) * chore(lints): Add amplify_lints package (aws-amplify#808) * Add lints package * Add missing readme item * feat(auth): add options to resendSignUpCode (aws-amplify#738) * feat(auth): add options to resendSignUpCode * fix: add type export for sing up code options * chore: fix formatting issues * chore: update comments for consistency * chore: rename private method names for consistency * chore: update comments for consistency * chore: make ResendSignUpCodeOptions abstract * chore: update ResendSignUpCodeRequest serializeAsMap * chore: update ios test * chore update doc comments * chore: bump amplify-android to 1.24.0 * chore: remove star import * feat(auth): add support for options to resetPassword, confirmResetPassword (aws-amplify#743) * chore: rename confirmPassword to confirmResetPassword * chore: fix grammatical errors in docs * feat(auth): add options support for resetPassword, confirmResetPassword * chore: rename private methods for consistency * chore: update comments for consistency * chore: update ios tests * chore: fix comments * chore: handle deprecated methods * separate out deprecated class * chore: update serializeAsMap for consistency * chore: properly deprecate confirmPassword * chore: add deprecation annotation to interface * chore: bump amplify-android to 1.24.0 * chore: apply suggestions from code review * feat(auth): add options to updateAttribute, updateAttributes, resendUserAttributeConfirmationCode (aws-amplify#775) * chore: move attribute types to new dir * chore: rename fetch attribute classes * feat: add client metadata to user attribute methods * deprecate renamed types * chore: update comments for consistency * chore: rename methods, tests * chore: break out depracted classes, make new classes abstract * chore: revert star import * chore: bump amplify-android to 1.24.0 * chore: apply suggestions from code review * chore: remove empty line * Clean up pinpoint * Clean up pinpoint dart * Enable CI for analytics * Add iOS linting * Clean up * Update iOS script * Fix Android melos script * Update CI order * Add Dart lints to API * Apply Android/iOS lints to API * Add GraphQL subscription stream and tests * fix(amplify_datastore): ios send modelProviderVersion (aws-amplify#439) Co-authored-by: Hui Zhao <[email protected]> * feat(datastore): Add start and stop APIs (aws-amplify#811) * Convert file to LF mode * feat(datastore): Add start and stop APIs * Add docs * Resolve comments * chore(analytics): Apply lints (aws-amplify#810) * Clean up pinpoint * Clean up pinpoint dart * Enable CI for analytics * Add iOS linting * Clean up * Update iOS script * Fix Android melos script * Update CI order * Small changes * Update type * Add iOS whitespace rules * Update type * Apply updated rules * feat(auth): OIDC/Lambda Support (aws-amplify#777) * OIDC/Lambda support * Clean up * Fix iOS test * Add unit tests * Fix Android test * Fix Android tests * Refactor and remove AuthToken from the public API * Remove concurrent guards * Clean up * chore: upgrade amplify-android 1.24.1 (aws-amplify#829) * fix(datastore): return null for list field in nested model (aws-amplify#843) * fix(datastore): Better loggin on unhandled DataStoreHubEvent (aws-amplify#647) * fix(datastore): Better loggin on unhandled DataStoreHubEvent * Print unrecognized event details in iOS * Resolve comments * chore(api): Apply lints (aws-amplify#812) * Clean up pinpoint * Clean up pinpoint dart * Update iOS script * Update CI order * Add Dart lints to API * Apply Android/iOS lints to API * Rename uuid * Small changes * Fix scripts * Clean up * Fix unit tests * Continue impl * Fix android unit tests * Remove duplicate lint check * Fix analytics app * Adjust java options * Bump java RAM * Remove concurrency * Disable gradle daemon * Update gradle properties * Update gradle config * Revert "Update gradle config" This reverts commit a43ad29. * Revert gradle changes * Disable gradle daemon * Add kotlin style flags * Disable gradle daemon * Bump JVM memory * Change daemon setting * Adjust JVM memory * Lint debug only * Fix API tests * Bump deps and fix coverage script * Fix Gradle version * Revert unnecessary changes * Update melos postclean files * Enable fatal infos * Fix analyze scope * Fix postbootstrap * Fix missing sample app * Fix order in CI * Revert add sample app * Revert order change * Revert license date * Revert "Revert license date" This reverts commit 1b93b3f. * Clean up * Add multi-subscription w/ tests * Refactor tests * Fix native tests * Fix onEstablished * Fix docs * Add unknown exception tests * Clean up * Revert multi & add tests * Fix merge errors * Fix async dependency Co-authored-by: Hui Zhao <[email protected]> Co-authored-by: Chris F <[email protected]> Co-authored-by: José Sánchez <[email protected]> Co-authored-by: Travis Sheppard <[email protected]> Co-authored-by: Jordan Nelson <[email protected]> Co-authored-by: Kyle <[email protected]>
haverchuck · Sep 14, 2021 · f25b64e · f25b64e
1 parent d8f8485
commit f25b64e
Show file tree

Hide file tree

Showing 21 changed files with 721 additions and 231 deletions.
diff --git a/...c/main/kotlin/com/amazonaws/amplify/amplify_analytics_pinpoint/AmplifyAnalyticsBuilder.kt b/...c/main/kotlin/com/amazonaws/amplify/amplify_analytics_pinpoint/AmplifyAnalyticsBuilder.kt
@@ -146,4 +146,4 @@ class AmplifyAnalyticsBuilder {
             return locationBuilder.build()
         }
     }
-}
+}
diff --git a/...mplify_api/android/src/main/kotlin/com/amazonaws/amplify/amplify_api/FlutterApiRequest.kt b/...mplify_api/android/src/main/kotlin/com/amazonaws/amplify/amplify_api/FlutterApiRequest.kt
@@ -134,4 +134,4 @@ object FlutterApiRequest {
             )
         }
     }
-}
+}
diff --git a/...mplify_api/android/src/main/kotlin/com/amazonaws/amplify/amplify_api/FlutterGraphQLApi.kt b/...mplify_api/android/src/main/kotlin/com/amazonaws/amplify/amplify_api/FlutterGraphQLApi.kt
@@ -235,10 +235,11 @@ class FlutterGraphQLApi {
             }
 
             val errorCallback = Consumer<ApiException> {
-                if (id.isNotEmpty()) OperationsManager.removeOperation(id)
+                OperationsManager.removeOperation(id)
                 if (established) {
                     graphqlSubscriptionStreamHandler.sendError(
                         "ApiException",
+                        id,
                         ExceptionUtil.createSerializedError(it)
                     )
                 } else {
@@ -252,7 +253,7 @@ class FlutterGraphQLApi {
             }
 
             val disconnectionCallback = Action {
-                if (id.isNotEmpty()) OperationsManager.removeOperation(id)
+                OperationsManager.removeOperation(id)
                 LOG.debug("Subscription has been closed successfully")
                 graphqlSubscriptionStreamHandler.sendEvent(
                     null,

diff --git a/...oid/src/main/kotlin/com/amazonaws/amplify/amplify_api/GraphQLSubscriptionStreamHandler.kt b/...oid/src/main/kotlin/com/amazonaws/amplify/amplify_api/GraphQLSubscriptionStreamHandler.kt
@@ -46,13 +46,13 @@ class GraphQLSubscriptionStreamHandler : EventChannel.StreamHandler {
         }
     }
 
-    fun sendError(errorCode: String, details: Map<String, Any?>) {
+    fun sendError(errorCode: String, id: String, details: Map<String, Any?>) {
         handler.post {
             eventSink?.error(
                 errorCode,
-                ExceptionMessages.defaultFallbackExceptionMessage,
+                id,
                 details
             )
         }
     }
-}
+}
diff --git a/..._api/android/src/main/kotlin/com/amazonaws/amplify/amplify_api/rest_api/FlutterRestApi.kt b/..._api/android/src/main/kotlin/com/amazonaws/amplify/amplify_api/rest_api/FlutterRestApi.kt
@@ -303,4 +303,4 @@ object FlutterRestApi {
     ): RestOperation? {
         return Amplify.API.patch(apiName, restOptions, restConsumer, exceptionConsumer)
     }
-}
+}
diff --git a/...lify_api/android/src/test/kotlin/com/amazonaws/amplify/amplify_api/GraphQLApiUnitTests.kt b/...lify_api/android/src/test/kotlin/com/amazonaws/amplify/amplify_api/GraphQLApiUnitTests.kt
@@ -588,6 +588,7 @@ class GraphQLApiUnitTests {
         verify(mockStreamHandler, times(1))
             .sendError(
                 "ApiException",
+                id,
                 ExceptionUtil.createSerializedError(apiException)
             )
     }
@@ -757,4 +758,4 @@ class GraphQLApiUnitTests {
         modifiersField.setInt(field, field.modifiers and Modifier.FINAL.inv())
         field.set(null, newValue)
     }
-}
+}
diff --git a/packages/amplify_api/example/ios/unit_tests/GraphQLApiUnitTests.swift b/packages/amplify_api/example/ios/unit_tests/GraphQLApiUnitTests.swift
@@ -315,7 +315,7 @@ class GraphQLApiUnitTests: XCTestCase {
                 XCTFail()
             }
 
-            override func sendError(errorCode: String, details: [String: Any]) {
+            override func sendError(errorCode: String, id: String, details: [String: Any]) {
                 XCTFail()
             }
         }
@@ -355,7 +355,7 @@ class GraphQLApiUnitTests: XCTestCase {
                 XCTFail()
             }
 
-            override func sendError(errorCode: String, details: [String: Any]) {
+            override func sendError(errorCode: String, id: String, details: [String: Any]) {
                 XCTFail()
             }
         }
@@ -437,7 +437,7 @@ class GraphQLApiUnitTests: XCTestCase {
         }
 
         class MockStreamHandler: GraphQLSubscriptionsStreamHandler {
-            override func sendError(errorCode: String, details: [String: Any]) {
+            override func sendError(errorCode: String, id: String, details: [String: Any]) {
                             eventSentExp?.fulfill()
                             XCTAssertEqual("ApiException", errorCode)
                             XCTAssertEqual("test error", details["message"] as! String)
@@ -480,7 +480,7 @@ class GraphQLApiUnitTests: XCTestCase {
                 XCTFail()
             }
 
-            override func sendError(errorCode: String, details: [String: Any]) {
+            override func sendError(errorCode: String, id: String, details: [String: Any]) {
                 XCTFail()
             }
         }
@@ -524,7 +524,7 @@ class GraphQLApiUnitTests: XCTestCase {
                 XCTFail()
             }
 
-            override func sendError(errorCode: String, details: [String: Any]) {
+            override func sendError(errorCode: String, id: String, details: [String: Any]) {
                 XCTFail()
             }
         }

diff --git a/packages/amplify_api/example/lib/graphql_api_view.dart b/packages/amplify_api/example/lib/graphql_api_view.dart
@@ -29,40 +29,29 @@ class GraphQLApiView extends StatefulWidget {
 
 class _GraphQLApiViewState extends State<GraphQLApiView> {
   String _result = '';
-  late Function _unsubscribe;
+  void Function()? _unsubscribe;
   late GraphQLOperation _lastOperation;
 
   Future<void> subscribe() async {
     String graphQLDocument = '''subscription MySubscription {
-        onCreateBlog {
-          id
-          name
-          createdAt
-        }
-      }''';
-    var operation = Amplify.API.subscribe(
-        request: GraphQLRequest<String>(document: graphQLDocument),
-        onData: (event) {
-          print('Subscription event data received: ${event.data}');
-        },
-        onEstablished: () {
-          print('Subscription established');
-        },
-        onError: (dynamic e) {
-          print('Error occurred');
-          print(e);
-        },
-        onDone: () {
-          print('Subscription has been closed successfully');
-        });
-
-    void unsubscribe() {
-      operation.cancel();
+    onCreateBlog {
+      id
+      name
+      createdAt
     }
+  }''';
+    final Stream<GraphQLResponse<String>> operation = Amplify.API.subscribe(
+      GraphQLRequest<String>(document: graphQLDocument),
+      onEstablished: () => print('Subscription established'),
+    );
 
-    setState(() {
-      _unsubscribe = unsubscribe;
-    });
+    try {
+      await for (var event in operation) {
+        print('Subscription event data received: ${event.data}');
+      }
+    } on Exception catch (e) {
+      print('Error in subscription stream: $e');
+    }
   }
 
   Future<void> query() async {
@@ -153,7 +142,10 @@ class _GraphQLApiViewState extends State<GraphQLApiView> {
         const Padding(padding: EdgeInsets.all(10.0)),
         Center(
           child: ElevatedButton(
-            onPressed: widget.isAmplifyConfigured ? () => _unsubscribe() : null,
+            onPressed: () => setState(() {
+              _unsubscribe?.call();
+              _unsubscribe = null;
+            }),
             child: const Text('Unsubscribe'),
           ),
         ),

diff --git a/packages/amplify_api/ios/Classes/FlutterApiResponse.swift b/packages/amplify_api/ios/Classes/FlutterApiResponse.swift
@@ -124,14 +124,20 @@ class FlutterApiResponse {
         case .transformationError(_, let error):
             print("Received a partially successful GraphQL subscription event with a transformation error: \(error)")
             let details = FlutterApiErrorHandler.createSerializedError(error: error)
-            graphQLSubscriptionsStreamHandler.sendError(errorCode: "ApiException", details: details)
+            graphQLSubscriptionsStreamHandler.sendError(
+                errorCode: "ApiException",
+                id: id,
+                details: details)
         case .unknown(let errorDescription, let recoverySuggestion, let error):
             print("An unknown error occured: \(errorDescription)")
             let details = FlutterApiErrorHandler.createSerializedError(
                 error: APIError.unknown(errorDescription,
                                         recoverySuggestion,
                                         error))
-            graphQLSubscriptionsStreamHandler.sendError(errorCode: "ApiException", details: details)
+            graphQLSubscriptionsStreamHandler.sendError(
+                errorCode: "ApiException",
+                id: id,
+                details: details)
         }
     }
 }
diff --git a/packages/amplify_api/ios/Classes/FlutterGraphQLApi.swift b/packages/amplify_api/ios/Classes/FlutterGraphQLApi.swift
@@ -194,7 +194,8 @@ class FlutterGraphQLApi {
                         ]
                         graphQLSubscriptionsStreamHandler.sendEvent(
                             payload: payload,
-                            id: id, type: GraphQLSubscriptionEventTypes.DATA)
+                            id: id,
+                            type: GraphQLSubscriptionEventTypes.DATA)
                     case .failure(let errorResponse):
                         FlutterApiResponse.handleGraphQLErrorResponseEvent(
                             graphQLSubscriptionsStreamHandler: graphQLSubscriptionsStreamHandler,
@@ -220,7 +221,10 @@ class FlutterGraphQLApi {
                     }
                     if established {
                         let details = FlutterApiErrorHandler.createSerializedError(error: apiError)
-                        graphQLSubscriptionsStreamHandler.sendError(errorCode: "ApiException", details: details)
+                        graphQLSubscriptionsStreamHandler.sendError(
+                            errorCode: "ApiException",
+                            id: id,
+                            details: details)
                     } else {
                         FlutterApiErrorHandler.handleApiError(error: apiError, flutterResult: flutterResult)
                     }

diff --git a/packages/amplify_api/ios/Classes/GraphQLSubscriptionsStreamHandler.swift b/packages/amplify_api/ios/Classes/GraphQLSubscriptionsStreamHandler.swift
@@ -34,10 +34,10 @@ public class GraphQLSubscriptionsStreamHandler: NSObject, FlutterStreamHandler {
         eventSink?(result)
     }
 
-    func sendError(errorCode: String, details: [String: Any]) {
+    func sendError(errorCode: String, id: String, details: [String: Any]) {
         let flutterError = FlutterError(
             code: errorCode,
-            message: ErrorMessages.defaultFallbackErrorMessage,
+            message: id,
             details: details)
         eventSink?(flutterError)
     }

diff --git a/packages/amplify_api/lib/amplify_api.dart b/packages/amplify_api/lib/amplify_api.dart
@@ -62,18 +62,11 @@ class AmplifyAPI extends APIPluginInterface {
   }
 
   @override
-  GraphQLSubscriptionOperation<T> subscribe<T>(
-      {required GraphQLRequest<T> request,
-      required Function(GraphQLResponse<T>) onData,
-      Function()? onEstablished,
-      Function(dynamic)? onError,
-      Function()? onDone}) {
-    return _instance.subscribe(
-        request: request,
-        onEstablished: onEstablished,
-        onData: onData,
-        onError: onError,
-        onDone: onDone);
+  Stream<GraphQLResponse<T>> subscribe<T>(
+    GraphQLRequest<T> request, {
+    void Function()? onEstablished,
+  }) {
+    return _instance.subscribe(request, onEstablished: onEstablished);
   }
 
   // ====== RestAPI ======

diff --git a/packages/amplify_api/lib/graphql/graphql_subscription_event.dart b/packages/amplify_api/lib/graphql/graphql_subscription_event.dart
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2021 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+import 'package:amplify_api/amplify_api.dart';
+
+/// GraphQL subscription event types.
+enum GraphQLSubscriptionEventType {
+  /// Triggered when data is received from a GraphQL subscription.
+  data,
+
+  /// Triggered when a GraphQL subscription completes either due to cancellation
+  /// or a platform error.
+  done,
+
+  /// Triggered when a platform error occurs for a subscription.
+  ///
+  /// Currently only valid in Dart.
+  error,
+}
+
+/// Helper functions for [GraphQLSubscriptionEventType].
+extension GraphQLSubscriptionEventTypeX on GraphQLSubscriptionEventType {
+  /// Converts a platform string to a [GraphQLSubscriptionEventType].
+  static GraphQLSubscriptionEventType fromString(String type) {
+    switch (type) {
+      case 'DATA':
+        return GraphQLSubscriptionEventType.data;
+      case 'DONE':
+        return GraphQLSubscriptionEventType.done;
+      default:
+        throw UnsupportedError('Unsupported type: $type');
+    }
+  }
+}
+
+/// {@template graphql_subscription_event}
+/// An event which occurs on a GraphQL subscription.
+/// {@endtemplate}
+class GraphQLSubscriptionEvent {
+  /// The ID of the subscription.
+  final String subscriptionId;
+
+  /// The type of event.
+  final GraphQLSubscriptionEventType type;
+
+  /// The GraphQL response, if a data event.
+  final GraphQLResponse<String?>? rawResponse;
+
+  /// Platform error, if an error event.
+  final ApiException? error;
+
+  /// {@macro graphql_subscription_event}
+  const GraphQLSubscriptionEvent({
+    required this.subscriptionId,
+    required this.type,
+    this.rawResponse,
+    this.error,
+  });
+
+  /// Deserializes a platform channel event map.
+  static GraphQLSubscriptionEvent fromJson(Map json) {
+    GraphQLResponse<String?>? rawResponse;
+    final payload = json['payload'] as Map?;
+    if (payload != null) {
+      rawResponse = GraphQLResponse.raw(
+        data: payload['data'] as String?,
+        errors: (payload['errors'] as List?)
+            ?.cast<Map>()
+            .map((error) => GraphQLResponseError.fromJson(
+                  error.cast<String, dynamic>(),
+                ))
+            .toList(),
+      );
+    }
+    return GraphQLSubscriptionEvent(
+      subscriptionId: json['id'] as String,
+      type: GraphQLSubscriptionEventTypeX.fromString(json['type'] as String),
+      rawResponse: rawResponse,
+    );
+  }
+}