Support subscriptions

This is experimental support.
The implementation is based on conduit and is boring. There is a new
resolver data constructor that should create a source event stream. The
executor receives the events, pipes them through the normal execution
and puts them into the response stream which is returned to the user.

- Tests are missing.
- The executor should check field value resolver on subscription types.
- The graphql function should probably return (Either
  ResponseEventStream Response), but I'm not sure about this. It will
  make the usage more complicated if no subscriptions are involved, but
  with the current API implementing subscriptions is more
  difficult than it should be.

2 files changed, 103 insertions, 8 deletions

diff --git a/src/Language/GraphQL/Execute/Execution.hs b/src/Language/GraphQL/Execute/Execution.hs
index e9ba4a7..fe4ad82 100644
--- a/src/Language/GraphQL/Execute/Execution.hs
+++ b/src/Language/GraphQL/Execute/Execution.hs
@@ -3,7 +3,9 @@
 {-# LANGUAGE ViewPatterns #-}
 
 module Language.GraphQL.Execute.Execution
-    ( executeSelectionSet
+    ( coerceArgumentValues
+    , collectFields
+    , executeSelectionSet
     ) where
 
 import Control.Monad.Trans.Class (lift)
@@ -32,10 +34,10 @@ import Prelude hiding (null)
 resolveFieldValue :: Monad m
     => Type.Value
     -> Type.Subs
-    -> Type.ResolverT m a
-    -> m (Either Text a)
-resolveFieldValue result args =
-    flip runReaderT context . runExceptT . Type.runResolverT
+    -> Type.Resolve m
+    -> m (Either Text Type.Value)
+resolveFieldValue result args resolver =
+    flip runReaderT context $ runExceptT resolver
   where
     context = Type.Context
         { Type.arguments = Type.Arguments args
@@ -101,12 +103,12 @@ instanceOf objectType (AbstractUnionType unionType) =
     go unionMemberType acc = acc || objectType == unionMemberType
 
 executeField :: (Monad m, Serialize a)
-    => Out.Field m
+    => Out.Resolver m
     -> Type.Value
     -> NonEmpty (Transform.Field m)
     -> CollectErrsT m a
-executeField fieldDefinition prev fields = do
-    let Out.Field _ fieldType argumentDefinitions resolver = fieldDefinition
+executeField (Out.ValueResolver fieldDefinition resolver) prev fields = do
+    let Out.Field _ fieldType argumentDefinitions = fieldDefinition
     let (Transform.Field _ _ arguments' _ :| []) = fields
     case coerceArgumentValues argumentDefinitions arguments' of
         Nothing -> addErrMsg "Argument coercing failed."
@@ -115,6 +117,7 @@ executeField fieldDefinition prev fields = do
             case answer of
                 Right result -> completeValue fieldType fields result
                 Left errorMessage -> addErrMsg errorMessage
+executeField _ _ _ = addErrMsg "No field value resolver specified."
 
 completeValue :: (Monad m, Serialize a)
     => Out.Type m
diff --git a/src/Language/GraphQL/Execute/Subscribe.hs b/src/Language/GraphQL/Execute/Subscribe.hs
new file mode 100644
index 0000000..ee9b116
--- /dev/null
+++ b/src/Language/GraphQL/Execute/Subscribe.hs
@@ -0,0 +1,92 @@
+{- This Source Code Form is subject to the terms of the Mozilla Public License,
+   v. 2.0. If a copy of the MPL was not distributed with this file, You can
+   obtain one at https://mozilla.org/MPL/2.0/. -}
+
+{-# LANGUAGE ExplicitForAll #-}
+{-# LANGUAGE OverloadedStrings #-}
+module Language.GraphQL.Execute.Subscribe
+    ( subscribe
+    ) where
+
+import Conduit
+import Control.Monad.Trans.Except (ExceptT(..), runExceptT)
+import Control.Monad.Trans.Reader (ReaderT(..), runReaderT)
+import Data.HashMap.Strict (HashMap)
+import qualified Data.HashMap.Strict as HashMap
+import qualified Data.Map.Strict as Map
+import qualified Data.List.NonEmpty as NonEmpty
+import Data.Sequence (Seq(..))
+import Data.Text (Text)
+import Language.GraphQL.AST (Name)
+import Language.GraphQL.Execute.Coerce
+import Language.GraphQL.Execute.Execution
+import qualified Language.GraphQL.Execute.Transform as Transform
+import Language.GraphQL.Error
+import qualified Language.GraphQL.Type.Definition as Definition
+import qualified Language.GraphQL.Type as Type
+import qualified Language.GraphQL.Type.Out as Out
+import Language.GraphQL.Type.Schema
+
+-- This is actually executeMutation, but we don't distinguish between queries
+-- and mutations yet.
+subscribe :: (Monad m, Serialize a)
+    => HashMap Name (Type m)
+    -> Out.ObjectType m
+    -> Seq (Transform.Selection m)
+    -> m (Either Text (ResponseEventStream m a))
+subscribe types' objectType fields = do
+    sourceStream <- createSourceEventStream types' objectType fields
+    traverse (mapSourceToResponseEvent types' objectType fields) sourceStream
+
+mapSourceToResponseEvent :: (Monad m, Serialize a)
+    => HashMap Name (Type m)
+    -> Out.ObjectType m
+    -> Seq (Transform.Selection m)
+    -> Out.SourceEventStream m
+    -> m (ResponseEventStream m a)
+mapSourceToResponseEvent types' subscriptionType fields sourceStream = pure
+    $ sourceStream
+    .| mapMC (executeSubscriptionEvent types' subscriptionType fields)
+
+createSourceEventStream :: Monad m
+    => HashMap Name (Type m)
+    -> Out.ObjectType m
+    -> Seq (Transform.Selection m)
+    -> m (Either Text (Out.SourceEventStream m))
+createSourceEventStream _types subscriptionType@(Out.ObjectType _ _ _ fieldTypes) fields
+    | [fieldGroup] <- Map.elems groupedFieldSet
+    , Transform.Field _ fieldName arguments' _ <- NonEmpty.head fieldGroup
+    , resolverT <- fieldTypes HashMap.! fieldName
+    , Out.EventStreamResolver fieldDefinition _ resolver <- resolverT
+    , Out.Field _ _fieldType argumentDefinitions <- fieldDefinition =
+        case coerceArgumentValues argumentDefinitions arguments' of
+            Nothing -> pure $ Left "Argument coercion failed."
+            Just argumentValues ->
+                resolveFieldEventStream Type.Null argumentValues resolver
+    | otherwise = pure $ Left "Subscription contains more than one field."
+  where
+    groupedFieldSet = collectFields subscriptionType fields
+
+resolveFieldEventStream :: Monad m
+    => Type.Value
+    -> Type.Subs
+    -> ExceptT Text (ReaderT Out.Context m) (Out.SourceEventStream m)
+    -> m (Either Text (Out.SourceEventStream m))
+resolveFieldEventStream result args resolver =
+    flip runReaderT context $ runExceptT resolver
+  where
+    context = Type.Context
+        { Type.arguments = Type.Arguments args
+        , Type.values = result
+        }
+
+-- This is actually executeMutation, but we don't distinguish between queries
+-- and mutations yet.
+executeSubscriptionEvent :: (Monad m, Serialize a)
+    => HashMap Name (Type m)
+    -> Out.ObjectType m
+    -> Seq (Transform.Selection m)
+    -> Definition.Value
+    -> m (Response a)
+executeSubscriptionEvent types' objectType fields initialValue =
+    runCollectErrs types' $ executeSelectionSet initialValue objectType fields