servant-subscriber 0.5.0.2 → 0.5.0.3
raw patch · 4 files changed
+42/−58 lines, 4 files
Files
- Readme.md +1/−1
- servant-subscriber.cabal +1/−1
- src/Servant/Subscriber/Client.hs +35/−54
- src/Servant/Subscriber/Types.hs +5/−2
Readme.md view
@@ -3,7 +3,7 @@ [](https://travis-ci.org/eskimor/servant-subscriber) -Servant-subscriber enables you clients to subscribe to resources in your servant-api (an API endpoint).+Servant-subscriber enables your clients to subscribe to resources in your servant-api (an API endpoint). Servant-subscriber will then notify the client via a WebSocket connection whenever the resource changes, thus the client can easily stay up to date with a resource.
servant-subscriber.cabal view
@@ -1,5 +1,5 @@ name: servant-subscriber-version: 0.5.0.2+version: 0.5.0.3 synopsis: When REST is not enough ... description: Please see Readme.md homepage: http://github.com/eskimor/servant-subscriber#readme
src/Servant/Subscriber/Client.hs view
@@ -5,6 +5,7 @@ import Control.Concurrent.Async+import Control.Lens import Control.Concurrent.STM (STM, atomically, retry) import Control.Concurrent.STM.TVar import Control.Exception (SomeException, displayException)@@ -19,6 +20,7 @@ import Data.Map (Map) import qualified Data.Map.Strict as Map import Data.Monoid ((<>))+import Data.Foldable (traverse_) import Data.Set (Set) import qualified Data.Set as Set import qualified Data.Text as T@@ -26,13 +28,14 @@ import qualified Data.Text.IO as T import qualified Network.WebSockets as WS import Network.WebSockets.Connection as WS+import Debug.Trace (trace) import Servant.Subscriber.Backend import Servant.Subscriber.Request import Servant.Subscriber.Response as Resp import Servant.Subscriber.Types as S -type ClientMonitors = Map Path StatusMonitor+type ClientMonitors = Map HttpRequest StatusMonitor data Client api = Client { subscriber :: !(Subscriber api)@@ -44,10 +47,9 @@ } data StatusMonitor = StatusMonitor {- monitorPath :: !Path-, requests :: !(Set HttpRequest)+ request :: !HttpRequest , monitor :: !(TVar (RefCounted ResourceStatus))-, oldStatus :: !ResourceStatus+, oldStatus :: !(Maybe ResourceStatus) -- Nothing when added so we get a notification in any case. } data Snapshot = Snapshot {@@ -55,9 +57,12 @@ , fullMonitor :: StatusMonitor } -snapshotOld :: Snapshot -> ResourceStatus+snapshotOld :: Snapshot -> Maybe ResourceStatus snapshotOld = oldStatus . fullMonitor +monitorPath :: StatusMonitor -> Path+monitorPath = httpPath . request+ toSnapshot :: StatusMonitor -> STM Snapshot toSnapshot mon = do current <- readTVar $ monitor mon@@ -66,8 +71,8 @@ , fullMonitor = mon } -snapshotRequests :: Snapshot -> Set HttpRequest-snapshotRequests = requests . fullMonitor+snapshotRequest :: Snapshot -> HttpRequest+snapshotRequest = request . fullMonitor fromWebSocket :: Subscriber api -> IORef (IO ()) -> WS.Connection -> STM (Client api) fromWebSocket sub myRef c = do@@ -106,41 +111,19 @@ addRequest :: HttpRequest -> Client api -> STM () addRequest req c = do let path = httpPath req+ trace ("Added request for PATH:" <> show path) $ pure () 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) }-- 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+ tState <- subscribe path sub+ modifyTVar' (monitors c) $ at req .~ Just (StatusMonitor req tState Nothing) -- | 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- 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+ traverse_ (unsubscribeMonitor sub) $ monitors'^.at req+ modifyTVar' (monitors c) $ at req .~ Nothing -- | Does not remove the monitor - use removeRequest if you want this! unsubscribeMonitor :: Subscriber api -> StatusMonitor -> STM ()@@ -157,7 +140,7 @@ req <- readRequest c case req of Nothing -> writeResponse c ParseError- Just (Subscribe httpReq) -> handleSubscribe b c httpReq+ Just (Subscribe httpReq) -> handleSubscribe c httpReq Just (Unsubscribe path) -> handleUnsubscribe c path Just (SetPongRequest httpReq) -> do let doIt = doRequestIgnoreResult httpReq@@ -172,13 +155,10 @@ doRequestIgnoreResult :: HttpRequest -> IO () doRequestIgnoreResult req' = void $ requestResource b req' (const (pure ResponseReceived)) -handleSubscribe :: Backend backend => backend -> Client api -> HttpRequest -> IO ()-handleSubscribe b c req = getServerResponse b req $ \ response -> do- mapM_ (writeResponse c) =<< case response of- (Resp.Modified _ _) -> do- atomically $ addRequest req c- return [ Resp.Subscribed req, response ]- _ -> return [ response ]+handleSubscribe :: Client api -> HttpRequest -> IO ()+handleSubscribe c req = do+ atomically $ addRequest req c+ writeResponse c $ Resp.Subscribed req handleUnsubscribe :: Client api -> HttpRequest -> IO () handleUnsubscribe c req = do@@ -191,10 +171,10 @@ changes <- atomically $ monitorChanges c mapM_ (handleUpdates b c) changes -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+handleUpdates :: Backend backend => backend -> Client api -> (HttpRequest, ResourceStatus) -> IO ()+handleUpdates b c (req, event) = case event of+ S.Deleted -> writeResponse c $ Resp.Deleted (httpPath req)+ ( S.Modified _ ) -> handleModified b c req handleModified :: Backend backend => backend -> Client api -> HttpRequest -> IO ()@@ -213,6 +193,7 @@ let status = statusCode . httpStatus $ httpResponse let response = Resp.httpBody httpResponse T.putStrLn $ "Got server response body: " <> T.decodeUtf8 (BS.toStrict . encode $ response)+ T.putStrLn $ "For request: " <> T.pack (show req) let isGoodStatus = status >= 200 && status < 300 -- We only accept success sendResponse $ if isGoodStatus then@@ -221,7 +202,7 @@ Resp.HttpRequestFailed req httpResponse return ResponseReceived -monitorChanges :: Client api -> STM [(Set HttpRequest, ResourceStatus)]+monitorChanges :: Client api -> STM [(HttpRequest, ResourceStatus)] monitorChanges c = do snapshots <- mapM toSnapshot . Map.elems =<< readTVar (monitors c) let result = getChanges snapshots@@ -233,26 +214,26 @@ return result -getChanges :: [Snapshot] -> [(Set HttpRequest, ResourceStatus)]+getChanges :: [Snapshot] -> [(HttpRequest, ResourceStatus)] getChanges = map toChangeReport . filter monitorChanged monitorChanged :: Snapshot -> Bool-monitorChanged m = snapshotCurrent m /= snapshotOld m+monitorChanged m = Just (snapshotCurrent m) /= snapshotOld m -toChangeReport :: Snapshot -> (Set HttpRequest, ResourceStatus)-toChangeReport m = (snapshotRequests m, snapshotCurrent m)+toChangeReport :: Snapshot -> (HttpRequest, ResourceStatus)+toChangeReport m = (snapshotRequest m, snapshotCurrent m) updateMonitors :: [Snapshot] -> [StatusMonitor] updateMonitors = map updateOldStatus . filter ((/= S.Deleted) . snapshotCurrent) updateOldStatus :: Snapshot -> StatusMonitor updateOldStatus m = (fullMonitor m) {- oldStatus = snapshotCurrent m+ oldStatus = Just $ snapshotCurrent m } monitorsFromList :: [StatusMonitor] -> ClientMonitors monitorsFromList ms = let- paths = map (monitorPath) ms- assList = zip paths ms+ reqs = map (request) ms+ assList = zip reqs ms in Map.fromList assList
src/Servant/Subscriber/Types.hs view
@@ -14,13 +14,15 @@ import Data.Map (Map) import qualified Data.Map as Map import Data.Proxy+import Data.Monoid import Data.String (IsString, fromString) import Data.Text (Text) import qualified Data.Text as T import GHC.Generics-import Network.URI (URI (..), pathSegments)+import Network.URI (URI (..), pathSegments, unEscapeString) import Servant.Utils.Links (IsElem, HasLink, MkLink, safeLink) import System.FilePath.Posix (splitPath)+import Debug.Trace (trace) import Servant.Subscriber.Subscribable @@ -90,8 +92,9 @@ -> (MkLink endpoint -> URI) -> STM () notify subscriber event pEndpoint getLink = do- let mkPath = Path . map T.pack . pathSegments . getLink+ let mkPath = Path . map (T.pack . unEscapeString) . pathSegments . getLink let resource = mkPath $ safeLink (Proxy :: Proxy api) pEndpoint+ trace ("NOTIFIED PATH:" <> show resource) (pure ()) modifyState event resource subscriber -- | Version of notify that lives in 'IO' - for your convenience.