zeromq-haskell 0.8 → 0.8.1

raw patch · 21 files changed

+481/−320 lines, 21 filesdep +HUnitdep +test-frameworkdep +test-framework-hunitPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: HUnit, test-framework, test-framework-hunit, zeromq-haskell

API changes (from Hackage documentation)

Files

• README.md +3/−0
• examples/display.hs +23/−0
• examples/perf/local_lat.hs +31/−0
• examples/perf/local_thr.hs +49/−0
• examples/perf/remote_lat.hs +36/−0
• examples/perf/remote_thr.hs +27/−0
• examples/poll.hs +31/−0
• examples/prompt.hs +25/−0
• examples/queue.hs +80/−0
• src/System/ZMQ.hs +23/−10
• src/System/ZMQ/Base.hsc +15/−4
• test/display.hs +0/−23
• test/perf/local_lat.hs +0/−31
• test/perf/local_thr.hs +0/−49
• test/perf/remote_lat.hs +0/−36
• test/perf/remote_thr.hs +0/−27
• test/poll.hs +0/−31
• test/prompt.hs +0/−25
• test/queue.hs +0/−80
• tests/tests.hs +112/−0
• zeromq-haskell.cabal +26/−4

@@ -5,6 +5,9 @@  This software currently has *beta* status, i.e. it had seen limited testing. +Version 0.8.1 - zmqVersion has been renamed to version and reports the+runtime version of 0MQ, instead of the compile time version macros.+ Version 0.8.0 - zeromq-haskell can now be compiled either against 0MQ 2.x (default) or 0MQ 3.x. 

@@ -0,0 +1,23 @@+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import Control.Exception+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB++main :: IO ()+main = do+    args <- getArgs+    when (length args < 1) $ do+        hPutStrLn stderr "usage: display <address> [<address>, ...]"+        exitFailure+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Sub $ \s -> do+            ZMQ.subscribe s ""+            mapM (ZMQ.connect s) args+            forever $ do+                line <- ZMQ.receive s []+                SB.putStrLn line+                hFlush stdout+

@@ -0,0 +1,31 @@+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB++main :: IO ()+main = do+    args <- getArgs+    when (length args /= 3) $ do+        hPutStrLn stderr usage+        exitFailure+    let bindTo = args !! 0+        size   = read $ args !! 1+        rounds = read $ args !! 2+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Rep $ \s -> do+            ZMQ.bind s bindTo+            loop s rounds size+  where+    loop s r sz = unless (r <= 0) $ do+        msg <- ZMQ.receive s []+        when (SB.length msg /= sz) $+            error "message of incorrect size received"+        ZMQ.send s msg []+        loop s (r - 1) sz++usage :: String+usage = "usage: local_lat <bind-to> <message-size> <roundtrip-count>"+

@@ -0,0 +1,49 @@+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import Data.Time.Clock+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB+import Text.Printf++main :: IO ()+main = do+    args <- getArgs+    when (length args /= 3) $ do+        hPutStrLn stderr usage+        exitFailure+    let bindTo = args !! 0+        size   = read $ args !! 1 :: Int+        count  = read $ args !! 2 :: Int+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Sub $ \s -> do+            ZMQ.subscribe s ""+            ZMQ.bind s bindTo+            receive s size+            start <- getCurrentTime+            loop s (count - 1) size+            end <- getCurrentTime+            printStat start end size count+  where+    receive s sz = do+        msg <- ZMQ.receive s []+        when (SB.length msg /= sz) $+            error "message of incorrect size received"++    loop s c sz = unless (c <= 0) $ do+        receive s sz+        loop s (c - 1) sz++    printStat start end size count = do+        let elapsed = fromRational . toRational $ diffUTCTime end start :: Double+            through = fromIntegral count / elapsed+            mbits   = (through * fromIntegral size * 8) / 1000000+        printf "message size: %d [B]\n" size+        printf "message count: %d\n" count+        printf "mean throughput: %.3f [msg/s]\n" through+        printf "mean throughput: %.3f [Mb/s]\n" mbits++usage :: String+usage = "usage: local_thr <bind-to> <message-size> <message-count>"+

