diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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. diff --git a/examples/display.hs b/examples/display.hs new file mode 100644 --- /dev/null +++ b/examples/display.hs @@ -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
[
, ...]" + 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 + diff --git a/examples/perf/local_lat.hs b/examples/perf/local_lat.hs new file mode 100644 --- /dev/null +++ b/examples/perf/local_lat.hs @@ -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 " + diff --git a/examples/perf/local_thr.hs b/examples/perf/local_thr.hs new file mode 100644 --- /dev/null +++ b/examples/perf/local_thr.hs @@ -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 " + diff --git a/examples/perf/remote_lat.hs b/examples/perf/remote_lat.hs new file mode 100644 --- /dev/null +++ b/examples/perf/remote_lat.hs @@ -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 " + diff --git a/examples/perf/remote_thr.hs b/examples/perf/remote_thr.hs new file mode 100644 --- /dev/null +++ b/examples/perf/remote_thr.hs @@ -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 " + diff --git a/examples/poll.hs b/examples/poll.hs new file mode 100644 --- /dev/null +++ b/examples/poll.hs @@ -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 " + diff --git a/examples/prompt.hs b/examples/prompt.hs new file mode 100644 --- /dev/null +++ b/examples/prompt.hs @@ -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
" + 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) [] + diff --git a/examples/queue.hs b/examples/queue.hs new file mode 100644 --- /dev/null +++ b/examples/queue.hs @@ -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 () diff --git a/src/System/ZMQ.hs b/src/System/ZMQ.hs --- a/src/System/ZMQ.hs +++ b/src/System/ZMQ.hs @@ -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. diff --git a/src/System/ZMQ/Base.hsc b/src/System/ZMQ/Base.hsc --- a/src/System/ZMQ/Base.hsc +++ b/src/System/ZMQ/Base.hsc @@ -17,11 +17,19 @@ #include -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 diff --git a/test/display.hs b/test/display.hs deleted file mode 100644 --- a/test/display.hs +++ /dev/null @@ -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
[
, ...]" - 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 - diff --git a/test/perf/local_lat.hs b/test/perf/local_lat.hs deleted file mode 100644 --- a/test/perf/local_lat.hs +++ /dev/null @@ -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 " - diff --git a/test/perf/local_thr.hs b/test/perf/local_thr.hs deleted file mode 100644 --- a/test/perf/local_thr.hs +++ /dev/null @@ -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 " - diff --git a/test/perf/remote_lat.hs b/test/perf/remote_lat.hs deleted file mode 100644 --- a/test/perf/remote_lat.hs +++ /dev/null @@ -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 " - diff --git a/test/perf/remote_thr.hs b/test/perf/remote_thr.hs deleted file mode 100644 --- a/test/perf/remote_thr.hs +++ /dev/null @@ -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 " - diff --git a/test/poll.hs b/test/poll.hs deleted file mode 100644 --- a/test/poll.hs +++ /dev/null @@ -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 " - diff --git a/test/prompt.hs b/test/prompt.hs deleted file mode 100644 --- a/test/prompt.hs +++ /dev/null @@ -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
" - 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) [] - diff --git a/test/queue.hs b/test/queue.hs deleted file mode 100644 --- a/test/queue.hs +++ /dev/null @@ -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 () diff --git a/tests/tests.hs b/tests/tests.hs new file mode 100644 --- /dev/null +++ b/tests/tests.hs @@ -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 "" diff --git a/zeromq-haskell.cabal b/zeromq-haskell.cabal --- a/zeromq-haskell.cabal +++ b/zeromq-haskell.cabal @@ -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 +