sparrow (empty) → 0.0.0
raw patch · 12 files changed
+1436/−0 lines, 12 filesdep +aesondep +aeson-attoparsecdep +asyncsetup-changed
Dependencies added: aeson, aeson-attoparsec, async, attoparsec, attoparsec-uri, base, bytestring, deepseq, exceptions, extractable-singleton, hashable, http-client, http-client-tls, http-types, list-t, monad-control, monad-control-aligned, mtl, nested-routes, path, path-extra, pred-trie, sparrow, stm, strict, text, tmapchan, tmapmvar, transformers, unordered-containers, urlpath, uuid, wai-middleware-content-type, wai-transformers, websockets, websockets-simple, wuss
Files
- LICENSE +30/−0
- README.md +1/−0
- Setup.hs +2/−0
- sparrow.cabal +126/−0
- src/Web/Dependencies/Sparrow.hs +13/−0
- src/Web/Dependencies/Sparrow/Client.hs +273/−0
- src/Web/Dependencies/Sparrow/Client/Types.hs +94/−0
- src/Web/Dependencies/Sparrow/Server.hs +281/−0
- src/Web/Dependencies/Sparrow/Server/Types.hs +324/−0
- src/Web/Dependencies/Sparrow/Session.hs +30/−0
- src/Web/Dependencies/Sparrow/Types.hs +260/−0
- test/Spec.hs +2/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2015, Athan Clark++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Athan Clark nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,1 @@+# sparrow
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ sparrow.cabal view
@@ -0,0 +1,126 @@+-- This file has been generated from package.yaml by hpack version 0.21.2.+--+-- see: https://github.com/sol/hpack+--+-- hash: 9de471f943693374ab3f8bbd43048dc68612170c4a0b4ee96359ed8112207bca++name: sparrow+version: 0.0.0+synopsis: Unified streaming dependency management for web apps+description: Please see the README on Github at <https://git.localcooking.com/tooling/sparrow#readme>+category: Web+author: Athan Clark+maintainer: athan.clark@localcooking.com+copyright: 2018 Local Cooking Inc.+license: BSD3+license-file: LICENSE+build-type: Simple+cabal-version: >= 1.10++extra-source-files:+ README.md++source-repository head+ type: git+ location: git://git.localcooking.com/tooling/sparrow.git++library+ exposed-modules:+ Web.Dependencies.Sparrow+ Web.Dependencies.Sparrow.Client+ Web.Dependencies.Sparrow.Client.Types+ Web.Dependencies.Sparrow.Server+ Web.Dependencies.Sparrow.Server.Types+ Web.Dependencies.Sparrow.Session+ Web.Dependencies.Sparrow.Types+ other-modules:+ Paths_sparrow+ hs-source-dirs:+ src+ ghc-options: -Wall+ build-depends:+ aeson+ , aeson-attoparsec+ , async+ , attoparsec+ , attoparsec-uri >=0.0.4+ , base >=4.7 && <5+ , bytestring+ , deepseq+ , exceptions+ , extractable-singleton+ , hashable+ , http-client+ , http-client-tls+ , http-types+ , list-t+ , monad-control+ , monad-control-aligned+ , mtl+ , nested-routes >=9.0.0+ , path+ , path-extra >=0.0.6+ , pred-trie >=0.6.0.1+ , stm+ , strict+ , text+ , tmapchan+ , tmapmvar >=0.0.4+ , transformers+ , unordered-containers+ , urlpath >=8.0.1+ , uuid+ , wai-middleware-content-type >=0.6.0+ , wai-transformers+ , websockets >=0.12.3.1+ , websockets-simple >=0.1.1+ , wuss+ default-language: Haskell2010++test-suite sparrow-test+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ other-modules:+ Paths_sparrow+ hs-source-dirs:+ test+ ghc-options: -Wall -threaded -rtsopts -Wall -with-rtsopts=-N+ build-depends:+ aeson+ , aeson-attoparsec+ , async+ , attoparsec+ , attoparsec-uri >=0.0.4+ , base >=4.7 && <5+ , bytestring+ , deepseq+ , exceptions+ , extractable-singleton+ , hashable+ , http-client+ , http-client-tls+ , http-types+ , list-t+ , monad-control+ , monad-control-aligned+ , mtl+ , nested-routes >=9.0.0+ , path+ , path-extra >=0.0.6+ , pred-trie >=0.6.0.1+ , sparrow+ , stm+ , strict+ , text+ , tmapchan+ , tmapmvar >=0.0.4+ , transformers+ , unordered-containers+ , urlpath >=8.0.1+ , uuid+ , wai-middleware-content-type >=0.6.0+ , wai-transformers+ , websockets >=0.12.3.1+ , websockets-simple >=0.1.1+ , wuss+ default-language: Haskell2010
+ src/Web/Dependencies/Sparrow.hs view
@@ -0,0 +1,13 @@+module Web.Dependencies.Sparrow+ ( module Web.Dependencies.Sparrow.Types+ , module Web.Dependencies.Sparrow.Server+ , module Web.Dependencies.Sparrow.Server.Types+ , module Web.Dependencies.Sparrow.Client+ , module Web.Dependencies.Sparrow.Client.Types+ ) where++import Web.Dependencies.Sparrow.Types (Server, ServerArgs (..), ServerReturn (..), ServerContinue (..), Client, ClientArgs (..), ClientReturn (..), Topic (..), Broadcast)+import Web.Dependencies.Sparrow.Server+import Web.Dependencies.Sparrow.Server.Types (SparrowServerT)+import Web.Dependencies.Sparrow.Client+import Web.Dependencies.Sparrow.Client.Types (SparrowClientT)
+ src/Web/Dependencies/Sparrow/Client.hs view
@@ -0,0 +1,273 @@+{-# LANGUAGE+ NamedFieldPuns+ , OverloadedStrings+ , RankNTypes+ , ScopedTypeVariables+ , FlexibleContexts+ , QuasiQuotes+ #-}++module Web.Dependencies.Sparrow.Client where++import Web.Dependencies.Sparrow.Types+ ( Client, ClientArgs (..), ClientReturn (..)+ , Topic (..), WSIncoming (..), WSOutgoing (..), WithTopic (..), WithSessionID (..)+ )+import Web.Dependencies.Sparrow.Session (SessionID (..))+import Web.Dependencies.Sparrow.Client.Types+ ( SparrowClientT (..), Env (..), ask', RegisteredTopicSubscriptions+ , callReject, callOnReceive, registerSubscription, removeSubscription+ , SparrowClientException (..)+ )++import Data.URI (URI, printURI)+import Data.URI.Auth (URIAuth (..))+import Data.URI.Auth.Host (printURIAuthHost)+import Data.Url (packLocation)+import qualified Data.Text as T+import qualified Data.Strict.Maybe as Strict+import Data.Aeson (FromJSON, ToJSON, Value)+import qualified Data.Aeson as Aeson+import Data.Singleton.Class (Extractable (runSingleton))+import Data.UUID.V4 (nextRandom)+import qualified Data.HashMap.Strict as HM+import Data.HashSet (HashSet)+import qualified Data.HashSet as HS+import Control.Monad (forever, void)+import Control.Monad.Trans (MonadTrans (lift))+import Control.Monad.Trans.Control (MonadBaseControl)+import qualified Control.Monad.Trans.Control.Aligned as Aligned+import Control.Monad.IO.Class (MonadIO (liftIO))+import Control.Monad.Catch (MonadCatch, MonadThrow (..))+import Control.Monad.Trans.Reader (ReaderT (runReaderT))+import Control.Concurrent.Async (Async, async, cancel, wait)+import Control.Concurrent.STM ( TMVar, TVar, TChan, atomically, newEmptyTMVarIO, newTChanIO+ , readTChan, writeTChan, takeTMVar, putTMVar, tryTakeTMVar+ , newTVarIO, readTVar, modifyTVar')+import Control.Concurrent.STM.TMapMVar (newTMapMVar)+import Control.DeepSeq (NFData (rnf))+import Control.Exception (evaluate)+import Path (Path, Abs, Rel, File, toFilePath, parseRelFile, (</>), parent, dirname, absdir)+import Path.Extended (Location, fromPath, (<&>))+import System.IO.Unsafe (unsafePerformIO)+import Network.Wai.Trans (runClientAppT)+import Network.WebSockets (runClient)+import Network.WebSockets.Simple (WebSocketsApp (..), WebSocketsAppParams (..), toClientAppT, expBackoffStrategy)+import Network.WebSockets.Simple.PingPong (pingPong)+import Network.HTTP.Client (parseRequest, newManager, defaultManagerSettings, method, requestBody, responseBody, responseStatus, httpLbs, RequestBody (RequestBodyLBS))+import Network.HTTP.Client.TLS (newTlsManager)+import Network.HTTP.Types (status200)+import Wuss (runSecureClient)+++unpackClient :: forall m stM initIn initOut deltaIn deltaOut+ . MonadIO m+ => MonadThrow m+ => Aligned.MonadBaseControl IO m stM+ => Extractable stM+ => ToJSON initIn+ => FromJSON initOut+ => ToJSON deltaIn+ => FromJSON deltaOut+ => Topic+ -> Client m initIn initOut deltaIn deltaOut+ -> SparrowClientT m ()+unpackClient topic client = do+ env@Env{envSendInit,envSendDelta} <- ask'++ lift $ Aligned.liftBaseWith $ \runInBase -> do+ let runM :: forall b. m b -> IO b+ runM x = runSingleton <$> runInBase x++ threadVar <- newEmptyTMVarIO++ -- spawn a new thread for the client+ thread <- async $ runM $ client $ \ClientArgs{clientReceive,clientInitIn,clientOnReject} -> do+ -- ** invoke init+ mInitOut <- envSendInit topic (Aeson.toJSON clientInitIn)++ case mInitOut of+ Nothing -> do+ _ <- throwM InitOutFailed+ pure Nothing -- TODO throw error+ Just v -> case Aeson.fromJSON v of+ Aeson.Error e -> do+ _ <- throwM (InitOutDecodingError e)+ pure Nothing+ Aeson.Success (initOut :: initOut) -> do+ let clientUnsubscribe = do+ envSendDelta (WSUnsubscribe topic)+ liftIO $ do+ thread <- atomically $ do+ removeSubscription env topic+ takeTMVar threadVar+ cancel thread+ clientSendCurrent =+ envSendDelta . WSIncoming . WithTopic topic . Aeson.toJSON++ clientReturn = ClientReturn+ { clientSendCurrent+ , clientUnsubscribe+ , clientInitOut = initOut+ }++ liftIO $ atomically $+ let go :: Value -> m () -- deltaOut handler+ go v' = case Aeson.fromJSON v' of+ Aeson.Error e ->+ throwM (DeltaOutDecodingError e)+ Aeson.Success (deltaOut :: deltaOut) ->+ clientReceive clientReturn deltaOut+ in registerSubscription env topic+ go+ ( do clientOnReject+ liftIO $ do+ mThread <- atomically $ tryTakeTMVar threadVar+ case mThread of+ Nothing -> pure ()+ Just thread -> cancel thread+ )++ pure (Just clientReturn)++ atomically (putTMVar threadVar thread)+++allocateDependencies :: forall m stM a+ . MonadIO m+ => MonadBaseControl IO m+ => Aligned.MonadBaseControl IO m stM+ => MonadCatch m+ => Extractable stM+ => Bool -- ^ TLS+ -> URIAuth -- ^ Hostname+ -> SparrowClientT m a+ -> m ()+allocateDependencies tls auth@(URIAuth _ host port) SparrowClientT{runSparrowClientT} = Aligned.liftBaseWith $ \runInBase -> do+ let path = [absdir|/dependencies/|]++ runM :: forall b. m b -> IO b+ runM x = runSingleton <$> runInBase x++ httpURI :: Topic -> URI+ httpURI (Topic topic) =+ packLocation (Strict.Just (if tls then "https" else "http")) True auth $+ fromPath $ path </> unsafePerformIO (parseRelFile $ T.unpack $ T.intercalate "/" topic)++ sessionID <- SessionID <$> nextRandom+ let q = T.pack $ toFilePath $ dirname path+ file <- parseRelFile $ T.unpack $ T.take (T.length q - 1) q++ let runWS :: WebSocketsApp m (WSOutgoing (WithTopic Value)) (WSIncoming (WithTopic Value)) -> IO ()+ runWS x = do+ let loc :: Location Abs File+ loc = fromPath (parent path </> file)+ <&> ("sessionID", Just (show sessionID))++ f | tls = runSecureClient (T.unpack $ printURIAuthHost host) (Strict.maybe 80 fromIntegral port) (show loc)+ | otherwise = runClient (T.unpack $ printURIAuthHost host) (Strict.maybe 80 fromIntegral port) (show loc)++ x' <- runM $ pingPong ((10^6) * 10) x -- every 10 seconds+ f $ runClientAppT runM $ toClientAppT x'++ ( toWSThread :: TMVar (Async ())+ ) <- newEmptyTMVarIO+ ( toWS :: TChan (WSIncoming (WithTopic Value))+ ) <- newTChanIO+ ( envSubscriptions :: RegisteredTopicSubscriptions m+ ) <- atomically newTMapMVar+ ( pendingTopicsAdded :: TVar (HashSet Topic)+ ) <- newTVarIO HS.empty+ ( pendingTopicsRemoved :: TVar (HashSet Topic)+ ) <- newTVarIO HS.empty++ httpManager <-+ if tls+ then newTlsManager+ else newManager defaultManagerSettings+++ let env :: Env m+ env = Env+ { envSendInit = \topic initIn -> liftIO $ do+ atomically $ modifyTVar' pendingTopicsAdded (HS.insert topic)+ req <- parseRequest $ T.unpack $ printURI $ httpURI topic+ let req' = req+ { method = "POST"+ , requestBody = RequestBodyLBS $ Aeson.encode WithSessionID+ { withSessionIDSessionID = sessionID+ , withSessionIDContent = initIn+ }+ }++ response <- httpLbs req' httpManager++ if responseStatus response == status200+ then case Aeson.eitherDecode (responseBody response) of+ Right (initOut :: Value) -> pure (Just initOut)+ Left e -> do+ _ <- throwM (InitOutDecodingError e)+ pure Nothing+ else do+ _ <- throwM InitOutHTTPError+ pure Nothing+ , envSendDelta = \x -> do+ case x of+ WSUnsubscribe topic -> liftIO $ atomically $ modifyTVar' pendingTopicsRemoved (HS.insert topic)+ _ -> pure ()+ liftIO $ atomically $ writeTChan toWS x+ , envSubscriptions+ }+++ -- WebSocket+ ( attemptWS :: TChan ()+ ) <- newTChanIO++ -- start running first attempt+ atomically $ writeTChan attemptWS ()+ -- backoff strategy+ backoff <- expBackoffStrategy $ atomically $ writeTChan attemptWS ()+ -- main thread+ ws <- async $ forever $ do+ atomically (readTChan attemptWS)++ runWS WebSocketsApp+ { onOpen = \WebSocketsAppParams{send} -> do+ liftIO $ do+ thread <- async $ forever $ do+ outgoing <- atomically (readTChan toWS)+ runM (send outgoing)+ atomically (putTMVar toWSThread thread)++ , onReceive = \_ r -> do+ liftIO $ evaluate (rnf r)+ case r of+ WSTopicsSubscribed topics -> pure () -- TODO verify that it's the exact state, adjust via a `join` if necessary of unsubscribe and onReject+ WSTopicAdded topic -> do+ liftIO $ do+ exists <- atomically $ HS.member topic <$> readTVar pendingTopicsAdded+ if exists+ then atomically $ modifyTVar' pendingTopicsAdded (HS.delete topic)+ else throwM (UnexpectedAddedTopic topic)+ WSTopicRemoved topic -> do+ liftIO $ do+ exists <- atomically $ HS.member topic <$> readTVar pendingTopicsRemoved+ if exists+ then atomically $ modifyTVar' pendingTopicsRemoved (HS.delete topic)+ else throwM (UnexpectedRemovedTopic topic)+ WSTopicRejected topic -> callReject env topic+ WSDecodingError err -> throwM (NetworkingDecodingError err)+ WSOutgoing (WithTopic topic v) -> callOnReceive env topic v+ , onClose = \_ e ->+ liftIO $ do+ thread <- atomically (takeTMVar toWSThread)+ cancel thread+ backoff e+ }++ -- Start clients+ z <- runM (runReaderT runSparrowClientT env)+ evaluate z++ wait ws
+ src/Web/Dependencies/Sparrow/Client/Types.hs view
@@ -0,0 +1,94 @@+{-# LANGUAGE+ GeneralizedNewtypeDeriving+ , DeriveFunctor+ , DeriveGeneric+ , NamedFieldPuns+ , FlexibleInstances+ , UndecidableInstances+ , MultiParamTypeClasses+ , TypeFamilies+ #-}++module Web.Dependencies.Sparrow.Client.Types where++import Web.Dependencies.Sparrow.Types (WSIncoming, WithTopic, Topic)++import Data.Aeson (Value)+import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as HM+import Control.Monad.Reader (ReaderT (..), MonadReader (..))+import Control.Monad.Trans (MonadTrans (..))+import Control.Monad.IO.Class (MonadIO (..))+import Control.Monad.Writer (MonadWriter)+import Control.Monad.State (MonadState)+import Control.Monad.Catch (MonadCatch, MonadThrow, MonadMask, Exception)+import Control.Concurrent.STM (STM, atomically, TVar, readTVar, modifyTVar')+import Control.Concurrent.STM.TMapMVar (TMapMVar)+import qualified Control.Concurrent.STM.TMapMVar as TMapMVar+import GHC.Generics (Generic)+++-- data ClientRefs m = ClientRefs+-- { clientContinue :: Maybe Value {-initOut-} -> m ()+-- }++type RegisteredTopicSubscriptions m =+ TMapMVar Topic+ ( Value -> m () -- `deltaOut` received from init+ , m () -- onReject+ )+++registerSubscription :: Env m -> Topic -> (Value -> m ()) -> m () -> STM ()+registerSubscription Env{envSubscriptions} topic onDeltaOut onReject =+ TMapMVar.insert envSubscriptions topic (onDeltaOut,onReject)++removeSubscription :: Env m -> Topic -> STM ()+removeSubscription Env{envSubscriptions} =+ TMapMVar.delete envSubscriptions++callReject :: MonadIO m => Env m -> Topic -> m ()+callReject Env{envSubscriptions} topic = do+ (_,onReject) <- liftIO $ atomically $ TMapMVar.lookup envSubscriptions topic+ onReject++callOnReceive :: MonadIO m => Env m -> Topic -> Value -> m ()+callOnReceive Env{envSubscriptions} topic v = do+ (onReceive,_) <- liftIO $ atomically $ TMapMVar.observe envSubscriptions topic+ onReceive v+++data Env m = Env+ { envSendDelta :: WSIncoming (WithTopic Value) -> m ()+ , envSendInit :: Topic -> Value -> m (Maybe Value)+ , envSubscriptions :: {-# UNPACK #-} !(RegisteredTopicSubscriptions m)+ }+++newtype SparrowClientT m a = SparrowClientT+ { runSparrowClientT :: ReaderT (Env m) m a+ } deriving (Functor, Applicative, Monad, MonadIO, MonadWriter w, MonadState s, MonadCatch, MonadThrow, MonadMask)++instance MonadReader r m => MonadReader r (SparrowClientT m) where+ ask = lift ask+ local f (SparrowClientT (ReaderT x)) = SparrowClientT $ ReaderT $ \r -> local f (x r)++instance MonadTrans SparrowClientT where+ lift = SparrowClientT . lift+++data SparrowClientException+ = InitOutFailed+ | InitOutDecodingError String+ | DeltaOutDecodingError String+ | InitOutHTTPError+ | UnexpectedAddedTopic Topic+ | UnexpectedRemovedTopic Topic+ | NetworkingDecodingError String+ deriving (Show, Generic)++instance Exception SparrowClientException+++ask' :: Applicative m => SparrowClientT m (Env m)+ask' = SparrowClientT (ReaderT pure)
+ src/Web/Dependencies/Sparrow/Server.hs view
@@ -0,0 +1,281 @@+{-# LANGUAGE+ RankNTypes+ , ScopedTypeVariables+ , NamedFieldPuns+ , GADTs+ , FlexibleContexts+ , OverloadedStrings+ , ConstraintKinds+ #-}++module Web.Dependencies.Sparrow.Server where++import Web.Dependencies.Sparrow.Types+ ( Server+ , ServerArgs (..)+ , ServerReturn (..)+ , ServerContinue (..)+ , Topic (..)+ , WithSessionID (..)+ , WithTopic (..)+ , InitResponse (..)+ , WSHTTPResponse (..)+ , WSIncoming (..)+ , WSOutgoing (..)+ )+import Web.Dependencies.Sparrow.Session (SessionID (..))+import Web.Dependencies.Sparrow.Server.Types+ ( SparrowServerT+ , Env (..)+ , ask'+ , tell'+ , execSparrowServerT+ , execSparrowServerT'+ , sendTo+ , unsafeRegisterReceive+ , registerOnUnsubscribe+ , registerOnOpenThread+ , registerInvalidator+ , broadcaster+ , getCurrentRegisteredTopics+ , getCallReceive+ , killOnOpenThread+ , killAllOnOpenThreads+ , callOnUnsubscribe+ , callAllOnUnsubscribe+ , unregisterReceive+ , unregisterSession+ , addSubscriber+ , delSubscriber+ , delSubscriberFromAllTopics+ , SparrowServerException (..)+ )++import Web.Routes.Nested (Match, UrlChunks, RouterT (..), ExtrudeSoundly)+import qualified Web.Routes.Nested as NR+import Data.Aeson (FromJSON, ToJSON (toJSON), Value)+import qualified Data.Aeson as Aeson+import Data.Trie.Pred.Interface.Types (Singleton (singleton), Extrude (extrude))+import Data.Monoid ((<>))+import qualified Data.UUID as UUID+import Data.Singleton.Class (Extractable (runSingleton))+import Data.Proxy (Proxy (..))+import Control.Monad (join, forever)+import Control.Monad.Trans (lift)+import Control.Monad.State (modify')+import Control.Monad.IO.Class (MonadIO (..))+import Control.Monad.Catch (MonadCatch, MonadThrow (..))+import Control.Monad.Trans.Control (MonadBaseControl)+import qualified Control.Monad.Trans.Control.Aligned as Aligned+import Control.Concurrent.Async (Async, async, cancel)+import Control.Concurrent.STM (TMVar, atomically, newEmptyTMVarIO, tryTakeTMVar, putTMVar)+import qualified Control.Concurrent.STM.TMapChan.Hash as TMapChan+import Control.Exception (evaluate)+import Control.DeepSeq (NFData (rnf))+import Network.Wai.Trans (MiddlewareT, strictRequestBody, queryString, websocketsOrT)+import Network.Wai.Middleware.ContentType.Json (jsonOnly)+import Network.WebSockets (defaultConnectionOptions)+import Network.WebSockets.Simple (WebSocketsApp (..), WebSocketsAppParams (..), toServerAppT)+import Network.WebSockets.Simple.PingPong (pingPong)+import Network.HTTP.Types (status400)+++++-- | Called per-connection+unpackServer :: forall m stM http initIn initOut deltaIn deltaOut+ . MonadIO m+ => Aligned.MonadBaseControl IO m stM+ => Extractable stM+ => FromJSON initIn+ => ToJSON initOut+ => FromJSON deltaIn+ => ToJSON deltaOut+ => Topic+ -> Server m initIn initOut deltaIn deltaOut+ -> SparrowServerT http m (MiddlewareT m)+unpackServer topic server = do+ env <- ask'+++ -- register topic's invalidator for `deltaIn` globally+ liftIO $ atomically $+ registerInvalidator env topic (Proxy :: Proxy deltaIn)++ pure $ \app req resp -> do+ -- attempt decoding POST data+ b <- liftIO (strictRequestBody req)+ case Aeson.decode b of+ Nothing -> resp (jsonOnly (InitBadEncoding b :: InitResponse ()) status400 [])+ Just WithSessionID+ { withSessionIDContent+ , withSessionIDSessionID+ } -> case Aeson.fromJSON withSessionIDContent of -- decode as `initIn`+ Aeson.Error e -> resp (jsonOnly (InitDecodingError e :: InitResponse ()) status400 [])+ Aeson.Success (initIn :: initIn) -> do++ -- ## invoke Server+ mContinue <- server initIn+ case mContinue of+ Nothing -> resp (jsonOnly (InitRejected :: InitResponse ()) status400 [])+ Just ServerContinue{serverContinue,serverOnUnsubscribe} -> do++ liftIO $ atomically $ do+ registerOnUnsubscribe env withSessionIDSessionID topic serverOnUnsubscribe++ -- notify client of added subscription+ sendTo env withSessionIDSessionID (WSTopicAdded topic)++ let serverArgs :: ServerArgs m deltaOut+ serverArgs = ServerArgs+ { serverDeltaReject = liftIO $ do+ atomically $ do+ unregisterReceive env withSessionIDSessionID topic+ delSubscriber env topic withSessionIDSessionID++ sendTo env withSessionIDSessionID (WSTopicRejected topic)+ killOnOpenThread env withSessionIDSessionID topic+ , serverSendCurrent =+ liftIO . atomically . sendTo env withSessionIDSessionID . WSOutgoing . WithTopic topic . toJSON+ }++ ServerReturn+ { serverInitOut+ , serverOnOpen+ , serverOnReceive+ } <- serverContinue (broadcaster env)++ -- register onReceive+ -- TODO security policy for consuming sessionIDs, expiring & pending+ -- check if currently "used", reject if topic is already subscribed+ -- - can have pending sessionIDs... for a while+ -- check if currently subscribed to that topic+ liftIO $ atomically $ do+ unsafeRegisterReceive env withSessionIDSessionID topic+ (\v -> case Aeson.fromJSON v of+ Aeson.Error _ -> Nothing+ Aeson.Success (x :: deltaIn) -> Just (serverOnReceive serverArgs x)+ )+ addSubscriber env topic withSessionIDSessionID++ thread <- Aligned.liftBaseWith $ \runInBase ->+ async $ (\x -> runSingleton <$> runInBase x) $ serverOnOpen serverArgs++ liftIO $ atomically $+ -- register onOpen thread+ registerOnOpenThread env withSessionIDSessionID topic thread++ (NR.action $ NR.post $ NR.json serverInitOut) app req resp+++match :: Monad m+ => Match xs' xs childHttp resultHttp+ => UrlChunks xs+ -> childHttp+ -> SparrowServerT resultHttp m ()+match ts http =+ tell' (singleton ts http)++++type MatchGroup xs' xs childHttp resultHttp =+ ( ExtrudeSoundly xs' xs childHttp resultHttp+ )+++matchGroup :: Monad m+ => MatchGroup xs' xs childHttp resultHttp+ => UrlChunks xs+ -> SparrowServerT childHttp m ()+ -> SparrowServerT resultHttp m ()+matchGroup ts x = do+ env <- ask'+ http <- lift (execSparrowServerT' env x)+ tell' (extrude ts http)++++serveDependencies :: forall m stM sec a+ . MonadBaseControl IO m+ => Aligned.MonadBaseControl IO m stM+ => Extractable stM+ => MonadIO m+ => MonadCatch m+ => SparrowServerT (MiddlewareT m) m a+ -> m (RouterT (MiddlewareT m) sec m ())+serveDependencies server = Aligned.liftBaseWith $ \runInBase -> do+ let runM :: forall b. m b -> IO b+ runM x = runSingleton <$> runInBase x++ (httpTrie,env@Env{envSessionsOutgoing}) <- runM (execSparrowServerT server)++ evaluate (rnf httpTrie)++ pure $ NR.matchGroup (NR.l_ "dependencies" NR.</> NR.o_) $ do+ -- websocket+ NR.matchHere $ \app req resp -> case join (lookup "sessionID" (queryString req))+ >>= UUID.fromASCIIBytes of+ Nothing -> resp (jsonOnly NoSessionID status400 [])+ Just sID -> do+ let sessionID = SessionID sID++ -- For listening on the outgoing TMapChan envSessionsOutgoing+ (outgoingListener :: TMVar (Async ())) <- liftIO newEmptyTMVarIO++ let wsApp :: WebSocketsApp m (WSIncoming (WithTopic Value)) (WSOutgoing (WithTopic Value))+ wsApp = WebSocketsApp+ { onOpen = \WebSocketsAppParams{send} -> do+ liftIO $ do+ -- spawn and store TMapChan envSessionsOutgoing listener+ listener <- async $ forever $ do+ x <- atomically (TMapChan.lookup envSessionsOutgoing sessionID)+ runM (send x)+ atomically $ putTMVar outgoingListener listener++ initSubs <- liftIO $ atomically $ getCurrentRegisteredTopics env sessionID++ send (WSTopicsSubscribed initSubs)++ , onReceive = \WebSocketsAppParams{send} r -> do+ case r of+ WSUnsubscribe topic -> do+ liftIO $ atomically $ do+ unregisterReceive env sessionID topic+ delSubscriber env topic sessionID++ callOnUnsubscribe env sessionID topic++ -- update client of removed subscription+ send (WSTopicRemoved topic)++ liftIO $ killOnOpenThread env sessionID topic++ WSIncoming (WithTopic topic x) -> do+ mEff <- liftIO $ atomically $ getCallReceive env sessionID topic x+ case mEff of+ Nothing -> throwM (NoHandlerForTopic topic)+ Just eff -> eff++ , onClose = \_ _ -> do+ liftIO $ do+ -- kill TMapChan envSessionsOutgoing listener+ mListener <- atomically (tryTakeTMVar outgoingListener)+ case mListener of+ Nothing -> pure ()+ Just listener -> cancel listener++ atomically $ do+ unregisterSession env sessionID+ delSubscriberFromAllTopics env sessionID++ callAllOnUnsubscribe env sessionID+ liftIO $ killAllOnOpenThreads env sessionID+ }++ wsApp' <- pingPong ((10^6) * 10) wsApp -- every 10 seconds++ (websocketsOrT runM defaultConnectionOptions (toServerAppT wsApp')) app req resp++ -- RESTful initIn -> initOut endpoints+ RouterT (modify' (<> NR.Tries httpTrie mempty mempty))+
+ src/Web/Dependencies/Sparrow/Server/Types.hs view
@@ -0,0 +1,324 @@+{-# LANGUAGE+ GeneralizedNewtypeDeriving+ , DeriveFunctor+ , DeriveGeneric+ , TupleSections+ , RankNTypes+ , ScopedTypeVariables+ , NamedFieldPuns+ , FlexibleInstances+ , MultiParamTypeClasses+ , UndecidableInstances+ #-}++module Web.Dependencies.Sparrow.Server.Types+ ( SparrowServerT+ , Env (..)+ , newEnv+ , execSparrowServerT+ , execSparrowServerT'+ , tell'+ , ask'+ , unsafeBroadcastTopic+ , unsafeRegisterReceive+ , sendTo+ , registerOnUnsubscribe+ , registerOnOpenThread+ , registerInvalidator+ , broadcaster+ , getCurrentRegisteredTopics+ , getCallReceive+ , killOnOpenThread+ , killAllOnOpenThreads+ , callOnUnsubscribe+ , callAllOnUnsubscribe+ , unregisterReceive+ , unregisterSession+ , addSubscriber+ , delSubscriber+ , delSubscriberFromAllTopics+ , SparrowServerException (..)+ ) where++import Web.Dependencies.Sparrow.Types (Topic (..), Broadcast, WithTopic (..), WSOutgoing (WSOutgoing))+import Web.Dependencies.Sparrow.Session (SessionID)++import Data.Text (Text)+import Data.Trie.Pred.Base (RootedPredTrie)+import Data.Monoid ((<>))+import Data.Maybe (fromMaybe)+import Data.Proxy (Proxy (..))+import Data.Foldable (sequenceA_)+import Data.Aeson (FromJSON, Value)+import qualified Data.Aeson as Aeson+import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as HM+import Data.HashSet (HashSet)+import qualified Data.HashSet as HS+import Control.Monad (forM_)+import Control.Monad.Reader (ReaderT (..), runReaderT, ask, MonadReader (..))+import Control.Monad.State (StateT, execStateT, modify', MonadState (..))+import Control.Monad.Writer (MonadWriter)+import Control.Monad.IO.Class (MonadIO (liftIO))+import Control.Monad.Trans (MonadTrans (lift))+import Control.Monad.Catch (Exception, MonadCatch, MonadThrow, MonadMask)+import Control.Concurrent.Async (Async, cancel)+import Control.Concurrent.STM (STM, atomically, TVar, newTVarIO, readTVar, modifyTVar')+import Control.Concurrent.STM.TMapChan.Hash (TMapChan, newTMapChan)+import qualified Control.Concurrent.STM.TMapChan.Hash as TMapChan+import Control.Concurrent.STM.TMapMVar.Hash (TMapMVar, newTMapMVar)+import qualified Control.Concurrent.STM.TMapMVar.Hash as TMapMVar+import GHC.Generics (Generic)++++type SessionsOutgoing = TMapChan SessionID (WSOutgoing (WithTopic Value))++sendTo :: Env m -> SessionID -> WSOutgoing (WithTopic Value) -> STM ()+sendTo Env{envSessionsOutgoing} =+ TMapChan.insert envSessionsOutgoing+++type RegisteredReceive m = -- STMMap.Map SessionID (STMMap.Map Topic (Value -> Maybe (m ())))+ TMapMVar SessionID (TMapMVar Topic (Value -> Maybe (m ())))+++-- unsafe because receiver isn't type-caste to the expected Topic+unsafeRegisterReceive :: MonadIO m+ => Env m -> SessionID -> Topic -> (Value -> Maybe (m ())) -> STM ()+unsafeRegisterReceive Env{envRegisteredReceive} sID topic f = do+ mTopics <- TMapMVar.tryObserve envRegisteredReceive sID+ case mTopics of+ Nothing -> do+ topics <- newTMapMVar+ TMapMVar.insert topics topic f+ TMapMVar.insert envRegisteredReceive sID topics+ -- ks <- getCurrentRegisteredTopics env sID+ -- putStrLn $ " - unsafeRegisterReceive: Topics...: " ++ show ks+ Just topics ->+ TMapMVar.insertForce topics topic f+++unregisterReceive :: Env m -> SessionID -> Topic -> STM ()+unregisterReceive Env{envRegisteredReceive} sID topic = do+ mTopics <- TMapMVar.tryObserve envRegisteredReceive sID+ case mTopics of+ Nothing -> pure ()+ Just topics -> TMapMVar.delete topics topic++unregisterSession :: Env m -> SessionID -> STM ()+unregisterSession Env{envRegisteredReceive} =+ TMapMVar.delete envRegisteredReceive+++getCallReceive :: MonadIO m+ => Env m -> SessionID -> Topic -> Value -> STM (Maybe (m ()))+getCallReceive Env{envRegisteredReceive} sID topic v = do+ topics <- TMapMVar.observe envRegisteredReceive sID+ onReceive <- TMapMVar.observe topics topic+ pure (onReceive v)++getCurrentRegisteredTopics :: Env m -> SessionID -> STM [Topic]+getCurrentRegisteredTopics Env{envRegisteredReceive} sID = do+ mTopics <- TMapMVar.tryObserve envRegisteredReceive sID+ case mTopics of+ Nothing -> pure []+ Just topics -> TMapMVar.keys topics++type RegisteredTopicInvalidators = TVar (HashMap Topic (Value -> Maybe String))++registerInvalidator :: forall deltaIn m+ . FromJSON deltaIn+ => Env m -> Topic -> Proxy deltaIn -> STM ()+registerInvalidator Env{envRegisteredTopicInvalidators} topic Proxy =+ let go v = case Aeson.fromJSON v of+ Aeson.Error e -> Just e+ Aeson.Success (_ :: deltaIn) -> Nothing+ in modifyTVar' envRegisteredTopicInvalidators (HM.insert topic go)++getValidator :: Env m -> Topic -> STM (Maybe (Value -> Maybe String))+getValidator Env{envRegisteredTopicInvalidators} topic =+ HM.lookup topic <$> readTVar envRegisteredTopicInvalidators++type RegisteredTopicSubscribers = TVar (HashMap Topic (HashSet SessionID))++addSubscriber :: Env m -> Topic -> SessionID -> STM ()+addSubscriber Env{envRegisteredTopicSubscribers} topic sID =+ modifyTVar' envRegisteredTopicSubscribers+ (HM.alter (maybe (Just (HS.singleton sID)) (Just . HS.insert sID)) topic)++delSubscriber :: Env m -> Topic -> SessionID -> STM ()+delSubscriber Env{envRegisteredTopicSubscribers} topic sID =+ let go xs+ | xs == HS.singleton sID = Nothing+ | otherwise = Just (HS.delete sID xs)+ in modifyTVar' envRegisteredTopicSubscribers+ (HM.alter (maybe Nothing go) topic)++delSubscriberFromAllTopics :: Env m -> SessionID -> STM ()+delSubscriberFromAllTopics env@Env{envRegisteredTopicSubscribers} sID = do+ allTopics <- HM.keys <$> readTVar envRegisteredTopicSubscribers+ forM_ allTopics (\topic -> delSubscriber env topic sID)++getSubscribers :: Env m -> Topic -> STM [SessionID]+getSubscribers Env{envRegisteredTopicSubscribers} topic =+ maybe [] HS.toList . HM.lookup topic <$> readTVar envRegisteredTopicSubscribers++type RegisteredOnUnsubscribe m = TVar (HashMap SessionID (HashMap Topic (m ())))++registerOnUnsubscribe :: Env m -> SessionID -> Topic -> m () -> STM ()+registerOnUnsubscribe Env{envRegisteredOnUnsubscribe} sID topic eff = do+ xs <- readTVar envRegisteredOnUnsubscribe+ let topics = fromMaybe HM.empty (HM.lookup sID xs)+ modifyTVar' envRegisteredOnUnsubscribe (HM.insert sID (HM.insert topic eff topics))++callOnUnsubscribe :: MonadIO m => Env m -> SessionID -> Topic -> m ()+callOnUnsubscribe Env{envRegisteredOnUnsubscribe} sID topic = do+ mEff <- liftIO $ atomically $ do+ xs <- readTVar envRegisteredOnUnsubscribe+ case HM.lookup sID xs of+ Nothing -> pure Nothing+ Just topics -> do+ let x = HM.lookup topic topics+ modifyTVar' envRegisteredOnUnsubscribe (HM.adjust (HM.delete topic) sID)+ pure x+ case mEff of+ Nothing -> pure ()+ Just eff -> eff++callAllOnUnsubscribe :: MonadIO m => Env m -> SessionID -> m ()+callAllOnUnsubscribe Env{envRegisteredOnUnsubscribe} sID = do+ effs <- liftIO $ atomically $ do+ xs <- readTVar envRegisteredOnUnsubscribe+ case HM.lookup sID xs of+ Nothing -> pure []+ Just topics -> do+ let effs = HM.elems topics+ modifyTVar' envRegisteredOnUnsubscribe (HM.delete sID)+ pure effs+ sequenceA_ effs++type RegisteredOnOpenThreads =+ TVar (HashMap SessionID (HashMap Topic (Async ())))++registerOnOpenThread :: Env m -> SessionID -> Topic -> Async () -> STM ()+registerOnOpenThread Env{envRegisteredOnOpenThreads} sID topic thread = do+ xs <- readTVar envRegisteredOnOpenThreads+ let topics = fromMaybe HM.empty (HM.lookup sID xs)+ modifyTVar' envRegisteredOnOpenThreads+ (HM.insert sID (HM.insert topic thread topics))++killOnOpenThread :: MonadIO m => Env m -> SessionID -> Topic -> IO ()+killOnOpenThread Env{envRegisteredOnOpenThreads} sID topic = do+ mThread <- atomically $ do+ xs <- readTVar envRegisteredOnOpenThreads+ case HM.lookup sID xs of+ Nothing -> pure Nothing+ Just topics -> do+ let x = HM.lookup topic topics+ modifyTVar' envRegisteredOnOpenThreads (HM.adjust (HM.delete topic) sID)+ pure x+ case mThread of+ Nothing -> pure ()+ Just thread -> cancel thread+++killAllOnOpenThreads :: MonadIO m => Env m -> SessionID -> IO ()+killAllOnOpenThreads Env{envRegisteredOnOpenThreads} sID = do+ threads <- atomically $ do+ xs <- readTVar envRegisteredOnOpenThreads+ case HM.lookup sID xs of+ Nothing -> pure []+ Just topics -> do+ let x = HM.elems topics+ modifyTVar' envRegisteredOnOpenThreads (HM.delete sID)+ pure x+ forM_ threads cancel+++data Env m = Env+ { envSessionsOutgoing :: {-# UNPACK #-} !SessionsOutgoing+ , envRegisteredReceive :: {-# UNPACK #-} !(RegisteredReceive m)+ , envRegisteredTopicInvalidators :: {-# UNPACK #-} !RegisteredTopicInvalidators+ , envRegisteredTopicSubscribers :: {-# UNPACK #-} !RegisteredTopicSubscribers+ , envRegisteredOnUnsubscribe :: {-# UNPACK #-} !(RegisteredOnUnsubscribe m)+ , envRegisteredOnOpenThreads :: {-# UNPACK #-} !RegisteredOnOpenThreads+ }+++newEnv :: IO (Env m)+newEnv = Env+ <$> atomically newTMapChan+ <*> atomically newTMapMVar+ <*> newTVarIO HM.empty+ <*> newTVarIO HM.empty+ <*> newTVarIO HM.empty+ <*> newTVarIO HM.empty+++unsafeBroadcastTopic :: MonadIO m => Env m -> Topic -> Value -> m ()+unsafeBroadcastTopic env t v =+ liftIO $ atomically $ do+ ss <- getSubscribers env t+ forM_ ss (\sessionID -> sendTo env sessionID (WSOutgoing (WithTopic t v)))+++broadcaster :: MonadIO m => Env m -> Broadcast m+broadcaster env = \topic -> do+ mInvalidator <- liftIO $ atomically $ getValidator env topic+ case mInvalidator of+ Nothing -> pure Nothing+ Just invalidator -> pure $ Just $ \v -> case invalidator v of+ Just _ -> Nothing -- is invalid+ Nothing -> Just (unsafeBroadcastTopic env topic v)+++-- | Monoid for WriterT (as StateT)+type Paper http = RootedPredTrie Text http+++newtype SparrowServerT http m a = SparrowServerT+ { runSparrowServerT :: ReaderT (Env m) (StateT (Paper http) m) a+ } deriving (Functor, Applicative, Monad, MonadIO, MonadWriter w, MonadCatch, MonadThrow, MonadMask)++instance MonadTrans (SparrowServerT http) where+ lift x = SparrowServerT (lift (lift x))++instance MonadReader r m => MonadReader r (SparrowServerT http m) where+ ask = lift ask+ local f (SparrowServerT (ReaderT g)) = SparrowServerT $ ReaderT $ \env -> local f (g env)++instance MonadState s m => MonadState s (SparrowServerT http m) where+ get = lift get+ put x = lift (put x)++++data SparrowServerException+ = NoHandlerForTopic Topic+ deriving (Show, Generic)++instance Exception SparrowServerException++++execSparrowServerT :: MonadIO m+ => SparrowServerT http m a+ -> m (Paper http, Env m)+execSparrowServerT x = do+ env <- liftIO newEnv+ (,env) <$> execSparrowServerT' env x++execSparrowServerT' :: Monad m+ => Env m+ -> SparrowServerT http m a+ -> m (Paper http)+execSparrowServerT' env (SparrowServerT x) =+ execStateT (runReaderT x env) mempty+++tell' :: Monad m => Paper http -> SparrowServerT http m ()+tell' x = SparrowServerT (lift (modify' (<> x)))++ask' :: Monad m => SparrowServerT http m (Env m)+ask' = SparrowServerT ask
+ src/Web/Dependencies/Sparrow/Session.hs view
@@ -0,0 +1,30 @@+{-# LANGUAGE+ GeneralizedNewtypeDeriving+ , DeriveGeneric+ #-}++module Web.Dependencies.Sparrow.Session where++import Data.UUID (UUID, fromString, toString)+import Data.Text (unpack)+import Data.Aeson (FromJSON (..), ToJSON (..), Value (String))+import Data.Aeson.Types (typeMismatch)+import Data.Hashable (Hashable)+import Control.DeepSeq (NFData)+import GHC.Generics (Generic)+++newtype SessionID = SessionID {getSessionID :: UUID}+ deriving (Eq, Hashable, Generic, NFData)++instance Show SessionID where+ show (SessionID x) = toString x++instance FromJSON SessionID where+ parseJSON (String x) = case fromString (unpack x) of+ Just y -> pure (SessionID y)+ Nothing -> fail "Not a UUID"+ parseJSON x = typeMismatch "SessionID" x++instance ToJSON SessionID where+ toJSON (SessionID x) = toJSON (toString x)
+ src/Web/Dependencies/Sparrow/Types.hs view
@@ -0,0 +1,260 @@+{-# LANGUAGE+ DeriveGeneric+ , GeneralizedNewtypeDeriving+ , OverloadedStrings+ , RecordWildCards+ #-}++module Web.Dependencies.Sparrow.Types where++import Web.Dependencies.Sparrow.Session (SessionID)++import Data.Hashable (Hashable)+import Data.Text (Text, intercalate)+import qualified Data.Text.Lazy.Encoding as LT+import qualified Data.ByteString.Lazy as LBS+import Data.Aeson (ToJSON (..), FromJSON (..), Value (String, Object), (.=), object, (.:))+import Data.Aeson.Types (typeMismatch)+import Data.Aeson.Attoparsec (attoAeson)+import Data.Attoparsec.Text (Parser, takeWhile1, char, sepBy)+import Control.Applicative ((<|>))+import Control.DeepSeq (NFData)+import GHC.Generics (Generic)+++-- * Conceptual++-- ** Server++data ServerArgs m deltaOut = ServerArgs+ { serverDeltaReject :: m ()+ , serverSendCurrent :: deltaOut -> m ()+ }++data ServerReturn m initOut deltaIn deltaOut = ServerReturn+ { serverInitOut :: initOut+ , serverOnOpen :: ServerArgs m deltaOut+ -> m ()+ -- ^ only after initOut is provided can we send deltas - invoked once, and should+ -- return a totally 'Control.Concurrent.Async.link'ed thread (if spawned)+ -- to kill with the subscription dies+ , serverOnReceive :: ServerArgs m deltaOut+ -> deltaIn -> m () -- ^ invoked for each receive+ }++data ServerContinue m initOut deltaIn deltaOut = ServerContinue+ { serverContinue :: Broadcast m -> m (ServerReturn m initOut deltaIn deltaOut)+ , serverOnUnsubscribe :: m ()+ }++type Server m initIn initOut deltaIn deltaOut =+ initIn -> m (Maybe (ServerContinue m initOut deltaIn deltaOut))++++-- ** Client++data ClientReturn m initOut deltaIn = ClientReturn+ { clientSendCurrent :: deltaIn -> m () -- was vs. can't be successful?+ , clientInitOut :: initOut+ , clientUnsubscribe :: m ()+ }++data ClientArgs m initIn initOut deltaIn deltaOut = ClientArgs+ { clientReceive :: ClientReturn m initOut deltaIn -> deltaOut -> m ()+ , clientInitIn :: initIn+ , clientOnReject :: m () -- ^ From a delta rejection, not init one+ }++type Client m initIn initOut deltaIn deltaOut =+ ( ClientArgs m initIn initOut deltaIn deltaOut+ -> m (Maybe (ClientReturn m initOut deltaIn))+ ) -> m ()+++-- ** Topic++newtype Topic = Topic {getTopic :: [Text]}+ deriving (Eq, Ord, Generic, Hashable, Show, NFData)++instance FromJSON Topic where+ parseJSON = attoAeson (Topic <$> breaker)+ where+ breaker :: Parser [Text]+ breaker = takeWhile1 (/= '/') `sepBy` char '/'++instance ToJSON Topic where+ toJSON (Topic xs) = String (intercalate "/" xs)+++-- ** Broadcast++type Broadcast m = Topic -> m (Maybe (Value -> Maybe (m ())))+++-- * JSON Encodings++++data WithSessionID a = WithSessionID+ { withSessionIDSessionID :: {-# UNPACK #-} !SessionID+ , withSessionIDContent :: a+ } deriving (Eq, Show, Generic)++instance NFData a => NFData (WithSessionID a)++instance ToJSON a => ToJSON (WithSessionID a) where+ toJSON WithSessionID{..} = object+ [ "sessionID" .= withSessionIDSessionID+ , "content" .= withSessionIDContent+ ]++instance FromJSON a => FromJSON (WithSessionID a) where+ parseJSON (Object o) = WithSessionID <$> (o .: "sessionID") <*> (o .: "content")+ parseJSON x = typeMismatch "WithSessionID" x++data WithTopic a = WithTopic+ { withTopicTopic :: !Topic+ , withTopicContent :: a+ } deriving (Eq, Show, Generic)++instance NFData a => NFData (WithTopic a)++instance ToJSON a => ToJSON (WithTopic a) where+ toJSON WithTopic{..} = object+ [ "topic" .= withTopicTopic+ , "content" .= withTopicContent+ ]++instance FromJSON a => FromJSON (WithTopic a) where+ parseJSON (Object o) = WithTopic <$> (o .: "topic") <*> (o .: "content")+ parseJSON x = typeMismatch "WithTopic" x+++data InitResponse a+ = InitBadEncoding !LBS.ByteString+ | InitDecodingError !String -- when manually decoding the content, casted+ | InitRejected+ | InitResponse a+ deriving (Eq, Show, Generic)++instance NFData a => NFData (InitResponse a)++instance ToJSON a => ToJSON (InitResponse a) where+ toJSON x = case x of+ InitBadEncoding y -> object ["error" .= object ["badRequest" .= LT.decodeUtf8 y]]+ InitDecodingError y -> object ["error" .= object ["decoding" .= y]]+ InitRejected -> object ["error" .= String "rejected"]+ InitResponse y -> object ["content" .= y]++instance FromJSON a => FromJSON (InitResponse a) where+ parseJSON json = case json of+ Object o -> do+ let error' = do+ json' <- o .: "error"+ case json' of+ String x+ | x == "rejected" -> pure InitRejected+ | otherwise -> fail'+ Object o' -> do+ let badEncoding = (InitBadEncoding . LT.encodeUtf8) <$> o' .: "badRequest"+ decodingError = InitDecodingError <$> o' .: "decoding"+ badEncoding <|> decodingError+ _ -> fail'+ response = InitResponse <$> o .: "content"+ error' <|> response+ _ -> fail'+ where+ fail' = typeMismatch "InitResponse" json+++data WSHTTPResponse+ = NoSessionID+ deriving (Eq, Show, Generic)++instance NFData WSHTTPResponse++instance ToJSON WSHTTPResponse where+ toJSON x = case x of+ NoSessionID -> object ["error" .= String "no sessionID query parameter"]++instance FromJSON WSHTTPResponse where+ parseJSON json = case json of+ Object o -> do+ json' <- o .: "error"+ case json' of+ String x+ | x == "no sessionID query parameter" -> pure NoSessionID+ | otherwise -> fail'+ _ -> fail'+ _ -> fail'+ where+ fail' = typeMismatch "WSHTTPResponse" json+++data WSIncoming a+ = WSUnsubscribe+ { wsUnsubscribeTopic :: !Topic+ }+ | WSIncoming a+ deriving (Eq, Show, Generic)++instance NFData a => NFData (WSIncoming a)++instance ToJSON a => ToJSON (WSIncoming a) where+ toJSON x = case x of+ WSUnsubscribe topic -> object ["unsubscribe" .= topic]+ WSIncoming y -> object ["content" .= y]++instance FromJSON a => FromJSON (WSIncoming a) where+ parseJSON (Object o) = do+ let unsubscribe = WSUnsubscribe <$> o .: "unsubscribe"+ incoming = WSIncoming <$> o .: "content"+ unsubscribe <|> incoming+ parseJSON x = typeMismatch "WSIncoming" x++data WSOutgoing a+ = WSTopicsSubscribed [Topic]+ | WSTopicAdded !Topic+ | WSTopicRemoved !Topic+ | WSTopicRejected !Topic+ | WSDecodingError !String+ | WSOutgoing a+ deriving (Eq, Show, Generic)++instance NFData a => NFData (WSOutgoing a)++instance ToJSON a => ToJSON (WSOutgoing a) where+ toJSON x = case x of+ WSDecodingError e -> object ["error" .= object ["decoding" .= e]]+ WSTopicsSubscribed subs -> object ["subs" .= object ["init" .= subs]]+ WSTopicAdded sub -> object ["subs" .= object ["add" .= sub]]+ WSTopicRemoved sub -> object ["subs" .= object ["del" .= sub]]+ WSTopicRejected sub -> object ["subs" .= object ["reject" .= sub]]+ WSOutgoing y -> object ["content" .= y]++instance FromJSON a => FromJSON (WSOutgoing a) where+ parseJSON json = case json of+ Object o -> do+ let content = WSOutgoing <$> o .: "content"+ error' = do+ json' <- o .: "error"+ case json' of+ Object o' -> do+ let decodingError = WSDecodingError <$> o' .: "decoding"+ decodingError+ _ -> fail'+ subs = do+ json' <- o .: "subs"+ case json' of+ Object o' -> do+ let subsInit = WSTopicsSubscribed <$> o' .: "init"+ subAdd = WSTopicAdded <$> o' .: "add"+ subDel = WSTopicRemoved <$> o' .: "del"+ subReject = WSTopicRejected <$> o' .: "reject"+ subsInit <|> subAdd <|> subDel <|> subReject+ _ -> fail'+ content <|> error' <|> subs+ _ -> fail'+ where+ fail' = typeMismatch "WSOutgoing" json
+ test/Spec.hs view
@@ -0,0 +1,2 @@+main :: IO ()+main = putStrLn "Test suite not yet implemented"