diff --git a/sparrow.cabal b/sparrow.cabal --- a/sparrow.cabal +++ b/sparrow.cabal @@ -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 category: Web diff --git a/src/Web/Dependencies/Sparrow/Client.hs b/src/Web/Dependencies/Sparrow/Client.hs --- a/src/Web/Dependencies/Sparrow/Client.hs +++ b/src/Web/Dependencies/Sparrow/Client.hs @@ -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/|] diff --git a/src/Web/Dependencies/Sparrow/Client/Types.hs b/src/Web/Dependencies/Sparrow/Client/Types.hs --- a/src/Web/Dependencies/Sparrow/Client/Types.hs +++ b/src/Web/Dependencies/Sparrow/Client/Types.hs @@ -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) diff --git a/src/Web/Dependencies/Sparrow/Server.hs b/src/Web/Dependencies/Sparrow/Server.hs --- a/src/Web/Dependencies/Sparrow/Server.hs +++ b/src/Web/Dependencies/Sparrow/Server.hs @@ -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 diff --git a/src/Web/Dependencies/Sparrow/Server/Types.hs b/src/Web/Dependencies/Sparrow/Server/Types.hs --- a/src/Web/Dependencies/Sparrow/Server/Types.hs +++ b/src/Web/Dependencies/Sparrow/Server/Types.hs @@ -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 diff --git a/src/Web/Dependencies/Sparrow/Types.hs b/src/Web/Dependencies/Sparrow/Types.hs --- a/src/Web/Dependencies/Sparrow/Types.hs +++ b/src/Web/Dependencies/Sparrow/Types.hs @@ -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)