diff --git a/echoworker.hs b/echoworker.hs deleted file mode 100644 --- a/echoworker.hs +++ /dev/null @@ -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 { } diff --git a/lib/System/Network/ZMQ/MDP/Client.hs b/lib/System/Network/ZMQ/MDP/Client.hs --- a/lib/System/Network/ZMQ/MDP/Client.hs +++ b/lib/System/Network/ZMQ/MDP/Client.hs @@ -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 diff --git a/lib/System/Network/ZMQ/MDP/Util.hs b/lib/System/Network/ZMQ/MDP/Util.hs new file mode 100644 --- /dev/null +++ b/lib/System/Network/ZMQ/MDP/Util.hs @@ -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" diff --git a/lib/System/Network/ZMQ/MDP/Worker.hs b/lib/System/Network/ZMQ/MDP/Worker.hs --- a/lib/System/Network/ZMQ/MDP/Worker.hs +++ b/lib/System/Network/ZMQ/MDP/Worker.hs @@ -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" diff --git a/majordomo.cabal b/majordomo.cabal --- a/majordomo.cabal +++ b/majordomo.cabal @@ -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 diff --git a/mdp_client.hs b/mdp_client.hs deleted file mode 100644 --- a/mdp_client.hs +++ /dev/null @@ -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 - diff --git a/src/echoworker.hs b/src/echoworker.hs new file mode 100644 --- /dev/null +++ b/src/echoworker.hs @@ -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) diff --git a/src/mdp_client.hs b/src/mdp_client.hs new file mode 100644 --- /dev/null +++ b/src/mdp_client.hs @@ -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 +