sparrow 0.0.0 → 0.0.1.1
raw patch · 6 files changed
+66/−53 lines, 6 files
Files
- sparrow.cabal +2/−2
- src/Web/Dependencies/Sparrow/Client.hs +5/−5
- src/Web/Dependencies/Sparrow/Client/Types.hs +15/−12
- src/Web/Dependencies/Sparrow/Server.hs +16/−14
- src/Web/Dependencies/Sparrow/Server/Types.hs +17/−12
- src/Web/Dependencies/Sparrow/Types.hs +11/−8
sparrow.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 9de471f943693374ab3f8bbd43048dc68612170c4a0b4ee96359ed8112207bca+-- hash: 957956460f140925a30f23e180c41fb3a32e4af820b0a5146fe91a3d3ec4d22c name: sparrow-version: 0.0.0+version: 0.0.1.1 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
src/Web/Dependencies/Sparrow/Client.hs view
@@ -69,8 +69,8 @@ => FromJSON initOut => ToJSON deltaIn => FromJSON deltaOut- => Topic- -> Client m initIn initOut deltaIn deltaOut+ => Topic -- ^ Dependency name+ -> Client m initIn initOut deltaIn deltaOut -- ^ Handler -> SparrowClientT m () unpackClient topic client = do env@Env{envSendInit,envSendDelta} <- ask'@@ -83,13 +83,13 @@ -- spawn a new thread for the client thread <- async $ runM $ client $ \ClientArgs{clientReceive,clientInitIn,clientOnReject} -> do- -- ** invoke init+ -- invoke init mInitOut <- envSendInit topic (Aeson.toJSON clientInitIn) case mInitOut of Nothing -> do _ <- throwM InitOutFailed- pure Nothing -- TODO throw error+ pure Nothing Just v -> case Aeson.fromJSON v of Aeson.Error e -> do _ <- throwM (InitOutDecodingError e)@@ -141,7 +141,7 @@ => Extractable stM => Bool -- ^ TLS -> URIAuth -- ^ Hostname- -> SparrowClientT m a+ -> SparrowClientT m a -- ^ All dependencies -> m () allocateDependencies tls auth@(URIAuth _ host port) SparrowClientT{runSparrowClientT} = Aligned.liftBaseWith $ \runInBase -> do let path = [absdir|/dependencies/|]
src/Web/Dependencies/Sparrow/Client/Types.hs view
@@ -28,10 +28,9 @@ import GHC.Generics (Generic) --- data ClientRefs m = ClientRefs--- { clientContinue :: Maybe Value {-initOut-} -> m ()--- } +-- * Internal Machinery+ type RegisteredTopicSubscriptions m = TMapMVar Topic ( Value -> m () -- `deltaOut` received from init@@ -58,12 +57,7 @@ onReceive v -data Env m = Env- { envSendDelta :: WSIncoming (WithTopic Value) -> m ()- , envSendInit :: Topic -> Value -> m (Maybe Value)- , envSubscriptions :: {-# UNPACK #-} !(RegisteredTopicSubscriptions m)- }-+-- * Context newtype SparrowClientT m a = SparrowClientT { runSparrowClientT :: ReaderT (Env m) m a@@ -76,7 +70,19 @@ instance MonadTrans SparrowClientT where lift = SparrowClientT . lift +data Env m = Env+ { envSendDelta :: WSIncoming (WithTopic Value) -> m ()+ , envSendInit :: Topic -> Value -> m (Maybe Value)+ , envSubscriptions :: {-# UNPACK #-} !(RegisteredTopicSubscriptions m)+ } +ask' :: Applicative m => SparrowClientT m (Env m)+ask' = SparrowClientT (ReaderT pure)+++-- * Exceptions++ data SparrowClientException = InitOutFailed | InitOutDecodingError String@@ -89,6 +95,3 @@ instance Exception SparrowClientException --ask' :: Applicative m => SparrowClientT m (Env m)-ask' = SparrowClientT (ReaderT pure)
src/Web/Dependencies/Sparrow/Server.hs view
@@ -91,13 +91,12 @@ => ToJSON initOut => FromJSON deltaIn => ToJSON deltaOut- => Topic- -> Server m initIn initOut deltaIn deltaOut+ => Topic -- ^ Name of Dependency+ -> Server m initIn initOut deltaIn deltaOut -- ^ Handler for all clients -> 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)@@ -158,20 +157,22 @@ ) addSubscriber env topic withSessionIDSessionID - thread <- Aligned.liftBaseWith $ \runInBase ->- async $ (\x -> runSingleton <$> runInBase x) $ serverOnOpen serverArgs+ mThread <- serverOnOpen serverArgs - liftIO $ atomically $- -- register onOpen thread- registerOnOpenThread env withSessionIDSessionID topic thread+ case mThread of+ Nothing -> pure ()+ Just thread -> liftIO $ atomically $+ -- register onOpen thread+ registerOnOpenThread env withSessionIDSessionID topic thread (NR.action $ NR.post $ NR.json serverInitOut) app req resp +-- | Match an individual dependency match :: Monad m => Match xs' xs childHttp resultHttp- => UrlChunks xs- -> childHttp+ => UrlChunks xs -- ^ Should match the dependency name+ -> childHttp -- ^ 'Network.Wai.Trans.MiddlewareT', or a function to one -> SparrowServerT resultHttp m () match ts http = tell' (singleton ts http)@@ -183,10 +184,11 @@ ) +-- | Group together a set of dependencies matchGroup :: Monad m => MatchGroup xs' xs childHttp resultHttp- => UrlChunks xs- -> SparrowServerT childHttp m ()+ => UrlChunks xs -- ^ Common 'Topic' prefix+ -> SparrowServerT childHttp m () -- ^ Set of handlers -> SparrowServerT resultHttp m () matchGroup ts x = do env <- ask'@@ -194,14 +196,14 @@ tell' (extrude ts http) -+-- | Host dependencies and websocket 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+ => SparrowServerT (MiddlewareT m) m a -- ^ Dependencies -> m (RouterT (MiddlewareT m) sec m ()) serveDependencies server = Aligned.liftBaseWith $ \runInBase -> do let runM :: forall b. m b -> IO b
src/Web/Dependencies/Sparrow/Server/Types.hs view
@@ -12,32 +12,39 @@ #-} module Web.Dependencies.Sparrow.Server.Types- ( SparrowServerT+ ( -- * Context+ SparrowServerT , Env (..) , newEnv , execSparrowServerT , execSparrowServerT' , tell' , ask'- , unsafeBroadcastTopic+ , -- * Internal Machinery+ -- ** Outgoing Per-Session+ unsafeBroadcastTopic , unsafeRegisterReceive , sendTo- , registerOnUnsubscribe- , registerOnOpenThread+ , -- ** Continuation Registration+ registerOnUnsubscribe , registerInvalidator , broadcaster , getCurrentRegisteredTopics , getCallReceive- , killOnOpenThread- , killAllOnOpenThreads , callOnUnsubscribe , callAllOnUnsubscribe- , unregisterReceive+ , -- ** Thread Management+ registerOnOpenThread+ , killOnOpenThread+ , killAllOnOpenThreads+ , -- ** Bookkeeping+ unregisterReceive , unregisterSession , addSubscriber , delSubscriber , delSubscriberFromAllTopics- , SparrowServerException (..)+ , -- * Exceptions+ SparrowServerException (..) ) where import Web.Dependencies.Sparrow.Types (Topic (..), Broadcast, WithTopic (..), WSOutgoing (WSOutgoing))@@ -145,7 +152,7 @@ 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)+ (HM.alter (Just . maybe (HS.singleton sID) (HS.insert sID)) topic) delSubscriber :: Env m -> Topic -> SessionID -> STM () delSubscriber Env{envRegisteredTopicSubscribers} topic sID =@@ -182,9 +189,7 @@ let x = HM.lookup topic topics modifyTVar' envRegisteredOnUnsubscribe (HM.adjust (HM.delete topic) sID) pure x- case mEff of- Nothing -> pure ()- Just eff -> eff+ fromMaybe (pure ()) mEff callAllOnUnsubscribe :: MonadIO m => Env m -> SessionID -> m () callAllOnUnsubscribe Env{envRegisteredOnUnsubscribe} sID = do
src/Web/Dependencies/Sparrow/Types.hs view
@@ -10,7 +10,7 @@ import Web.Dependencies.Sparrow.Session (SessionID) import Data.Hashable (Hashable)-import Data.Text (Text, intercalate)+import Data.Text (Text, intercalate, unpack) import qualified Data.Text.Lazy.Encoding as LT import qualified Data.ByteString.Lazy as LBS import Data.Aeson (ToJSON (..), FromJSON (..), Value (String, Object), (.=), object, (.:))@@ -19,6 +19,7 @@ import Data.Attoparsec.Text (Parser, takeWhile1, char, sepBy) import Control.Applicative ((<|>)) import Control.DeepSeq (NFData)+import Control.Concurrent.Async (Async) import GHC.Generics (Generic) @@ -34,10 +35,9 @@ 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+ -> m (Maybe (Async ()))+ -- ^ invoked once, and should return a 'Control.Concurrent.Async.link'ed long-lived thread+ -- to kill when the subscription dies , serverOnReceive :: ServerArgs m deltaOut -> deltaIn -> m () -- ^ invoked for each receive }@@ -55,7 +55,7 @@ -- ** Client data ClientReturn m initOut deltaIn = ClientReturn- { clientSendCurrent :: deltaIn -> m () -- was vs. can't be successful?+ { clientSendCurrent :: deltaIn -> m () , clientInitOut :: initOut , clientUnsubscribe :: m () }@@ -63,7 +63,7 @@ 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+ , clientOnReject :: m () -- ^ Run if the server decides to randomly kick the client } type Client m initIn initOut deltaIn deltaOut =@@ -75,7 +75,10 @@ -- ** Topic newtype Topic = Topic {getTopic :: [Text]}- deriving (Eq, Ord, Generic, Hashable, Show, NFData)+ deriving (Eq, Ord, Generic, Hashable, NFData)++instance Show Topic where+ show (Topic x) = unpack (intercalate "/" x) instance FromJSON Topic where parseJSON = attoAeson (Topic <$> breaker)