diff --git a/Network/Wai/Handler/WebSockets.hs b/Network/Wai/Handler/WebSockets.hs --- a/Network/Wai/Handler/WebSockets.hs +++ b/Network/Wai/Handler/WebSockets.hs @@ -4,71 +4,96 @@ , interceptWith ) where -import Control.Monad.IO.Class (liftIO) -import Data.ByteString (ByteString) -import Data.Char (toLower) -import qualified Data.ByteString.Char8 as S -import Data.Conduit -import qualified Data.Enumerator as E -import qualified Network.Wai as Wai -import qualified Network.Wai.Handler.Warp as Warp -import qualified Network.WebSockets as WS +import Control.Monad (forever) +import Control.Monad.IO.Class (liftIO) +import Control.Concurrent (forkIO, threadDelay) +import Control.Exception (SomeException (..), handle) +import Blaze.ByteString.Builder (Builder) +import qualified Blaze.ByteString.Builder as Builder +import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as BC +import Data.Char (toLower) +import Data.Conduit +import qualified Network.Wai as Wai +import qualified Network.Wai.Handler.Warp as Warp +import qualified Network.WebSockets as WS +import qualified Network.WebSockets.Connection as WS +import System.IO.Streams (InputStream, OutputStream) +import qualified System.IO.Streams as Streams +-------------------------------------------------------------------------------- -- | For use with 'settingsIntercept' from the Warp web server. -intercept :: WS.Protocol p - => (WS.Request -> WS.WebSockets p ()) +intercept :: WS.ServerApp -> Wai.Request -> Maybe (Source (ResourceT IO) ByteString -> Warp.Connection -> ResourceT IO ()) -intercept = interceptWith WS.defaultWebSocketsOptions +intercept = interceptWith WS.defaultConnectionOptions +-------------------------------------------------------------------------------- -- | Variation of 'intercept' which allows custom options. -interceptWith :: WS.Protocol p - => WS.WebSocketsOptions - -> (WS.Request -> WS.WebSockets p ()) +interceptWith :: WS.ConnectionOptions + -> WS.ServerApp -> Wai.Request -> Maybe (Source (ResourceT IO) ByteString -> Warp.Connection -> ResourceT IO ()) -interceptWith opts app req = case lookup "upgrade" $ Wai.requestHeaders req of +interceptWith opts app req = case lookup "upgrade" (Wai.requestHeaders req) of Just s - | S.map toLower s == "websocket" -> Just $ runWebSockets opts req' app + | BC.map toLower s == "websocket" -> Just $ runWebSockets opts req' app | otherwise -> Nothing _ -> Nothing - where - req' = WS.RequestHttpPart (Wai.rawPathInfo req) (Wai.requestHeaders req) - (Wai.isSecure req) + where + req' = WS.RequestHead (Wai.rawPathInfo req) (Wai.requestHeaders req) (Wai.isSecure req) --- | Internal function to run the WebSocket iteratee using the conduit library -runWebSockets :: WS.Protocol p - => WS.WebSocketsOptions - -> WS.RequestHttpPart - -> (WS.Request -> WS.WebSockets p ()) +-------------------------------------------------------------------------------- +---- | Internal function to run the WebSocket io-streams using the conduit library +runWebSockets :: WS.ConnectionOptions + -> WS.RequestHead + -> WS.ServerApp -> Source (ResourceT IO) ByteString -> Warp.Connection -> ResourceT IO () -runWebSockets opts req app source conn = do - step <- liftIO $ E.runIteratee $ WS.runWebSocketsWith opts req app send - source $$ sink (E.returnI step) - where - send = iterConnection conn +runWebSockets opts req app _ conn = do - sink iter = await >>= maybe (close iter) (push iter) + (is, os) <- liftIO $ connectionToStreams conn - push iter bs = do - step <- liftIO $ E.runIteratee $ E.enumList 1 [bs] E.$$ iter - case step of - E.Continue _ -> sink $ E.returnI step - E.Yield out inp -> maybe (return ()) leftover (streamToMaybe inp) >> return out - E.Error e -> liftIO $ monadThrow e - close iter = do - _ <- liftIO $ E.runIteratee $ E.enumEOF E.$$ iter - return () + let pc = WS.PendingConnection + { WS.pendingOptions = opts + , WS.pendingRequest = req + , WS.pendingOnAccept = forkPingThread + , WS.pendingIn = is + , WS.pendingOut = os + } -iterConnection :: Warp.Connection -> E.Iteratee ByteString IO () -iterConnection c = E.continue go - where - go (E.Chunks []) = E.continue go - go (E.Chunks cs) = E.tryIO (Warp.connSendMany c cs) >> E.continue go - go E.EOF = E.continue go + liftIO $ app pc -streamToMaybe :: E.Stream S.ByteString -> Maybe S.ByteString -streamToMaybe E.EOF = Nothing -streamToMaybe (E.Chunks bs) = Just $ S.concat bs +-------------------------------------------------------------------------------- +-- | Start a ping thread in the background +forkPingThread :: WS.Connection -> IO () +forkPingThread conn = do + _ <- forkIO pingThread + return () + where + pingThread = handle ignore $ forever $ do + WS.sendPing conn (BC.pack "ping") + threadDelay $ 30 * 1000 * 1000 + + ignore :: SomeException -> IO () + ignore _ = return () + +------------------------------------------------------------------------------ +-- | Converts a 'Connection' to an 'InputStream' \/ 'OutputStream' pair. Note that, +-- as is usually the case in @io-streams@, writing a 'Nothing' to the generated +-- 'OutputStream' does not cause the underlying 'Connection' to be closed. +connectionToStreams :: Warp.Connection + -> IO (InputStream ByteString, OutputStream Builder) +connectionToStreams connection = do + is <- Streams.makeInputStream input + os <- Streams.makeOutputStream output + return $! (is, os) + + where + input = do + s <- Warp.connRecv connection + return $! if BC.null s then Nothing else Just s + + output Nothing = return $! () + output (Just s') = if BC.null s then return $! () else Warp.connSendAll connection s + where s = Builder.toByteString s' diff --git a/server.lhs b/server.lhs new file mode 100644 --- /dev/null +++ b/server.lhs @@ -0,0 +1,153 @@ +websockets example +================== + +This is the Haskell implementation of the example for the WebSockets library. We +implement a simple multi-user chat program. A live demo of the example is +available [here](http://jaspervdj.be/websockets-example). In order to understand +this example, keep the [reference](http://jaspervdj.be/websockets/reference) +nearby to check out the functions we use. + +> {-# LANGUAGE OverloadedStrings, TemplateHaskell #-} +> import Data.Char (isPunctuation, isSpace) +> import Data.Monoid (mappend) +> import Data.Text (Text) +> import Control.Exception (fromException, handle) +> import Control.Monad (forM_, forever) +> import Control.Concurrent (MVar, newMVar, modifyMVar_, readMVar) +> import Control.Monad.IO.Class (liftIO) +> import qualified Data.Text as T +> import qualified Data.Text.IO as T + +> import qualified Network.WebSockets as WS +> import qualified Network.Wai +> import qualified Network.Wai.Handler.Warp as Warp +> import qualified Network.Wai.Handler.WebSockets as WaiWS +> import qualified Network.Wai.Application.Static as Static +> import Data.FileEmbed (embedDir) + +We represent a client by his username and a 'WS.Connection'. We will see how we obtain this 'WS.Connection' later on. + +> type Client = (Text, WS.Connection) + +The state kept on the server is simply a list of connected clients. We've added +an alias and some utility functions, so it will be easier to extend this state +later on. + +> type ServerState = [Client] + +Create a new, initial state + +> newServerState :: ServerState +> newServerState = [] + +Get the number of active clients + +> numClients :: ServerState -> Int +> numClients = length + +Check if a user already exists (based on username) + +> clientExists :: Client -> ServerState -> Bool +> clientExists client = any ((== fst client) . fst) + +Add a client (first, you should verify the client is not already connected using +'clientExists') + +> addClient :: Client -> ServerState -> ServerState +> addClient client clients = client : clients + +Remove a client + +> removeClient :: Client -> ServerState -> ServerState +> removeClient client = filter ((/= fst client) . fst) + +Send a message to all clients, and log it on stdout. + +> broadcast :: Text -> ServerState -> IO () +> broadcast message clients = do +> T.putStrLn message +> forM_ clients $ \(_, conn) -> WS.sendTextData conn message + +The main function first creates a new state for the server, then spawns the +actual server. For this purpose, we use the simple server provided by +'WS.runServer'. + +> main :: IO () +> main = do +> putStrLn "http://localhost:9160/client.html" +> state <- newMVar newServerState +> Warp.runSettings Warp.defaultSettings +> { Warp.settingsPort = 9160 +> , Warp.settingsIntercept = WaiWS.intercept (application state) +> } staticApp + +> staticApp :: Network.Wai.Application +> staticApp = Static.staticApp $ Static.embeddedSettings $(embedDir "static") + +When a client connects, we accept the connection, regardless of the path. + +> application :: MVar ServerState -> WS.ServerApp +> application state pending = do +> conn <- WS.acceptRequest pending + +When a client is succesfully connected, we read the first message. This should +be in the format of "Hi, I am Jasper", where Jasper is the requested username. + +> msg <- WS.receiveData conn +> clients <- liftIO $ readMVar state +> case msg of + +Check that the first message has the right format + +> _ | not (prefix `T.isPrefixOf` msg) -> +> WS.sendTextData conn ("Wrong announcement" :: Text) + +Check the validity of the username + +> | any ($ fst client) +> [T.null, T.any isPunctuation, T.any isSpace] -> +> WS.sendTextData conn ("Name cannot " `mappend` +> "contain punctuation or whitespace, and " `mappend` +> "cannot be empty" :: Text) + +Check that the given username is not already taken + +> | clientExists client clients -> +> WS.sendTextData conn ("User already exists" :: Text) + +All is right! + +> | otherwise -> do + +We send a "Welcome!", according to our own little protocol. We add the client to +the list and broadcast the fact that he has joined. Then, we give control to the +'talk' function. + +> liftIO $ modifyMVar_ state $ \s -> do +> let s' = addClient client s +> WS.sendTextData conn $ +> "Welcome! Users: " `mappend` +> T.intercalate ", " (map fst s) +> broadcast (fst client `mappend` " joined") s' +> return s' +> talk conn state client +> where +> prefix = "Hi! I am " +> client = (T.drop (T.length prefix) msg, conn) + +The talk function continues to read messages from a single client until he +disconnects. All messages are broadcasted to the other clients. + +> talk :: WS.Connection -> MVar ServerState -> Client -> IO () +> talk conn state client@(user, _) = handle catchDisconnect $ +> forever $ do +> msg <- WS.receiveData conn +> liftIO $ readMVar state >>= broadcast +> (user `mappend` ": " `mappend` msg) +> where +> catchDisconnect e = case fromException e of +> Just WS.ConnectionClosed -> liftIO $ modifyMVar_ state $ \s -> do +> let s' = removeClient client s +> broadcast (user `mappend` " disconnected") s' +> return s' +> _ -> return () diff --git a/wai-websockets.cabal b/wai-websockets.cabal --- a/wai-websockets.cabal +++ b/wai-websockets.cabal @@ -1,9 +1,9 @@ Name: wai-websockets -Version: 1.3.1.3 +Version: 1.3.2.0 Synopsis: Provide a bridge betweeen WAI and the websockets package. License: MIT License-file: LICENSE -Author: Michael Snoyman, Jasper Van der Jeugt +Author: Michael Snoyman, Jasper Van der Jeugt, Ting-Yen Lai Maintainer: michael@snoyman.com Homepage: http://github.com/yesodweb/wai Category: Web, Yesod @@ -20,17 +20,40 @@ Build-Depends: base >= 3 && < 5 , bytestring >= 0.9.1.4 , conduit >= 0.5 && < 1.1 - , wai >= 1.3 && < 2.1 - , enumerator >= 0.4.8 && < 0.5 - , network-enumerator >= 0.1.2 && < 0.2 + , wai >= 1.3 && < 1.5 , blaze-builder >= 0.2.1.4 && < 0.4 , case-insensitive >= 0.2 , network >= 2.2.1.5 , transformers >= 0.2 && < 0.4 - , websockets >= 0.6 && < 0.8 - , warp >= 1.3 && < 2.1 + , websockets >= 0.8 + , warp >= 1.3 && < 1.4 + , io-streams >= 1.1 && < 1.2 Exposed-modules: Network.Wai.Handler.WebSockets ghc-options: -Wall + +Executable wai-websockets-example + if flag(example) + buildable: True + else + buildable: False + Build-Depends: base >= 3 && < 5 + , conduit + , wai-websockets + , websockets + , warp + , wai + , wai-app-static + , bytestring + , case-insensitive + , blaze-builder + , transformers + , network + , text + , file-embed + , io-streams + + ghc-options: -Wall -threaded + main-is: server.lhs source-repository head type: git