diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +# Version 2.0.0 + + * Add instance for WebsocketSource. + * Update to Servant 0.16. + * Move from Conduit to ConduitT. + # Version 1.1.0 * Compatibility with servant-0.12. diff --git a/examples/Echo.hs b/examples/Echo.hs --- a/examples/Echo.hs +++ b/examples/Echo.hs @@ -1,20 +1,27 @@ -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE TypeOperators #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE TypeOperators #-} module Main where -import Servant.API.WebSocketConduit (WebSocketConduit) +import Servant.API.WebSocketConduit (WebSocketConduit, WebSocketSource) -import Data.Aeson (Value) -import Data.Conduit (Conduit) +import Control.Concurrent (threadDelay) +import Control.Monad (forever) +import Control.Monad.IO.Class (MonadIO (..)) +import Data.Aeson (Value (..)) +import Data.Conduit (ConduitT, yield) +import Data.Text (Text) import Network.Wai (Application) import Network.Wai.Handler.Warp (run) -import Servant (Proxy (..), Server, serve) +import Servant ((:<|>) (..), (:>), Proxy (..), Server, serve) import qualified Data.Conduit.List as CL -type API = WebSocketConduit Value Value +type API = "echo" :> WebSocketConduit Value Value + :<|> "hello" :> WebSocketSource Text + startApp :: IO () startApp = do putStrLn "Starting server on http://localhost:8080" @@ -27,10 +34,15 @@ api = Proxy server :: Server API -server = echo +server = echo :<|> hello -echo :: Monad m => Conduit Value m Value +echo :: Monad m => ConduitT Value Value m () echo = CL.map id + +hello :: MonadIO m => ConduitT () Text m () +hello = forever $ do + yield "hello world" + liftIO $ threadDelay 1000000 main :: IO () main = startApp diff --git a/servant-websockets.cabal b/servant-websockets.cabal --- a/servant-websockets.cabal +++ b/servant-websockets.cabal @@ -1,5 +1,5 @@ name: servant-websockets -version: 1.1.0 +version: 2.0.0 homepage: https://github.com/moesenle/servant-websockets#readme synopsis: Small library providing WebSocket endpoints for servant. description: Small library providing WebSocket endpoints for servant. @@ -24,6 +24,7 @@ , conduit , exceptions , resourcet + , monad-control , servant-server , text , wai @@ -42,6 +43,7 @@ , conduit , servant-server , servant-websockets + , text , wai , warp default-language: Haskell2010 diff --git a/src/Servant/API/WebSocket.hs b/src/Servant/API/WebSocket.hs --- a/src/Servant/API/WebSocket.hs +++ b/src/Servant/API/WebSocket.hs @@ -11,9 +11,10 @@ import Data.Proxy (Proxy (..)) import Network.Wai.Handler.WebSockets (websocketsOr) import Network.WebSockets (Connection, PendingConnection, acceptRequest, defaultConnectionOptions) -import Servant.Server (HasServer (..), ServantErr (..), ServerT, runHandler) +import Servant.Server (HasServer (..), ServerError (..), ServerT, runHandler) import Servant.Server.Internal.Router (leafRouter) -import Servant.Server.Internal.RoutingApplication (RouteResult (..), runDelayed) +import Servant.Server.Internal.RouteResult (RouteResult (..)) +import Servant.Server.Internal.Delayed (runDelayed) -- | Endpoint for defining a route to provide a web socket. The -- handler function gets an already negotiated websocket 'Connection' @@ -51,11 +52,11 @@ runApp a = acceptRequest >=> \c -> void (runHandler $ a c) - backupApp respond _ _ = respond $ Fail ServantErr { errHTTPCode = 426 - , errReasonPhrase = "Upgrade Required" - , errBody = mempty - , errHeaders = mempty - } + backupApp respond _ _ = respond $ Fail ServerError { errHTTPCode = 426 + , errReasonPhrase = "Upgrade Required" + , errBody = mempty + , errHeaders = mempty + } -- | Endpoint for defining a route to provide a web socket. The @@ -96,8 +97,8 @@ runApp a c = void (runHandler $ a c) - backupApp respond _ _ = respond $ Fail ServantErr { errHTTPCode = 426 - , errReasonPhrase = "Upgrade Required" - , errBody = mempty - , errHeaders = mempty - } + backupApp respond _ _ = respond $ Fail ServerError { errHTTPCode = 426 + , errReasonPhrase = "Upgrade Required" + , errBody = mempty + , errHeaders = mempty + } diff --git a/src/Servant/API/WebSocketConduit.hs b/src/Servant/API/WebSocketConduit.hs --- a/src/Servant/API/WebSocketConduit.hs +++ b/src/Servant/API/WebSocketConduit.hs @@ -1,4 +1,5 @@ {-# LANGUAGE CPP #-} +{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} @@ -9,22 +10,25 @@ import Control.Concurrent (newEmptyMVar, putMVar, takeMVar) import Control.Concurrent.Async (race_) -import Control.Monad (forever, (>=>)) +import Control.Monad (forever, void, (>=>)) import Control.Monad.Catch (handle) import Control.Monad.IO.Class (liftIO) -import Control.Monad.Trans.Resource (ResourceT, runResourceT) +import Control.Monad.Trans.Control (MonadBaseControl) +import Control.Monad.Trans.Resource (MonadUnliftIO, ResourceT, runResourceT) import Data.Aeson (FromJSON, ToJSON, decode, encode) import Data.ByteString.Lazy (fromStrict) -import Data.Conduit (Conduit, runConduitRes, yieldM, (.|)) +import Data.Conduit (ConduitT, runConduitRes, yieldM, (.|)) import Data.Proxy (Proxy (..)) import Data.Text (Text) +import Data.Void (Void) import Network.Wai.Handler.WebSockets (websocketsOr) -import Network.WebSockets (ConnectionException, acceptRequest, defaultConnectionOptions, - forkPingThread, receiveData, receiveDataMessage, sendClose, - sendTextData) -import Servant.Server (HasServer (..), ServantErr (..), ServerT) +import Network.WebSockets (Connection, ConnectionException, acceptRequest, + defaultConnectionOptions, forkPingThread, receiveData, + receiveDataMessage, sendClose, sendTextData) +import Servant.Server (HasServer (..), ServerError (..), ServerT) import Servant.Server.Internal.Router (leafRouter) -import Servant.Server.Internal.RoutingApplication (RouteResult (..), runDelayed) +import Servant.Server.Internal.RouteResult (RouteResult (..)) +import Servant.Server.Internal.Delayed (runDelayed) import qualified Data.Conduit.List as CL @@ -46,7 +50,7 @@ -- > server :: Server WebSocketApi -- > server = echo -- > where --- > echo :: Monad m => Conduit Value m Value +-- > echo :: Monad m => ConduitT Value Value m () -- > echo = CL.map id -- > -- @@ -56,7 +60,7 @@ instance (FromJSON i, ToJSON o) => HasServer (WebSocketConduit i o) ctx where - type ServerT (WebSocketConduit i o) m = Conduit i (ResourceT IO) o + type ServerT (WebSocketConduit i o) m = ConduitT i o (ResourceT IO) () #if MIN_VERSION_servant_server(0,12,0) hoistServerWithContext _ _ _ svr = svr @@ -69,27 +73,79 @@ websocketsOr defaultConnectionOptions (runWSApp cond) - (backupApp respond) + (\_ _ -> respond $ Fail upgradeRequired) request (respond . Route) go _ respond (Fail e) = respond $ Fail e go _ respond (FailFatal e) = respond $ FailFatal e runWSApp cond = acceptRequest >=> \c -> handle (\(_ :: ConnectionException) -> return ()) $ do - forkPingThread c 10 i <- newEmptyMVar - race_ (forever $ receiveData c >>= putMVar i) $ do - runConduitRes $ forever (yieldM . liftIO $ takeMVar i) - .| CL.mapMaybe (decode . fromStrict) - .| cond - .| CL.mapM_ (liftIO . sendTextData c . encode) - sendClose c ("Out of data" :: Text) - -- After sending the close message, we keep receiving packages - -- (and drop them) until the connection is actually closed, - -- which is indicated by an exception. - forever $ receiveDataMessage c + race_ (forever $ receiveData c >>= putMVar i) $ + runConduitWebSocket c $ + forever (yieldM . liftIO $ takeMVar i) + .| CL.mapMaybe (decode . fromStrict) + .| cond + .| CL.mapM_ (liftIO . sendTextData c . encode) - backupApp respond _ _ = respond $ Fail ServantErr { errHTTPCode = 426 - , errReasonPhrase = "Upgrade Required" - , errBody = mempty - , errHeaders = mempty - } +-- | Endpoint for defining a route to provide a websocket. In contrast +-- to the 'WebSocketConduit', this endpoint only produces data. It can +-- be useful when implementing web sockets that simply just send data +-- to clients. +-- +-- Example: +-- +-- > import Data.Text (Text) +-- > import qualified Data.Conduit.List as CL +-- > +-- > type WebSocketApi = "hello" :> WebSocketSource Text +-- > +-- > server :: Server WebSocketApi +-- > server = hello +-- > where +-- > hello :: Monad m => Conduit Text m () +-- > hello = yield $ Just "hello" +-- > +-- +data WebSocketSource o + +instance ToJSON o => HasServer (WebSocketSource o) ctx where + + type ServerT (WebSocketSource o) m = ConduitT () o (ResourceT IO) () + +#if MIN_VERSION_servant_server(0,12,0) + hoistServerWithContext _ _ _ svr = svr +#endif + + route Proxy _ app = leafRouter $ \env request respond -> runResourceT $ + runDelayed app env request >>= liftIO . go request respond + where + go request respond (Route cond) = + websocketsOr + defaultConnectionOptions + (runWSApp cond) + (\_ _ -> respond $ Fail upgradeRequired) + request (respond . Route) + go _ respond (Fail e) = respond $ Fail e + go _ respond (FailFatal e) = respond $ FailFatal e + + runWSApp cond = acceptRequest >=> \c -> handle (\(_ :: ConnectionException) -> return ()) $ + race_ (forever . void $ (receiveData c :: IO Text)) $ + runConduitWebSocket c $ cond .| CL.mapM_ (liftIO . sendTextData c . encode) + +runConduitWebSocket :: (MonadBaseControl IO m, MonadUnliftIO m) => Connection -> ConduitT () Void (ResourceT m) () -> m () +runConduitWebSocket c a = do + liftIO $ forkPingThread c 10 + void $ runConduitRes a + liftIO $ do + sendClose c ("Out of data" :: Text) + -- After sending the close message, we keep receiving packages + -- (and drop them) until the connection is actually closed, + -- which is indicated by an exception. + forever $ receiveDataMessage c + +upgradeRequired :: ServerError +upgradeRequired = ServerError { errHTTPCode = 426 + , errReasonPhrase = "Upgrade Required" + , errBody = mempty + , errHeaders = mempty + }