@@ -0,0 +1,36 @@+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import Data.Time.Clock+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB++main :: IO ()+main = do+    args <- getArgs+    when (length args /= 3) $ do+        hPutStrLn stderr usage+        exitFailure+    let connTo  = args !! 0+        size    = read $ args !! 1+        rounds  = read $ args !! 2+        message = SB.replicate size 0x65+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Req $ \s -> do+            ZMQ.connect s connTo+            start <- getCurrentTime+            loop s rounds message+            end <- getCurrentTime+            print (diffUTCTime end start)+  where+    loop s r msg = unless (r <= 0) $ do+        ZMQ.send s msg []+        msg' <- ZMQ.receive s []+        when (SB.length msg' /= SB.length msg) $+            error "message of incorrect size received"+        loop s (r - 1) msg++usage :: String+usage = "usage: remote_lat <connect-to> <message-size> <roundtrip-count>"+

@@ -0,0 +1,27 @@+import Control.Monad+import Control.Concurrent+import System.IO+import System.Exit+import System.Environment+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB++main :: IO ()+main = do+    args <- getArgs+    when (length args /= 3) $ do+        hPutStrLn stderr usage+        exitFailure+    let connTo  = args !! 0+        size    = read $ args !! 1+        count   = read $ args !! 2+        message = SB.replicate size 0x65+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Pub $ \s -> do+            ZMQ.connect s connTo+            replicateM_ count $ ZMQ.send s message []+            threadDelay 10000000++usage :: String+usage = "usage: remote_thr <connect-to> <message-size> <message-count>"+

@@ -0,0 +1,31 @@+import Control.Applicative+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB++main :: IO ()+main = do+    args <- getArgs+    when (length args /= 1) $ do+        hPutStrLn stderr usage+        exitFailure+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Rep $ \s -> do+            let bindTo = head args+                toPoll = [ZMQ.S s ZMQ.In]+            ZMQ.bind s bindTo+            forever $+                ZMQ.poll toPoll 1000000 >>= receive+ where+    receive []               = return ()+    receive ((ZMQ.S s e):ss) = do+        msg <- ZMQ.receive s []+        ZMQ.send s msg []+        receive ss++usage :: String+usage = "usage: poll <bind-to>"+

