websockets 0.8.2.6 → 0.9.0.0
raw patch · 17 files changed
+502/−219 lines, 17 filesdep −io-streamsdep ~attoparsecdep ~bytestringdep ~network
Dependencies removed: io-streams
Dependency ranges changed: attoparsec, bytestring, network, random
Files
- CHANGELOG +6/−0
- src/Network/WebSockets.hs +7/−0
- src/Network/WebSockets/Client.hs +24/−19
- src/Network/WebSockets/Connection.hs +93/−34
- src/Network/WebSockets/Http.hs +13/−7
- src/Network/WebSockets/Hybi13.hs +46/−40
- src/Network/WebSockets/Hybi13/Demultiplex.hs +25/−13
- src/Network/WebSockets/Protocol.hs +11/−10
- src/Network/WebSockets/Server.hs +40/−22
- src/Network/WebSockets/Stream.hs +114/−0
- src/Network/WebSockets/Types.hs +19/−7
- tests/haskell/Network/WebSockets/Handshake/Tests.hs +30/−12
- tests/haskell/Network/WebSockets/Http/Tests.hs +1/−1
- tests/haskell/Network/WebSockets/Server/Tests.hs +39/−11
- tests/haskell/Network/WebSockets/Tests.hs +23/−18
- tests/haskell/Network/WebSockets/Tests/Util.hs +5/−18
- websockets.cabal +6/−7
CHANGELOG view
@@ -1,3 +1,9 @@+- 0.9.0.0+ * Bump various dependencies+ * Remove io-streams dependency+ * New close mechanism+ * More flexible API interface+ - 0.8.2.6 * Bump QuickCheck dependency
src/Network/WebSockets.hs view
@@ -4,7 +4,9 @@ ( -- * Incoming connections and handshaking PendingConnection , pendingRequest+ , AcceptRequest(..) , acceptRequest+ , acceptRequestWith , rejectRequest -- * Main connection type@@ -29,6 +31,7 @@ , Headers , Request (..) , RequestHead (..)+ , getRequestSubprotocols , Response (..) , ResponseHead (..) @@ -47,6 +50,10 @@ , ServerApp , runServer , runServerWith++ -- * Utilities for writing your own server+ , makeListenSocket+ , makePendingConnection -- * Running a client , ClientApp
src/Network/WebSockets/Client.hs view
@@ -12,19 +12,19 @@ -------------------------------------------------------------------------------- import qualified Blaze.ByteString.Builder as Builder-import Control.Exception (finally)-import qualified Data.ByteString as B+import Control.Exception (finally, throw)+import Data.IORef (newIORef) import qualified Data.Text as T import qualified Data.Text.Encoding as T import qualified Network.Socket as S-import qualified System.IO.Streams as Streams-import qualified System.IO.Streams.Attoparsec as Streams -------------------------------------------------------------------------------- import Network.WebSockets.Connection import Network.WebSockets.Http import Network.WebSockets.Protocol+import Network.WebSockets.Stream (Stream)+import qualified Network.WebSockets.Stream as Stream import Network.WebSockets.Types @@ -72,7 +72,7 @@ -------------------------------------------------------------------------------- runClientWithStream- :: (Streams.InputStream B.ByteString, Streams.OutputStream B.ByteString)+ :: Stream -- ^ Stream -> String -- ^ Host@@ -85,23 +85,28 @@ -> ClientApp a -- ^ Client application -> IO a-runClientWithStream (sIn, sOut) host path opts customHeaders app = do+runClientWithStream stream host path opts customHeaders app = do -- Create the request and send it- request <- createRequest protocol bHost bPath False customHeaders- bOut <- Streams.builderStream sOut- Streams.write (Just $ encodeRequestHead request) bOut- Streams.write (Just Builder.flush) bOut- response <- Streams.parseFromStream decodeResponseHead sIn+ request <- createRequest protocol bHost bPath False customHeaders+ Stream.write stream (Builder.toLazyByteString $ encodeRequestHead request)+ mbResponse <- Stream.parse stream decodeResponseHead+ response <- case mbResponse of+ Just response -> return response+ Nothing -> throw $ OtherHandshakeException $+ "Network.WebSockets.Client.runClientWithStream: no handshake " +++ "response from server" -- Note that we pattern match to evaluate the result here Response _ _ <- return $ finishResponse protocol request response- mIn <- decodeMessages protocol sIn- mOut <- encodeMessages protocol ClientConnection bOut+ parse <- decodeMessages protocol stream+ write <- encodeMessages protocol ClientConnection stream+ sentRef <- newIORef False app Connection- { connectionOptions = opts- , connectionType = ClientConnection- , connectionProtocol = protocol- , connectionIn = mIn- , connectionOut = mOut+ { connectionOptions = opts+ , connectionType = ClientConnection+ , connectionProtocol = protocol+ , connectionParse = parse+ , connectionWrite = write+ , connectionSentClose = sentRef } where protocol = defaultProtocol -- TODO@@ -118,5 +123,5 @@ -> ClientApp a -- ^ Client application -> IO a runClientWithSocket sock host path opts customHeaders app = do- stream <- Streams.socketToStreams sock+ stream <- Stream.makeSocketStream sock runClientWithStream stream host path opts customHeaders app
src/Network/WebSockets/Connection.hs view
@@ -2,7 +2,9 @@ {-# LANGUAGE OverloadedStrings #-} module Network.WebSockets.Connection ( PendingConnection (..)+ , AcceptRequest(..) , acceptRequest+ , acceptRequestWith , rejectRequest , Connection (..)@@ -18,23 +20,27 @@ , sendTextData , sendBinaryData , sendClose+ , sendCloseCode , sendPing ) where ---------------------------------------------------------------------------------import Blaze.ByteString.Builder (Builder) import qualified Blaze.ByteString.Builder as Builder import Control.Exception (throw)+import Control.Monad (unless) import qualified Data.ByteString as B+import Data.IORef (IORef, newIORef, readIORef,+ writeIORef) import Data.List (find)-import System.IO.Streams (InputStream, OutputStream)-import qualified System.IO.Streams as Streams+import Data.Word (Word16) -------------------------------------------------------------------------------- import Network.WebSockets.Http import Network.WebSockets.Protocol+import Network.WebSockets.Stream (Stream)+import qualified Network.WebSockets.Stream as Stream import Network.WebSockets.Types @@ -42,45 +48,59 @@ -- | A new client connected to the server. We haven't accepted the connection -- yet, though. data PendingConnection = PendingConnection- { pendingOptions :: ConnectionOptions+ { pendingOptions :: !ConnectionOptions -- ^ Options, passed as-is to the 'Connection'- , pendingRequest :: RequestHead+ , pendingRequest :: !RequestHead -- ^ Useful for e.g. inspecting the request path.- , pendingOnAccept :: Connection -> IO ()+ , pendingOnAccept :: !(Connection -> IO ()) -- ^ One-shot callback fired when a connection is accepted, i.e., *after* -- the accepting response is sent to the client.- , pendingIn :: InputStream B.ByteString- -- ^ Input stream- , pendingOut :: OutputStream Builder- -- ^ Output stream+ , pendingStream :: !Stream+ -- ^ Input/output stream } --------------------------------------------------------------------------------+data AcceptRequest = AcceptRequest+ { acceptSubprotocol :: !(Maybe B.ByteString)+ -- ^ The subprotocol to speak with the client. If 'pendingSubprotcols' is+ -- non-empty, 'acceptSubprotocol' must be one of the subprotocols from the+ -- list.+ }+++-------------------------------------------------------------------------------- -- | Utility sendResponse :: PendingConnection -> Response -> IO ()-sendResponse pc rsp = do- Streams.write (Just (encodeResponse rsp)) (pendingOut pc)- Streams.write (Just Builder.flush) (pendingOut pc)+sendResponse pc rsp = Stream.write (pendingStream pc)+ (Builder.toLazyByteString (encodeResponse rsp)) -------------------------------------------------------------------------------- acceptRequest :: PendingConnection -> IO Connection-acceptRequest pc = case find (flip compatible request) protocols of+acceptRequest pc = acceptRequestWith pc $ AcceptRequest Nothing+++--------------------------------------------------------------------------------+acceptRequestWith :: PendingConnection -> AcceptRequest -> IO Connection+acceptRequestWith pc ar = case find (flip compatible request) protocols of Nothing -> do sendResponse pc $ response400 versionHeader "" throw NotSupported Just protocol -> do- let response = finishRequest protocol request+ let subproto = maybe [] (\p -> [("Sec-WebSocket-Protocol", p)]) $ acceptSubprotocol ar+ response = finishRequest protocol request subproto sendResponse pc response- msgIn <- decodeMessages protocol (pendingIn pc)- msgOut <- encodeMessages protocol ServerConnection (pendingOut pc)+ parse <- decodeMessages protocol (pendingStream pc)+ write <- encodeMessages protocol ServerConnection (pendingStream pc)+ sentRef <- newIORef False let connection = Connection- { connectionOptions = pendingOptions pc- , connectionType = ServerConnection- , connectionProtocol = protocol- , connectionIn = msgIn- , connectionOut = msgOut+ { connectionOptions = pendingOptions pc+ , connectionType = ServerConnection+ , connectionProtocol = protocol+ , connectionParse = parse+ , connectionWrite = write+ , connectionSentClose = sentRef } pendingOnAccept pc connection@@ -98,17 +118,23 @@ -------------------------------------------------------------------------------- data Connection = Connection- { connectionOptions :: ConnectionOptions- , connectionType :: ConnectionType- , connectionProtocol :: Protocol- , connectionIn :: InputStream Message- , connectionOut :: OutputStream Message+ { connectionOptions :: !ConnectionOptions+ , connectionType :: !ConnectionType+ , connectionProtocol :: !Protocol+ , connectionParse :: !(IO (Maybe Message))+ , connectionWrite :: !(Message -> IO ())+ , connectionSentClose :: !(IORef Bool)+ -- ^ According to the RFC, both the client and the server MUST send+ -- a close control message to each other. Either party can initiate+ -- the first close message but then the other party must respond. Finally,+ -- the server is in charge of closing the TCP connection. This IORef tracks+ -- if we have sent a close message and are waiting for the peer to respond. } -------------------------------------------------------------------------------- data ConnectionOptions = ConnectionOptions- { connectionOnPong :: IO ()+ { connectionOnPong :: !(IO ()) } @@ -122,21 +148,33 @@ -------------------------------------------------------------------------------- receive :: Connection -> IO Message receive conn = do- mmsg <- Streams.read (connectionIn conn)- case mmsg of+ mbMsg <- connectionParse conn+ case mbMsg of Nothing -> throw ConnectionClosed Just msg -> return msg -------------------------------------------------------------------------------- -- | Receive an application message. Automatically respond to control messages.+--+-- When the peer sends a close control message, an exception of type 'CloseRequest'+-- is thrown. The peer can send a close control message either to initiate a+-- close or in response to a close message we have sent to the peer. In either+-- case the 'CloseRequest' exception will be thrown. The RFC specifies that+-- the server is responsible for closing the TCP connection, which should happen+-- after receiving the 'CloseRequest' exception from this function.+--+-- This will throw 'ConnectionClosed' if the TCP connection dies unexpectedly. receiveDataMessage :: Connection -> IO DataMessage receiveDataMessage conn = do msg <- receive conn case msg of DataMessage am -> return am ControlMessage cm -> case cm of- Close _ -> throw ConnectionClosed+ Close i closeMsg -> do+ hasSentClose <- readIORef $ connectionSentClose conn+ unless hasSentClose $ send conn msg+ throw $ CloseRequest i closeMsg Pong _ -> do connectionOnPong (connectionOptions conn) receiveDataMessage conn@@ -157,7 +195,11 @@ -------------------------------------------------------------------------------- send :: Connection -> Message -> IO ()-send conn msg = Streams.write (Just msg) (connectionOut conn)+send conn msg = do+ case msg of+ (ControlMessage (Close _ _)) -> writeIORef (connectionSentClose conn) True+ _ -> return ()+ connectionWrite conn msg --------------------------------------------------------------------------------@@ -179,9 +221,26 @@ ----------------------------------------------------------------------------------- | Send a friendly close message+-- | Send a friendly close message. Note that after sending this message,+-- you should still continue calling 'receiveDataMessage' to process any+-- in-flight messages. The peer will eventually respond with a close control+-- message of its own which will cause 'receiveDataMessage' to throw the+-- 'CloseRequest' exception. This exception is when you can finally consider+-- the connection closed. sendClose :: WebSocketsData a => Connection -> a -> IO ()-sendClose conn = send conn . ControlMessage . Close . toLazyByteString+sendClose conn = sendCloseCode conn 1000+++--------------------------------------------------------------------------------+-- | Send a friendly close message and close code. Similar to 'sendClose',+-- you should continue calling 'receiveDataMessage' until you receive a+-- 'CloseRequest' exception.+--+-- See <http://tools.ietf.org/html/rfc6455#section-7.4> for a list of close+-- codes.+sendCloseCode :: WebSocketsData a => Connection -> Word16 -> a -> IO ()+sendCloseCode conn code =+ send conn . ControlMessage . Close code . toLazyByteString --------------------------------------------------------------------------------
src/Network/WebSockets/Http.hs view
@@ -25,6 +25,7 @@ , getRequestHeader , getResponseHeader , getRequestSecWebSocketVersion+ , getRequestSubprotocols ) where @@ -34,8 +35,7 @@ import Control.Applicative (pure, (*>), (<$>), (<*), (<*>)) import Control.Exception (Exception, throw)-import Control.Monad.Error (Error (..))-import qualified Data.Attoparsec as A+import qualified Data.Attoparsec.ByteString as A import Data.ByteString (ByteString) import qualified Data.ByteString as B import Data.ByteString.Char8 ()@@ -106,11 +106,6 @@ ---------------------------------------------------------------------------------instance Error HandshakeException where- strMsg = OtherHandshakeException----------------------------------------------------------------------------------- instance Exception HandshakeException @@ -231,6 +226,17 @@ getRequestSecWebSocketVersion :: RequestHead -> Maybe B.ByteString getRequestSecWebSocketVersion p = lookup "Sec-WebSocket-Version" (requestHeaders p)+++--------------------------------------------------------------------------------+-- | List of subprotocols specified by the client, in order of preference.+-- If the client did not specify a list of subprotocols, this will be the+-- empty list.+getRequestSubprotocols :: RequestHead -> [B.ByteString]+getRequestSubprotocols rh = maybe [] parse mproto+ where+ mproto = lookup "Sec-WebSocket-Protocol" $ requestHeaders rh+ parse = filter (not . B.null) . BC.splitWith (\o -> o == ',' || o == ' ') --------------------------------------------------------------------------------
src/Network/WebSockets/Hybi13.hs view
@@ -19,10 +19,10 @@ import Control.Applicative (pure, (<$>)) import Control.Exception (throw) import Control.Monad (liftM)-import Data.Attoparsec (anyWord8)-import qualified Data.Attoparsec as A+import qualified Data.Attoparsec.ByteString as A import Data.Binary.Get (getWord16be, getWord64be, runGet)+import Data.Binary.Put (putWord16be, runPut) import Data.Bits ((.&.), (.|.)) import Data.ByteString (ByteString) import qualified Data.ByteString.Base64 as B64@@ -35,8 +35,6 @@ mempty) import Data.Tuple (swap) import System.Entropy as R-import qualified System.IO.Streams as Streams-import qualified System.IO.Streams.Attoparsec as Streams import System.Random (RandomGen, newStdGen) @@ -44,6 +42,8 @@ import Network.WebSockets.Http import Network.WebSockets.Hybi13.Demultiplex import Network.WebSockets.Hybi13.Mask+import Network.WebSockets.Stream (Stream)+import qualified Network.WebSockets.Stream as Stream import Network.WebSockets.Types @@ -54,12 +54,13 @@ -------------------------------------------------------------------------------- finishRequest :: RequestHead+ -> Headers -> Response-finishRequest reqHttp =+finishRequest reqHttp headers = let !key = getRequestHeader reqHttp "Sec-WebSocket-Key" !hash = hashKey key !encoded = B64.encode hash- in response101 [("Sec-WebSocket-Accept", encoded)] ""+ in response101 (("Sec-WebSocket-Accept", encoded):headers) "" --------------------------------------------------------------------------------@@ -94,26 +95,24 @@ ServerConnection -> (Nothing, gen) ClientConnection -> randomMask gen builder = encodeFrame mask $ case msg of- (ControlMessage (Close pl)) -> mkFrame CloseFrame pl- (ControlMessage (Ping pl)) -> mkFrame PingFrame pl- (ControlMessage (Pong pl)) -> mkFrame PongFrame pl- (DataMessage (Text pl)) -> mkFrame TextFrame pl- (DataMessage (Binary pl)) -> mkFrame BinaryFrame pl+ (ControlMessage (Close code pl)) -> mkFrame CloseFrame $+ runPut (putWord16be code) `mappend` pl+ (ControlMessage (Ping pl)) -> mkFrame PingFrame pl+ (ControlMessage (Pong pl)) -> mkFrame PongFrame pl+ (DataMessage (Text pl)) -> mkFrame TextFrame pl+ (DataMessage (Binary pl)) -> mkFrame BinaryFrame pl ---------------------------------------------------------------------------------encodeMessages :: ConnectionType- -> Streams.OutputStream B.Builder- -> IO (Streams.OutputStream Message)-encodeMessages conType bStream = do+encodeMessages+ :: ConnectionType+ -> Stream+ -> IO (Message -> IO ())+encodeMessages conType stream = do genRef <- newIORef =<< newStdGen- Streams.lockingOutputStream =<< Streams.makeOutputStream (next genRef)- where- next :: RandomGen g => IORef g -> Maybe Message -> IO ()- next _ Nothing = return ()- next genRef (Just msg) = do- build <- atomicModifyIORef genRef $ \s -> encodeMessage conType s msg- Streams.write (Just build) bStream+ return $ \msg -> do+ builder <- atomicModifyIORef genRef $ \s -> encodeMessage conType s msg+ Stream.write stream (B.toLazyByteString builder) --------------------------------------------------------------------------------@@ -148,39 +147,46 @@ ---------------------------------------------------------------------------------decodeMessages :: Streams.InputStream ByteString- -> IO (Streams.InputStream Message)-decodeMessages bsStream = do+decodeMessages+ :: Stream+ -> IO (IO (Maybe Message))+decodeMessages stream = do dmRef <- newIORef emptyDemultiplexState- Streams.makeInputStream $ next dmRef+ return $ go dmRef where- next dmRef = do- frame <- Streams.parseFromStream parseFrame bsStream- m <- atomicModifyIORef dmRef $ \s -> swap $ demultiplex s frame- maybe (next dmRef) (return . Just) m+ go dmRef = do+ mbFrame <- Stream.parse stream parseFrame+ case mbFrame of+ Nothing -> return Nothing+ Just frame -> do+ mbMsg <- atomicModifyIORef dmRef $+ \s -> swap $ demultiplex s frame+ case mbMsg of+ Nothing -> go dmRef+ Just msg -> return (Just msg) -------------------------------------------------------------------------------- -- | Parse a frame parseFrame :: A.Parser Frame parseFrame = do- byte0 <- anyWord8+ byte0 <- A.anyWord8 let fin = byte0 .&. 0x80 == 0x80 rsv1 = byte0 .&. 0x40 == 0x40 rsv2 = byte0 .&. 0x20 == 0x20 rsv3 = byte0 .&. 0x10 == 0x10 opcode = byte0 .&. 0x0f - let ft = case opcode of- 0x00 -> ContinuationFrame- 0x01 -> TextFrame- 0x02 -> BinaryFrame- 0x08 -> CloseFrame- 0x09 -> PingFrame- 0x0a -> PongFrame- _ -> error "Unknown opcode"+ ft <- case opcode of+ 0x00 -> return ContinuationFrame+ 0x01 -> return TextFrame+ 0x02 -> return BinaryFrame+ 0x08 -> return CloseFrame+ 0x09 -> return PingFrame+ 0x0a -> return PongFrame+ _ -> fail $ "Unknown opcode: " ++ show opcode - byte1 <- anyWord8+ byte1 <- A.anyWord8 let mask = byte1 .&. 0x80 == 0x80 lenflag = fromIntegral (byte1 .&. 0x7f)
src/Network/WebSockets/Hybi13/Demultiplex.hs view
@@ -1,6 +1,6 @@ -------------------------------------------------------------------------------- -- | Demultiplexing of frames into messages-{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveDataTypeable, OverloadedStrings #-} module Network.WebSockets.Hybi13.Demultiplex ( FrameType (..) , Frame (..)@@ -14,6 +14,7 @@ import Blaze.ByteString.Builder (Builder) import qualified Blaze.ByteString.Builder as B import Control.Exception (Exception, throw)+import Data.Binary.Get (runGet, getWord16be) import qualified Data.ByteString.Lazy as BL import Data.Monoid (mappend) import Data.Typeable (Typeable)@@ -59,14 +60,14 @@ -------------------------------------------------------------------------------- -- | Internal state used by the demultiplexer-newtype DemultiplexState = DemultiplexState- { unDemultiplexState :: Maybe (FrameType, Builder)- }+data DemultiplexState+ = EmptyDemultiplexState+ | DemultiplexState !FrameType !Builder -------------------------------------------------------------------------------- emptyDemultiplexState :: DemultiplexState-emptyDemultiplexState = DemultiplexState Nothing+emptyDemultiplexState = EmptyDemultiplexState --------------------------------------------------------------------------------@@ -75,18 +76,18 @@ -> (Maybe Message, DemultiplexState) demultiplex state (Frame fin _ _ _ tp pl) = case tp of -- Return control messages immediately, they have no influence on the state- CloseFrame -> (Just (ControlMessage (Close pl)), state)+ CloseFrame -> (Just (ControlMessage (uncurry Close parsedClose)), state) PingFrame -> (Just (ControlMessage (Ping pl)), state) PongFrame -> (Just (ControlMessage (Pong pl)), state) -- If we're dealing with a continuation...- ContinuationFrame -> case unDemultiplexState state of+ ContinuationFrame -> case state of -- We received a continuation but we don't have any state. Let's ignore -- this fragment...- Nothing -> (Nothing, DemultiplexState Nothing)+ EmptyDemultiplexState -> (Nothing, EmptyDemultiplexState) -- Append the payload to the state -- TODO: protect against overflows- Just (amt, b)- | not fin -> (Nothing, DemultiplexState (Just (amt, b')))+ DemultiplexState amt b+ | not fin -> (Nothing, DemultiplexState amt b') | otherwise -> case amt of TextFrame -> (Just (DataMessage (Text m)), e) BinaryFrame -> (Just (DataMessage (Binary m)), e)@@ -96,10 +97,21 @@ m = B.toLazyByteString b' TextFrame | fin -> (Just (DataMessage (Text pl)), e)- | otherwise -> (Nothing, DemultiplexState (Just (TextFrame, plb)))+ | otherwise -> (Nothing, DemultiplexState TextFrame plb) BinaryFrame | fin -> (Just (DataMessage (Binary pl)), e)- | otherwise -> (Nothing, DemultiplexState (Just (BinaryFrame, plb)))+ | otherwise -> (Nothing, DemultiplexState BinaryFrame plb) where- e = emptyDemultiplexState+ e = emptyDemultiplexState plb = B.fromLazyByteString pl++ -- The Close frame MAY contain a body (the "Application data" portion of the+ -- frame) that indicates a reason for closing, such as an endpoint shutting+ -- down, an endpoint having received a frame too large, or an endpoint+ -- having received a frame that does not conform to the format expected by+ -- the endpoint. If there is a body, the first two bytes of the body MUST+ -- be a 2-byte unsigned integer (in network byte order) representing a+ -- status code with value /code/ defined in Section 7.4.+ parsedClose+ | BL.length pl >= 2 = (runGet getWord16be pl, BL.drop 2 pl)+ | otherwise = (1000, "")
src/Network/WebSockets/Protocol.hs view
@@ -16,15 +16,14 @@ ---------------------------------------------------------------------------------import Blaze.ByteString.Builder (Builder) import Data.ByteString (ByteString) import qualified Data.ByteString as B-import qualified System.IO.Streams as Streams -------------------------------------------------------------------------------- import Network.WebSockets.Http import qualified Network.WebSockets.Hybi13 as Hybi13+import Network.WebSockets.Stream (Stream) import Network.WebSockets.Types @@ -57,7 +56,7 @@ ---------------------------------------------------------------------------------finishRequest :: Protocol -> RequestHead -> Response+finishRequest :: Protocol -> RequestHead -> Headers -> Response finishRequest Hybi13 = Hybi13.finishRequest @@ -67,19 +66,21 @@ ---------------------------------------------------------------------------------encodeMessages :: Protocol -> ConnectionType- -> Streams.OutputStream Builder- -> IO (Streams.OutputStream Message)+encodeMessages+ :: Protocol -> ConnectionType -> Stream+ -> IO (Message -> IO ()) encodeMessages Hybi13 = Hybi13.encodeMessages ---------------------------------------------------------------------------------decodeMessages :: Protocol -> Streams.InputStream B.ByteString- -> IO (Streams.InputStream Message)+decodeMessages+ :: Protocol -> Stream+ -> IO (IO (Maybe Message)) decodeMessages Hybi13 = Hybi13.decodeMessages ---------------------------------------------------------------------------------createRequest :: Protocol -> B.ByteString -> B.ByteString -> Bool -> Headers- -> IO RequestHead+createRequest+ :: Protocol -> B.ByteString -> B.ByteString -> Bool -> Headers+ -> IO RequestHead createRequest Hybi13 = Hybi13.createRequest
src/Network/WebSockets/Server.hs view
@@ -7,23 +7,24 @@ ( ServerApp , runServer , runServerWith+ , makeListenSocket+ , makePendingConnection ) where -------------------------------------------------------------------------------- import Control.Concurrent (forkIO)-import Control.Exception (finally)+import Control.Exception (finally, throw) import Control.Monad (forever) import Network.Socket (Socket) import qualified Network.Socket as S-import qualified System.IO.Streams.Attoparsec as Streams-import qualified System.IO.Streams.Builder as Streams-import qualified System.IO.Streams.Network as Streams -------------------------------------------------------------------------------- import Network.WebSockets.Connection import Network.WebSockets.Http+import qualified Network.WebSockets.Stream as Stream+import Network.WebSockets.Types --------------------------------------------------------------------------------@@ -47,12 +48,8 @@ -- | A version of 'runServer' which allows you to customize some options. runServerWith :: String -> Int -> ConnectionOptions -> ServerApp -> IO () runServerWith host port opts app = S.withSocketsDo $ do- sock <- S.socket S.AF_INET S.Stream S.defaultProtocol- _ <- S.setSocketOption sock S.ReuseAddr 1- host' <- S.inet_addr host- S.bindSocket sock (S.SockAddrInet (fromIntegral port) host')- S.listen sock 5- _ <- forever $ do+ sock <- makeListenSocket host port+ _ <- forever $ do -- TODO: top level handle (conn, _) <- S.accept sock _ <- forkIO $ finally (runApp conn opts app) (S.sClose conn)@@ -61,21 +58,42 @@ --------------------------------------------------------------------------------+-- | Create a standardized socket on which you can listen for incomming+-- connections. Should only be used for a quick and dirty solution! Should be+-- preceded by the call 'Network.Socket.withSocketsDo'.+makeListenSocket :: String -> Int -> IO Socket+makeListenSocket host port = do+ sock <- S.socket S.AF_INET S.Stream S.defaultProtocol+ _ <- S.setSocketOption sock S.ReuseAddr 1+ host' <- S.inet_addr host+ S.bindSocket sock (S.SockAddrInet (fromIntegral port) host')+ S.listen sock 5+ return sock+++-------------------------------------------------------------------------------- runApp :: Socket -> ConnectionOptions -> ServerApp -> IO () runApp socket opts app = do- (sIn, sOut) <- Streams.socketToStreams socket- bOut <- Streams.builderStream sOut- -- TODO: we probably want to send a 40x if the request is bad?- request <- Streams.parseFromStream (decodeRequestHead False) sIn- let pc = PendingConnection- { pendingOptions = opts- , pendingRequest = request- , pendingOnAccept = \_ -> return ()- , pendingIn = sIn- , pendingOut = bOut- }+ pending <- makePendingConnection socket opts+ app pending - app pc++--------------------------------------------------------------------------------+-- | Turns a socket, connected to some client, into a 'PendingConnection'.+makePendingConnection+ :: Socket -> ConnectionOptions -> IO PendingConnection+makePendingConnection sock opts = do+ stream <- Stream.makeSocketStream sock+ -- TODO: we probably want to send a 40x if the request is bad?+ mbRequest <- Stream.parse stream (decodeRequestHead False)+ case mbRequest of+ Nothing -> throw ConnectionClosed+ Just request -> return PendingConnection+ { pendingOptions = opts+ , pendingRequest = request+ , pendingOnAccept = \_ -> return ()+ , pendingStream = stream+ }
+ src/Network/WebSockets/Stream.hs view
@@ -0,0 +1,114 @@+--------------------------------------------------------------------------------+-- | Lightweight abstraction over an input/output stream.+module Network.WebSockets.Stream+ ( Stream+ , makeStream+ , makeSocketStream+ , makeEchoStream+ , parse+ , write+ , close+ ) where++import Control.Applicative ((<$>))+import qualified Control.Concurrent.Chan as Chan+import Control.Exception (throw)+import Control.Monad (forM_)+import qualified Data.Attoparsec.ByteString as Atto+import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as BL+import Data.IORef (IORef, newIORef, readIORef,+ writeIORef)+import qualified Network.Socket as S+import qualified Network.Socket.ByteString as SB (recv)+import qualified Network.Socket.ByteString.Lazy as SBL (sendAll)++import Network.WebSockets.Types+++--------------------------------------------------------------------------------+-- | State of the stream+data StreamState+ = Closed !B.ByteString -- Remainder+ | Open !B.ByteString -- Buffer+++--------------------------------------------------------------------------------+-- | Lightweight abstraction over an input/output stream.+data Stream = Stream+ { streamIn :: IO (Maybe B.ByteString)+ , streamOut :: (Maybe BL.ByteString -> IO ())+ , streamState :: !(IORef StreamState)+ }+++--------------------------------------------------------------------------------+makeStream+ :: IO (Maybe B.ByteString) -- ^ Reading+ -> (Maybe BL.ByteString -> IO ()) -- ^ Writing+ -> IO Stream -- ^ Resulting stream+makeStream i o = Stream i o <$> newIORef (Open B.empty)+++--------------------------------------------------------------------------------+makeSocketStream :: S.Socket -> IO Stream+makeSocketStream socket = makeStream receive send+ where+ receive = do+ bs <- SB.recv socket 1024+ return $ if B.null bs then Nothing else Just bs++ send Nothing = return ()+ send (Just bs) = SBL.sendAll socket bs+++--------------------------------------------------------------------------------+makeEchoStream :: IO Stream+makeEchoStream = do+ chan <- Chan.newChan+ makeStream (Chan.readChan chan) $ \mbBs -> case mbBs of+ Nothing -> Chan.writeChan chan Nothing+ Just bs -> forM_ (BL.toChunks bs) $ \c -> Chan.writeChan chan (Just c)+++--------------------------------------------------------------------------------+parse :: Stream -> Atto.Parser a -> IO (Maybe a)+parse stream parser = do+ state <- readIORef (streamState stream)+ case state of+ Closed remainder+ | B.null remainder -> return Nothing+ | otherwise -> go (Atto.parse parser remainder) True+ Open buffer+ | B.null buffer -> do+ mbBs <- streamIn stream+ case mbBs of+ Nothing -> do+ writeIORef (streamState stream) (Closed B.empty)+ return Nothing+ Just bs -> go (Atto.parse parser bs) False+ | otherwise -> go (Atto.parse parser buffer) False+ where+ -- Buffer is empty when entering this function.+ go (Atto.Done remainder x) closed = do+ writeIORef (streamState stream) $+ if closed then Closed remainder else Open remainder+ return (Just x)+ go (Atto.Partial f) closed+ | closed = go (f B.empty) True+ | otherwise = do+ mbBs <- streamIn stream+ case mbBs of+ Nothing -> go (f B.empty) True+ Just bs -> go (f bs) False+ go (Atto.Fail _ _ err) _ = throw (ParseException err)+++--------------------------------------------------------------------------------+write :: Stream -> BL.ByteString -> IO ()+write stream = streamOut stream . Just+++--------------------------------------------------------------------------------+close :: Stream -> IO ()+close stream = streamOut stream Nothing
src/Network/WebSockets/Types.hs view
@@ -22,6 +22,7 @@ import qualified Data.Text.Lazy as TL import qualified Data.Text.Lazy.Encoding as TL import Data.Typeable (Typeable)+import Data.Word (Word16) --------------------------------------------------------------------------------@@ -39,7 +40,7 @@ -------------------------------------------------------------------------------- -- | Different control messages data ControlMessage- = Close BL.ByteString+ = Close Word16 BL.ByteString | Ping BL.ByteString | Pong BL.ByteString deriving (Eq, Show)@@ -99,14 +100,25 @@ ----------------------------------------------------------------------------------- | The connection couldn't be established or broke down unexpectedly. thrown--- as an iteratee exception.+-- | Various exceptions that can occur while receiving or transmitting messages data ConnectionException- -- | the client unexpectedly closed the connection while we were trying to- -- receive some data.+ -- | The peer has requested that the connection be closed, and included+ -- a close code and a reason for closing. When receiving this exception,+ -- no more messages can be sent. Also, the server is responsible for+ -- closing the TCP connection once this exception is received. --- -- todo: Also want this for sending.- = ConnectionClosed+ -- See <http://tools.ietf.org/html/rfc6455#section-7.4> for a list of close+ -- codes.+ = CloseRequest Word16 BL.ByteString++ -- | The peer unexpectedly closed the connection while we were trying to+ -- receive some data. This is a violation of the websocket RFC since the+ -- TCP connection should only be closed after sending and receiving close+ -- control messages.+ | ConnectionClosed++ -- | The client sent garbage, i.e. we could not parse the WebSockets stream.+ | ParseException String deriving (Show, Typeable)
tests/haskell/Network/WebSockets/Handshake/Tests.hs view
@@ -12,38 +12,40 @@ import Data.IORef (newIORef, readIORef, writeIORef) import Data.Maybe (fromJust)-import qualified System.IO.Streams.Attoparsec as Streams-import qualified System.IO.Streams.Builder as Streams import Test.Framework (Test, testGroup) import Test.Framework.Providers.HUnit (testCase)-import Test.HUnit (Assertion, (@?=), assert)+import Test.HUnit (Assertion, assert, (@?=)) -------------------------------------------------------------------------------- import Network.WebSockets import Network.WebSockets.Connection import Network.WebSockets.Http+import qualified Network.WebSockets.Stream as Stream import Network.WebSockets.Tests.Util -------------------------------------------------------------------------------- tests :: Test tests = testGroup "Network.WebSockets.Handshake.Test"- [ testCase "handshake Hybi13" testHandshakeHybi13- , testCase "handshake reject" testHandshakeReject- , testCase "handshake Hybi9000" testHandshakeHybi9000+ [ testCase "handshake Hybi13" testHandshakeHybi13+ , testCase "handshake Hybi13 with subprotocols" testHandshakeHybi13WithProto+ , testCase "handshake reject" testHandshakeReject+ , testCase "handshake Hybi9000" testHandshakeHybi9000 ] -------------------------------------------------------------------------------- testHandshake :: RequestHead -> (PendingConnection -> IO a) -> IO ResponseHead testHandshake rq app = do- (is, os) <- makeChanPipe- os' <- Streams.builderStream os- _ <- forkIO $ do- _ <- app (PendingConnection defaultConnectionOptions rq nullify is os')+ echo <- Stream.makeEchoStream+ _ <- forkIO $ do+ _ <- app (PendingConnection defaultConnectionOptions rq nullify echo) return ()- Streams.parseFromStream decodeResponseHead is+ mbRh <- Stream.parse echo decodeResponseHead+ case mbRh of+ Nothing -> fail "testHandshake: No response"+ Just rh -> return rh where nullify _ = return () @@ -60,7 +62,7 @@ , ("Upgrade", "websocket") , ("Connection", "Upgrade") , ("Sec-WebSocket-Key", "x3JJHMbDL1EzLkh9GBhXDw==")- , ("Sec-WebSocket-Protocol", "chat")+ , ("Sec-WebSocket-Protocol", "chat, superchat") , ("Sec-WebSocket-Version", "13") , ("Origin", "http://example.com") ]@@ -79,7 +81,23 @@ message @?= "WebSocket Protocol Handshake" headers ! "Sec-WebSocket-Accept" @?= "HSmrc0sMlYUkAGmm5OPpG2HaGWk=" headers ! "Connection" @?= "Upgrade"+ lookup "Sec-WebSocket-Protocol" headers @?= Nothing +--------------------------------------------------------------------------------+testHandshakeHybi13WithProto :: Assertion+testHandshakeHybi13WithProto = do+ onAcceptFired <- newIORef False+ ResponseHead code message headers <- testHandshake rq13 $ \pc -> do+ getRequestSubprotocols (pendingRequest pc) @?= ["chat", "superchat"]+ acceptRequestWith pc {pendingOnAccept = \_ -> writeIORef onAcceptFired True}+ (AcceptRequest $ Just "superchat")++ readIORef onAcceptFired >>= assert+ code @?= 101+ message @?= "WebSocket Protocol Handshake"+ headers ! "Sec-WebSocket-Accept" @?= "HSmrc0sMlYUkAGmm5OPpG2HaGWk="+ headers ! "Connection" @?= "Upgrade"+ headers ! "Sec-WebSocket-Protocol" @?= "superchat" -------------------------------------------------------------------------------- testHandshakeReject :: Assertion
tests/haskell/Network/WebSockets/Http/Tests.hs view
@@ -6,7 +6,7 @@ ---------------------------------------------------------------------------------import qualified Data.Attoparsec as A+import qualified Data.Attoparsec.ByteString as A import qualified Data.ByteString.Char8 as BC import Test.Framework (Test, testGroup) import Test.Framework.Providers.HUnit (testCase)
tests/haskell/Network/WebSockets/Server/Tests.hs view
@@ -10,25 +10,25 @@ import Control.Applicative ((<$>)) import Control.Concurrent (forkIO, killThread, threadDelay)-import Control.Exception (SomeException, handle)-import Control.Monad (forM_, forever, replicateM)-import Data.IORef (newIORef, readIORef,+import Control.Exception (SomeException, handle, catch)+import Control.Monad (forM_, forever, replicateM, unless)+import Data.IORef (newIORef, readIORef, IORef, writeIORef) - -------------------------------------------------------------------------------- import qualified Data.ByteString.Lazy as BL import Data.Text (Text)-import System.Random (newStdGen) import Test.Framework (Test, testGroup) import Test.Framework.Providers.HUnit (testCase) import Test.HUnit (Assertion, assert, (@=?)) import Test.QuickCheck (Arbitrary, arbitrary) import Test.QuickCheck.Gen (Gen (..))+import Test.QuickCheck.Random (newQCGen) -------------------------------------------------------------------------------- import Network.WebSockets+import Network.WebSockets.Connection import Network.WebSockets.Tests.Util @@ -42,7 +42,7 @@ -------------------------------------------------------------------------------- testSimpleServerClient :: Assertion-testSimpleServerClient = withEchoServer 42940 $ do+testSimpleServerClient = withEchoServer 42940 "Bye" $ do texts <- map unArbitraryUtf8 <$> sample texts' <- retry $ runClient "127.0.0.1" 42940 "/chat" $ client texts texts @=? texts'@@ -52,12 +52,13 @@ forM_ texts (sendTextData conn) texts' <- replicateM (length texts) (receiveData conn) sendClose conn ("Bye" :: BL.ByteString)+ expectCloseException conn "Bye" return texts' -------------------------------------------------------------------------------- testOnPong :: Assertion-testOnPong = withEchoServer 42941 $ do+testOnPong = withEchoServer 42941 "Bye" $ do gotPong <- newIORef False let opts = defaultConnectionOptions { connectionOnPong = writeIORef gotPong True@@ -72,13 +73,15 @@ sendPing conn ("What's a fish without an eye?" :: Text) sendTextData conn ("A fsh!" :: Text) msg <- receiveData conn+ sendCloseCode conn 1000 ("Bye" :: BL.ByteString)+ expectCloseException conn "Bye" return $ "A fsh!" == (msg :: Text) -------------------------------------------------------------------------------- sample :: Arbitrary a => IO [a] sample = do- gen <- newStdGen+ gen <- newQCGen return $ (unGen arbitrary) gen 512 @@ -101,13 +104,16 @@ ---------------------------------------------------------------------------------withEchoServer :: Int -> IO a -> IO a-withEchoServer port action = do- serverThread <- forkIO $ retry $ runServer "0.0.0.0" port server+withEchoServer :: Int -> BL.ByteString -> IO a -> IO a+withEchoServer port expectedClose action = do+ cRef <- newIORef False+ serverThread <- forkIO $ retry $ runServer "0.0.0.0" port (\c -> server c `catch` handleClose cRef) waitSome result <- action waitSome killThread serverThread+ closeCalled <- readIORef cRef+ unless closeCalled $ error "Expecting the CloseRequest exception" return result where server :: ServerApp@@ -116,3 +122,25 @@ forever $ do msg <- receiveDataMessage conn sendDataMessage conn msg++ handleClose :: IORef Bool -> ConnectionException -> IO ()+ handleClose cRef (CloseRequest i msg) = do+ i @=? 1000+ msg @=? expectedClose+ writeIORef cRef True+ handleClose _ ConnectionClosed =+ error "Unexpected connection closed exception"+ handleClose _ (ParseException _) =+ error "Unexpected parse exception"+++--------------------------------------------------------------------------------+expectCloseException :: Connection -> BL.ByteString -> IO ()+expectCloseException conn msg = act `catch` handler+ where+ act = receiveDataMessage conn >> error "Expecting CloseRequest exception"+ handler (CloseRequest i msg') = do+ i @=? 1000+ msg' @=? msg+ handler ConnectionClosed = error "Unexpected connection closed"+ handler (ParseException _) = error "Unexpected parse exception"
tests/haskell/Network/WebSockets/Tests.hs view
@@ -9,11 +9,10 @@ -------------------------------------------------------------------------------- import qualified Blaze.ByteString.Builder as Builder import Control.Applicative ((<$>))-import Control.Monad (replicateM)+import Control.Monad (replicateM, forM_) import qualified Data.ByteString.Lazy as BL import Data.List (intersperse) import Data.Maybe (catMaybes)-import qualified System.IO.Streams as Streams import Test.Framework (Test, testGroup) import Test.Framework.Providers.QuickCheck2 (testProperty) import Test.HUnit ((@=?))@@ -28,6 +27,7 @@ import qualified Network.WebSockets.Hybi13 as Hybi13 import Network.WebSockets.Hybi13.Demultiplex import Network.WebSockets.Protocol+import qualified Network.WebSockets.Stream as Stream import Network.WebSockets.Tests.Util import Network.WebSockets.Types @@ -44,12 +44,11 @@ testSimpleEncodeDecode :: Protocol -> Property testSimpleEncodeDecode protocol = QC.monadicIO $ QC.forAllM QC.arbitrary $ \msgs -> QC.run $ do- (is, os) <- makeChanPipe- is' <- decodeMessages protocol is- os' <- encodeMessages protocol ClientConnection =<<- Streams.builderStream os- Streams.writeList msgs os'- msgs' <- catMaybes <$> replicateM (length msgs) (Streams.read is')+ echo <- Stream.makeEchoStream+ parse <- decodeMessages protocol echo+ write <- encodeMessages protocol ClientConnection echo+ forM_ msgs write+ msgs' <- catMaybes <$> replicateM (length msgs) parse msgs @=? msgs' @@ -57,27 +56,32 @@ testFragmentedHybi13 :: Property testFragmentedHybi13 = QC.monadicIO $ QC.forAllM QC.arbitrary $ \fragmented -> QC.run $ do- (is, os) <- makeChanPipe- is' <- Streams.filter isDataMessage =<< Hybi13.decodeMessages is- os' <- Streams.builderStream os+ echo <- Stream.makeEchoStream+ parse <- Hybi13.decodeMessages echo+ -- is' <- Streams.filter isDataMessage =<< Hybi13.decodeMessages is -- Simple hacky encoding of all frames- Streams.writeList- [ Hybi13.encodeFrame Nothing f+ mapM_ (Stream.write echo)+ [ Builder.toLazyByteString (Hybi13.encodeFrame Nothing f) | FragmentedMessage _ frames <- fragmented , f <- frames- ] os'- Streams.write (Just Builder.flush) os'- Streams.write Nothing os'+ ]+ Stream.close echo -- Check if we got all data- msgs <- catMaybes <$> replicateM (length fragmented) (Streams.read is')+ msgs <- filter isDataMessage <$> parseAll parse [msg | FragmentedMessage msg _ <- fragmented] @=? msgs where isDataMessage (ControlMessage _) = False isDataMessage (DataMessage _) = True + parseAll parse = do+ mbMsg <- parse+ case mbMsg of+ Just msg -> (msg :) <$> parseAll parse+ Nothing -> return [] + -------------------------------------------------------------------------------- instance Arbitrary FrameType where arbitrary = QC.elements@@ -108,8 +112,9 @@ instance Arbitrary Message where arbitrary = do payload <- BL.pack <$> arbitrary+ closecode <- arbitrary QC.elements- [ ControlMessage (Close payload)+ [ ControlMessage (Close closecode payload) , ControlMessage (Ping payload) , ControlMessage (Pong payload) , DataMessage (Text payload)
tests/haskell/Network/WebSockets/Tests/Util.hs view
@@ -3,19 +3,15 @@ ( ArbitraryUtf8 (..) , arbitraryUtf8 , arbitraryByteString- , makeChanPipe ) where ---------------------------------------------------------------------------------import Control.Applicative ((<$>), (<*>))-import Control.Concurrent.Chan (newChan)-import qualified Data.ByteString.Lazy as BL-import qualified Data.Text.Lazy as TL-import qualified Data.Text.Lazy.Encoding as TL-import System.IO.Streams (InputStream, OutputStream)-import qualified System.IO.Streams.Concurrent as Streams-import Test.QuickCheck (Arbitrary (..), Gen)+import Control.Applicative ((<$>))+import qualified Data.ByteString.Lazy as BL+import qualified Data.Text.Lazy as TL+import qualified Data.Text.Lazy.Encoding as TL+import Test.QuickCheck (Arbitrary (..), Gen) --------------------------------------------------------------------------------@@ -40,12 +36,3 @@ -------------------------------------------------------------------------------- arbitraryByteString :: Gen BL.ByteString arbitraryByteString = BL.pack <$> arbitrary-------------------------------------------------------------------------------------- | TODO: I added this function to the io-streams library but it isn't released--- yet, at some point we should be able to remove it here.-makeChanPipe :: IO (InputStream a, OutputStream a)-makeChanPipe = do- chan <- newChan- (,) <$> Streams.chanToInput chan <*> Streams.chanToOutput chan
websockets.cabal view
@@ -1,5 +1,5 @@ Name: websockets-Version: 0.8.2.6+Version: 0.9.0.0 Synopsis: A sensible and clean way to write WebSocket-capable servers in Haskell.@@ -50,6 +50,7 @@ Exposed-modules: Network.WebSockets Network.WebSockets.Connection+ Network.WebSockets.Stream -- Network.WebSockets.Util.PubSub TODO Other-modules:@@ -63,7 +64,7 @@ Network.WebSockets.Types Build-depends:- attoparsec >= 0.9 && < 0.13,+ attoparsec >= 0.10 && < 0.13, base >= 4 && < 5, base64-bytestring >= 0.1 && < 1.1, binary >= 0.5 && < 0.8,@@ -71,10 +72,9 @@ bytestring >= 0.9 && < 0.11, case-insensitive >= 0.3 && < 1.3, containers >= 0.3 && < 0.6,- io-streams >= 1.1 && < 1.2, mtl >= 2.0 && < 2.3,- network >= 2.3 && < 2.6,- random >= 1.0 && < 1.1,+ network >= 2.3 && < 2.7,+ random >= 1.0 && < 1.2, SHA >= 1.5 && < 1.7, text >= 0.10 && < 1.2, entropy >= 0.2.1 && < 0.4@@ -99,7 +99,7 @@ test-framework-hunit >= 0.2 && < 0.4, test-framework-quickcheck2 >= 0.2 && < 0.4, -- Copied from regular dependencies...- attoparsec >= 0.9 && < 0.13,+ attoparsec >= 0.10 && < 0.13, base >= 4 && < 5, base64-bytestring >= 0.1 && < 1.1, binary >= 0.5 && < 0.8,@@ -107,7 +107,6 @@ bytestring >= 0.9 && < 0.11, case-insensitive >= 0.3 && < 1.3, containers >= 0.3 && < 0.6,- io-streams >= 1.1 && < 1.2, mtl >= 2.0 && < 2.3, network >= 2.3 && < 2.6, random >= 1.0 && < 1.1,