servant-subscriber 0.2.0.1 → 0.3.0.0
raw patch · 7 files changed
+97/−235 lines, 7 filesdep −aeson-compatdep −base-compatdep −blaze-htmldep ~aesondep ~attoparsecdep ~basenew-component:exe:subscriber-psGenerator
Dependencies removed: aeson-compat, base-compat, blaze-html, blaze-markup, hspec, http-media, lucid, mtl, string-conversions
Dependency ranges changed: aeson, attoparsec, base, bytestring, time
Files
- app/PSGenerator.hs +0/−1
- servant-subscriber.cabal +2/−29
- src/Servant/Subscriber.hs +2/−2
- src/Servant/Subscriber/Client.hs +85/−65
- src/Servant/Subscriber/Response.hs +3/−12
- src/Servant/Subscriber/Subscribable.hs +5/−1
- test/Spec.hs +0/−125
app/PSGenerator.hs view
@@ -14,7 +14,6 @@ , mkSumType (Proxy :: Proxy Response) , mkSumType (Proxy :: Proxy HttpResponse) , mkSumType (Proxy :: Proxy Status)- , mkSumType (Proxy :: Proxy RequestError) , mkSumType (Proxy :: Proxy Path) ]
servant-subscriber.cabal view
@@ -1,5 +1,5 @@ name: servant-subscriber-version: 0.2.0.1+version: 0.3.0.0 synopsis: When REST is not enough ... description: Please see Readme.md homepage: http://github.com/eskimor/servant-subscriber#readme@@ -55,7 +55,7 @@ , websockets default-language: Haskell2010 -executable psGenerator+executable subscriber-psGenerator hs-source-dirs: app main-is: PSGenerator.hs ghc-options: -threaded -rtsopts -with-rtsopts=-N@@ -64,33 +64,6 @@ , purescript-bridge >= 0.6 && < 0.7 default-language: Haskell2010 -test-suite servant-subscriber-test- type: exitcode-stdio-1.0- hs-source-dirs: test- main-is: Spec.hs- build-depends:- aeson- , aeson-compat- , attoparsec >= 0.13.0.1- , base- , base-compat >= 0.9.0- , blaze-html >= 0.8.1.1- , blaze-markup >= 0.7.0.3- , bytestring >= 0.10.6.0- , directory >= 1.2.2.0- , http-media >= 0.6.3- , lucid- , mtl >= 2.2.1- , servant-foreign- , servant-server- , servant-subscriber- , string-conversions >= 0.4- , time >= 1.5.0.1- , wai- , warp- , hspec- ghc-options: -threaded -rtsopts -with-rtsopts=-N- default-language: Haskell2010 source-repository head type: git
src/Servant/Subscriber.hs view
@@ -53,8 +53,8 @@ let runLog = runLogging subscriber let handleWSConnection pending = do connection <- acceptRequest pending- forkPingThread connection 25- runLog . Client.run app subscriber <=< atomically . Client.fromWebSocket $ connection+ forkPingThread connection 28+ runLog . Client.run app <=< atomically . Client.fromWebSocket subscriber $ connection if Path (pathInfo req) == entryPoint subscriber then websocketsOr opts handleWSConnection app req sendResponse else app req sendResponse
src/Servant/Subscriber/Client.hs view
@@ -10,7 +10,7 @@ import Control.Exception.Lifted (finally, try) import Control.Monad import Control.Monad.IO.Class-import Control.Monad.Logger (MonadLogger, logDebug, logError, logInfo, monadLoggerLog)+import Control.Monad.Logger (MonadLogger, logDebug) import Control.Monad.Trans.Control (MonadBaseControl) import Data.Aeson import Data.Map (Map)@@ -26,8 +26,9 @@ type ClientMonitors = Map Path StatusMonitor -data Client = Client {- monitors :: !(TVar ClientMonitors)+data Client api = Client {+ subscriber :: !(Subscriber api)+ , monitors :: !(TVar ClientMonitors) , readRequest :: !(IO (Maybe Request)) , writeResponse :: !(Response -> IO ()) }@@ -54,16 +55,15 @@ , fullMonitor = mon } -- snapshotRequest :: Snapshot -> HttpRequest snapshotRequest = request . fullMonitor -fromWebSocket :: WS.Connection -> STM Client-fromWebSocket c = do+fromWebSocket :: Subscriber api -> WS.Connection -> STM (Client api)+fromWebSocket sub c = do ms <- newTVar Map.empty return Client {- monitors = ms+ subscriber = sub+ , monitors = ms , readRequest = do msg <- WS.receiveDataMessage c case msg of@@ -72,10 +72,12 @@ , writeResponse = sendDataMessage c . WS.Text . encode } -run :: (MonadLogger m, MonadBaseControl IO m, MonadIO m, Backend backend) => backend -> Subscriber api -> Client -> m ()-run b sub c = do+run :: (MonadLogger m, MonadBaseControl IO m, MonadIO m, Backend backend)+ => backend -> Client api -> m ()+run b c = do let- work = liftIO $ race_ (runMonitor b c) (handleRequests b sub c)+ sub = subscriber c+ work = liftIO $ race_ (runMonitor b c) (handleRequests b c) cleanup = liftIO . atomically $ do ms <- readTVar (monitors c) mapM_ (unsubscribeMonitor sub) ms@@ -84,6 +86,24 @@ 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)++-- | Unsubscribe from subscriber and also delete our monitor+removeMonitor :: Client api -> Path -> STM ()+removeMonitor c path = 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++-- | Does not remove the monitor - use removeMonitor if you want this! unsubscribeMonitor :: Subscriber api -> StatusMonitor -> STM () unsubscribeMonitor sub m = let@@ -92,75 +112,75 @@ in unsubscribe path mon sub -subscribeMonitor :: Subscriber api -> HttpRequest -> Client -> STM ()-subscribeMonitor sub req c = do- let path = httpPath req- tState <- subscribe path sub- stateVal <- refValue <$> readTVar tState- modifyTVar' (monitors c) $ Map.insert path (StatusMonitor req tState stateVal) -handleRequests :: Backend backend => backend -> Subscriber api -> Client -> IO ()-handleRequests b sub c = forever $ do+handleRequests :: Backend backend => backend -> Client api -> IO ()+handleRequests b c = forever $ do req <- readRequest c case req of- Nothing -> writeResponse c (RequestError ParseError)- Just (Subscribe req) -> handleSubscribe b sub c req- Just (Unsubscribe path) -> handleUnsubscribe b sub c path+ Nothing -> writeResponse c ParseError+ Just (Subscribe httpReq) -> handleSubscribe b c httpReq+ Just (Unsubscribe path) -> handleUnsubscribe c path -handleSubscribe :: Backend backend => backend -> Subscriber api -> Client -> HttpRequest -> IO ()-handleSubscribe b sub c req = void $ requestResource b req $ \ httpResponse -> do- let status = statusCode . httpStatus $ httpResponse- let path = httpPath req- let isGoodStatus = status >= 200 && status < 300 -- For now we only accept success- if isGoodStatus- then- mapM_ (writeResponse c) <=< atomically $ do- ms <- readTVar (monitors c)- case Map.lookup path ms of- Just _ -> return [RequestError (AlreadySubscribed path)]- Nothing -> do- subscribeMonitor sub req c- return [Resp.Subscribed path, Resp.Modified path httpResponse]- else- writeResponse c $ RequestError (HttpRequestFailed req httpResponse)- return ResponseReceived+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 ]+ _ -> return [ response ] -handleUnsubscribe :: Backend backend => backend -> Subscriber api -> Client -> Path -> IO ()-handleUnsubscribe b sub c path = writeResponse c <=< atomically $ do- ms <- readTVar (monitors c)- case Map.lookup path ms of- Nothing -> return $ RequestError (NoSuchSubscription path)- Just m -> do- unsubscribeMonitor sub m- modifyTVar (monitors c) $ Map.delete path- return $ Unsubscribed path+handleUnsubscribe :: Client api -> Path -> IO ()+handleUnsubscribe c path = do+ atomically $ removeMonitor c path+ writeResponse c $ Unsubscribed path -runMonitor :: Backend backend => backend -> Client -> IO ()++runMonitor :: Backend backend => backend -> Client api -> IO () runMonitor b c = forever $ do changes <- atomically $ monitorChanges c- mapM_ (sendUpdate b (writeResponse c)) changes+ mapM_ (handleUpdate b c) changes -sendUpdate :: Backend backend => backend -> (Response -> IO ()) -> (HttpRequest, ResourceStatus) -> IO ()-sendUpdate b sendResponse (req, S.Deleted) = sendResponse $ Resp.Deleted (httpPath req)-sendUpdate b sendResponse (req, S.Modified _) = sendServerResponse b req sendResponse+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 -sendServerResponse :: Backend backend => backend -> HttpRequest -> (Response -> IO ()) -> IO ()-sendServerResponse b req sendResponse = void $ requestResource b req+ handleResponse :: Response -> IO ()+ handleResponse resp = case resp of+ HttpRequestFailed _ _ -> do+ atomically $ removeMonitor c (httpPath req)+ sendResponse resp+ _ ->+ sendResponse 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- sendResponse $ Resp.Modified path httpResponse+ let isGoodStatus = status >= 200 && status < 300 -- We only accept success+ sendResponse $ if isGoodStatus+ then+ Resp.Modified path response+ else+ Resp.HttpRequestFailed req httpResponse return ResponseReceived -monitorChanges :: Client -> STM [(HttpRequest, ResourceStatus)]+monitorChanges :: Client api -> STM [(HttpRequest, ResourceStatus)] monitorChanges c = do- snapshots <- mapM toSnapshot . Map.elems =<< readTVar (monitors c)- let result = getChanges snapshots- if null result- then retry- else do- let newMonitors = monitorsFromList . updateMonitors $ snapshots- writeTVar (monitors c) newMonitors- return result+ snapshots <- mapM toSnapshot . Map.elems =<< readTVar (monitors c)+ let result = getChanges snapshots+ if null result+ then retry+ else do+ let newMonitors = monitorsFromList . updateMonitors $ snapshots+ writeTVar (monitors c) newMonitors+ return result getChanges :: [Snapshot] -> [(HttpRequest, ResourceStatus)]
src/Servant/Subscriber/Response.hs view
@@ -28,10 +28,11 @@ -- | Any message from the server is a Response. data Response = Subscribed !Path -- |< Resource was successfully subscribed- | Modified !Path !HttpResponse -- |< Can also be a non 2xx code, ServerError only triggers on the very first response, resulting in a failed subscription.+ | Modified !Path !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- | RequestError !RequestError+ | 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 instance ToJSON Response@@ -50,16 +51,6 @@ } deriving Generic instance ToJSON Status---- | Your subscription did not work out because:-data RequestError =- ParseError- | HttpRequestFailed !R.HttpRequest !HttpResponse -- |< The server replied with some none 2xx status code. Thus your subscription failed.- | NoSuchSubscription !Path- | AlreadySubscribed !Path deriving Generic--instance ToJSON RequestError- data ResponseBody = ResponseBody B.Builder deriving Generic
src/Servant/Subscriber/Subscribable.hs view
@@ -63,7 +63,11 @@ IsSubscribable e a = IsSubscribable' e a -type instance IsElem' (Subscribable :> e) (Subscribable :> s) = IsElem e s+type instance IsElem' sa (Subscribable :> sb) = SubscribableIsElem sa (Subscribable :> sb)++type family SubscribableIsElem endpoint api :: Constraint where+ SubscribableIsElem (Subscribable :> e) (Subscribable :> s) = IsElem e s+ SubscribableIsElem e (Subscribable :> s) = IsElem e s -- | A valid endpoint may only contain Symbols and captures & for convenince Subscribable: type family IsValidEndpoint endpoint :: Constraint where
− test/Spec.hs
@@ -1,125 +0,0 @@-{-# LANGUAGE DataKinds #-}-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeOperators #-}-{-# LANGUAGE FlexibleContexts #-}--- {-# LANGUAGE NoMonomorphismRestriction #-}---import Control.Monad.Except-import Data.Aeson.Compat-import Data.Aeson.Types-import Data.Attoparsec.ByteString-import Data.ByteString (ByteString)-import Data.List-import Data.Maybe-import Data.String.Conversions-import Data.Time.Calendar-import GHC.Generics-import Lucid-import Network.HTTP.Media ((//), (/:))-import Network.Wai-import Network.Wai.Handler.Warp-import Servant-import System.Directory-import Text.Blaze-import Text.Blaze.Html.Renderer.Utf8-import qualified Data.Aeson.Parser-import qualified Text.Blaze.Html-- -import Lib---type API = "position" :> Capture "x" Int :> Capture "y" Int :> Get '[JSON] Position- :<|> "hello" :> QueryParam "name" String :> Get '[JSON] HelloMessage- :<|> "marketing" :> ReqBody '[JSON] ClientInfo :> Post '[JSON] Email--data Position = Position- { xCoord :: Int- , yCoord :: Int- } deriving (Generic, Show)--instance ToJSON Position--newtype HelloMessage = HelloMessage { msg :: String }- deriving (Generic, Show)--instance ToJSON HelloMessage--data ClientInfo = ClientInfo- { clientName :: String- , clientEmail :: String- , clientAge :: Int- , clientInterestedIn :: [String]- } deriving Generic--instance FromJSON ClientInfo-instance ToJSON ClientInfo--data Email = Email- { from :: String- , to :: String- , subject :: String- , body :: String- } deriving Generic--instance ToJSON Email--emailForClient :: ClientInfo -> Email-emailForClient c = Email from' to' subject' body'-- where from' = "great@company.com"- to' = clientEmail c- subject' = "Hey " ++ clientName c ++ ", we miss you!"- body' = "Hi " ++ clientName c ++ ",\n\n"- ++ "Since you've recently turned " ++ show (clientAge c)- ++ ", have you checked out our latest "- ++ intercalate ", " (clientInterestedIn c)- ++ " products? Give us a visit!"--server3 :: ServerT API (ExceptT ServantErr IO)-server3 = position- :<|> hello- :<|> marketing-- where position :: Int -> Int -> ExceptT ServantErr IO Position- position x y = return (Position x y)-- hello :: Maybe String -> ExceptT ServantErr IO HelloMessage- hello mname = return . HelloMessage $ case mname of- Nothing -> "Hello, anonymous coward"- Just n -> "Hello, " ++ n-- marketing :: ClientInfo -> ExceptT ServantErr IO Email- marketing clientinfo = return (emailForClient clientinfo)--userAPI :: Proxy API-userAPI = Proxy--app3 :: Application-app3 = serve userAPI server3---callServer3 :: forall endpoint. (GetEndpoint API endpoint (PickLeftRight endpoint API))- => Proxy endpoint- -> ServerT endpoint (ExceptT ServantErr IO)-callServer3 pE = callHandler (Proxy :: Proxy API) server3 pE--main :: IO ()-main = do- -- Let's call handler by url!- let endpoint :: Proxy ("position" :> Capture "x" Int :> Capture "y" Int :> Get '[JSON] Position)- endpoint = Proxy- pos <- runExceptT $ callServer3 endpoint 9 2- print pos- let endpoint2 :: Proxy ("hello" :> QueryParam "name" String :> Get '[JSON] HelloMessage)- endpoint2 = Proxy- hello <- runExceptT $ callServer3 endpoint2 $ Just "Robert"- hello2 <- runExceptT $ callServer3 endpoint2 Nothing- print hello- print hello2