@@ -0,0 +1,25 @@+{-# LANGUAGE OverloadedStrings #-}+import Control.Applicative+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString.UTF8 as SB+import qualified Data.ByteString.Char8 as SB++main :: IO ()+main = do+    args <- getArgs+    when (length args /= 2) $ do+        hPutStrLn stderr "usage: prompt <address> <username>"+        exitFailure+    let addr = args !! 0+        name = SB.append (SB.fromString $ args !! 1) ": "+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Pub $ \s -> do+            ZMQ.bind s addr+            forever $ do+                line <- SB.fromString <$> getLine+                ZMQ.send s (SB.append name line) []+

@@ -0,0 +1,80 @@+-- Demo application for a ZeroMQ 'queue' device+--+-- Compile using:+--+-- ghc --make -threaded queue.hs++import Control.Concurrent (forkIO, threadDelay)+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)+import Control.Monad (forever, forM_, replicateM, replicateM_)+import qualified Data.ByteString.Char8 as SB+import qualified System.ZMQ as ZMQ++main :: IO ()+main = ZMQ.withContext 1 $ \context -> do+    lock <- newEmptyMVar+    _    <- forkIO $ launchQueue context lock+    _    <- takeMVar lock++    forM_ [0..numWorkers] $ \i ->+        forkIO $ launchWorker context i++    locks <- replicateM numClients newEmptyMVar+    forM_ (zip [0..numClients] locks) $ \(i, lock') ->+        forkIO $ launchClient context i lock'++    -- Wait untill all clients signal completion+    forM_ locks takeMVar++    -- our queue device is still running, and can't be killed...++  where+    numWorkers :: Int+    numWorkers = 5++    numClients :: Int+    numClients = 2 * numWorkers++    workersAddress :: String+    workersAddress = "inproc://workers"++    clientsAddress :: String+    clientsAddress = "tcp://127.0.0.1:5555"++    message :: SB.ByteString+    message = SB.replicate 10 '\0'++    delay :: Int+    delay = 1000000++    launchQueue :: ZMQ.Context -> MVar () -> IO ()+    launchQueue context lock =+        ZMQ.withSocket context ZMQ.Xreq $ \workers ->+        ZMQ.withSocket context ZMQ.Xrep $ \clients -> do+            ZMQ.bind workers workersAddress+            ZMQ.bind clients clientsAddress+            putMVar lock ()+            ZMQ.device ZMQ.Queue clients workers++    launchWorker :: ZMQ.Context -> Int -> IO ()+    launchWorker context i =+        ZMQ.withSocket context ZMQ.Rep $ \socket -> do+            ZMQ.connect socket workersAddress+            forever $ do+                request <- ZMQ.receive socket []+                putStrLn $+                    "Message received in worker " ++ show i ++ ": " +++                    SB.unpack request+                threadDelay delay -- Do some 'work'+                ZMQ.send socket message []+                putStrLn $ "Reply sent in worker " ++ show i++    launchClient :: ZMQ.Context -> Int -> MVar () -> IO ()+    launchClient context i lock = do+        ZMQ.withSocket context ZMQ.Req $ \socket -> do+            ZMQ.connect socket clientsAddress+            putStrLn $ "Sending message in client " ++ show i+            ZMQ.send socket (SB.pack $ show i) []+            _ <- ZMQ.receive socket []+            putStrLn $ "Reply received in client " ++ show i+        putMVar lock ()

@@ -29,8 +29,10 @@   , Pair(..)   , Pub(..)   , Sub(..)+#ifdef ZMQ3   , XPub(..)   , XSub(..)+#endif   , Req(..)   , Rep(..)   , XReq(..)@@ -61,7 +63,7 @@   , receive   , moreToReceive   , poll-  , System.ZMQ.zmqVersion+  , version      -- * Low-level functions   , init@@ -81,7 +83,7 @@ import Control.Exception import Control.Monad (unless, when) import Data.IORef (atomicModifyIORef)-import Foreign hiding (with)+import Foreign import Foreign.C.Error import Foreign.C.String import Foreign.C.Types (CInt, CShort)@@ -121,6 +123,7 @@ instance SType Sub where     zmqSocketType = const sub +#ifdef ZMQ3 -- | Same as 'Pub' except that you can receive subscriptions from the -- peers in form of incoming messages. Subscription message is a byte 1 -- (for subscriptions) or byte 0 (for unsubscriptions) followed by the@@ -137,6 +140,7 @@ data XSub = XSub instance SType XSub where     zmqSocketType = const xsub+#endif  -- | Socket to send requests and receive replies. Requests are -- load-balanced among all the peers. This socket type allows only an@@ -160,7 +164,7 @@ -- Replies received by this socket are tagged with a proper postfix -- that can be use to route the reply back to the original requester. -- /Compatible peer sockets/: 'Rep', 'Xrep'.-data XReq = Xreq+data XReq = XReq instance SType XReq where     zmqSocketType = const xrequest @@ -170,7 +174,7 @@ -- sending a reply via XREP socket the message should be tagged with a -- prefix from a corresponding request. -- /Compatible peer sockets/: 'Req', 'Xreq'.-data XRep = Xrep+data XRep = XRep instance SType XRep where     zmqSocketType = const xresponse @@ -218,22 +222,24 @@ class SubsType a  instance SubsType Sub+#ifdef ZMQ3 instance SubsType XSub+#endif  -- | The option to set on 0MQ sockets (cf. zmq_setsockopt and zmq_getsockopt -- manpages for details). data SocketOption =     Affinity        Word64    -- ^ ZMQ_AFFINITY-  | Backlog         Int       -- ^ ZMQ_BACKLOG+  | Backlog         CInt      -- ^ ZMQ_BACKLOG   | Events          PollEvent -- ^ ZMQ_EVENTS   | FD              Int       -- ^ ZMQ_FD   | Identity        String    -- ^ ZMQ_IDENTITY-  | Linger          Int       -- ^ ZMQ_LINGER+  | Linger          CInt      -- ^ ZMQ_LINGER   | Rate            Int64     -- ^ ZMQ_RATE   | ReceiveBuf      Word64    -- ^ ZMQ_RCVBUF   | ReceiveMore     Bool      -- ^ ZMQ_RCVMORE-  | ReconnectIVL    Int       -- ^ ZMQ_RECONNECT_IVL-  | ReconnectIVLMax Int       -- ^ ZMQ_RECONNECT_IVL_MAX+  | ReconnectIVL    CInt      -- ^ ZMQ_RECONNECT_IVL+  | ReconnectIVLMax CInt      -- ^ ZMQ_RECONNECT_IVL_MAX   | RecoveryIVL     Int64     -- ^ ZMQ_RECOVERY_IVL   | SendBuf         Word64    -- ^ ZMQ_SNDBUF #ifdef ZMQ2@@ -347,8 +353,15 @@ getMsgOption m (MoreMsgParts _) = MoreMsgParts <$> getIntMsgOpt m more #endif -zmqVersion :: (Int, Int, Int)-zmqVersion = B.zmqVersion+version :: IO (Int, Int, Int)+version =+    with 0 $ \major_ptr ->+    with 0 $ \minor_ptr ->+    with 0 $ \patch_ptr ->+        c_zmq_version major_ptr minor_ptr patch_ptr >>+        tupleUp <$> peek major_ptr <*> peek minor_ptr <*> peek patch_ptr+  where+    tupleUp a b c = (fromIntegral a, fromIntegral b, fromIntegral c)  -- | Initialize a 0MQ context (cf. zmq_init for details).  You should -- normally prefer to use 'with' instead.

@@ -17,11 +17,19 @@  #include <zmq.h> -zmqVersion :: (Int, Int, Int)-zmqVersion = ( #const ZMQ_VERSION_MAJOR-             , #const ZMQ_VERSION_MINOR-             , #const ZMQ_VERSION_PATCH )+-- Ensure Cabal flag matches compile time version of 0MQ+#ifdef ZMQ2+#if ZMQ_VERSION_MAJOR != 2+    ERROR__ZMQ2_Flag_does_not_match_0MQ_Version+#endif+#endif +#ifdef ZMQ3+#if ZMQ_VERSION_MAJOR != 3+    ERROR__ZMQ3_Flag_does_not_match_0MQ_Version+#endif+#endif+ newtype ZMQMsg = ZMQMsg { content :: Ptr () }  instance Storable ZMQMsg where@@ -167,6 +175,9 @@ #endif  -- general initialization++foreign import ccall unsafe "zmq.h zmq_version"+    c_zmq_version :: Ptr CInt -> Ptr CInt -> Ptr CInt -> IO ()  foreign import ccall unsafe "zmq.h zmq_init"     c_zmq_init :: CInt -> IO ZMQCtx

@@ -1,23 +0,0 @@-import Control.Monad-import System.IO-import System.Exit-import System.Environment-import Control.Exception-import qualified System.ZMQ as ZMQ-import qualified Data.ByteString as SB--main :: IO ()-main = do-    args <- getArgs-    when (length args < 1) $ do-        hPutStrLn stderr "usage: display <address> [<address>, ...]"-        exitFailure-    ZMQ.withContext 1 $ \c ->-        ZMQ.withSocket c ZMQ.Sub $ \s -> do-            ZMQ.subscribe s ""-            mapM (ZMQ.connect s) args-            forever $ do-                line <- ZMQ.receive s []-                SB.putStrLn line-                hFlush stdout-

@@ -1,31 +0,0 @@-import Control.Monad-import System.IO-import System.Exit-import System.Environment-import qualified System.ZMQ as ZMQ-import qualified Data.ByteString as SB--main :: IO ()-main = do-    args <- getArgs-    when (length args /= 3) $ do-        hPutStrLn stderr usage-        exitFailure-    let bindTo = args !! 0-        size   = read $ args !! 1-        rounds = read $ args !! 2-    ZMQ.withContext 1 $ \c ->-        ZMQ.withSocket c ZMQ.Rep $ \s -> do-            ZMQ.bind s bindTo-            loop s rounds size-  where-    loop s r sz = unless (r <= 0) $ do-        msg <- ZMQ.receive s []-        when (SB.length msg /= sz) $-            error "message of incorrect size received"-        ZMQ.send s msg []-        loop s (r - 1) sz--usage :: String-usage = "usage: local_lat <bind-to> <message-size> <roundtrip-count>"-

@@ -1,49 +0,0 @@-import Control.Monad-import System.IO-import System.Exit-import System.Environment-import Data.Time.Clock-import qualified System.ZMQ as ZMQ-import qualified Data.ByteString as SB-import Text.Printf--main :: IO ()-main = do-    args <- getArgs-    when (length args /= 3) $ do-        hPutStrLn stderr usage-        exitFailure-    let bindTo = args !! 0-        size   = read $ args !! 1 :: Int-        count  = read $ args !! 2 :: Int-    ZMQ.withContext 1 $ \c ->-        ZMQ.withSocket c ZMQ.Sub $ \s -> do-            ZMQ.subscribe s ""-            ZMQ.bind s bindTo-            receive s size-            start <- getCurrentTime-            loop s (count - 1) size-            end <- getCurrentTime-            printStat start end size count-  where-    receive s sz = do-        msg <- ZMQ.receive s []-        when (SB.length msg /= sz) $-            error "message of incorrect size received"--    loop s c sz = unless (c <= 0) $ do-        receive s sz-        loop s (c - 1) sz--    printStat start end size count = do-        let elapsed = fromRational . toRational $ diffUTCTime end start :: Double-            through = fromIntegral count / elapsed-            mbits   = (through * fromIntegral size * 8) / 1000000-        printf "message size: %d [B]\n" size-        printf "message count: %d\n" count-        printf "mean throughput: %.3f [msg/s]\n" through-        printf "mean throughput: %.3f [Mb/s]\n" mbits--usage :: String-usage = "usage: local_thr <bind-to> <message-size> <message-count>"-

@@ -1,36 +0,0 @@-import Control.Monad-import System.IO-import System.Exit-import System.Environment-import Data.Time.Clock-import qualified System.ZMQ as ZMQ-import qualified Data.ByteString as SB--main :: IO ()-main = do-    args <- getArgs-    when (length args /= 3) $ do-        hPutStrLn stderr usage-        exitFailure-    let connTo  = args !! 0-        size    = read $ args !! 1-        rounds  = read $ args !! 2-        message = SB.replicate size 0x65-    ZMQ.withContext 1 $ \c ->-        ZMQ.withSocket c ZMQ.Req $ \s -> do-            ZMQ.connect s connTo-            start <- getCurrentTime-            loop s rounds message-            end <- getCurrentTime-            print (diffUTCTime end start)-  where-    loop s r msg = unless (r <= 0) $ do-        ZMQ.send s msg []-        msg' <- ZMQ.receive s []-        when (SB.length msg' /= SB.length msg) $-            error "message of incorrect size received"-        loop s (r - 1) msg--usage :: String-usage = "usage: remote_lat <connect-to> <message-size> <roundtrip-count>"-

