Add Angel functionality for handling subscription streams

angel_graphql/lib/src/graphql_http.dart

@@ -1,3 +1,4 @@
+import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 import 'package:angel_framework/angel_framework.dart';
@@ -19,13 +20,30 @@ final Validator graphQlPostBody = new Validator({
 ///
 /// Follows the guidelines listed here:
 /// https://graphql.org/learn/serving-over-http/
-RequestHandler graphQLHttp(GraphQL graphQL) {
+RequestHandler graphQLHttp(GraphQL graphQL,
+    {Function(RequestContext, ResponseContext, Stream<Map<String, dynamic>>)
+        onSubscription}) {
   return (req, res) async {
     var globalVariables = <String, dynamic>{
       '__requestctx': req,
       '__responsectx': res,
     };
 
+    sendGraphQLResponse(result) async {
+      if (result is Stream<Map<String, dynamic>>) {
+        if (onSubscription == null) {
+          throw StateError(
+              'The GraphQL backend returned a Stream, but no `onSubscription` callback was provided.');
+        } else {
+          return await onSubscription(req, res, result);
+        }
+      }
+
+      return {
+        'data': result,
+      };
+    }
+
     executeMap(Map map) async {
       var body = await req.parseBody().then((_) => req.bodyAsMap);
       var text = body['query'] as String;
@@ -36,15 +54,13 @@ RequestHandler graphQLHttp(GraphQL graphQL) {
         variables = json.decode(variables as String);
       }
 
-      return {
+      return await sendGraphQLResponse(await graphQL.parseAndExecute(
-        'data': await graphQL.parseAndExecute(
         text,
         sourceUrl: 'input',
         operationName: operationName,
         variableValues: foldToStringDynamic(variables as Map),
         globalVariables: globalVariables,
-        ),
+      ));
-      };
     }
 
     try {
@@ -55,13 +71,11 @@ RequestHandler graphQLHttp(GraphQL graphQL) {
       } else if (req.method == 'POST') {
         if (req.headers.contentType?.mimeType == graphQlContentType.mimeType) {
           var text = await req.body.transform(utf8.decoder).join();
-          return {
+          return sendGraphQLResponse(await graphQL.parseAndExecute(
-            'data': await graphQL.parseAndExecute(
             text,
             sourceUrl: 'input',
             globalVariables: globalVariables,
-            ),
+          ));
-          };
         } else if (req.headers.contentType?.mimeType == 'application/json') {
           if (await validate(graphQlPostBody)(req, res) as bool) {
             return await executeMap(req.bodyAsMap);

angel_graphql/pubspec.yaml

@@ -7,6 +7,7 @@ environment:
   sdk: ">=2.0.0-dev <3.0.0"
 dependencies:
   angel_framework: ^2.0.0-alpha
+  angel_websocket: ^2.0.0
   angel_validate: ^2.0.0-alpha
   graphql_parser: ^1.0.0
   graphql_schema: ^1.0.0
@@ -15,3 +16,6 @@ dependencies:
 dev_dependencies:
   angel_serialize: ^2.0.0
   logging: ^0.11.0
+dependency_overrides:
+  graphql_server:
+    path: ../graphql_server