Stomp 0.1 → 0.1.1
raw patch · 3 files changed
+125/−76 lines, 3 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
+ Network.Stomp: beat :: Connection -> IO ()
+ Network.Stomp: send' :: Connection -> [(Destination, [Header], ByteString)] -> IO ()
+ Network.Stomp: server :: Connection -> String
+ Network.Stomp: session :: Connection -> String
Files
- Stomp.cabal +7/−2
- src/Network/Stomp.hs +95/−74
- test/Tests.hs +23/−0
Stomp.cabal view
@@ -1,5 +1,5 @@ Name: Stomp-Version: 0.1+Version: 0.1.1 Description: Stomp is a client library for communicating with message servers through the STOMP protocol (http://stomp.github.com/stomp-specification-1.1.html) License: BSD3@@ -11,10 +11,15 @@ Build-Type: Simple Category: Network Synopsis: Client library for Stomp brokers.-Cabal-Version: >=1.2+Cabal-Version: >=1.6 Extra-Source-Files: README test/Tests.hs examples/Example.hs++Source-Repository head+ Type: git+ Location: http://github.com/rukav/Stomp+ Library Hs-Source-Dirs: src Exposed-Modules: Network.Stomp
src/Network/Stomp.hs view
@@ -60,6 +60,7 @@ stomp', disconnect, send,+ send', subscribe, unsubscribe, ack,@@ -73,11 +74,15 @@ setExcpHandler, startSendBeat, startRecvBeat,+ beat, sendTimeout, recvTimeout, lastSend, lastRecv,- versions++ versions,+ session,+ server ) where @@ -101,6 +106,9 @@ import Control.Monad import System.IO +---+import qualified Network.Socket.ByteString.Lazy as L+ -- | Stomp frame record data Frame = Frame { command :: Command, @@ -144,6 +152,8 @@ data Connection = Connection { handle :: Handle, versions :: [Version], -- ^ accepted stomp versions+ session :: String, -- ^ session identifier+ server :: String, -- ^ stomp server info listener :: MVar ThreadId, -- ^ frames consumer thread sendBeat :: MVar ThreadId, -- ^ send heartbeat thread recvBeat :: MVar ThreadId, -- ^ recieve heartbeat thread@@ -155,7 +165,7 @@ closed :: MVar (Maybe StompException), -- ^ closed connection flag disconReq :: MVar String, -- ^ disconnect receipt request disconResp :: MVar (), -- ^ disconnect receipt response - excpHandle :: MVar (Maybe (StompException -> IO ())) -- ^ exception handler in the frames consumer+ excpHandle :: MVar (Maybe (StompException -> IO ())) -- ^ excp handler invoked by the frames consumer } data StompException @@ -198,7 +208,7 @@ vers = intercalate "," $ map go vs go (v,v') = show v ++ '.':show v' --- | connect to the stomp 1.0 broker using hostname and port+-- | connect to the stomp 1.1 broker using hostname and port stomp' :: Host -> PortNumber -> [Header] -> IO Connection stomp' = mkConnection STOMP @@ -217,11 +227,19 @@ modifyMVar_ (closed con) $ \x -> return (Just $ fromMaybe (ConnectionError "Connection closed") x) -{- send message to the destination. -The size of the message should be equal to the value of 'content-length' header. --}+-- | send message to the destination.+-- | The header 'content-length' is automatically set by this module. send :: Connection -> Destination -> [Header] -> BL.ByteString -> IO ()-send con dest hs body = sendFrame con (Frame (CC SEND) hs' body)+send con dest hs body = sendFrame con $ mkSendFrame (dest, hs, body)++-- | send group of messages to the destination.+-- | The header 'content-length' is automatically set by this module. +send' :: Connection -> [(Destination, [Header], BL.ByteString)] -> IO ()+send' con xs = sendFrames con $ map mkSendFrame xs++-- | create SEND frame+mkSendFrame :: (Destination, [Header], BL.ByteString) -> Frame+mkSendFrame (dest, hs, body) = Frame (CC SEND) hs' body where hs' = hs ++ [("destination", dest)] ++ clh clh = maybe hdr (\_->[]) $ lookup "content-length" hs hdr = [("content-length", show $ BL.length body)]@@ -272,9 +290,13 @@ when (cmd /= CONNECTED) $ E.throwIO $ ConnectionError (BL.unpack body) let (sendBeat, recvBeat) = getBeats hs hs'- return con {recvTimeout = recvBeat, sendTimeout = sendBeat, versions = ver hs'} - where ver h = maybe [(1,0)] parseVer $ lookup "version" h-+ return con { recvTimeout = recvBeat + , sendTimeout = sendBeat + , versions = maybe [(1,0)] parseVer $ lookup "version" hs'+ , session = fromMaybe [] $ lookup "session" hs' + , server = fromMaybe [] $ lookup "server" hs'+ } + -- | parse versions supported by the broker parseVer :: String -> [Version] parseVer vs @@ -309,7 +331,8 @@ dRcpt <- newEmptyMVar dLock <- newEmptyMVar eHndl <- newMVar Nothing - return $ Connection h [] lstnr sBeat rBeat 0 0 lSend lRecv sLock close dRcpt dLock eHndl+ return $ Connection h [] [] [] lstnr sBeat rBeat 0 0 lSend + lRecv sLock close dRcpt dLock eHndl -- | create the authority value stompAuth :: String -> Maybe URIAuth@@ -370,8 +393,7 @@ tid <- forkIO $ E.finally (consumeFrames c fun) (tryPutMVar (disconResp c) ())- tryPutMVar (listener c) tid- return ()+ void $ tryPutMVar (listener c) tid -- | For any incoming frame the user callback will be invoked consumeFrames :: Connection -> (Frame -> IO ()) -> IO ()@@ -394,20 +416,24 @@ -- | send frame buffer to the server sendFrame :: Connection -> Frame -> IO ()-sendFrame con f = sendBuf con (strict $ runPut $ putFrame f)+sendFrame con f = sendBuf con (putFrames [f]) ++-- | send batch of frames to the server+sendFrames :: Connection -> [Frame] -> IO ()+sendFrames con fs = sendBuf con (putFrames fs)++-- | serialize frames+putFrames :: [Frame] -> BL.ByteString+putFrames fs = runPut $ mapM_ put fs where - putFrame frame@(Frame (CC cmd) hs body) = do- mapM_ (putByteString . BU.fromString)- [show cmd, "\n", hdrToStr cmd hs, "\n"]- unless (BL.null body) $ - putLazyByteString body- putWord8 0x00- strict x = BS.concat (BL.toChunks x)+ put (Frame (CC cmd) hs body) = do+ mapM_ (putByteString . BU.fromString)+ [show cmd, "\n", hdrToStr cmd hs, "\n"]+ unless (BL.null body) $ putLazyByteString body+ putWord8 0x00 -{- unparse frame headers. -All frames except the CONNECT and CONNECTED frames will also escape any -colon or newline octets--}+-- | unparse frame headers. +-- | All frames except the CONNECT and CONNECTED frames will also escape any colon or newline octets hdrToStr :: ClientCommand -> [Header] -> String hdrToStr _ [] = [] hdrToStr CONNECT hs = hdrToStr' id hs@@ -417,53 +443,48 @@ where go (x,y) = f x ++ ":" ++ f y -- | send buffer to the handle-sendBuf :: Connection -> BS.ByteString -> IO ()+sendBuf :: Connection -> BL.ByteString -> IO () sendBuf con bs = withMVar (closed con) $ \c -> if isJust c then E.throwIO (fromJust c) else E.catch (withMVar (sockLock con) $ \_ -> do- BS.hPut (handle con) bs+ BS.hPut (handle con) (BS.concat $ BL.toChunks bs) hFlush (handle con) beatTime (lastSend con)) (\(e :: E.IOException) -> E.throwIO $ StompIOError e) --------- Receive frame --- | reads incoming frame from handle+-- | receives incoming frame receiveFrame :: Connection -> IO Frame receiveFrame con = do- cmd <- readCommand con- headers <- readHeaders con cmd- body <- readBody con headers- beatTime (lastRecv con)- return (Frame cmd headers body)---- | read line from handle-readLine :: Handle -> IO String-readLine h = fmap BU.toString (BS.hGetLine h)---- | read stomp command or broker heartbeat from handle-readCommand :: Connection -> IO Command-readCommand con = do eof <- hIsEOF (handle con)- if eof then - E.throwIO $ ConnectionError "Connection closed by broker"+ if eof then+ E.throwIO $ ConnectionError "Connection closed by broker" else do- l <- readLine (handle con)- let l' = dropWhile (=='\x00') l - if null l' then do -- broker heartbeat- beatTime (lastRecv con)- readCommand con- else return $ SC (read l' :: ServerCommand)+ frame <- getFrame (handle con)+ beatTime (lastRecv con)+ maybe (receiveFrame con) return frame --- | read stomp headers from handle -readHeaders :: Connection -> Command -> IO [Header]-readHeaders con cmd = do- l <- readLine (handle con)+-- | reads frame from the handle+getFrame :: Handle -> IO (Maybe Frame)+getFrame h = do+ line <- liftM (dropWhile (=='\x00')) (readLine h) + if null line then return Nothing+ else do+ let cmd = SC (read line :: ServerCommand)+ headers <- readHeaders h cmd+ body <- readBody h headers+ return (Just $ Frame cmd headers body)++-- | reads stomp headers+readHeaders :: Handle -> Command -> IO [Header]+readHeaders h cmd = do+ l <- readLine h if null l then return []- else do hs <- readHeaders con cmd+ else do hs <- readHeaders h cmd return (header (unesc l):hs) case cmd of (SC CONNECTED) -> return (header l:hs)@@ -472,11 +493,10 @@ header x = let (name, val) = break (==':') x in (name, tail val) --- read stomp message with length 'content-length' or until frame terminator-readBody :: Connection -> [Header] -> IO BL.ByteString-readBody con hs = maybe (readTill h) (readBuf h) len+-- reads stomp message with length 'content-length' or until frame terminator+readBody :: Handle -> [Header] -> IO BL.ByteString+readBody h hs = maybe (readTill h) (readBuf h) len where - h = handle con len = lookup "content-length" hs readBuf h x = do bs <- BL.hGet h (read x :: Int)@@ -489,10 +509,11 @@ ch <- BL.hGet h 1 if ch == term then return B.empty else liftM (B.append $ B.fromLazyByteString ch) (readTill' h)+ term = BL.singleton '\x00' --- | stomp frame terminator-term :: BL.ByteString-term = BL.pack "\x00"+-- | read line from handle+readLine :: Handle -> IO String+readLine h = fmap BU.toString (BS.hGetLine h) -------- Escaping @@ -514,21 +535,19 @@ -------- Heartbeat -{- get heartbeats values from the client and broker headers-The receiver tolerance is 100%.--}+-- | get heartbeats values from the client and broker headers. The receiver tolerance is 100%. getBeats :: [Header] -> [Header] -> (Int, Int) getBeats xs ys = (getBeat cs sr, 2 * getBeat cr ss) where getBeat x y | x /= 0 && y /= 0 = max x y | otherwise = 0 - (cs,cr) = beat xs- (ss,sr) = beat ys+ (cs,cr) = mkBeat xs+ (ss,sr) = mkBeat ys -- | parse heartbeat-beat :: [Header] -> (Int,Int)-beat = maybe (0,0) parseBeat . lookup "heart-beat" +mkBeat :: [Header] -> (Int,Int)+mkBeat = maybe (0,0) parseBeat . lookup "heart-beat" where parseBeat = parse . break (==',') parse (x,y) = (read x :: Int, read (tail y) :: Int) @@ -540,6 +559,10 @@ b <- tryPutMVar v now unless b $ modifyMVar_ v $ \_ -> return now +-- | send a single newline byte to the server+beat :: Connection -> IO ()+beat con = sendBuf con (BL.fromChunks [BU.fromString "\n\x00"])+ -- | periodically send heartbeat data to the server clientBeat :: Connection -> IO () clientBeat con = @@ -550,7 +573,7 @@ let diff = floor $ diffUTCTime now prev * 1000 let delay = sendTimeout con if diff >= delay then do- sendBuf con (BU.fromString "\n\x00")+ beat con threadDelay (1000 * delay) else threadDelay (1000 * (delay - diff)) clientBeat con)@@ -579,8 +602,7 @@ tid <- forkIO $ E.finally (clientBeat c) (tryPutMVar (disconResp c) ())- tryPutMVar (sendBeat c) tid- return ()+ void $ tryPutMVar (sendBeat c) tid -- | fork receive heartbeat thread startRecvBeat :: Connection -> IO ()@@ -590,6 +612,5 @@ tid <- forkIO $ E.finally (brokerBeat c) (tryPutMVar (disconResp c) ())- tryPutMVar (recvBeat c) tid- return ()- + void $ tryPutMVar (recvBeat c) tid+
test/Tests.hs view
@@ -20,6 +20,7 @@ , testCase "stomp" testStomp , testCase "disconnect" testDisconnect , testCase "send" testSend+ , testCase "send'" testBatchSend , testCase "subscribe" testSubscribe , testCase "unsubscribe" testUnsubscribe , testCase "ack" testAck@@ -31,6 +32,7 @@ [ testCase "receipt" testReceipt , testCase "sendbeat" testSendbeat , testCase "recvbeat" testRecvbeat+ , testCase "beat" testBeat , testCase "errors" testErrors , testCase "encode" testEncode ],@@ -82,6 +84,17 @@ assertEqual "Receipt for send command" receipt "1") (disconnect con []) +testBatchSend = do+ initDest dest subid+ (con, resp, _) <- mkStomp+ E.finally+ (do subscribe con dest subid []+ send' con $ replicate 50 (dest, [], testMsg)+ forM_ [1..50] $ \_ -> do+ r <- readChan' resp+ assertEqual "Invalid message received" (body (fromJust r)) testMsg)+ (disconnect con [])+ testSubscribe = do initDest dest subid (con, resp, _) <- mkStomp@@ -223,6 +236,16 @@ where isBrokerError (Just (BrokerError _)) = True isBrokerError _ = False++testBeat = do+ initDest dest subid+ (con, resp, excp) <- mkStomp+ E.finally+ (when (maximum (versions con) > (1,0)) $ do + beat con+ r <- readChan' resp+ assertBool "Unexpected response after beat" (isNothing r))+ (disconnect con []) testErrors = do E.catch