servant-subscriber 0.4.0.0 → 0.5.0.0
raw patch · 6 files changed
+127/−70 lines, 6 files
Files
- servant-subscriber.cabal +1/−1
- src/Servant/Subscriber.hs +21/−10
- src/Servant/Subscriber/Backend/Wai.hs +20/−8
- src/Servant/Subscriber/Client.hs +49/−22
- src/Servant/Subscriber/Request.hs +19/−6
- src/Servant/Subscriber/Response.hs +17/−23
servant-subscriber.cabal view
@@ -1,5 +1,5 @@ name: servant-subscriber-version: 0.4.0.0+version: 0.5.0.0 synopsis: When REST is not enough ... description: Please see Readme.md homepage: http://github.com/eskimor/servant-subscriber#readme
src/Servant/Subscriber.hs view
@@ -26,35 +26,46 @@ import Control.Concurrent.STM import Control.Monad-import qualified Data.Map as Map+import Data.IORef (IORef, newIORef, readIORef)+import qualified Data.Map as Map import Data.Proxy-import Network.URI (URI(..), pathSegments)+import Network.URI (URI (..), pathSegments) import Network.Wai import Network.Wai.Handler.WebSockets-import Network.WebSockets.Connection as WS+import Network.WebSockets.Connection as WS import Servant.Server import Servant.Subscriber.Subscribable-import Servant.Utils.Links (IsElem, HasLink, MkLink, safeLink)+import Servant.Utils.Links (HasLink, IsElem, MkLink,+ safeLink) -import Servant.Subscriber.Backend.Wai ()-import qualified Servant.Subscriber.Client as Client+import Servant.Subscriber.Backend.Wai ()+import qualified Servant.Subscriber.Client as Client import Servant.Subscriber.Types makeSubscriber :: Path -> LogRunner -> STM (Subscriber api)-makeSubscriber entryPoint logRunner = do+makeSubscriber entryPoint' logRunner = do state <- newTVar Map.empty- return $ Subscriber state entryPoint logRunner+ return $ Subscriber state entryPoint' logRunner serveSubscriber :: forall api. (HasServer api '[]) => Subscriber api -> Server api -> Application serveSubscriber subscriber server req sendResponse = do let app = serve (Proxy :: Proxy api) server- let opts = defaultConnectionOptions+ (pongHandler, myRef) <- makePongAction+ let opts = defaultConnectionOptions {+ connectionOnPong = pongHandler+ } let runLog = runLogging subscriber let handleWSConnection pending = do connection <- acceptRequest pending forkPingThread connection 28- runLog . Client.run app <=< atomically . Client.fromWebSocket subscriber $ connection+ runLog . Client.run app <=< atomically . Client.fromWebSocket subscriber myRef $ connection if Path (pathInfo req) == entryPoint subscriber then websocketsOr opts handleWSConnection app req sendResponse else app req sendResponse++makePongAction :: IO (IO (), IORef (IO ()))+makePongAction = do+ myRef <- newIORef $ return ()+ let action = join $ readIORef myRef+ return (action, myRef)
src/Servant/Subscriber/Backend/Wai.hs view
@@ -1,5 +1,6 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE OverloadedStrings #-} -- | Backend of accessing a wai server:@@ -7,12 +8,14 @@ import qualified Blaze.ByteString.Builder as B import Data.Aeson-import qualified Data.ByteString as BS+import qualified Data.ByteString as BS import Data.IORef-import qualified Data.Text.Encoding as T-import qualified Network.HTTP.Types as H-import qualified Network.Wai as Wai-import qualified Network.Wai.Internal as Wai+import Data.Monoid ((<>))+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import qualified Network.HTTP.Types as H+import qualified Network.Wai as Wai+import qualified Network.Wai.Internal as Wai import Servant.Subscriber.Backend import Servant.Subscriber.Request as Req@@ -40,12 +43,12 @@ , Wai.rawPathInfo = B.toByteString . H.encodePathSegments . toSegments . httpPath $ r , Wai.queryString = H.queryTextToQuery . httpQuery $ r , Wai.rawQueryString = B.toByteString . H.renderQueryText True . httpQuery $ r- , Wai.requestHeaders = toHTTPHeaders . Req.httpHeaders $ r+ , Wai.requestHeaders = toHTTPHeaders . (<> standardHeaders) . Req.httpHeaders $ r , Wai.requestBody = waiBody , Wai.requestBodyLength = Wai.KnownLength . fromIntegral . BS.length $ encodedBody } where- encodedBody = B.toByteString . fromEncoding . toEncoding . Req.httpBody $ r+ encodedBody = T.encodeUtf8 . runRequestBody . Req.httpBody $ r mkWaiRequestBody :: BS.ByteString -> IO (IO BS.ByteString) mkWaiRequestBody b = do@@ -60,6 +63,15 @@ fromWaiResponse (Wai.ResponseBuilder status headers builder)= HttpResponse { httpStatus = fromHTTPStatus status , Res.httpHeaders = fromHTTPHeaders headers- , Res.httpBody = ResponseBody builder+ , Res.httpBody = T.decodeUtf8 . B.toByteString $ builder } fromWaiResponse _ = error "I am sorry - this 'Response' type is not yet implemented in servant-subscriber!"++standardHeaders :: RequestHeaders+standardHeaders = [+ ("Accept", "application/json")+ , ("Content-Type", "application/json")+ ]++lengthHeader :: BS.ByteString -> RequestHeader+lengthHeader body = ( "Content-Length", T.pack . show . BS.length $ body )
src/Servant/Subscriber/Client.hs view
@@ -1,38 +1,46 @@ {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-} module Servant.Subscriber.Client where import Control.Concurrent.Async-import Control.Concurrent.STM (STM, atomically, retry)+import Control.Concurrent.STM (STM, atomically, retry) import Control.Concurrent.STM.TVar-import Control.Exception (displayException, SomeException)-import Control.Exception.Lifted (finally, try)+import Control.Exception (SomeException, displayException)+import Control.Exception.Lifted (finally, try) import Control.Monad import Control.Monad.IO.Class-import Control.Monad.Logger (MonadLogger, logDebug)-import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad.Logger (MonadLogger, logDebug)+import Control.Monad.Trans.Control (MonadBaseControl) 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 qualified Data.ByteString.Lazy as BS+import Data.IORef (IORef, writeIORef)+import Data.Map (Map)+import qualified Data.Map.Strict as Map+import Data.Monoid ((<>))+import Data.Set (Set)+import qualified Data.Set as Set+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import qualified Data.Text.IO as T+import qualified Network.WebSockets as WS import Network.WebSockets.Connection as WS import Servant.Subscriber.Backend import Servant.Subscriber.Request import Servant.Subscriber.Response as Resp-import Servant.Subscriber.Types as S+import Servant.Subscriber.Types as S type ClientMonitors = Map Path StatusMonitor data Client api = Client {- subscriber :: !(Subscriber api)- , monitors :: !(TVar ClientMonitors)- , readRequest :: !(IO (Maybe Request))- , writeResponse :: !(Response -> IO ())+ subscriber :: !(Subscriber api)+ , monitors :: !(TVar ClientMonitors)+ , readRequest :: !(IO (Maybe Request))+ , writeResponse :: !(Response -> IO ())+ , pongCommandRef :: !(IORef (IO ()))+ , closeCommandRef :: !(TVar (IO ())) -- |< TVar so actions on it can happen in STM } data StatusMonitor = StatusMonitor {@@ -61,9 +69,10 @@ snapshotRequests :: Snapshot -> Set HttpRequest snapshotRequests = requests . fullMonitor -fromWebSocket :: Subscriber api -> WS.Connection -> STM (Client api)-fromWebSocket sub c = do+fromWebSocket :: Subscriber api -> IORef (IO ()) -> WS.Connection -> STM (Client api)+fromWebSocket sub myRef c = do ms <- newTVar Map.empty+ closeVar <- newTVar (return ()) return Client { subscriber = sub , monitors = ms@@ -73,6 +82,8 @@ WS.Text bs -> return $ decode bs WS.Binary _ -> error "Sorry - binary connections currently unsupported!" , writeResponse = sendDataMessage c . WS.Text . encode+ , pongCommandRef = myRef+ , closeCommandRef = closeVar } run :: (MonadLogger m, MonadBaseControl IO m, MonadIO m, Backend backend)@@ -81,9 +92,12 @@ let sub = subscriber c work = liftIO $ race_ (runMonitor b c) (handleRequests b c)- cleanup = liftIO . atomically $ do- ms <- readTVar (monitors c)- mapM_ (unsubscribeMonitor sub) ms+ cleanup = do+ close <- liftIO . atomically $ do+ ms <- readTVar (monitors c)+ mapM_ (unsubscribeMonitor sub) ms+ readTVar (closeCommandRef c)+ liftIO close r <- try $ finally work cleanup case r of Left e -> $logDebug $ T.pack $ displayException (e :: SomeException)@@ -132,7 +146,7 @@ unsubscribeMonitor :: Subscriber api -> StatusMonitor -> STM () unsubscribeMonitor sub m = let- path = monitorPath m + path = monitorPath m mon = monitor m in unsubscribe path mon sub@@ -145,6 +159,18 @@ Nothing -> writeResponse c ParseError Just (Subscribe httpReq) -> handleSubscribe b c httpReq Just (Unsubscribe path) -> handleUnsubscribe c path+ Just (SetPongRequest httpReq) -> do+ let doIt = doRequestIgnoreResult httpReq+ writeIORef (pongCommandRef c) doIt+ doIt+ writeResponse c $ Subscribed httpReq+ Just (SetCloseRequest httpReq) -> do+ let doIt = doRequestIgnoreResult httpReq+ atomically $ writeTVar (closeCommandRef c) doIt+ writeResponse c $ Subscribed httpReq+ where+ 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@@ -186,6 +212,7 @@ $ \ httpResponse -> do let status = statusCode . httpStatus $ httpResponse let response = Resp.httpBody httpResponse+ T.putStrLn $ "Got server response body: " <> T.decodeUtf8 (BS.toStrict . encode $ response) let isGoodStatus = status >= 200 && status < 300 -- We only accept success sendResponse $ if isGoodStatus then
src/Servant/Subscriber/Request.hs view
@@ -7,11 +7,11 @@ import Data.Aeson import Data.Bifunctor-import qualified Data.CaseInsensitive as Case-import Data.Text (Text)-import qualified Data.Text.Encoding as T+import qualified Data.CaseInsensitive as Case+import Data.Text (Text)+import qualified Data.Text.Encoding as T import GHC.Generics-import qualified Network.HTTP.Types as H+import qualified Network.HTTP.Types as H import Servant.Subscriber.Types @@ -27,6 +27,13 @@ data Request = Subscribe !HttpRequest | Unsubscribe !HttpRequest+-- | A request that should be issued whenever a websocket pong is received.+-- In addition to every websocket pong the request also gets issued+-- immediately upon receival. Bot `SetPongRequest` and `SetCloseRequest` will+-- be confirmed with a `Subscribed` response, but any return value of the+-- request won't be delivered.+ | SetPongRequest !HttpRequest+ | SetCloseRequest !HttpRequest -- |< A request that should be issued when the websocket connection closes for whatever reason. deriving Generic instance FromJSON Request@@ -39,20 +46,26 @@ , httpHeaders :: RequestHeaders , httpQuery :: H.QueryText , httpBody :: RequestBody-} deriving ( Generic, Eq, Ord )+} deriving ( Generic, Eq, Ord, Show ) instance FromJSON HttpRequest instance ToJSON HttpRequest -newtype RequestBody = RequestBody Text deriving (Generic, ToJSON, FromJSON, Eq, Ord)+newtype RequestBody = RequestBody Text deriving (Generic, ToJSON, FromJSON, Eq, Ord, Show) +runRequestBody :: RequestBody -> Text+runRequestBody (RequestBody t) = t+ toHTTPHeader :: RequestHeader -> H.Header toHTTPHeader = bimap (Case.mk . T.encodeUtf8) T.encodeUtf8 toHTTPHeaders :: RequestHeaders -> H.RequestHeaders toHTTPHeaders = map toHTTPHeader +-- | This is ugly - but I don't care for now. requestPath :: Request -> Path requestPath req = httpPath $ case req of Subscribe hReq -> hReq Unsubscribe hReq -> hReq+ SetPongRequest hReq -> hReq+ SetCloseRequest hReq -> hReq
src/Servant/Subscriber/Response.hs view
@@ -2,20 +2,21 @@ module Servant.Subscriber.Response where -import qualified Blaze.ByteString.Builder as B-import qualified Blaze.ByteString.Builder.Char8 as B+import qualified Blaze.ByteString.Builder as B+import qualified Blaze.ByteString.Builder.Char8 as B import Data.Aeson-import Data.Aeson.Parser (value)-import Data.Aeson.Types (unsafeToEncoding)-import Data.Attoparsec.ByteString (parseOnly)+import Data.Aeson.Parser (value)+import Data.Aeson.Types (unsafeToEncoding)+import Data.Attoparsec.ByteString (parseOnly) import Data.Bifunctor-import qualified Data.CaseInsensitive as Case-import Data.Monoid ((<>))-import Data.Text (Text)-import qualified Data.Text as T-import qualified Data.Text.Encoding as T+import qualified Data.ByteString.Lazy as BS+import qualified Data.CaseInsensitive as Case+import Data.Monoid ((<>))+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as T import GHC.Generics-import qualified Network.HTTP.Types as H+import qualified Network.HTTP.Types as H import Servant.Server import qualified Servant.Subscriber.Request as R@@ -40,27 +41,20 @@ data HttpResponse = HttpResponse { httpStatus :: !Status , httpHeaders :: !ResponseHeaders-, httpBody :: ResponseBody-} deriving (Generic)+, httpBody :: !ResponseBody+} deriving Generic instance ToJSON HttpResponse data Status = Status { statusCode :: !Int , statusMessage :: !Text-} deriving Generic+} deriving (Generic, Show) instance ToJSON Status -data ResponseBody = ResponseBody B.Builder deriving Generic+type ResponseBody = Text -instance ToJSON ResponseBody where- toJSON (ResponseBody b) = getValue $ parseOnly value (B.toByteString (wrapInString b))- where- getValue r = case r of- Left e -> error e- Right r -> r- toEncoding (ResponseBody b) = unsafeToEncoding . wrapInString $ b fromHTTPHeader :: H.Header -> ResponseHeader fromHTTPHeader = bimap (T.decodeUtf8 . Case.original) T.decodeUtf8@@ -78,7 +72,7 @@ fromServantError err = HttpResponse { httpStatus = Status (errHTTPCode err) (T.pack $ errReasonPhrase err) , httpHeaders = fromHTTPHeaders . errHeaders $ err-, httpBody = ResponseBody . B.fromLazyByteString . errBody $ err+, httpBody = T.decodeUtf8 . BS.toStrict . errBody $ err } wrapInString :: B.Builder -> B.Builder