diff --git a/Stomp.cabal b/Stomp.cabal --- a/Stomp.cabal +++ b/Stomp.cabal @@ -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 diff --git a/src/Network/Stomp.hs b/src/Network/Stomp.hs --- a/src/Network/Stomp.hs +++ b/src/Network/Stomp.hs @@ -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 + diff --git a/test/Tests.hs b/test/Tests.hs --- a/test/Tests.hs +++ b/test/Tests.hs @@ -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