servant-subscriber 0.3.0.0 → 0.4.0.0
raw patch · 5 files changed
+84/−64 lines, 5 filesdep ~purescript-bridge
Dependency ranges changed: purescript-bridge
Files
- Readme.md +1/−4
- servant-subscriber.cabal +2/−2
- src/Servant/Subscriber/Client.hs +72/−49
- src/Servant/Subscriber/Request.hs +6/−6
- src/Servant/Subscriber/Response.hs +3/−3
Readme.md view
@@ -13,11 +13,8 @@ Still missing: - - Client code generation with servant-purescript for subscriptions.- - Client side code (purescript-subscriber) currently only contains a very low-level API which- is also about to change. - Documentation, blog post.- - Tests+ - Tests I will now continue with my actual project which uses servant-subscriber and add the missing parts when they come along.
servant-subscriber.cabal view
@@ -1,5 +1,5 @@ name: servant-subscriber-version: 0.3.0.0+version: 0.4.0.0 synopsis: When REST is not enough ... description: Please see Readme.md homepage: http://github.com/eskimor/servant-subscriber#readme@@ -61,7 +61,7 @@ ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: base , servant-subscriber- , purescript-bridge >= 0.6 && < 0.7+ , purescript-bridge >= 0.6 default-language: Haskell2010
src/Servant/Subscriber/Client.hs view
@@ -15,6 +15,8 @@ import Data.Aeson import Data.Map (Map) import qualified Data.Map.Strict as Map+import Data.Set (Set)+import qualified Data.Set as Set import qualified Data.Text as T import qualified Network.WebSockets as WS import Network.WebSockets.Connection as WS@@ -34,9 +36,10 @@ } data StatusMonitor = StatusMonitor {- request :: !HttpRequest-, monitor :: !(TVar (RefCounted ResourceStatus))-, oldStatus :: !ResourceStatus+ monitorPath :: !Path+, requests :: !(Set HttpRequest)+, monitor :: !(TVar (RefCounted ResourceStatus))+, oldStatus :: !ResourceStatus } data Snapshot = Snapshot {@@ -55,8 +58,8 @@ , fullMonitor = mon } -snapshotRequest :: Snapshot -> HttpRequest-snapshotRequest = request . fullMonitor+snapshotRequests :: Snapshot -> Set HttpRequest+snapshotRequests = requests . fullMonitor fromWebSocket :: Subscriber api -> WS.Connection -> STM (Client api) fromWebSocket sub c = do@@ -86,28 +89,50 @@ Left e -> $logDebug $ T.pack $ displayException (e :: SomeException) Right _ -> return () -addMonitor :: HttpRequest -> Client api -> STM ()-addMonitor req c = do- let path = httpPath req- let sub = subscriber c- tState <- subscribe path sub- stateVal <- refValue <$> readTVar tState- modifyTVar' (monitors c) $ Map.insert path (StatusMonitor req tState stateVal)+addRequest :: HttpRequest -> Client api -> STM ()+addRequest req c = do+ let path = httpPath req+ let sub = subscriber c+ monitors' <- readTVar $ monitors c+ let alreadySubscribed = Map.member path monitors'+ let+ insertRequest :: StatusMonitor -> StatusMonitor+ insertRequest mon = mon { requests = Set.insert req (requests mon) } --- | Unsubscribe from subscriber and also delete our monitor-removeMonitor :: Client api -> Path -> STM ()-removeMonitor c path = do+ modifyTVar' (monitors c)+ =<< if alreadySubscribed+ then+ return $ Map.adjust insertRequest path+ else do+ tState <- subscribe path sub+ stateVal <- refValue <$> readTVar tState+ return $ Map.insert path $ StatusMonitor path (Set.singleton req) tState stateVal++-- | Remove a Request, also unsubscribes from subscriber and deletes our monitor+-- if it was the last Request for the given path.+removeRequest :: Client api -> HttpRequest -> STM ()+removeRequest c req = do let sub = subscriber c- ms <- readTVar (monitors c)- forM_ ( Map.lookup path ms ) $ \m -> do- unsubscribeMonitor sub m- modifyTVar (monitors c) $ Map.delete path+ let path = httpPath req+ monitors' <- readTVar (monitors c)+ let+ deleteRequest :: StatusMonitor -> StatusMonitor+ deleteRequest mon = mon { requests = Set.delete req (requests mon) }+ forM_ ( Map.lookup path monitors' ) $ \mon -> do+ let newMon = deleteRequest mon+ modifyTVar' (monitors c)+ =<< if Set.null (requests newMon)+ then do+ unsubscribeMonitor sub mon+ return $ Map.delete path+ else+ return $ Map.adjust deleteRequest path --- | Does not remove the monitor - use removeMonitor if you want this!+-- | Does not remove the monitor - use removeRequest if you want this! unsubscribeMonitor :: Subscriber api -> StatusMonitor -> STM () unsubscribeMonitor sub m = let- path = httpPath . request $ m+ path = monitorPath m mon = monitor m in unsubscribe path mon sub@@ -123,55 +148,53 @@ handleSubscribe :: Backend backend => backend -> Client api -> HttpRequest -> IO () handleSubscribe b c req = getServerResponse b req $ \ response -> do- let path = httpPath req mapM_ (writeResponse c) =<< case response of (Resp.Modified _ _) -> do- atomically $ addMonitor req c- return [ Resp.Subscribed path, response ]+ atomically $ addRequest req c+ return [ Resp.Subscribed req, response ] _ -> return [ response ] -handleUnsubscribe :: Client api -> Path -> IO ()-handleUnsubscribe c path = do- atomically $ removeMonitor c path- writeResponse c $ Unsubscribed path+handleUnsubscribe :: Client api -> HttpRequest -> IO ()+handleUnsubscribe c req = do+ atomically $ removeRequest c req+ writeResponse c $ Unsubscribed req runMonitor :: Backend backend => backend -> Client api -> IO () runMonitor b c = forever $ do changes <- atomically $ monitorChanges c- mapM_ (handleUpdate b c) changes+ mapM_ (handleUpdates b c) changes -handleUpdate :: Backend backend => backend -> Client api -> (HttpRequest, ResourceStatus) -> IO ()-handleUpdate b c (req, event) = case event of- S.Deleted -> sendResponse $ Resp.Deleted (httpPath req)- ( S.Modified _ ) -> getServerResponse b req handleResponse- where- sendResponse :: Response -> IO ()- sendResponse = writeResponse c+handleUpdates :: Backend backend => backend -> Client api -> (Set HttpRequest, ResourceStatus) -> IO ()+handleUpdates b c (reqs, event) = case event of+ S.Deleted -> writeResponse c $ Resp.Deleted (httpPath . Set.elemAt 0 $ reqs) -- |< elemAt is partial - but we should never have an empty set, so this should be fine! - Check, check double check - test suite?! Or change parameters to this function.+ ( S.Modified _ ) -> mapM_ (handleModified b c) reqs ++handleModified :: Backend backend => backend -> Client api -> HttpRequest -> IO ()+handleModified b c req = getServerResponse b req handleResponse+ where handleResponse :: Response -> IO ()- handleResponse resp = case resp of- HttpRequestFailed _ _ -> do- atomically $ removeMonitor c (httpPath req)- sendResponse resp- _ ->- sendResponse resp+ handleResponse resp = do+ case resp of+ HttpRequestFailed _ _ -> atomically $ removeRequest c req+ _ -> return ()+ writeResponse c resp getServerResponse :: Backend backend => backend -> HttpRequest -> (Response -> IO ()) -> IO () getServerResponse b req sendResponse = void $ requestResource b req $ \ httpResponse -> do let status = statusCode . httpStatus $ httpResponse let response = Resp.httpBody httpResponse- let path = httpPath req let isGoodStatus = status >= 200 && status < 300 -- We only accept success sendResponse $ if isGoodStatus then- Resp.Modified path response+ Resp.Modified req response else Resp.HttpRequestFailed req httpResponse return ResponseReceived -monitorChanges :: Client api -> STM [(HttpRequest, ResourceStatus)]+monitorChanges :: Client api -> STM [(Set HttpRequest, ResourceStatus)] monitorChanges c = do snapshots <- mapM toSnapshot . Map.elems =<< readTVar (monitors c) let result = getChanges snapshots@@ -183,14 +206,14 @@ return result -getChanges :: [Snapshot] -> [(HttpRequest, ResourceStatus)]+getChanges :: [Snapshot] -> [(Set HttpRequest, ResourceStatus)] getChanges = map toChangeReport . filter monitorChanged monitorChanged :: Snapshot -> Bool monitorChanged m = snapshotCurrent m /= snapshotOld m -toChangeReport :: Snapshot -> (HttpRequest, ResourceStatus)-toChangeReport m = (snapshotRequest m, snapshotCurrent m)+toChangeReport :: Snapshot -> (Set HttpRequest, ResourceStatus)+toChangeReport m = (snapshotRequests m, snapshotCurrent m) updateMonitors :: [Snapshot] -> [StatusMonitor] updateMonitors = map updateOldStatus . filter ((/= S.Deleted) . snapshotCurrent)@@ -202,7 +225,7 @@ monitorsFromList :: [StatusMonitor] -> ClientMonitors monitorsFromList ms = let- paths = map (httpPath . request) ms+ paths = map (monitorPath) ms assList = zip paths ms in Map.fromList assList
src/Servant/Subscriber/Request.hs view
@@ -24,10 +24,9 @@ type RequestHeaders = [RequestHeader] -- | Any message from the client is a 'Request':--- Currently you can only subscribe to 'GET' method endpoints. data Request = Subscribe !HttpRequest- | Unsubscribe !Path+ | Unsubscribe !HttpRequest deriving Generic instance FromJSON Request@@ -40,12 +39,12 @@ , httpHeaders :: RequestHeaders , httpQuery :: H.QueryText , httpBody :: RequestBody-} deriving Generic+} deriving ( Generic, Eq, Ord ) instance FromJSON HttpRequest instance ToJSON HttpRequest -newtype RequestBody = RequestBody Text deriving (Generic, ToJSON, FromJSON)+newtype RequestBody = RequestBody Text deriving (Generic, ToJSON, FromJSON, Eq, Ord) toHTTPHeader :: RequestHeader -> H.Header toHTTPHeader = bimap (Case.mk . T.encodeUtf8) T.encodeUtf8@@ -54,5 +53,6 @@ toHTTPHeaders = map toHTTPHeader requestPath :: Request -> Path-requestPath (Subscribe req) = httpPath req-requestPath (Unsubscribe path) = path+requestPath req = httpPath $ case req of+ Subscribe hReq -> hReq+ Unsubscribe hReq -> hReq
src/Servant/Subscriber/Response.hs view
@@ -27,10 +27,10 @@ -- | Any message from the server is a Response. data Response =- Subscribed !Path -- |< Resource was successfully subscribed- | Modified !Path !ResponseBody -- |< If the full response is needed an additional FullSubscribe command with an appropriate additional response type will need to be added. + Subscribed !R.HttpRequest -- |< Resource was successfully subscribed+ | Modified !R.HttpRequest !ResponseBody -- |< If the full response is needed an additional FullSubscribe command with an appropriate additional response type will need to be added. | Deleted !Path- | Unsubscribed !Path+ | Unsubscribed !R.HttpRequest | HttpRequestFailed !R.HttpRequest !HttpResponse -- |< The server replied with some none 2xx status code. Thus your subscription failed. | ParseError -- |< Your request could not be parsed. deriving Generic