@@ -1,27 +0,0 @@-import Control.Monad-import Control.Concurrent-import System.IO-import System.Exit-import System.Environment-import qualified System.ZMQ as ZMQ-import qualified Data.ByteString as SB--main :: IO ()-main = do-    args <- getArgs-    when (length args /= 3) $ do-        hPutStrLn stderr usage-        exitFailure-    let connTo  = args !! 0-        size    = read $ args !! 1-        count   = read $ args !! 2-        message = SB.replicate size 0x65-    ZMQ.withContext 1 $ \c ->-        ZMQ.withSocket c ZMQ.Pub $ \s -> do-            ZMQ.connect s connTo-            replicateM_ count $ ZMQ.send s message []-            threadDelay 10000000--usage :: String-usage = "usage: remote_thr <connect-to> <message-size> <message-count>"-

@@ -1,31 +0,0 @@-import Control.Applicative-import Control.Monad-import System.IO-import System.Exit-import System.Environment-import qualified System.ZMQ as ZMQ-import qualified Data.ByteString as SB--main :: IO ()-main = do-    args <- getArgs-    when (length args /= 1) $ do-        hPutStrLn stderr usage-        exitFailure-    ZMQ.withContext 1 $ \c ->-        ZMQ.withSocket c ZMQ.Rep $ \s -> do-            let bindTo = head args-                toPoll = [ZMQ.S s ZMQ.In]-            ZMQ.bind s bindTo-            forever $-                ZMQ.poll toPoll 1000000 >>= receive- where-    receive []               = return ()-    receive ((ZMQ.S s e):ss) = do-        msg <- ZMQ.receive s []-        ZMQ.send s msg []-        receive ss--usage :: String-usage = "usage: poll <bind-to>"-

