diff --git a/morpheus-graphql-subscriptions.cabal b/morpheus-graphql-subscriptions.cabal --- a/morpheus-graphql-subscriptions.cabal +++ b/morpheus-graphql-subscriptions.cabal @@ -1,11 +1,11 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.35.0. +-- This file has been generated from package.yaml by hpack version 0.36.0. -- -- see: https://github.com/sol/hpack name: morpheus-graphql-subscriptions -version: 0.27.3 +version: 0.28.0 synopsis: Morpheus GraphQL Subscriptions description: Build GraphQL APIs with your favourite functional language! category: web, graphql, subscriptions @@ -42,9 +42,9 @@ build-depends: aeson >=1.4.4 && <3.0.0 , base >=4.7.0 && <5.0.0 - , bytestring >=0.10.4 && <0.12.0 - , morpheus-graphql-app >=0.27.0 && <0.28.0 - , morpheus-graphql-core >=0.27.0 && <0.28.0 + , bytestring >=0.10.4 && <0.15.0 + , morpheus-graphql-app >=0.28.0 && <0.29.0 + , morpheus-graphql-core >=0.28.0 && <0.29.0 , mtl >=2.0.0 && <3.0.0 , relude >=0.3.0 && <2.0.0 , text >=1.2.3 && <3.0.0 diff --git a/src/Data/Morpheus/Subscriptions.hs b/src/Data/Morpheus/Subscriptions.hs --- a/src/Data/Morpheus/Subscriptions.hs +++ b/src/Data/Morpheus/Subscriptions.hs @@ -13,6 +13,7 @@ module Data.Morpheus.Subscriptions ( webSocketsApp, httpPubApp, + ApolloMessageType (..), PubApp (..), SubApp (..), Event (..), @@ -28,6 +29,9 @@ ( App, MapAPI (..), runApp, + ) +import Data.Morpheus.Subscriptions.Apollo + ( ApolloMessageType (..), ) import Data.Morpheus.Subscriptions.Event ( Event (..), diff --git a/src/Data/Morpheus/Subscriptions/Apollo.hs b/src/Data/Morpheus/Subscriptions/Apollo.hs --- a/src/Data/Morpheus/Subscriptions/Apollo.hs +++ b/src/Data/Morpheus/Subscriptions/Apollo.hs @@ -11,10 +11,12 @@ toApolloResponse, Validation, ApolloSubscription (..), + ApolloMessageType (..), ) where import Control.Applicative (Applicative (..)) +import Control.Monad.Fail (fail) import Control.Monad.IO.Class (MonadIO (..)) import Data.Aeson ( FromJSON (..), @@ -24,6 +26,7 @@ encode, pairs, withObject, + withText, (.:), (.:?), (.=), @@ -65,8 +68,11 @@ pendingRequest, ) import Prelude - ( Show, + ( Eq, + Show, String, + mempty, + return, ($), (.), ) @@ -75,16 +81,19 @@ data ApolloSubscription payload = ApolloSubscription { apolloId :: Maybe ID, - apolloType :: Text, + apolloType :: ApolloMessageType, apolloPayload :: Maybe payload } deriving (Show, Generic) -instance FromJSON a => FromJSON (ApolloSubscription a) where +instance (FromJSON a) => FromJSON (ApolloSubscription a) where parseJSON = withObject "ApolloSubscription" objectParser where objectParser o = - ApolloSubscription <$> o .:? "id" <*> o .: "type" <*> o .:? "payload" + ApolloSubscription + <$> o .:? "id" + <*> o .: "type" + <*> o .:? "payload" data RequestPayload = RequestPayload { payloadOperationName :: Maybe FieldName, @@ -102,12 +111,23 @@ <*> o .:? "query" <*> o .:? "variables" -instance ToJSON a => ToJSON (ApolloSubscription a) where +instance (ToJSON a) => ToJSON (ApolloSubscription a) where toEncoding (ApolloSubscription id' type' payload') = - pairs $ "id" .= id' <> "type" .= type' <> "payload" .= payload' + pairs $ + encodeMaybe "id" id' + <> "type" .= type' + <> encodeMaybe "payload" payload' + where + -- Messages should only include these fields when they have real values, + -- for example the MessageAck response should only include the type and optionally + -- extraneous data in the payload. + -- Aeson < 2.0.0 has Keys as Text, >= 2.0.0 has Data.Aeson.Key.Key + -- encodeMaybe :: ToJSON b => Text -> Maybe b -> Series + encodeMaybe _ Nothing = Prelude.mempty + encodeMaybe k (Just v) = k .= v acceptApolloRequest :: - MonadIO m => + (MonadIO m) => PendingConnection -> m Connection acceptApolloRequest pending = @@ -122,17 +142,61 @@ where apolloProtocol ["graphql-subscriptions"] = AcceptRequest (Just "graphql-subscriptions") [] - apolloProtocol ["graphql-ws"] = AcceptRequest (Just "graphql-ws") [] + apolloProtocol ["graphql-ws"] = + AcceptRequest (Just "graphql-ws") [] + apolloProtocol ["graphql-transport-ws"] = + AcceptRequest (Just "graphql-transport-ws") [] apolloProtocol _ = AcceptRequest Nothing [] -toApolloResponse :: ID -> GQLResponse -> ByteString -toApolloResponse sid val = - encode $ ApolloSubscription (Just sid) "data" (Just val) +toApolloResponse :: ApolloMessageType -> Maybe ID -> Maybe GQLResponse -> ByteString +toApolloResponse responseType sid_myb val_myb = + encode $ ApolloSubscription sid_myb responseType val_myb +data ApolloMessageType + = GqlConnectionAck + | GqlConnectionError + | GqlData + | GqlError + | GqlComplete + | GqlConnectionInit + | GqlSubscribe + | GqlPing + | GqlPong + deriving (Eq, Show, Generic) + +instance FromJSON ApolloMessageType where + parseJSON = withText "ApolloMessageType" txtParser + where + txtParser "connection_ack" = return GqlConnectionAck + txtParser "connection_error" = return GqlConnectionError + txtParser "next" = return GqlData + txtParser "error" = return GqlError + txtParser "complete" = return GqlComplete + txtParser "connection_init" = return GqlConnectionInit + txtParser "subscribe" = return GqlSubscribe + txtParser "ping" = return GqlPing + txtParser "pong" = return GqlPong + txtParser _ = fail "Invalid type encountered." + +instance ToJSON ApolloMessageType where + toEncoding = toEncoding . apolloResponseToProtocolMsgType + +apolloResponseToProtocolMsgType :: ApolloMessageType -> Text +apolloResponseToProtocolMsgType GqlConnectionAck = "connection_ack" +apolloResponseToProtocolMsgType GqlConnectionError = "connection_error" +apolloResponseToProtocolMsgType GqlConnectionInit = "connection_init" +apolloResponseToProtocolMsgType GqlData = "next" +apolloResponseToProtocolMsgType GqlError = "error" +apolloResponseToProtocolMsgType GqlComplete = "complete" +apolloResponseToProtocolMsgType GqlSubscribe = "subscribe" +apolloResponseToProtocolMsgType GqlPing = "ping" +apolloResponseToProtocolMsgType GqlPong = "pong" + data ApolloAction = SessionStop ID | SessionStart ID GQLRequest | ConnectionInit + | Ping type Validation = Either ByteString @@ -143,18 +207,23 @@ validateReq = either (Left . pack) validateSub ------------------------------------- validateSub :: ApolloSubscription RequestPayload -> Validation ApolloAction - validateSub ApolloSubscription {apolloType = "connection_init"} = + validateSub ApolloSubscription {apolloType = GqlConnectionInit} = pure ConnectionInit - validateSub ApolloSubscription {apolloType = "start", apolloId, apolloPayload} = + validateSub ApolloSubscription {apolloType = GqlPing} = + pure Ping + validateSub ApolloSubscription {apolloType = GqlSubscribe, apolloId, apolloPayload} = do sessionId <- validateSession apolloId payload <- validatePayload apolloPayload pure $ SessionStart sessionId payload - validateSub ApolloSubscription {apolloType = "stop", apolloId} = + validateSub ApolloSubscription {apolloType = GqlComplete, apolloId} = SessionStop <$> validateSession apolloId validateSub ApolloSubscription {apolloType} = - Left $ "Unknown Request type \"" <> pack (unpack apolloType) <> "\"." - -------------------------------------------- + Left $ + "Unknown Request type \"" + <> pack (unpack $ apolloResponseToProtocolMsgType apolloType) + <> "\"." + validateSession :: Maybe ID -> Validation ID validateSession = maybe (Left "\"id\" was not provided") Right ------------------------------------- diff --git a/src/Data/Morpheus/Subscriptions/ClientConnectionStore.hs b/src/Data/Morpheus/Subscriptions/ClientConnectionStore.hs --- a/src/Data/Morpheus/Subscriptions/ClientConnectionStore.hs +++ b/src/Data/Morpheus/Subscriptions/ClientConnectionStore.hs @@ -39,7 +39,8 @@ KeyOf (..), ) import Data.Morpheus.Subscriptions.Apollo - ( toApolloResponse, + ( ApolloMessageType (..), + toApolloResponse, ) import Data.Morpheus.Subscriptions.Event (Event (..)) import Data.Morpheus.Types.IO (GQLResponse) @@ -162,7 +163,7 @@ sessionId ClientSession { sessionChannel, - sessionCallback = fmap (toApolloResponse sid) . resolver + sessionCallback = fmap (toApolloResponse GqlData (Just sid) . Just) . resolver } clientSessions, clientConnections = HM.adjust (addConnectionSession sid) cid clientConnections, diff --git a/src/Data/Morpheus/Subscriptions/Stream.hs b/src/Data/Morpheus/Subscriptions/Stream.hs --- a/src/Data/Morpheus/Subscriptions/Stream.hs +++ b/src/Data/Morpheus/Subscriptions/Stream.hs @@ -36,6 +36,7 @@ ) import Data.Morpheus.Subscriptions.Apollo ( ApolloAction (..), + ApolloMessageType (..), apolloFormat, toApolloResponse, ) @@ -88,6 +89,10 @@ } -> ApiContext SUB event m +data WSOutputEvent e m + = WSUpdate (Updates e m) + | WSMessage ByteString + data Output (api :: API) @@ -95,7 +100,7 @@ (m :: Type -> Type) where SubOutput :: - { streamWS :: ApiContext SUB e m -> m (Either ByteString [Updates e m]) + { streamWS :: ApiContext SUB e m -> m (Either ByteString [WSOutputEvent e m]) } -> Output SUB e m PubOutput :: @@ -117,13 +122,16 @@ execute Publish {} = apolloError ["websocket can only handle subscriptions, not mutations"] - execute (Subscribe ch subRes) = Right $ startSession ch subRes session + execute (Subscribe ch subRes) = + Right . WSUpdate $ startSession ch subRes session -------------------------- - unfoldR Success {result = (events, _)} = traverse execute events - unfoldR Failure {errors} = apolloError (toList errors) + unfoldR Success {result = (events, _)} = + traverse execute events + unfoldR Failure {errors} = + apolloError (toList errors) -------------------------- apolloError :: [GQLError] -> Either ByteString a - apolloError = Left . toApolloResponse (sid session) . Errors + apolloError = Left . toApolloResponse GqlError (Just $ sid session) . Just . Errors handleWSRequest :: ( Monad m, @@ -143,16 +151,19 @@ handle = either (liftWS . Left) handleAction -------------------------------------------------- -- handleAction :: ApolloAction -> Stream SUB e m - handleAction ConnectionInit = liftWS $ Right [] + handleAction ConnectionInit = do + liftWS $ Right [WSMessage $ toApolloResponse GqlConnectionAck Nothing Nothing] + handleAction Ping = do + liftWS $ Right [WSMessage $ toApolloResponse GqlPong Nothing Nothing] handleAction (SessionStart sessionId request) = handleResponseStream (SessionID clientId sessionId) (gqlApp request) handleAction (SessionStop sessionId) = liftWS $ - Right [endSession (SessionID clientId sessionId)] + Right [WSUpdate $ endSession (SessionID clientId sessionId)] liftWS :: Applicative m => - Either ByteString [Updates e m] -> + Either ByteString [WSOutputEvent e m] -> Output SUB e m liftWS = SubOutput . const . pure @@ -163,7 +174,11 @@ m () runStreamWS scope@SubContext {callback} SubOutput {streamWS} = streamWS scope - >>= either callback (traverse_ (run scope)) + >>= either callback (traverse_ eventRunner) + where + -- eventRunner :: Monad m => WSOutputEvent e m -> m () + eventRunner (WSUpdate updates) = run scope updates + eventRunner (WSMessage msg) = callback msg runStreamHTTP :: (Monad m) => @@ -189,7 +204,7 @@ handle ws@SubContext {listener, callback} = do let runS (SubOutput x) = x ws bla <- listener >>= runS . handleWSRequest app clientId - pure $ (Updates (insertConnection clientId callback) :) <$> bla + pure $ ((WSUpdate $ Updates (insertConnection clientId callback)) :) <$> bla toOutStream app (Request req) = PubOutput $ handleResponseHTTP (app req)