Merge pull request graphql-dotnet#431 from graphql-dotnet/subscriptions2

Add additional subscription support

Author: joemcbride

src/GraphQL.Tests/Subscription/SubscriptionSchema.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Collections.Concurrent;
 using System.Linq;
 using System.Reactive.Linq;
@@ -19,7 +19,7 @@ public ChatSchema(IChat chat)
         }
     }
 
-    public class ChatSubscriptions : ObjectGraphType<object>
+    public class ChatSubscriptions : ObjectGraphType
     {
         private readonly IChat _chat;
 

src/GraphQL/Builders/FieldBuilder.cs

@@ -1,6 +1,7 @@
 using System;
 using GraphQL.Types;
 using GraphQL.Resolvers;
+using GraphQL.Subscription;
 
 namespace GraphQL.Builders
 {
@@ -14,18 +15,18 @@ public static FieldBuilder<TSourceType, TReturnType> Create<TSourceType, TReturn
 
     public class FieldBuilder<TSourceType, TReturnType>
     {
-        private readonly FieldType _fieldType;
+        private readonly EventStreamFieldType _fieldType;
 
-        public FieldType FieldType => _fieldType;
+        public EventStreamFieldType FieldType => _fieldType;
 
-        private FieldBuilder(FieldType fieldType)
+        private FieldBuilder(EventStreamFieldType fieldType)
         {
             _fieldType = fieldType;
         }
 
         public static FieldBuilder<TSourceType, TReturnType> Create(Type type = null)
         {
-            var fieldType = new FieldType
+            var fieldType = new EventStreamFieldType
             {
                 Type = type,
                 Arguments = new QueryArguments(),
@@ -106,5 +107,11 @@ public FieldBuilder<TSourceType, TReturnType> Configure(Action<FieldType> config
             configure(FieldType);
             return this;
         }
+
+        public FieldBuilder<TSourceType, TReturnType> Subscribe(Func<ResolveEventStreamContext<TSourceType>, IObservable<TReturnType>> subscribe)
+        {
+            FieldType.Subscriber = new EventStreamResolver<TSourceType, TReturnType>(subscribe);
+            return this;
+        }
     }
 }

src/GraphQL/Execution/DocumentExecuter.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Collections;
 using System.Collections.Generic;
 using System.Linq;
@@ -81,14 +81,49 @@ public Task<ExecutionResult> ExecuteAsync(Action<ExecutionOptions> configure)
             return ExecuteAsync(options);
         }
 
+        protected virtual ExecutionResult InitializeResult(ExecutionOptions config)
+        {
+            var result = new ExecutionResult { Query = config.Query, ExposeExceptions = config.ExposeExceptions };
+            return result;
+        }
+
+        protected virtual async Task OnExecution(ExecutionOptions config, ExecutionContext context, ExecutionResult result)
+        {
+            var task = ExecuteOperationAsync(context).ConfigureAwait(false);
+
+            foreach (var listener in config.Listeners)
+            {
+                await listener.BeforeExecutionAwaitedAsync(config.UserContext, config.CancellationToken).ConfigureAwait(false);
+            }
+
+            result.Data = await task;
+        }
+
+        protected virtual void OnValidationError(IValidationResult validationResult, ExecutionResult result)
+        {
+            result.Data = null;
+            result.Errors = validationResult.Errors;
+        }
+
+        protected virtual void OnError(Exception exception, ExecutionResult result)
+        {
+            if (result.Errors == null)
+            {
+                result.Errors = new ExecutionErrors();
+            }
+
+            result.Data = null;
+            result.Errors.Add(new ExecutionError(exception.Message, exception));
+        }
+
         public async Task<ExecutionResult> ExecuteAsync(ExecutionOptions config)
         {
             var metrics = new Metrics();
             metrics.Start(config.OperationName);
 
             config.Schema.FieldNameConverter = config.FieldNameConverter;
 
-            var result = new ExecutionResult { Query = config.Query, ExposeExceptions = config.ExposeExceptions };
+            var result = InitializeResult(config);
             try
             {
                 if (!config.Schema.Initialized)
@@ -166,14 +201,7 @@ await listener.AfterValidationAsync(
                             await listener.BeforeExecutionAsync(config.UserContext, config.CancellationToken).ConfigureAwait(false);
                         }
 
-                        var task = ExecuteOperationAsync(context).ConfigureAwait(false);
-
-                        foreach (var listener in config.Listeners)
-                        {
-                            await listener.BeforeExecutionAwaitedAsync(config.UserContext, config.CancellationToken).ConfigureAwait(false);
-                        }
-
-                        result.Data = await task;
+                        await OnExecution(config, context, result);
 
                         foreach (var listener in config.Listeners)
                         {
@@ -188,21 +216,14 @@ await listener.AfterValidationAsync(
                 }
                 else
                 {
-                    result.Data = null;
-                    result.Errors = validationResult.Errors;
+                    OnValidationError(validationResult, result);
                 }
 
                 return result;
             }
             catch (Exception exc)
             {
-                if (result.Errors == null)
-                {
-                    result.Errors = new ExecutionErrors();
-                }
-
-                result.Data = null;
-                result.Errors.Add(new ExecutionError(exc.Message, exc));
+                OnError(exc, result);
                 return result;
             }
             finally
@@ -237,7 +258,7 @@ public ExecutionContext BuildExecutionContext(
             return context;
         }
 
-        protected Operation GetOperation(string operationName, Document document)
+        protected virtual Operation GetOperation(string operationName, Document document)
         {
             var operation = !string.IsNullOrWhiteSpace(operationName)
                 ? document.Operations.WithName(operationName)

src/GraphQL/Resolvers/EventStreamResolver.cs

@@ -1,5 +1,4 @@
-using System;
-using GraphQL.Execution;
+using System;
 using GraphQL.Subscription;
 
 namespace GraphQL.Resolvers
@@ -24,4 +23,25 @@ IObservable<object> IEventStreamResolver.Subscribe(ResolveEventStreamContext con
             return (IObservable<object>)Subscribe(context);
         }
     }
+
+    public class EventStreamResolver<TSourceType, TReturnType> : IEventStreamResolver<TReturnType>
+    {
+        private readonly Func<ResolveEventStreamContext<TSourceType>, IObservable<TReturnType>> _subscriber;
+
+        public EventStreamResolver(
+            Func<ResolveEventStreamContext<TSourceType>, IObservable<TReturnType>> subscriber)
+        {
+            _subscriber = subscriber ?? throw new ArgumentNullException(nameof(subscriber));
+        }
+
+        public IObservable<TReturnType> Subscribe(ResolveEventStreamContext context)
+        {
+            return _subscriber(context.As<TSourceType>());
+        }
+
+        IObservable<object> IEventStreamResolver.Subscribe(ResolveEventStreamContext context)
+        {
+            return (IObservable<object>)Subscribe(context);
+        }
+    }
 }

src/GraphQL/Resolvers/IEventStreamResolver.cs

@@ -1,5 +1,4 @@
-using System;
-using GraphQL.Execution;
+using System;
 using GraphQL.Subscription;
 
 namespace GraphQL.Resolvers

src/GraphQL/Subscription/ResolveEventStreamContext.cs

@@ -1,8 +1,38 @@
-using GraphQL.Types;
+using GraphQL.Types;
 
 namespace GraphQL.Subscription
 {
-    public class ResolveEventStreamContext : ResolveFieldContext<object>
+    public class ResolveEventStreamContext<T> : ResolveFieldContext<T>
     {
+        public ResolveEventStreamContext() { }
+
+        public ResolveEventStreamContext(ResolveEventStreamContext context)
+        {
+            Source = (T)context.Source;
+            FieldName = context.FieldName;
+            FieldAst = context.FieldAst;
+            FieldDefinition = context.FieldDefinition;
+            ReturnType = context.ReturnType;
+            ParentType = context.ParentType;
+            Arguments = context.Arguments;
+            Schema = context.Schema;
+            Document = context.Document;
+            Fragments = context.Fragments;
+            RootValue = context.RootValue;
+            UserContext = context.UserContext;
+            Operation = context.Operation;
+            Variables = context.Variables;
+            CancellationToken = context.CancellationToken;
+            Metrics = context.Metrics;
+            Errors = context.Errors;
+        }
+    }
+
+    public class ResolveEventStreamContext : ResolveEventStreamContext<object>
+    {
+        internal ResolveEventStreamContext<TSourceType> As<TSourceType>()
+        {
+            return new ResolveEventStreamContext<TSourceType>(this);
+        }
     }
 }