majordomo 0.1.1 → 0.1.2
raw patch · 8 files changed
+128/−124 lines, 8 filesdep ~basedep ~majordomo
Dependency ranges changed: base, majordomo
Files
- echoworker.hs +0/−31
- lib/System/Network/ZMQ/MDP/Client.hs +5/−19
- lib/System/Network/ZMQ/MDP/Util.hs +32/−0
- lib/System/Network/ZMQ/MDP/Worker.hs +21/−35
- majordomo.cabal +9/−5
- mdp_client.hs +0/−34
- src/echoworker.hs +27/−0
- src/mdp_client.hs +34/−0
− echoworker.hs
@@ -1,31 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}-module Main where-import Data.ByteString hiding (map)--import qualified System.Network.ZMQ.MDP.Worker as W-import Data.Foldable-import Control.Concurrent.Thread.Group as TG-import System.Posix.Signals-import qualified Control.Concurrent as CC--threaded :: [IO ()] -> IO ()-threaded actions = do- tg <- TG.new- tids <- mapM (TG.forkIO tg) actions- _ <- installHandler sigINT (CatchOnce $ do- Prelude.putStrLn "worker caught an interrupt"- forM_ tids ((\x -> print x >> CC.killThread x) . fst)- ) Nothing- Prelude.putStrLn "waiting..."- TG.wait tg- Prelude.putStrLn "all dead"--main :: IO ()-main = threaded $ flip map [1..4] $ \tid ->- W.withWorker "tcp://127.0.0.1:5555" "echo"- (\x -> return ("hi there, " `append` x))-- - -- withContext 1 $ \c ->- -- threaded $ flip map [1..4] $ \tid ->- -- W.start W.defaultWorker { }
lib/System/Network/ZMQ/MDP/Client.hs view
@@ -9,12 +9,16 @@ withClientSocket ) where +-- libraries import Data.ByteString.Char8 import qualified System.ZMQ as Z import System.ZMQ hiding(send) import Control.Applicative import System.Timeout +-- friends+import System.Network.ZMQ.MDP.Util+ data Protocol = MDCP01 data Response = Response { protocol :: Protocol,@@ -46,27 +50,9 @@ case maybeprot of Nothing -> return $ Left ClientTimedOut Just "MDPC01" -> do- res <- Response MDCP01 <$> receive sock [] <*> receiveTillEmpty sock+ res <- Response MDCP01 <$> receive sock [] <*> receiveUntilEnd sock return $ Right res _ -> return $ Left ClientBadProtocol where sock = clientSocket mdpcs ------ Helper functions-----receiveTillEmpty :: Socket a -> IO [ByteString]-receiveTillEmpty sock = do- more <- Z.moreToReceive sock- if more- then (:) <$> receive sock [] <*> receiveTillEmpty sock- else return []------- Sends a multipart message----sendAll :: Socket a -> [ByteString] -> IO ()-sendAll _sock [] = return ()-sendAll sock (m:[]) = Z.send sock m []-sendAll sock (m:ms) = Z.send sock m [SndMore] >> sendAll sock ms
+ lib/System/Network/ZMQ/MDP/Util.hs view
@@ -0,0 +1,32 @@+{-# LANGUAGE OverloadedStrings #-}+module System.Network.ZMQ.MDP.Util where++import System.ZMQ as Z+import Data.ByteString.Char8 (ByteString)+import qualified Data.ByteString.Char8 as BS+import Control.Applicative++-- Receive until no more messages+receiveUntilEnd :: Socket a -> IO [ByteString]+receiveUntilEnd sock = do+ more <- Z.moreToReceive sock+ if more+ then (:) <$> Z.receive sock [] <*> receiveUntilEnd sock+ else return []++-- Receive until an empty frame+receiveUntilEmpty :: Socket a -> IO [ByteString]+receiveUntilEmpty sock = do+ frame <- Z.receive sock []+ if BS.null frame+ then return []+ else (frame:) <$> receiveUntilEmpty sock++--+-- Sends a multipart message+--+sendAll :: Socket a -> [ByteString] -> IO ()+sendAll sock = go+ where go [x] = Z.send sock x []+ go (x:xs) = Z.send sock x [SndMore] >> go xs+ go [] = error "empty send not allowed"
lib/System/Network/ZMQ/MDP/Worker.hs view
@@ -1,5 +1,7 @@ {-# LANGUAGE OverloadedStrings #-} module System.Network.ZMQ.MDP.Worker where++-- libraries import Data.ByteString.Char8 import Data.Int import Prelude hiding (putStr, putStrLn)@@ -16,6 +18,9 @@ import Control.Concurrent import System.Timeout +-- friends+import System.Network.ZMQ.MDP.Util+ data Protocol = WORKER_PROTOCOL instance Show Protocol where show WORKER_PROTOCOL = "MDPW01"@@ -43,29 +48,12 @@ parseCommand "\005" = Just DISCONNECT parseCommand _ = Nothing - type MDError = ByteString - -read_till_empty :: Socket a -> IO [ByteString]-read_till_empty sock = do- frame <- Z.receive sock []- if frame == ""- then return []- else (frame:) <$> read_till_empty sock---send_all :: Socket a -> [ByteString] -> IO ()-send_all sock = go- where go [l] = Z.send sock l []- go (x:xs) = Z.send sock x [SndMore] >> go xs- go [] = error "empty send not allowed"---send_to_broker :: Socket a -> ResponseCode -> [ByteString] ->+sendToBroker :: Socket a -> ResponseCode -> [ByteString] -> [ByteString] -> IO ()-send_to_broker sock cmd option message =- send_all sock $ ["",+sendToBroker sock cmd option message =+ sendAll sock $ ["", pack $ show WORKER_PROTOCOL, pack $ show cmd] ++ option ++ message @@ -80,10 +68,10 @@ -- FIXME might we ever want to send a multi-part body? ---send_response :: Socket a -> Response -> IO ()-send_response sock resp = send_to_broker sock REPLY (envelope resp) (body resp)+sendResponse :: Socket a -> Response -> IO ()+sendResponse sock resp = sendToBroker sock REPLY (envelope resp) (body resp) --- send_to_broker sock REPLY Nothing [reply])+-- sendToBroker sock REPLY Nothing [reply]) whileJust :: Monad m => (b -> m (Maybe b)) -> b -> m b whileJust action seed = action seed >>= maybe (return seed) (whileJust action)@@ -102,7 +90,7 @@ broker :: String, context :: System.ZMQ.Context, svc :: ByteString,- handler :: ByteString -> IO ByteString+ handler :: [ByteString] -> IO [ByteString] } epoch :: UTCTime@@ -112,9 +100,9 @@ lIVENESS = 3 -withWorker :: String -> ByteString -> (ByteString -> IO ByteString) -> IO ()+withWorker :: String -> ByteString -> ([ByteString] -> IO [ByteString]) -> IO () withWorker broker_ service_ io =- withContext 1 $ \c -> + withContext 1 $ \c -> start WorkerState { broker = broker_, context = c, svc = service_,@@ -124,15 +112,13 @@ heartbeat = 2, reconnect = 2 }- - withBroker :: (Socket XReq -> WorkerState a -> IO b) -> WorkerState t -> IO b withBroker go worker = withSocket (context worker) XReq $ \sock -> do loggedPut ( "connecting to broker " ++ broker worker) connect sock (broker worker)- send_to_broker sock READY [svc worker] []+ sendToBroker sock READY [svc worker] [] now <- getCurrentTime let time = addUTCTime (fromIntegral $ heartbeat worker) now loggedPut ("beat at:" ++ show time)@@ -181,7 +167,7 @@ -- loggedPut $ "beat at " ++ show (heartbeat_at worker) if {-# SCC "time_comparison" #-} time > heartbeat_at worker then do --loggedPut "sending heartbeat"- {-# SCC "postcheck_send" #-} send_to_broker sock WORKER_HEARTBEAT [] []+ {-# SCC "postcheck_send" #-} sendToBroker sock WORKER_HEARTBEAT [] [] --loggedPut "sent heartbeat!" {-# SCC "postcheck_return" #-} return $ Just $ updateWorkerTime worker time else return $ Just worker@@ -201,11 +187,11 @@ let new_worker = worker { liveness = lIVENESS } case command of Just REQUEST -> do- addresses <- read_till_empty sock- msg <- zrecv- reply_string <- (handler worker) msg- send_response sock Response { envelope = addresses,- body = [reply_string] }+ addresses <- receiveUntilEmpty sock+ msgs <- receiveUntilEnd sock+ replyString <- handler worker msgs+ sendResponse sock Response { envelope = addresses,+ body = replyString } return $ Just new_worker Just HEARTBEAT -> do -- loggedPut "handling a heartbeat"
majordomo.cabal view
@@ -1,5 +1,5 @@ Name: majordomo-Version: 0.1.1+Version: 0.1.2 Synopsis: Majordomo protocol for ZeroMQ Description: The Majordomo Protocol (MDP) defines a reliable service-oriented request-reply dialog between a@@ -13,37 +13,41 @@ License: BSD3 License-file: LICENSE-Extra-Source-Files: README.md Author: Mark Wotton, Sean Seefried Maintainer: mark@ninjablocks.com, sean@ninjablocks.com Category: Network Build-type: Simple Cabal-version: >= 1.8+Extra-Source-Files: README.md Source-Repository head type: git location: git://github.com/ninjablocks/majordomo.git Executable mdp_client Main-is: mdp_client.hs- Build-depends: majordomo,base, bytestring, cmdargs+ Build-depends: majordomo ==0.1.2, bytestring, cmdargs,base+ hs-source-dirs: src ghc-options: -O2 -threaded Executable echoworker Main-is: echoworker.hs ghc-options: -O2 -threaded ghc-prof-options: -auto-all- Build-depends: majordomo,base, bytestring, threads, unix+ hs-source-dirs: src+ Build-depends: majordomo ==0.1.2,bytestring, + threads, unix,base Library -- Modules exported by the library. Exposed-modules: System.Network.ZMQ.MDP.Worker, System.Network.ZMQ.MDP.Client+ Other-modules: System.Network.ZMQ.MDP.Util -- Packages needed in order to build this package. hs-source-dirs: lib ghc-prof-options: -auto-all ghc-options: -O2- Build-depends: base >= 2 && <= 4.5,+ Build-depends: base >= 2 && <= 4.5.0.0, zeromq-haskell >= 0.8.4, bytestring, monad-loops, time, old-locale
− mdp_client.hs
@@ -1,34 +0,0 @@-{-# LANGUAGE OverloadedStrings, DeriveDataTypeable #-}-module Main where--import Prelude hiding(getContents, putStr, putStrLn)-import Data.ByteString.Char8 as BS-import qualified System.Network.ZMQ.MDP.Client as C-import Data.List as L-import System.Console.CmdArgs.Implicit--data Client = Client {broker :: String,- service::String, - message_parts::[String]} deriving (Show, Data, Typeable)---client :: Client-client = Client{ broker = def &= argPos 0 ,- service = def &= argPos 1 ,- message_parts = def &= args- } &= summary "connect to a 0mq server"---main :: IO ()-main = do- s <- cmdArgs client- -- print s- if (L.null $ message_parts s)- then putStrLn "must send at least one argument"- else C.withClientSocket (broker s) $ \sock -> do- res <- C.sendAndReceive sock (pack $ service s) (Prelude.map pack $ message_parts s)- putStr $ case res of- Left C.ClientTimedOut -> "Timed out!"- Left C.ClientBadProtocol -> "Bad protocol!"- Right l -> BS.concat . L.intersperse "\n" $ C.response l-
+ src/echoworker.hs view
@@ -0,0 +1,27 @@+{-# LANGUAGE OverloadedStrings #-}+module Main where+import Data.ByteString hiding (map)++import qualified System.Network.ZMQ.MDP.Worker as W+import Data.Foldable+import Control.Concurrent.Thread.Group as TG+import System.Posix.Signals+import qualified Control.Concurrent as CC+import qualified Data.ByteString.Char8 as BS++threaded :: [IO ()] -> IO ()+threaded actions = do+ tg <- TG.new+ tids <- mapM (TG.forkIO tg) actions+ _ <- installHandler sigINT (CatchOnce $ do+ Prelude.putStrLn "worker caught an interrupt"+ forM_ tids ((\x -> print x >> CC.killThread x) . fst)+ ) Nothing+ Prelude.putStrLn "waiting..."+ TG.wait tg+ Prelude.putStrLn "all dead"++main :: IO ()+main = threaded $ flip map [1..4] $ \tid ->+ W.withWorker "tcp://127.0.0.1:5773" "echo"+ (\msgs -> return $ "hi there, ":msgs)
+ src/mdp_client.hs view
@@ -0,0 +1,34 @@+{-# LANGUAGE OverloadedStrings, DeriveDataTypeable #-}+module Main where++import Prelude hiding(getContents, putStr, putStrLn)+import Data.ByteString.Char8 as BS+import qualified System.Network.ZMQ.MDP.Client as C+import Data.List as L+import System.Console.CmdArgs.Implicit++data Client = Client {broker :: String,+ service::String, + message_parts::[String]} deriving (Show, Data, Typeable)+++client :: Client+client = Client{ broker = def &= argPos 0 ,+ service = def &= argPos 1 ,+ message_parts = def &= args+ } &= summary "connect to a 0mq server"+++main :: IO ()+main = do+ s <- cmdArgs client+ -- print s+ if (L.null $ message_parts s)+ then putStrLn "must send at least one argument"+ else C.withClientSocket (broker s) $ \sock -> do+ res <- C.sendAndReceive sock (pack $ service s) (Prelude.map pack $ message_parts s)+ putStr $ case res of+ Left C.ClientTimedOut -> "Timed out!"+ Left C.ClientBadProtocol -> "Bad protocol!"+ Right l -> BS.concat . L.intersperse "\n" $ C.response l+