packages feed

morpheus-graphql-subscriptions (empty) → 0.16.0

raw patch · 12 files changed

+1142/−0 lines, 12 filesdep +aesondep +basedep +bytestring

Dependencies added: aeson, base, bytestring, directory, morpheus-graphql-core, mtl, relude, tasty, tasty-hunit, text, transformers, unliftio-core, unordered-containers, uuid, websockets

Files

+ LICENSE view
@@ -0,0 +1,21 @@+MIT License++Copyright (c) 2019 Daviti Nalchevanidze++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE+SOFTWARE.
+ README.md view
@@ -0,0 +1,3 @@+# Morpheus GraphQL Subscriptions++provides tools for GraphQL subscriptions
+ changelog.md view
@@ -0,0 +1,3 @@+# Changelog++## 0.16.0 - 05.11.2020
+ morpheus-graphql-subscriptions.cabal view
@@ -0,0 +1,83 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.33.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: cf18d3ec9c4893a697af764d63a2bd1467bfaf0502d3b88cbfd85d6c30977355++name:           morpheus-graphql-subscriptions+version:        0.16.0+synopsis:       Morpheus GraphQL Subscriptions+description:    Build GraphQL APIs with your favourite functional language!+category:       web, graphql, subscriptions+homepage:       https://morpheusgraphql.com+bug-reports:    https://github.com/nalchevanidze/morpheus-graphql/issues+author:         Daviti Nalchevanidze+maintainer:     d.nalchevanidze@gmail.com+copyright:      (c) 2019 Daviti Nalchevanidze+license:        MIT+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    changelog.md+    README.md++source-repository head+  type: git+  location: https://github.com/nalchevanidze/morpheus-graphql++library+  exposed-modules:+      Data.Morpheus.Subscriptions+      Data.Morpheus.Subscriptions.Internal+  other-modules:+      Data.Morpheus.Subscriptions.Apollo+      Data.Morpheus.Subscriptions.ClientConnectionStore+      Data.Morpheus.Subscriptions.Event+      Data.Morpheus.Subscriptions.Stream+      Data.Morpheus.Subscriptions.WebSockets+      Paths_morpheus_graphql_subscriptions+  hs-source-dirs:+      src+  ghc-options: -Wall+  build-depends:+      aeson >=1.4.4.0 && <=1.6+    , base >=4.7 && <5+    , bytestring >=0.10.4 && <0.11+    , morpheus-graphql-core >=0.16.0 && <0.17.0+    , mtl >=2.0 && <=3.0+    , relude >=0.3.0+    , text >=1.2.3.0 && <1.3+    , transformers >=0.3.0.0 && <1.0+    , unliftio-core >=0.0.1 && <=0.4+    , unordered-containers >=0.2.8.0 && <0.3+    , uuid >=1.0 && <=1.4+    , websockets >=0.11.0 && <=1.0+  default-language: Haskell2010++test-suite morpheus-subscription-test+  type: exitcode-stdio-1.0+  main-is: Spec.hs+  other-modules:+      Paths_morpheus_graphql_subscriptions+  hs-source-dirs:+      test+  ghc-options: -Wall+  build-depends:+      aeson+    , base >=4.7 && <5+    , bytestring >=0.10.4 && <0.11+    , directory >=1.0+    , morpheus-graphql-core >=0.16.0 && <0.17.0+    , mtl >=2.0 && <=3.0+    , relude >=0.3.0+    , tasty+    , tasty-hunit+    , text >=1.2.3.0 && <1.3+    , transformers >=0.3.0.0 && <1.0+    , unliftio-core >=0.0.1 && <=0.4+    , unordered-containers >=0.2.8.0 && <0.3+    , uuid >=1.0 && <=1.4+    , websockets >=0.11.0 && <=1.0+  default-language: Haskell2010
+ src/Data/Morpheus/Subscriptions.hs view
@@ -0,0 +1,105 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE NoImplicitPrelude #-}++-- |  GraphQL Wai Server Applications+module Data.Morpheus.Subscriptions+  ( webSocketsApp,+    httpPubApp,+    PubApp (..),+    SubApp (..),+    Event (..),+    ServerApp,+    Hashable,+  )+where++import Control.Monad.IO.Unlift+  ( MonadUnliftIO,+  )+import Data.Morpheus.Core+  ( App,+    runApp,+  )+import Data.Morpheus.Subscriptions.Event+  ( Event (..),+    runEvents,+  )+import Data.Morpheus.Subscriptions.Internal+  ( ApiContext (..),+    Input (..),+    connectionThread,+    initDefaultStore,+    publishEventWith,+    runStreamHTTP,+    streamApp,+  )+import Data.Morpheus.Subscriptions.WebSockets+  ( webSocketsWrapper,+  )+import Data.Morpheus.Types.IO (MapAPI (..))+import Network.WebSockets+  ( ServerApp,+  )+import Relude++httpPubApp ::+  ( PubApp e,+    MapAPI a b,+    MonadIO m+  ) =>+  [e -> m ()] ->+  App e m ->+  a ->+  m b+httpPubApp = runPubApp++-- | Wai WebSocket Server App for GraphQL subscriptions+webSocketsApp ::+  ( MonadUnliftIO m,+    MonadIO m,+    SubApp ServerApp e+  ) =>+  App e m ->+  m (ServerApp, e -> m ())+webSocketsApp = runSubApp++class SubApp app e where+  runSubApp :: (MonadIO m, MonadUnliftIO m) => App e m -> m (app, e -> m ())++class PubApp e where+  runPubApp :: (MonadIO m, MapAPI a b) => [e -> m ()] -> App e m -> a -> m b++instance (Show ch, Eq ch, Hashable ch) => SubApp ServerApp (Event ch con) where+  runSubApp app = do+    store <- initDefaultStore+    wsApp <- webSocketsWrapper store (connectionThread app)+    pure+      ( wsApp,+        publishEventWith store+      )++instance (Show ch, Eq ch, Hashable ch) => PubApp (Event ch con) where+  runPubApp [] app = runApp app+  runPubApp callbacks app =+    mapAPI $+      runStreamHTTP PubContext {eventPublisher = runEvents callbacks}+        . streamApp app+        . Request++instance SubApp ServerApp () where+  runSubApp _ =+    pure+      ( const $ pure (),+        const $ pure ()+      )++instance PubApp () where+  runPubApp _ = runApp
+ src/Data/Morpheus/Subscriptions/Apollo.hs view
@@ -0,0 +1,167 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE NoImplicitPrelude #-}++module Data.Morpheus.Subscriptions.Apollo+  ( ApolloAction (..),+    apolloFormat,+    acceptApolloRequest,+    toApolloResponse,+    Validation,+  )+where++import Control.Applicative (Applicative (..))+import Control.Monad.IO.Class (MonadIO (..))+import Data.Aeson+  ( (.:),+    (.:?),+    (.=),+    FromJSON (..),+    ToJSON (..),+    Value (..),+    eitherDecode,+    encode,+    pairs,+    withObject,+  )+import Data.ByteString.Lazy.Char8+  ( ByteString,+    pack,+  )+import Data.Either+  ( Either (..),+    either,+  )+import Data.Functor ((<$>))+import Data.Maybe+  ( Maybe (..),+    maybe,+  )+import Data.Morpheus.Types.IO+  ( GQLRequest (..),+    GQLResponse,+  )+import Data.Morpheus.Types.Internal.AST+  ( FieldName,+    Token,+  )+import Data.Semigroup ((<>))+import Data.Text+  ( Text,+    unpack,+  )+import GHC.Generics (Generic)+import Network.WebSockets+  ( AcceptRequest (..),+    Connection,+    PendingConnection,+    RequestHead,+    acceptRequestWith,+    getRequestSubprotocols,+    pendingRequest,+  )+import Prelude+  ( ($),+    (.),+    Show,+    String,+  )++type ID = Text++data ApolloSubscription payload = ApolloSubscription+  { apolloId :: Maybe ID,+    apolloType :: Text,+    apolloPayload :: Maybe payload+  }+  deriving (Show, Generic)++instance FromJSON a => FromJSON (ApolloSubscription a) where+  parseJSON = withObject "ApolloSubscription" objectParser+    where+      objectParser o =+        ApolloSubscription <$> o .:? "id" <*> o .: "type" <*> o .:? "payload"++data RequestPayload = RequestPayload+  { payloadOperationName :: Maybe FieldName,+    payloadQuery :: Maybe Token,+    payloadVariables :: Maybe Value+  }+  deriving (Show, Generic)++instance FromJSON RequestPayload where+  parseJSON = withObject "ApolloPayload" objectParser+    where+      objectParser o =+        RequestPayload <$> o .:? "operationName" <*> o .:? "query"+          <*> o .:? "variables"++instance ToJSON a => ToJSON (ApolloSubscription a) where+  toEncoding (ApolloSubscription id' type' payload') =+    pairs $ "id" .= id' <> "type" .= type' <> "payload" .= payload'++acceptApolloRequest ::+  MonadIO m =>+  PendingConnection ->+  m Connection+acceptApolloRequest pending =+  liftIO $+    acceptRequestWith+      pending+      (acceptApolloSubProtocol (pendingRequest pending))++acceptApolloSubProtocol :: RequestHead -> AcceptRequest+acceptApolloSubProtocol reqHead =+  apolloProtocol (getRequestSubprotocols reqHead)+  where+    apolloProtocol ["graphql-subscriptions"] =+      AcceptRequest (Just "graphql-subscriptions") []+    apolloProtocol ["graphql-ws"] = AcceptRequest (Just "graphql-ws") []+    apolloProtocol _ = AcceptRequest Nothing []++toApolloResponse :: ID -> GQLResponse -> ByteString+toApolloResponse sid val =+  encode $ ApolloSubscription (Just sid) "data" (Just val)++data ApolloAction+  = SessionStop ID+  | SessionStart ID GQLRequest+  | ConnectionInit++type Validation = Either ByteString++apolloFormat :: ByteString -> Validation ApolloAction+apolloFormat = validateReq . eitherDecode+  where+    validateReq :: Either String (ApolloSubscription RequestPayload) -> Validation ApolloAction+    validateReq = either (Left . pack) validateSub+    -------------------------------------+    validateSub :: ApolloSubscription RequestPayload -> Validation ApolloAction+    validateSub ApolloSubscription {apolloType = "connection_init"} =+      pure ConnectionInit+    validateSub ApolloSubscription {apolloType = "start", apolloId, apolloPayload} =+      do+        sessionId <- validateSession apolloId+        payload <- validatePayload apolloPayload+        pure $ SessionStart sessionId payload+    validateSub ApolloSubscription {apolloType = "stop", apolloId} =+      SessionStop <$> validateSession apolloId+    validateSub ApolloSubscription {apolloType} =+      Left $ "Unknown Request type \"" <> pack (unpack apolloType) <> "\"."+    --------------------------------------------+    validateSession :: Maybe ID -> Validation ID+    validateSession = maybe (Left "\"id\" was not provided") Right+    -------------------------------------+    validatePayload = maybe (Left "\"payload\" was not provided") validatePayloadContent+    -------------------------------------+    validatePayloadContent+      RequestPayload+        { payloadQuery,+          payloadOperationName = operationName,+          payloadVariables = variables+        } = do+        query <- maybe (Left "\"payload.query\" was not provided") Right payloadQuery+        pure $ GQLRequest {query, operationName, variables}
+ src/Data/Morpheus/Subscriptions/ClientConnectionStore.hs view
@@ -0,0 +1,256 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE NoImplicitPrelude #-}++module Data.Morpheus.Subscriptions.ClientConnectionStore+  ( SessionID (..),+    ClientConnectionStore,+    ClientConnection,+    Updates (..),+    startSession,+    endSession,+    empty,+    insertConnection,+    delete,+    publish,+    toList,+    connectionSessionIds,+    storedChannels,+    storedSessions,+  )+where++import Data.ByteString.Lazy.Char8 (ByteString)+import qualified Data.HashMap.Lazy as HM+import Data.Morpheus.Internal.Utils+  ( Collection (..),+    KeyOf (..),+  )+import Data.Morpheus.Subscriptions.Apollo+  ( toApolloResponse,+  )+import Data.Morpheus.Subscriptions.Event (Event (..))+import Data.Morpheus.Types.IO (GQLResponse)+import Data.Morpheus.Types.Internal.Resolving+  ( EventHandler (..),+  )+import Data.UUID (UUID)+import Relude hiding+  ( ByteString,+    Show,+    empty,+    show,+    toList,+  )+import Prelude (Show (..))++data SessionID = SessionID+  { cid :: UUID,+    sid :: Text+  }+  deriving (Show, Generic, Eq, Hashable)++data ClientConnection (m :: * -> *) = ClientConnection+  { connectionId :: UUID,+    connectionCallback :: ByteString -> m (),+    -- one connection can have multiple subscription session+    connectionSessionIds :: [Text]+  }++addConnectionSession ::+  Text ->+  ClientConnection m ->+  ClientConnection m+addConnectionSession+  sid+  ClientConnection {..} = ClientConnection {connectionSessionIds = connectionSessionIds <> [sid], ..}++data ClientSession e (m :: * -> *) = ClientSession+  { sessionChannel :: Channel e,+    sessionCallback :: e -> m ByteString+  }++instance Show (ClientSession e m) where+  show ClientSession {} = "ClientSession"++instance Show (ClientConnection m) where+  show ClientConnection {connectionId, connectionSessionIds} =+    "ClientConnection { id = "+      <> show connectionId+      <> ", sessions = "+      <> show connectionSessionIds+      <> " }"++mapAt :: (Eq k, Hashable k) => c -> (a -> c) -> k -> HashMap k a -> c+mapAt fallback f key = maybe fallback f . HM.lookup key++publish ::+  ( Monad m,+    Eq channel,+    Hashable channel,+    Show channel+  ) =>+  Event channel content ->+  ClientConnectionStore (Event channel content) m ->+  m ()+publish event@Event {channels} ClientConnectionStore {activeChannels, clientConnections, clientSessions} =+  traverse_ onChannel channels+  where+    onChannel ch = mapAt (pure ()) sendBy ch activeChannels+    sendBy = traverse_ sendByChannel+    sendByChannel sid = mapAt (pure ()) sendMessage sid clientSessions+      where+        sendMessage ClientSession {sessionCallback} = sessionCallback event >>= upd+        upd = mapAt cantFindConnection connectionCallback (cid sid) clientConnections+        cantFindConnection _ = pure ()++newtype Updates e (m :: * -> *) = Updates+  { _runUpdate :: ClientConnectionStore e m -> ClientConnectionStore e m+  }++endSession ::+  ( Eq ch,+    Hashable ch+  ) =>+  SessionID ->+  Updates (Event ch con) m+endSession sessionId@SessionID {sid, cid} = Updates endSub+  where+    endSub ::+      ( Eq ch,+        Hashable ch+      ) =>+      ClientConnectionStore (Event ch con) m ->+      ClientConnectionStore (Event ch con) m+    endSub ClientConnectionStore {..} =+      ClientConnectionStore+        { clientConnections = HM.adjust (removeSessionId sid) cid clientConnections,+          clientSessions = HM.delete sessionId clientSessions,+          activeChannels = removeActiveChannel sessionId activeChannels+        }++removeSessionId :: Text -> ClientConnection m -> ClientConnection m+removeSessionId sid conn =+  conn+    { connectionSessionIds = filter (/= sid) (connectionSessionIds conn)+    }++startSession ::+  ( Monad m,+    Eq (Channel e),+    Hashable (Channel e)+  ) =>+  Channel e ->+  (e -> m GQLResponse) ->+  SessionID ->+  Updates e m+startSession sessionChannel resolver sessionId@SessionID {cid, sid} = Updates startSub+  where+    startSub ClientConnectionStore {..} =+      ClientConnectionStore+        { clientSessions =+            HM.insert+              sessionId+              ClientSession+                { sessionChannel,+                  sessionCallback = fmap (toApolloResponse sid) . resolver+                }+              clientSessions,+          clientConnections = HM.adjust (addConnectionSession sid) cid clientConnections,+          activeChannels = addActiveChannel sessionChannel sessionId activeChannels+        }++addActiveChannel ::+  (Eq ch, Hashable ch) =>+  ch ->+  SessionID ->+  HashMap ch [SessionID] ->+  HashMap ch [SessionID]+addActiveChannel sessionChannel sessionId = HM.alter update sessionChannel+  where+    update Nothing = Just [sessionId]+    update (Just ids) = Just (ids <> [sessionId])++removeActiveChannel ::+  (Eq ch, Hashable ch) =>+  SessionID ->+  HashMap ch [SessionID] ->+  HashMap ch [SessionID]+removeActiveChannel sessionId = fmap update+  where+    update = filter (/= sessionId)++-- stores active client connections+-- every registered client has ID+-- when client connection is closed client(including all its subscriptions) can By removed By its ID+data ClientConnectionStore e (m :: * -> *) where+  ClientConnectionStore ::+    { clientConnections :: HashMap UUID (ClientConnection m),+      clientSessions :: HashMap SessionID (ClientSession (Event channel content) m),+      activeChannels :: HashMap channel [SessionID]+    } ->+    ClientConnectionStore (Event channel content) m++deriving instance+  Show e =>+  Show (ClientConnectionStore (Event e c) m)++type StoreMap e m =+  ClientConnectionStore e m ->+  ClientConnectionStore e m++toList :: ClientConnectionStore (Event channel content) m -> [(UUID, ClientConnection m)]+toList = HM.toList . clientConnections++storedSessions :: ClientConnectionStore (Event channel content) m -> [(SessionID, ClientSession (Event channel content) m)]+storedSessions = HM.toList . clientSessions++storedChannels :: ClientConnectionStore (Event channel content) m -> [(channel, [SessionID])]+storedChannels = HM.toList . activeChannels++instance KeyOf UUID (ClientConnection m) where+  keyOf = connectionId++instance Collection (ClientConnection m) (ClientConnectionStore (Event ch con) m) where+  empty = ClientConnectionStore empty HM.empty HM.empty+  singleton x = ClientConnectionStore (singleton x) HM.empty HM.empty++mapConnections ::+  ( HashMap UUID (ClientConnection m) -> HashMap UUID (ClientConnection m)+  ) ->+  ClientConnectionStore e m ->+  ClientConnectionStore e m+mapConnections f ClientConnectionStore {..} =+  ClientConnectionStore+    { clientConnections = f clientConnections,+      ..+    }++-- returns original store, if connection with same id already exist+insertConnection ::+  UUID ->+  (ByteString -> m ()) ->+  StoreMap e m+insertConnection connectionId connectionCallback =+  mapConnections (HM.insertWith (const id) connectionId c)+  where+    c =+      ClientConnection+        { connectionId,+          connectionCallback,+          connectionSessionIds = empty+        }++delete ::+  UUID ->+  StoreMap e m+delete key = mapConnections (HM.delete key)
+ src/Data/Morpheus/Subscriptions/Event.hs view
@@ -0,0 +1,31 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE NoImplicitPrelude #-}++module Data.Morpheus.Subscriptions.Event+  ( Event (..),+    runEvents,+  )+where++import Data.Morpheus.Types.Internal.Resolving+  ( EventHandler (..),+  )+import Relude++data Event ch con = Event+  { channels :: [ch],+    content :: con+  }++instance EventHandler (Event ch con) where+  type Channel (Event ch con) = ch+  getChannels = channels++runEvents ::+  (Foldable t, Applicative f) =>+  t (event -> f b) ->+  event ->+  f ()+runEvents fs e = traverse_ (e &) fs
+ src/Data/Morpheus/Subscriptions/Internal.hs view
@@ -0,0 +1,160 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE NoImplicitPrelude #-}++module Data.Morpheus.Subscriptions.Internal+  ( connect,+    disconnect,+    connectionThread,+    runStreamWS,+    runStreamHTTP,+    ApiContext (..),+    Input (..),+    PUB,+    SUB,+    Store (..),+    ClientConnectionStore,+    acceptApolloRequest,+    SessionID,+    publish,+    initDefaultStore,+    publishEventWith,+    empty,+    toList,+    connectionSessionIds,+    storedSessions,+    storedChannels,+    streamApp,+  )+where++import Control.Concurrent+  ( modifyMVar_,+  )+import Control.Exception (finally)+import Control.Monad.IO.Unlift+  ( MonadUnliftIO,+    withRunInIO,+  )+-- MORPHEUS++import Data.Morpheus.Core+  ( App,+    runAppStream,+  )+import Data.Morpheus.Internal.Utils+  ( empty,+  )+import Data.Morpheus.Subscriptions.Apollo+  ( acceptApolloRequest,+  )+import Data.Morpheus.Subscriptions.ClientConnectionStore+  ( ClientConnectionStore,+    SessionID,+    connectionSessionIds,+    delete,+    publish,+    storedChannels,+    storedSessions,+    toList,+  )+import Data.Morpheus.Subscriptions.Event+  ( Event,+  )+import Data.Morpheus.Subscriptions.Stream+  ( ApiContext (..),+    Input (..),+    Output,+    PUB,+    SUB,+    runStreamHTTP,+    runStreamWS,+    toOutStream,+  )+import Data.UUID.V4 (nextRandom)+import Relude hiding (empty, toList)++connect :: MonadIO m => m (Input SUB)+connect = InitConnection <$> liftIO nextRandom++disconnect :: ApiContext SUB e m -> Input SUB -> m ()+disconnect SubContext {updateStore} (InitConnection clientID) =+  updateStore (delete clientID)++-- | PubSubStore interface+-- shared GraphQL state between __websocket__ and __http__ server,+-- you can define your own store if you provide write and read methods+-- to work properly Morpheus needs all entries of ClientConnectionStore (+ client Callbacks)+-- that why it is recomended that you use many local ClientStores on evenry server node+-- rathen then single centralized Store.+data Store e m = Store+  { readStore :: m (ClientConnectionStore e m),+    writeStore :: (ClientConnectionStore e m -> ClientConnectionStore e m) -> m ()+  }++publishEventWith ::+  ( MonadIO m,+    Eq channel,+    Hashable channel,+    Show channel+  ) =>+  Store (Event channel cont) m ->+  Event channel cont ->+  m ()+publishEventWith store event = readStore store >>= publish event++-- | initializes empty GraphQL state+initDefaultStore ::+  ( MonadIO m,+    MonadIO m2+  ) =>+  m2 (Store (Event ch con) m)+initDefaultStore = do+  store <- liftIO $ newMVar empty+  pure+    Store+      { readStore = liftIO $ readMVar store,+        writeStore = \changes -> liftIO $ modifyMVar_ store (return . changes)+      }++finallyM :: MonadUnliftIO m => m () -> m () -> m ()+finallyM loop end = withRunInIO $ \runIO -> finally (runIO loop) (runIO end)++connectionThread ::+  ( MonadUnliftIO m,+    Eq ch,+    Hashable ch+  ) =>+  App (Event ch con) m ->+  ApiContext SUB (Event ch con) m ->+  m ()+connectionThread api scope = do+  input <- connect+  finallyM+    (connectionLoop api scope input)+    (disconnect scope input)++connectionLoop ::+  (Monad m, Eq ch, Hashable ch) =>+  App (Event ch con) m ->+  ApiContext SUB (Event ch con) m ->+  Input SUB ->+  m ()+connectionLoop app scope =+  forever+    . runStreamWS scope+    . streamApp app++streamApp ::+  (Eq ch, Monad m, Hashable ch) =>+  App (Event ch con) m ->+  Input api ->+  Output api (Event ch con) m+streamApp app = toOutStream (runAppStream app)
+ src/Data/Morpheus/Subscriptions/Stream.hs view
@@ -0,0 +1,232 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE NoImplicitPrelude #-}++module Data.Morpheus.Subscriptions.Stream+  ( toOutStream,+    runStreamWS,+    runStreamHTTP,+    ApiContext (..),+    Input (..),+    Output,+    API (..),+    PUB,+    SUB,+  )+where++import Data.ByteString.Lazy.Char8 (ByteString)+import Data.Morpheus.Error+  ( globalErrorMessage,+  )+import Data.Morpheus.Internal.Utils+  ( failure,+  )+import Data.Morpheus.Subscriptions.Apollo+  ( ApolloAction (..),+    apolloFormat,+    toApolloResponse,+  )+import Data.Morpheus.Subscriptions.ClientConnectionStore+  ( ClientConnectionStore,+    SessionID (..),+    Updates (..),+    endSession,+    insertConnection,+    startSession,+  )+import Data.Morpheus.Subscriptions.Event (Event (..))+import Data.Morpheus.Types.IO+  ( GQLRequest (..),+    GQLResponse (..),+  )+import Data.Morpheus.Types.Internal.AST+  ( GQLErrors,+    VALID,+    Value (..),+  )+import Data.Morpheus.Types.Internal.Resolving+  ( Channel,+    ResponseEvent (..),+    ResponseStream,+    Result (..),+    ResultT (..),+    runResultT,+  )+import Data.UUID (UUID)+import Relude hiding (ByteString)++data API = PUB | SUB++type SUB = 'SUB++type PUB = 'PUB++data+  Input+    (api :: API)+  where+  InitConnection :: UUID -> Input SUB+  Request :: GQLRequest -> Input PUB++run :: ApiContext SUB e m -> Updates e m -> m ()+run SubContext {updateStore} (Updates changes) = updateStore changes++data ApiContext (api :: API) event (m :: * -> *) where+  PubContext ::+    { eventPublisher :: event -> m ()+    } ->+    ApiContext PUB event m+  SubContext ::+    { listener :: m ByteString,+      callback :: ByteString -> m (),+      updateStore :: (ClientConnectionStore event m -> ClientConnectionStore event m) -> m ()+    } ->+    ApiContext SUB event m++data+  Output+    (api :: API)+    e+    (m :: * -> *)+  where+  SubOutput ::+    { streamWS :: ApiContext SUB e m -> m (Either ByteString [Updates e m])+    } ->+    Output SUB e m+  PubOutput ::+    { streamHTTP :: ApiContext PUB e m -> m GQLResponse+    } ->+    Output PUB e m++handleResponseStream ::+  ( Monad m,+    Eq (Channel e),+    Hashable (Channel e)+  ) =>+  SessionID ->+  ResponseStream e m (Value VALID) ->+  Output SUB e m+handleResponseStream session (ResultT res) =+  SubOutput $ const $ unfoldR <$> res+  where+    execute Publish {} = apolloError $ globalErrorMessage "websocket can only handle subscriptions, not mutations"+    execute (Subscribe ch subRes) = Right $ startSession ch subRes session+    --------------------------+    unfoldR Success {events} = traverse execute events+    unfoldR Failure {errors} = apolloError errors+    --------------------------+    apolloError :: GQLErrors -> Either ByteString a+    apolloError = Left . toApolloResponse (sid session) . Errors++handleWSRequest ::+  ( Monad m,+    Functor m,+    Eq ch,+    Hashable ch+  ) =>+  ( GQLRequest ->+    ResponseStream (Event ch con) m (Value VALID)+  ) ->+  UUID ->+  ByteString ->+  Output SUB (Event ch con) m+handleWSRequest gqlApp clientId = handle . apolloFormat+  where+    --handle :: Applicative m => Validation ApolloAction -> Stream SUB e m+    handle = either (liftWS . Left) handleAction+    --------------------------------------------------+    -- handleAction :: ApolloAction -> Stream SUB e m+    handleAction ConnectionInit = liftWS $ Right []+    handleAction (SessionStart sessionId request) =+      handleResponseStream (SessionID clientId sessionId) (gqlApp request)+    handleAction (SessionStop sessionId) =+      liftWS $+        Right [endSession (SessionID clientId sessionId)]++liftWS ::+  Applicative m =>+  Either ByteString [Updates e m] ->+  Output SUB e m+liftWS = SubOutput . const . pure++runStreamWS ::+  (Monad m) =>+  ApiContext SUB e m ->+  Output SUB e m ->+  m ()+runStreamWS scope@SubContext {callback} SubOutput {streamWS} =+  streamWS scope+    >>= either callback (traverse_ (run scope))++runStreamHTTP ::+  (Monad m) =>+  ApiContext PUB e m ->+  Output PUB e m ->+  m GQLResponse+runStreamHTTP scope PubOutput {streamHTTP} =+  streamHTTP scope++toOutStream ::+  ( Monad m,+    Eq ch,+    Hashable ch+  ) =>+  ( GQLRequest ->+    ResponseStream (Event ch con) m (Value VALID)+  ) ->+  Input api ->+  Output api (Event ch con) m+toOutStream app (InitConnection clientId) =+  SubOutput handle+  where+    handle ws@SubContext {listener, callback} = do+      let runS (SubOutput x) = x ws+      bla <- listener >>= runS . handleWSRequest app clientId+      pure $ (Updates (insertConnection clientId callback) :) <$> bla+toOutStream app (Request req) =+  PubOutput $ handleResponseHTTP (app req)++handleResponseHTTP ::+  ( Monad m+  ) =>+  ResponseStream e m (Value VALID) ->+  ApiContext PUB e m ->+  m GQLResponse+handleResponseHTTP+  res+  PubContext {eventPublisher} = runResultT (handleRes res execute) >>= runResult+    where+      runResult Success {result, events} = traverse_ eventPublisher events $> Data result+      runResult Failure {errors} = pure $ Errors errors+      execute (Publish event) = pure event+      execute Subscribe {} = failure (globalErrorMessage "http server can't handle subscription")++handleRes ::+  (Monad m) =>+  ResponseStream e m a ->+  (ResponseEvent e m -> ResultT e' m e') ->+  ResultT e' m a+handleRes res execute = ResultT $ runResultT res >>= runResultT . unfoldRes execute++unfoldRes ::+  (Monad m) =>+  (e -> ResultT e' m e') ->+  Result e a ->+  ResultT e' m a+unfoldRes execute Success {result, warnings, events} =+  traverse execute events+    >>= ( ResultT . pure+            . Success+              result+              warnings+        )+unfoldRes _ Failure {errors} = ResultT $ pure $ Failure {errors}
+ src/Data/Morpheus/Subscriptions/WebSockets.hs view
@@ -0,0 +1,67 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE NoImplicitPrelude #-}++module Data.Morpheus.Subscriptions.WebSockets+  ( webSocketsWrapper,+  )+where++import Control.Monad.IO.Unlift+  ( MonadUnliftIO,+    withRunInIO,+  )+import Data.Morpheus.Subscriptions.Internal+  ( ApiContext (..),+    SUB,+    Store (..),+    acceptApolloRequest,+  )+import Network.WebSockets+  ( Connection,+    ServerApp,+    receiveData,+    sendTextData,+  )+import qualified Network.WebSockets as WS+import Relude++-- support old version of Websockets+pingThread :: Connection -> IO () -> IO ()++#if MIN_VERSION_websockets(0,12,6)+pingThread connection = WS.withPingThread connection 30 (return ())+#else+pingThread connection = (WS.forkPingThread connection 30 >>)+#endif++defaultWSScope :: MonadIO m => Store e m -> Connection -> ApiContext SUB e m+defaultWSScope Store {writeStore} connection =+  SubContext+    { listener = liftIO (receiveData connection),+      callback = liftIO . sendTextData connection,+      updateStore = writeStore+    }++webSocketsWrapper ::+  (MonadUnliftIO m, MonadIO m) =>+  Store e m ->+  (ApiContext SUB e m -> m ()) ->+  m ServerApp+webSocketsWrapper store handler =+  withRunInIO $+    \runIO ->+      pure $+        \pending -> do+          conn <- acceptApolloRequest pending+          pingThread+            conn+            $ runIO (handler (defaultWSScope store conn))
+ test/Spec.hs view
@@ -0,0 +1,14 @@+{-# LANGUAGE NoImplicitPrelude #-}++module Main+  ( main,+  )+where++import Prelude+  ( IO,+    pure,+  )++main :: IO ()+main = pure ()