@@ -1,25 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}-import Control.Applicative-import Control.Monad-import System.IO-import System.Exit-import System.Environment-import qualified System.ZMQ as ZMQ-import qualified Data.ByteString.UTF8 as SB-import qualified Data.ByteString.Char8 as SB--main :: IO ()-main = do-    args <- getArgs-    when (length args /= 2) $ do-        hPutStrLn stderr "usage: prompt <address> <username>"-        exitFailure-    let addr = args !! 0-        name = SB.append (SB.fromString $ args !! 1) ": "-    ZMQ.withContext 1 $ \c ->-        ZMQ.withSocket c ZMQ.Pub $ \s -> do-            ZMQ.bind s addr-            forever $ do-                line <- SB.fromString <$> getLine-                ZMQ.send s (SB.append name line) []-

@@ -1,80 +0,0 @@--- Demo application for a ZeroMQ 'queue' device------ Compile using:------ ghc --make -threaded queue.hs--import Control.Concurrent (forkIO, threadDelay)-import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)-import Control.Monad (forever, forM_, replicateM, replicateM_)-import qualified Data.ByteString.Char8 as SB-import qualified System.ZMQ as ZMQ--main :: IO ()-main = ZMQ.withContext 1 $ \context -> do-    lock <- newEmptyMVar-    _    <- forkIO $ launchQueue context lock-    _    <- takeMVar lock--    forM_ [0..numWorkers] $ \i ->-        forkIO $ launchWorker context i--    locks <- replicateM numClients newEmptyMVar-    forM_ (zip [0..numClients] locks) $ \(i, lock') ->-        forkIO $ launchClient context i lock'--    -- Wait untill all clients signal completion-    forM_ locks takeMVar--    -- our queue device is still running, and can't be killed...--  where-    numWorkers :: Int-    numWorkers = 5--    numClients :: Int-    numClients = 2 * numWorkers--    workersAddress :: String-    workersAddress = "inproc://workers"--    clientsAddress :: String-    clientsAddress = "tcp://127.0.0.1:5555"--    message :: SB.ByteString-    message = SB.replicate 10 '\0'--    delay :: Int-    delay = 1000000--    launchQueue :: ZMQ.Context -> MVar () -> IO ()-    launchQueue context lock =-        ZMQ.withSocket context ZMQ.Xreq $ \workers ->-        ZMQ.withSocket context ZMQ.Xrep $ \clients -> do-            ZMQ.bind workers workersAddress-            ZMQ.bind clients clientsAddress-            putMVar lock ()-            ZMQ.device ZMQ.Queue clients workers--    launchWorker :: ZMQ.Context -> Int -> IO ()-    launchWorker context i =-        ZMQ.withSocket context ZMQ.Rep $ \socket -> do-            ZMQ.connect socket workersAddress-            forever $ do-                request <- ZMQ.receive socket []-                putStrLn $-                    "Message received in worker " ++ show i ++ ": " ++-                    SB.unpack request-                threadDelay delay -- Do some 'work'-                ZMQ.send socket message []-                putStrLn $ "Reply sent in worker " ++ show i--    launchClient :: ZMQ.Context -> Int -> MVar () -> IO ()-    launchClient context i lock = do-        ZMQ.withSocket context ZMQ.Req $ \socket -> do-            ZMQ.connect socket clientsAddress-            putStrLn $ "Sending message in client " ++ show i-            ZMQ.send socket (SB.pack $ show i) []-            _ <- ZMQ.receive socket []-            putStrLn $ "Reply received in client " ++ show i-        putMVar lock ()

@@ -0,0 +1,112 @@+import Test.Framework (defaultMain, testGroup)+import Test.Framework (Test)+import Test.Framework.Providers.HUnit++import qualified System.ZMQ as ZMQ++main :: IO ()+main = defaultMain tests++tests :: [Test]+tests = [+    testGroup "Misc" [+          testCase "ZMQ Context" test_context+    ]+  , testGroup "Socket Creation" [+          testCase "Push" (test_socket ZMQ.Push)+        , testCase "Pull" (test_socket ZMQ.Pull)+        , testCase "XRep" (test_socket ZMQ.XRep)+        , testCase "XReq" (test_socket ZMQ.XReq)+        , testCase "Rep"  (test_socket ZMQ.Rep)+        , testCase "Req"  (test_socket ZMQ.Req)+        , testCase "Sub"  (test_socket ZMQ.Sub)+        , testCase "Pub"  (test_socket ZMQ.Pub)+        , testCase "Pair" (test_socket ZMQ.Pair)+--ifdef ZMQ2+        , testCase "Down" (test_socket ZMQ.Down)+        , testCase "Up"   (test_socket ZMQ.Up)+--endif+    ]+  , testGroup "Socket Option (get)" [+          testCase "Affinity"        (test_option_get (ZMQ.Affinity undefined))+        , testCase "Backlog"         (test_option_get (ZMQ.Backlog undefined))+        , testCase "Events"          (test_option_get (ZMQ.Events undefined))+        , testCase "FD"              (test_option_get (ZMQ.FD undefined))+        , testCase "Identity"        (test_option_get (ZMQ.Identity undefined))+        , testCase "Linger"          (test_option_get (ZMQ.Linger undefined))+        , testCase "Rate"            (test_option_get (ZMQ.Rate undefined))+        , testCase "ReceiveBuf"      (test_option_get (ZMQ.ReceiveBuf undefined))+        , testCase "ReceiveMore"     (test_option_get (ZMQ.ReceiveMore undefined))+        , testCase "ReconnectIVL"    (test_option_get (ZMQ.ReconnectIVL undefined))+        , testCase "ReconnectIVLMax" (test_option_get (ZMQ.ReconnectIVLMax undefined))+        , testCase "ReconnectIVL"    (test_option_get (ZMQ.RecoveryIVL undefined))+        , testCase "SendBuf"         (test_option_get (ZMQ.SendBuf undefined))+--ifdef ZMQ2+        , testCase "HighWM"          (test_option_get (ZMQ.HighWM undefined))+        , testCase "McastLoop"       (test_option_get (ZMQ.McastLoop undefined))+        , testCase "RecoveryIVLMsec" (test_option_get (ZMQ.RecoveryIVLMsec undefined))+        , testCase "Swap"            (test_option_get (ZMQ.Swap undefined))+--endif+    ]+  , testGroup "Socket Option (set)" [+          testCase "Affinity"        (test_option_set (ZMQ.Affinity 0))+        , testCase "Backlog"         (test_option_set (ZMQ.Backlog 100))+        , testCase "Events"          (test_option_set (ZMQ.Events undefined))+        , testCase "FD"              (test_option_set (ZMQ.FD undefined))+        , testCase "Identity"        (test_option_set (ZMQ.Identity "id"))+        , testCase "Linger"          (test_option_set (ZMQ.Linger (-1)))+        , testCase "Rate"            (test_option_set (ZMQ.Rate 100))+        , testCase "ReceiveBuf"      (test_option_set (ZMQ.ReceiveBuf 0))+        , testCase "ReceiveMore"     (test_option_set (ZMQ.ReceiveMore undefined))+        , testCase "ReconnectIVL"    (test_option_set (ZMQ.ReconnectIVL 100))+        , testCase "ReconnectIVLMax" (test_option_set (ZMQ.ReconnectIVLMax 0))+        , testCase "ReconnectIVL"    (test_option_set (ZMQ.RecoveryIVL 100))+        , testCase "SendBuf"         (test_option_set (ZMQ.SendBuf 0))+--ifdef ZMQ2+        , testCase "HighWM"          (test_option_set (ZMQ.HighWM 0))+        , testCase "McastLoop"       (test_option_set (ZMQ.McastLoop True))+        , testCase "RecoveryIVLMsec" (test_option_set (ZMQ.RecoveryIVLMsec 10))+        , testCase "Swap"            (test_option_set (ZMQ.Swap 0))+--endif+    ]+  , testGroup "Pub/Sub" [+          testCase "Subscribe Sub"    (test_subscribe ZMQ.Sub)+        , testCase "Unsubscribe Sub"  (test_unsubscribe ZMQ.Sub)+    ]+  ]++test_context :: IO ()+test_context = do+    c1 <- ZMQ.init 1+    c2 <- ZMQ.init 3+    ZMQ.term c1+    ZMQ.term c2++test_socket :: ZMQ.SType a => a -> IO ()+test_socket ty =+    ZMQ.withContext 1 $ \c ->+        ZMQ.socket c ty >>= ZMQ.close++test_option_get :: ZMQ.SocketOption -> IO ()+test_option_get o =+    ZMQ.withContext 1 $ \c ->+    ZMQ.withSocket c ZMQ.Sub $ \s ->+        ZMQ.getOption s o >> return ()++test_option_set :: ZMQ.SocketOption -> IO ()+test_option_set o =+    ZMQ.withContext 1 $ \c ->+    ZMQ.withSocket c ZMQ.Sub $ \s ->+        ZMQ.setOption s o++test_subscribe :: (ZMQ.SubsType a, ZMQ.SType a) => a -> IO ()+test_subscribe ty =+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ty $ \s ->+            ZMQ.subscribe s ""++test_unsubscribe :: (ZMQ.SubsType a, ZMQ.SType a) => a -> IO ()+test_unsubscribe ty =+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ty $ \s ->+            ZMQ.unsubscribe s ""

@@ -1,5 +1,5 @@ name:               zeromq-haskell-version:            0.8+version:            0.8.1 synopsis:           bindings to zeromq category:           System, FFI license:            MIT@@ -10,9 +10,13 @@ homepage:           http://github.com/twittner/zeromq-haskell/ stability:          experimental tested-With:        GHC == 7.0.3-cabal-version:      >= 1.6.0+cabal-version:      >= 1.8 build-type:         Simple-extra-source-files: README.md, AUTHORS, test/*.hs, test/perf/*.hs+extra-source-files: README.md+                  , AUTHORS+                  , examples/*.hs+                  , examples/perf/*.hs+                  , tests/*.hs description:     The 0MQ lightweight messaging kernel is a library which extends     the standard socket interfaces with features traditionally provided@@ -22,7 +26,7 @@     multiple transport protocols and more.      This library provides the Haskell language binding to 0MQ and-    support 0MQ 2.x as well as 3.x (use -fzmq2 or -fzmq3 to select+    supports 0MQ 2.x as well as 3.x (use -fzmq2 or -fzmq3 to select     between the two).  flag zmq2@@ -57,4 +61,22 @@         extra-libraries:  zmq, pthread     else         extra-libraries:  zmq++test-suite zeromq-haskell-tests+    type:             exitcode-stdio-1.0+    hs-source-dirs:   tests+    main-is:          tests.hs+    build-depends:    zeromq-haskell+                    , base >= 3 && < 5+                    , containers+                    , bytestring+                    , test-framework >= 0.4+                    , test-framework-hunit >= 0.2+                    , HUnit+    ghc-options:      -Wall -threaded -rtsopts++source-repository head+    type:             git+    location:         https://github.com/twittner/zeromq-haskell+