diff --git a/servant-subscriber.cabal b/servant-subscriber.cabal --- a/servant-subscriber.cabal +++ b/servant-subscriber.cabal @@ -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 diff --git a/src/Servant/Subscriber.hs b/src/Servant/Subscriber.hs --- a/src/Servant/Subscriber.hs +++ b/src/Servant/Subscriber.hs @@ -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) diff --git a/src/Servant/Subscriber/Backend/Wai.hs b/src/Servant/Subscriber/Backend/Wai.hs --- a/src/Servant/Subscriber/Backend/Wai.hs +++ b/src/Servant/Subscriber/Backend/Wai.hs @@ -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 ) diff --git a/src/Servant/Subscriber/Client.hs b/src/Servant/Subscriber/Client.hs --- a/src/Servant/Subscriber/Client.hs +++ b/src/Servant/Subscriber/Client.hs @@ -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 diff --git a/src/Servant/Subscriber/Request.hs b/src/Servant/Subscriber/Request.hs --- a/src/Servant/Subscriber/Request.hs +++ b/src/Servant/Subscriber/Request.hs @@ -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 diff --git a/src/Servant/Subscriber/Response.hs b/src/Servant/Subscriber/Response.hs --- a/src/Servant/Subscriber/Response.hs +++ b/src/Servant/Subscriber/Response.hs @@ -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