zeromq-haskell 0.7.1 → 0.8.4
raw patch · 23 files changed
Files
- README.md +15/−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 +174/−214
- src/System/ZMQ/Base.hsc +52/−46
- src/System/ZMQ/Internal.hs +21/−9
- 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/System/ZMQ/Test/Properties.hs +120/−0
- tests/tests.hs +7/−0
- zeromq-haskell.cabal +56/−21
README.md view
@@ -5,6 +5,21 @@ This software currently has *beta* status, i.e. it had seen limited testing. +Version 0.8.4 - Various fixes to work with GHC 7.4.1.++Version 0.8.3 - Derive Read in SocketOption and PollEvent.++Version 0.8.2 - Revert changes to support 0MQ 2.x as well as 3.x in a single+package. Instead git branches will be used to track the various 0MQ versions+and a separate zeromq-haskell-3 library will be released to hackage.+Also this releases handles EINTR properly in socket option setting/getting.++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.+ Version 0.7.1 - Removes unix dependency Verison 0.7.0 - Changes semantics of poll to return a list of the same
+ examples/display.hs view
@@ -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+
+ examples/perf/local_lat.hs view
@@ -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>"+
+ examples/perf/local_thr.hs view
@@ -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>"+
+ examples/perf/remote_lat.hs view
@@ -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>"+
+ examples/perf/remote_thr.hs view
@@ -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>"+
+ examples/poll.hs view
@@ -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>"+
+ examples/prompt.hs view
@@ -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) []+
+ examples/queue.hs view
@@ -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 ()
src/System/ZMQ.hs view
@@ -1,7 +1,7 @@ {-# LANGUAGE ExistentialQuantification #-} -- | -- Module : System.ZMQ--- Copyright : (c) 2010 Toralf Wittner+-- Copyright : (c) 2010-2011 Toralf Wittner -- License : MIT -- Maintainer : toralf.wittner@gmail.com -- Stability : experimental@@ -9,56 +9,61 @@ -- -- 0MQ haskell binding. The API closely follows the C-API of 0MQ with -- the main difference that sockets are typed.--- The documentation of the individual socket types and socket options--- is copied from 0MQ's man pages authored by Martin Sustrik.+-- The documentation of the individual socket types is copied from+-- 0MQ's man pages authored by Martin Sustrik. For details please+-- refer to http://api.zeromq.org module System.ZMQ ( - Size,- Context,- Socket,- Flag(..),- SocketOption(..),- Poll(..),- Timeout,- PollEvent(..),- Device(..),+ Size+ , Context+ , Socket+ , Flag(..)+ , SocketOption(..)+ , Poll(..)+ , Timeout+ , PollEvent(..) - SType,- SubsType,- Pair(..),- Pub(..),- Sub(..),- Req(..),- Rep(..),- XReq(..),- XRep(..),- Pull(..),- Push(..),- Up(..),- Down(..),+ , SType+ , SubsType+ , Pair(..)+ , Pub(..)+ , Sub(..)+ , Req(..)+ , Rep(..)+ , XReq(..)+ , XRep(..)+ , Dealer(..)+ , Router(..)+ , Pull(..)+ , Push(..)+ , Up(..)+ , Down(..) - withContext,- withSocket,- setOption,- getOption,- System.ZMQ.subscribe,- System.ZMQ.unsubscribe,- bind,- connect,- send,- send',- receive,- moreToReceive,- poll,- device,+ , withContext+ , withSocket+ , setOption+ , getOption+ , System.ZMQ.subscribe+ , System.ZMQ.unsubscribe+ , bind+ , connect+ , send+ , send'+ , receive+ , moreToReceive+ , poll+ , version -- * Low-level functions- init,- term,- socket,- close,+ , init+ , term+ , socket+ , close + , Device(..)+ , device+ ) where import Prelude hiding (init)@@ -66,23 +71,20 @@ import Control.Exception import Control.Monad (unless, when) import Data.IORef (atomicModifyIORef)-import Data.Int-import System.ZMQ.Base-import qualified System.ZMQ.Base as B-import System.ZMQ.Internal-import Foreign hiding (with)+import Foreign import Foreign.C.Error import Foreign.C.String import Foreign.C.Types (CInt, CShort) import qualified Data.ByteString as SB import qualified Data.ByteString.Lazy as LB-import System.Mem.Weak (addFinalizer) import System.Posix.Types (Fd(..))+import System.ZMQ.Base+import qualified System.ZMQ.Base as B+import System.ZMQ.Internal import GHC.Conc (threadWaitRead, threadWaitWrite) --- Socket types:-+-- | Socket types. class SType a where zmqSocketType :: a -> ZMQSocketType @@ -130,21 +132,29 @@ -- 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 +data Dealer = Dealer+instance SType Dealer where+ zmqSocketType = const dealer+ -- | Special socket type to be used in request/reply middleboxes -- such as zmq_queue(7). Requests received using this socket are already -- properly tagged with prefix identifying the original requester. When -- 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 --- | A socket of type ZMQ_PULL is used by a pipeline node to receive+data Router = Router+instance SType Router where+ zmqSocketType = const router++-- | A socket of type Pull is used by a pipeline node to receive -- messages from upstream pipeline nodes. Messages are fair-queued from -- among all connected upstream nodes. The zmq_send() function is not -- implemented for this socket type.@@ -152,12 +162,12 @@ instance SType Pull where zmqSocketType = const pull --- | A socket of type ZMQ_PUSH is used by a pipeline node to send messages+-- | A socket of type Push is used by a pipeline node to send messages -- to downstream pipeline nodes. Messages are load-balanced to all connected -- downstream nodes. The zmq_recv() function is not implemented for this -- socket type. ----- When a ZMQ_PUSH socket enters an exceptional state due to having reached+-- When a Push socket enters an exceptional state due to having reached -- the high water mark for all downstream nodes, or if there are no -- downstream nodes at all, then any zmq_send(3) operations on the socket -- shall block until the exceptional state ends or at least one downstream@@ -182,94 +192,32 @@ instance SType Down where zmqSocketType = const downstream --- Subscribable:-+-- | Subscribable. class SubsType a+ instance SubsType Sub --- | The option to set on 0MQ sockets (descriptions reproduced here from--- zmq_setsockopt(3) (cf. man zmq_setsockopt for further details)).------ [@HighWM@] High watermark for the message pipes associated with the--- socket. The water mark cannot be exceeded. If the messages--- don't fit into the pipe emergency mechanisms of the--- particular socket type are used (block, drop etc.)--- If HWM is set to zero, there are no limits for the content--- of the pipe.--- /Default/: 0------ [@Swap@] Swap allows the pipe to exceed high watermark. However,--- the data are written to the disk rather than held in the memory.--- Until high watermark is exceeded there is no disk activity involved--- though. The value of the option defines maximal size of the swap file.--- /Default/: 0------ [@Affinity@] Affinity defines which threads in the thread pool will--- be used to handle newly created sockets. This way you can dedicate--- some of the threads (CPUs) to a specific work. Value of 0 means no--- affinity. Work is distributed fairly among the threads in the--- thread pool. For non-zero values, the lowest bit corresponds to the--- thread 1, second lowest bit to the thread 2 etc. Thus, value of 3--- means that from now on newly created sockets will handle I/O activity--- exclusively using threads no. 1 and 2.--- /Default/: 0------ [@Identity@] Identity of the socket. Identity is important when--- restarting applications. If the socket has no identity, each run of--- the application is completely separated from other runs. However,--- with identity application reconnects to existing infrastructure--- left by the previous run. Thus it may receive messages that were--- sent in the meantime, it shares pipe limits with the previous run etc.--- /Default/: NULL------ [@Rate@] This option applies only to sending side of multicast--- transports (pgm & udp). It specifies maximal outgoing data rate that--- an individual sender socket can send.--- /Default/: 100------ [@RecoveryIVL@] This option applies only to multicast transports--- (pgm & udp). It specifies how long can the receiver socket survive--- when the sender is inaccessible. Keep in mind that large recovery--- intervals at high data rates result in very large recovery buffers,--- meaning that you can easily overload your box by setting say 1 minute--- recovery interval at 1Gb/s rate (requires 7GB in-memory buffer).--- /Default/: 10------ [@McastLoop@] This option applies only to multicast transports--- (pgm & udp). Value of 1 means that the mutlicast packets can be--- received on the box they were sent from. Setting the value to 0--- disables the loopback functionality which can have negative impact on--- the performance. If possible, disable the loopback in production--- environments.--- /Default/: 1------ [@SendBuf@] Sets the underlying kernel transmit buffer size to the--- specified size. See SO_SNDBUF POSIX socket option. Value of zero--- means leaving the OS default unchanged.--- /Default/: 0------ [@ReceiveBuf@] Sets the underlying kernel receive buffer size to--- the specified size. See SO_RCVBUF POSIX socket option. Value of--- zero means leaving the OS default unchanged.--- /Default/: 0---+-- | The option to set on 0MQ sockets (cf. zmq_setsockopt and zmq_getsockopt+-- manpages for details). data SocketOption =- HighWM Word64 -- ^ ZMQ_HWM- | Swap Int64 -- ^ ZMQ_SWAP- | Affinity Word64 -- ^ ZMQ_AFFINITY- | Identity String -- ^ ZMQ_IDENTITY- | Rate Int64 -- ^ ZMQ_RATE- | RecoveryIVL Int64 -- ^ ZMQ_RECOVERY_IVL- | RecoveryIVLMsec Int64 -- ^ ZMQ_RECOVERY_IVL_MSEC- | McastLoop Int64 -- ^ ZMQ_MCAST_LOOP- | SendBuf Word64 -- ^ ZMQ_SNDBUF- | ReceiveBuf Word64 -- ^ ZMQ_RCVBUF- | FD CInt -- ^ ZMQ_FD- | Events Word32 -- ^ ZMQ_EVENTS- | Linger CInt -- ^ ZMQ_LINGER- | ReconnectIVL CInt -- ^ ZMQ_RECONNECT_IVL- | Backlog CInt -- ^ ZMQ_BACKLOG- deriving (Eq, Ord, Show)+ Affinity Word64 -- ^ ZMQ_AFFINITY+ | Backlog CInt -- ^ ZMQ_BACKLOG+ | Events PollEvent -- ^ ZMQ_EVENTS+ | FD CInt -- ^ ZMQ_FD+ | Identity String -- ^ ZMQ_IDENTITY+ | Linger CInt -- ^ ZMQ_LINGER+ | Rate Int64 -- ^ ZMQ_RATE+ | ReceiveBuf Word64 -- ^ ZMQ_RCVBUF+ | ReceiveMore Bool -- ^ ZMQ_RCVMORE+ | ReconnectIVL CInt -- ^ ZMQ_RECONNECT_IVL+ | ReconnectIVLMax CInt -- ^ ZMQ_RECONNECT_IVL_MAX+ | RecoveryIVL Int64 -- ^ ZMQ_RECOVERY_IVL+ | SendBuf Word64 -- ^ ZMQ_SNDBUF+ | HighWM Word64 -- ^ ZMQ_HWM+ | McastLoop Bool -- ^ ZMQ_MCAST_LOOP+ | RecoveryIVLMsec Int64 -- ^ ZMQ_RECOVERY_IVL_MSEC+ | Swap Int64 -- ^ ZMQ_SWAP+ deriving (Eq, Ord, Show, Read) -- | The events to wait for in poll (cf. man zmq_poll) data PollEvent =@@ -278,7 +226,7 @@ | InOut -- ^ ZMQ_POLLIN | ZMQ_POLLOUT | Native -- ^ ZMQ_POLLERR | None- deriving (Eq, Ord, Show)+ deriving (Eq, Ord, Show, Read) -- | Type representing a descriptor, poll is waiting for -- (either a 0MQ socket or a file descriptor) plus the type@@ -287,13 +235,64 @@ forall a. S (Socket a) PollEvent | F Fd PollEvent --- | Type representing ZeroMQ devices, as used with zmq_device-data Device =- Streamer -- ^ ZMQ_STREAMER- | Forwarder -- ^ ZMQ_FORWARDER- | Queue -- ^ ZMQ_QUEUE- deriving (Eq, Ord, Show)+-- | Set the given option on the socket. Please note that there are+-- certain combatibility constraints w.r.t the socket type (cf. man+-- zmq_setsockopt).+--+-- Please note that subscribe/unsubscribe is handled with separate+-- functions.+setOption :: Socket a -> SocketOption -> IO ()+setOption s (Affinity o) = setIntOpt s affinity o+setOption s (Backlog o) = setIntOpt s backlog o+setOption _ (Events _) = return () -- NOP+setOption _ (FD _) = return () -- NOP+setOption s (Identity o) = setStrOpt s identity o+setOption s (Linger o) = setIntOpt s linger o+setOption s (Rate o) = setIntOpt s rate o+setOption s (ReceiveBuf o) = setIntOpt s receiveBuf o+setOption _ (ReceiveMore _) = return () -- NOP+setOption s (ReconnectIVL o) = setIntOpt s reconnectIVL o+setOption s (ReconnectIVLMax o) = setIntOpt s reconnectIVLMax o+setOption s (RecoveryIVL o) = setIntOpt s recoveryIVL o+setOption s (SendBuf o) = setIntOpt s sendBuf o+setOption s (HighWM o) = setIntOpt s highWM o+setOption s (McastLoop o) = setBoolOpt s mcastLoop o+setOption s (RecoveryIVLMsec o) = setIntOpt s recoveryIVLMsec o+setOption s (Swap o) = setIntOpt s swap o +-- | Get the given socket option by passing in some dummy value of+-- that option. The actual value will be returned. Please note that+-- there are certain combatibility constraints w.r.t the socket+-- type (cf. man zmq_setsockopt).+getOption :: Socket a -> SocketOption -> IO SocketOption+getOption s (Affinity _) = Affinity <$> getIntOpt s affinity+getOption s (Backlog _) = Backlog <$> getIntOpt s backlog+getOption s (Events _) = Events . toEvent <$> getIntOpt s events+getOption s (FD _) = FD <$> getIntOpt s filedesc+getOption s (Identity _) = Identity <$> getStrOpt s identity+getOption s (Linger _) = Linger <$> getIntOpt s linger+getOption s (Rate _) = Rate <$> getIntOpt s rate+getOption s (ReceiveBuf _) = ReceiveBuf <$> getIntOpt s receiveBuf+getOption s (ReceiveMore _) = ReceiveMore <$> getBoolOpt s receiveMore+getOption s (ReconnectIVL _) = ReconnectIVL <$> getIntOpt s reconnectIVL+getOption s (ReconnectIVLMax _) = ReconnectIVLMax <$> getIntOpt s reconnectIVLMax+getOption s (RecoveryIVL _) = RecoveryIVL <$> getIntOpt s recoveryIVL+getOption s (SendBuf _) = SendBuf <$> getIntOpt s sendBuf+getOption s (HighWM _) = HighWM <$> getIntOpt s highWM+getOption s (McastLoop _) = McastLoop <$> getBoolOpt s mcastLoop+getOption s (RecoveryIVLMsec _) = RecoveryIVLMsec <$> getIntOpt s recoveryIVLMsec+getOption s (Swap _) = Swap <$> getIntOpt s swap++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. init :: Size -> IO Context@@ -304,7 +303,7 @@ -- | Terminate a 0MQ context (cf. zmq_term). You should normally -- prefer to use 'with' instead. term :: Context -> IO ()-term = throwErrnoIfMinus1_ "term" . c_zmq_term . ctx+term = throwErrnoIfMinus1Retry_ "term" . c_zmq_term . ctx -- | Run an action with a 0MQ context. The 'Context' supplied to your -- action will /not/ be valid after the action either returns or@@ -312,7 +311,7 @@ withContext :: Size -> (Context -> IO a) -> IO a withContext ioThreads act = bracket (throwErrnoIfNull "c_zmq_init" $ c_zmq_init (fromIntegral ioThreads))- (throwErrnoIfMinus1_ "c_zmq_term" . c_zmq_term)+ (throwErrnoIfMinus1Retry_ "c_zmq_term" . c_zmq_term) (act . Context) -- | Run an action with a 0MQ socket. The socket will be closed after running@@ -325,13 +324,8 @@ -- automatic socket closing and may be safer to use. socket :: SType a => Context -> a -> IO (Socket a) socket (Context c) t = do- let zt = typeVal . zmqSocketType $ t- s <- throwErrnoIfNull "socket" (c_zmq_socket c zt)- sock@(Socket _ status) <- mkSocket s- addFinalizer sock $ do- alive <- atomicModifyIORef status (\b -> (False, b))- when alive $ c_zmq_close s >> return () -- socket has not been closed yet- return sock+ let zt = typeVal . zmqSocketType $ t+ throwErrnoIfNull "socket" (c_zmq_socket c zt) >>= mkSocket -- | Close a 0MQ socket. 'withSocket' provides automatic socket closing and may -- be safer to use.@@ -340,50 +334,6 @@ alive <- atomicModifyIORef status (\b -> (False, b)) when alive $ throwErrnoIfMinus1_ "close" . c_zmq_close $ s --- | Set the given option on the socket. Please note that there are--- certain combatibility constraints w.r.t the socket type (cf. man--- zmq_setsockopt).------ Please note that subscribe/unsubscribe is handled with separate--- functions.-setOption :: Socket a -> SocketOption -> IO ()-setOption s (HighWM o) = setIntOpt s highWM o-setOption s (Swap o) = setIntOpt s swap o-setOption s (Affinity o) = setIntOpt s affinity o-setOption s (Identity o) = setStrOpt s identity o-setOption s (Rate o) = setIntOpt s rate o-setOption s (RecoveryIVL o) = setIntOpt s recoveryIVL o-setOption s (RecoveryIVLMsec o) = setIntOpt s recoveryIVLMsec o-setOption s (McastLoop o) = setIntOpt s mcastLoop o-setOption s (SendBuf o) = setIntOpt s sendBuf o-setOption s (ReceiveBuf o) = setIntOpt s receiveBuf o-setOption s (FD o) = setIntOpt s filedesc o-setOption s (Events o) = setIntOpt s events o-setOption s (Linger o) = setIntOpt s linger o-setOption s (ReconnectIVL o) = setIntOpt s reconnectIVL o-setOption s (Backlog o) = setIntOpt s backlog o---- | Get the given socket option by passing in some dummy value of--- that option. The actual value will be returned. Please note that--- there are certain combatibility constraints w.r.t the socket--- type (cf. man zmq_setsockopt).-getOption :: Socket a -> SocketOption -> IO SocketOption-getOption s (HighWM _) = HighWM <$> getIntOpt s highWM-getOption s (Swap _) = Swap <$> getIntOpt s swap-getOption s (Affinity _) = Affinity <$> getIntOpt s affinity-getOption s (Identity _) = Identity <$> getStrOpt s identity-getOption s (Rate _) = Rate <$> getIntOpt s rate-getOption s (RecoveryIVL _) = RecoveryIVL <$> getIntOpt s recoveryIVL-getOption s (RecoveryIVLMsec _) = RecoveryIVLMsec <$> getIntOpt s recoveryIVLMsec-getOption s (McastLoop _) = McastLoop <$> getIntOpt s mcastLoop-getOption s (SendBuf _) = SendBuf <$> getIntOpt s sendBuf-getOption s (ReceiveBuf _) = ReceiveBuf <$> getIntOpt s receiveBuf-getOption s (FD _) = FD <$> getIntOpt s filedesc-getOption s (Events _) = Events <$> getIntOpt s events-getOption s (Linger _) = Linger <$> getIntOpt s linger-getOption s (ReconnectIVL _) = ReconnectIVL <$> getIntOpt s reconnectIVL-getOption s (Backlog _) = Backlog <$> getIntOpt s backlog- -- | Subscribe Socket to given subscription. subscribe :: SubsType a => Socket a -> String -> IO () subscribe s = setStrOpt s B.subscribe@@ -429,7 +379,7 @@ receive sock fls = bracket messageInit messageClose $ \m -> onSocket "receive" sock $ \s -> do retry "receive" (waitRead sock) $- c_zmq_recv_unsafe s (msgPtr m) (combine (NoBlock : fls))+ c_zmq_recv s (msgPtr m) (combine (NoBlock : fls)) data_ptr <- c_zmq_msg_data (msgPtr m) size <- c_zmq_msg_size (msgPtr m) SB.packCStringLen (data_ptr, fromIntegral size)@@ -456,9 +406,9 @@ createPoll :: (ZMQPoll, Poll) -> Poll createPoll (zp, S (Socket s t) _) =- maybe (S (Socket s t) None) (S (Socket s t)) (toEvent . pRevents $ zp)+ S (Socket s t) (toEvent . fromIntegral . pRevents $ zp) createPoll (zp, F fd _) =- maybe (F fd None) (F fd) (toEvent . pRevents $ zp)+ F fd (toEvent . fromIntegral . pRevents $ zp) fromEvent :: PollEvent -> CShort fromEvent In = fromIntegral . pollVal $ pollIn@@ -467,26 +417,36 @@ fromEvent Native = fromIntegral . pollVal $ pollerr fromEvent None = 0 - toEvent :: CShort -> Maybe PollEvent- toEvent e | e == (fromIntegral . pollVal $ pollIn) = Just In- | e == (fromIntegral . pollVal $ pollOut) = Just Out- | e == (fromIntegral . pollVal $ pollInOut) = Just InOut- | e == (fromIntegral . pollVal $ pollerr) = Just Native- | otherwise = Nothing+toEvent :: Word32 -> PollEvent+toEvent e | e == (fromIntegral . pollVal $ pollIn) = In+ | e == (fromIntegral . pollVal $ pollOut) = Out+ | e == (fromIntegral . pollVal $ pollInOut) = InOut+ | e == (fromIntegral . pollVal $ pollerr) = Native+ | otherwise = None retry :: String -> IO () -> IO CInt -> IO () retry msg wait act = throwErrnoIfMinus1RetryMayBlock_ msg act wait wait' :: (Fd -> IO ()) -> ZMQPollEvent -> Socket a -> IO ()-wait' w f s = do (FD fd) <- getOption s (FD undefined)- w (Fd fd)- (Events evs) <- getOption s (Events undefined)- unless (testev evs) $ wait' w f s- where testev e = e .&. fromIntegral (pollVal f) /= 0+wait' w f s = do+ fd <- getIntOpt s filedesc+ w (Fd fd)+ evs <- getIntOpt s events :: IO Word32+ unless (testev evs) $+ wait' w f s+ where+ testev e = e .&. fromIntegral (pollVal f) /= 0 waitRead, waitWrite :: Socket a -> IO () waitRead = wait' threadWaitRead pollIn waitWrite = wait' threadWaitWrite pollOut++-- | Type representing ZeroMQ devices, as used with zmq_device+data Device =+ Streamer -- ^ ZMQ_STREAMER+ | Forwarder -- ^ ZMQ_FORWARDER+ | Queue -- ^ ZMQ_QUEUE+ deriving (Eq, Ord, Show) -- | Launch a ZeroMQ device (zmq_device). --
src/System/ZMQ/Base.hsc view
@@ -1,7 +1,7 @@ {-# LANGUAGE CPP, ForeignFunctionInterface #-} -- | -- Module : System.ZMQ.Base--- Copyright : (c) 2010 Toralf Wittner+-- Copyright : (c) 2010-2011 Toralf Wittner -- License : MIT -- Maintainer : toralf.wittner@gmail.com -- Stability : experimental@@ -10,13 +10,17 @@ module System.ZMQ.Base where -import Control.Applicative import Foreign import Foreign.C.Types import Foreign.C.String+import Control.Applicative #include <zmq.h> +#if ZMQ_VERSION_MAJOR != 2+ ERROR__Invalid_0MQ_Version+#endif+ newtype ZMQMsg = ZMQMsg { content :: Ptr () } instance Storable ZMQMsg where@@ -56,49 +60,52 @@ newtype ZMQSocketType = ZMQSocketType { typeVal :: CInt } deriving (Eq, Ord) -#{enum ZMQSocketType, ZMQSocketType,- pair = ZMQ_PAIR,- pub = ZMQ_PUB,- sub = ZMQ_SUB,- xpub = ZMQ_XPUB,- xsub = ZMQ_XSUB,- request = ZMQ_REQ,- response = ZMQ_REP,- xrequest = ZMQ_XREQ,- xresponse = ZMQ_XREP,- pull = ZMQ_PULL,- push = ZMQ_PUSH,- upstream = ZMQ_UPSTREAM,- downstream = ZMQ_DOWNSTREAM+#{enum ZMQSocketType, ZMQSocketType+ , pair = ZMQ_PAIR+ , pub = ZMQ_PUB+ , sub = ZMQ_SUB+ , xpub = ZMQ_XPUB+ , xsub = ZMQ_XSUB+ , request = ZMQ_REQ+ , response = ZMQ_REP+ , xrequest = ZMQ_XREQ+ , xresponse = ZMQ_XREP+ , dealer = ZMQ_DEALER+ , router = ZMQ_ROUTER+ , pull = ZMQ_PULL+ , push = ZMQ_PUSH+ , upstream = ZMQ_UPSTREAM+ , downstream = ZMQ_DOWNSTREAM } newtype ZMQOption = ZMQOption { optVal :: CInt } deriving (Eq, Ord) -#{enum ZMQOption, ZMQOption,- highWM = ZMQ_HWM,- swap = ZMQ_SWAP,- affinity = ZMQ_AFFINITY,- identity = ZMQ_IDENTITY,- subscribe = ZMQ_SUBSCRIBE,- unsubscribe = ZMQ_UNSUBSCRIBE,- rate = ZMQ_RATE,- recoveryIVL = ZMQ_RECOVERY_IVL,- recoveryIVLMsec = ZMQ_RECOVERY_IVL_MSEC,- mcastLoop = ZMQ_MCAST_LOOP,- sendBuf = ZMQ_SNDBUF,- receiveBuf = ZMQ_RCVBUF,- receiveMore = ZMQ_RCVMORE,- filedesc = ZMQ_FD,- events = ZMQ_EVENTS,- linger = ZMQ_LINGER,- reconnectIVL = ZMQ_RECONNECT_IVL,- backlog = ZMQ_BACKLOG+#{enum ZMQOption, ZMQOption+ , affinity = ZMQ_AFFINITY+ , backlog = ZMQ_BACKLOG+ , events = ZMQ_EVENTS+ , filedesc = ZMQ_FD+ , identity = ZMQ_IDENTITY+ , linger = ZMQ_LINGER+ , rate = ZMQ_RATE+ , receiveBuf = ZMQ_RCVBUF+ , receiveMore = ZMQ_RCVMORE+ , reconnectIVL = ZMQ_RECONNECT_IVL+ , reconnectIVLMax = ZMQ_RECONNECT_IVL_MAX+ , recoveryIVL = ZMQ_RECOVERY_IVL+ , sendBuf = ZMQ_SNDBUF+ , subscribe = ZMQ_SUBSCRIBE+ , unsubscribe = ZMQ_UNSUBSCRIBE+ , highWM = ZMQ_HWM+ , mcastLoop = ZMQ_MCAST_LOOP+ , recoveryIVLMsec = ZMQ_RECOVERY_IVL_MSEC+ , swap = ZMQ_SWAP } newtype ZMQFlag = ZMQFlag { flagVal :: CInt } deriving (Eq, Ord) -#{enum ZMQFlag, ZMQFlag,- noBlock = ZMQ_NOBLOCK+#{enum ZMQFlag, ZMQFlag+ , noBlock = ZMQ_NOBLOCK , sndMore = ZMQ_SNDMORE } @@ -119,8 +126,14 @@ deviceQueue = ZMQ_QUEUE } +foreign import ccall safe "zmq.h zmq_device"+ c_zmq_device :: CInt -> ZMQSocket -> ZMQSocket -> IO CInt+ -- 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 @@ -172,22 +185,15 @@ foreign import ccall unsafe "zmq.h zmq_connect" c_zmq_connect :: ZMQSocket -> CString -> IO CInt + foreign import ccall unsafe "zmq.h zmq_send" c_zmq_send :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt -foreign import ccall safe "zmq.h zmq_recv"- c_zmq_recv :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt- foreign import ccall unsafe "zmq.h zmq_recv"- c_zmq_recv_unsafe :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt+ c_zmq_recv :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt -- poll foreign import ccall safe "zmq.h zmq_poll" c_zmq_poll :: ZMQPollPtr -> CInt -> CLong -> IO CInt---- device--foreign import ccall safe "zmq.h zmq_device"- c_zmq_device :: CInt -> ZMQSocket -> ZMQSocket -> IO CInt
src/System/ZMQ/Internal.hs view
@@ -11,6 +11,7 @@ , messageClose , messageInit , messageInitSize+ , setBoolOpt , setIntOpt , setStrOpt , getBoolOpt@@ -23,7 +24,7 @@ ) where import Control.Applicative-import Control.Monad (foldM_)+import Control.Monad (foldM_, when) import Control.Exception import Data.IORef (IORef) @@ -35,7 +36,7 @@ import qualified Data.ByteString as SB import qualified Data.ByteString.Lazy as LB import qualified Data.ByteString.Unsafe as UB-import Data.IORef (newIORef)+import Data.IORef (newIORef, mkWeakIORef, readIORef) import System.ZMQ.Base @@ -47,7 +48,7 @@ -- [@NoBlock@] Send operation should be performed in non-blocking mode. -- If it cannot be performed immediatley an error will be thrown (errno -- is set to EAGAIN).-data Flag = NoBlock -- ^ ZMQ_NOBLOCK+data Flag = NoBlock -- ^ ZMQ_NOBLOCK (0MQ-2.x), ZMQ_DONTWAIT (0MQ-3.x) | SndMore -- ^ ZMQ_SNDMORE deriving (Eq, Ord, Show) @@ -56,7 +57,7 @@ -- | A 0MQ Socket. data Socket a = Socket {- _socket :: ZMQSocket+ _socket :: ZMQSocket , _sockLive :: IORef Bool } @@ -70,7 +71,14 @@ {-# INLINE onSocket #-} mkSocket :: ZMQSocket -> IO (Socket a)-mkSocket s = Socket s <$> newIORef True+mkSocket s = do+ ref <- newIORef True+ addFinalizer ref $ do+ alive <- readIORef ref+ when alive $ c_zmq_close s >> return ()+ return (Socket s ref)+ where+ addFinalizer r f = mkWeakIORef r f >> return () messageOf :: SB.ByteString -> IO Message messageOf b = UB.unsafeUseAsCStringLen b $ \(cstr, len) -> do@@ -109,16 +117,20 @@ c_zmq_msg_init_size ptr (fromIntegral s) return (Message ptr) +setBoolOpt :: Socket a -> ZMQOption -> Bool -> IO ()+setBoolOpt sock opt True = setIntOpt sock opt (1 :: Int64)+setBoolOpt sock opt False = setIntOpt sock opt (0 :: Int64)+ setIntOpt :: (Storable b, Integral b) => Socket a -> ZMQOption -> b -> IO () setIntOpt sock (ZMQOption o) i = onSocket "setIntOpt" sock $ \s ->- throwErrnoIfMinus1_ "setIntOpt" $ with i $ \ptr ->+ throwErrnoIfMinus1Retry_ "setIntOpt" $ with i $ \ptr -> c_zmq_setsockopt s (fromIntegral o) (castPtr ptr) (fromIntegral . sizeOf $ i) setStrOpt :: Socket a -> ZMQOption -> String -> IO () setStrOpt sock (ZMQOption o) str = onSocket "setStrOpt" sock $ \s ->- throwErrnoIfMinus1_ "setStrOpt" $ withCStringLen str $ \(cstr, len) ->+ throwErrnoIfMinus1Retry_ "setStrOpt" $ withCStringLen str $ \(cstr, len) -> c_zmq_setsockopt s (fromIntegral o) (castPtr cstr) (fromIntegral len)@@ -131,7 +143,7 @@ let i = 0 bracket (new i) free $ \iptr -> bracket (new (fromIntegral . sizeOf $ i :: CSize)) free $ \jptr -> do- throwErrnoIfMinus1_ "integralOpt" $+ throwErrnoIfMinus1Retry_ "getIntOpt" $ c_zmq_getsockopt s (fromIntegral o) (castPtr iptr) jptr peek iptr @@ -139,7 +151,7 @@ getStrOpt sock (ZMQOption o) = onSocket "getStrOpt" sock $ \s -> bracket (mallocBytes 255) free $ \bPtr -> bracket (new (255 :: CSize)) free $ \sPtr -> do- throwErrnoIfMinus1_ "getStrOpt" $+ throwErrnoIfMinus1Retry_ "getStrOpt" $ c_zmq_getsockopt s (fromIntegral o) (castPtr bPtr) sPtr peek sPtr >>= \len -> peekCStringLen (bPtr, fromIntegral len)
− test/display.hs
@@ -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-
− test/perf/local_lat.hs
@@ -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>"-
− test/perf/local_thr.hs
@@ -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>"-
− test/perf/remote_lat.hs
@@ -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>"-
− test/perf/remote_thr.hs
@@ -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>"-
− test/poll.hs
@@ -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>"-
− test/prompt.hs
@@ -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) []-
− test/queue.hs
@@ -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 ()
+ tests/System/ZMQ/Test/Properties.hs view
@@ -0,0 +1,120 @@+module System.ZMQ.Test.Properties where++import Control.Applicative+import Test.Framework (Test, testGroup)+import Test.Framework.Providers.QuickCheck2+import Test.QuickCheck+import Test.QuickCheck.Monadic++import Data.Int+import Data.Word+import Data.ByteString (ByteString)+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB+import qualified Data.ByteString.Char8 as CB++tests :: [Test]+tests = [+ testGroup "0MQ Socket Properties" [+ testProperty "get socket option (Push)" (prop_get_socket_option ZMQ.Push)+ , testProperty "get socket option (Pull)" (prop_get_socket_option ZMQ.Pull)+ , testProperty "get socket option (XRep)" (prop_get_socket_option ZMQ.XRep)+ , testProperty "get socket option (XReq)" (prop_get_socket_option ZMQ.XReq)+ , testProperty "get socket option (Rep)" (prop_get_socket_option ZMQ.Rep)+ , testProperty "get socket option (Req)" (prop_get_socket_option ZMQ.Req)+ , testProperty "get socket option (Sub)" (prop_get_socket_option ZMQ.Sub)+ , testProperty "get socket option (Pub)" (prop_get_socket_option ZMQ.Pub)+ , testProperty "get socket option (Pair)" (prop_get_socket_option ZMQ.Pair)+ , testProperty "get socket option (Down)" (prop_get_socket_option ZMQ.Down)+ , testProperty "get socket option (Up)" (prop_get_socket_option ZMQ.Up)++ , testProperty "set/get socket option (Push)" (prop_set_get_socket_option ZMQ.Push)+ , testProperty "set/get socket option (Pull)" (prop_set_get_socket_option ZMQ.Pull)+ , testProperty "set/get socket option (XRep)" (prop_set_get_socket_option ZMQ.XRep)+ , testProperty "set/get socket option (XReq)" (prop_set_get_socket_option ZMQ.XReq)+ , testProperty "set/get socket option (Rep)" (prop_set_get_socket_option ZMQ.Rep)+ , testProperty "set/get socket option (Req)" (prop_set_get_socket_option ZMQ.Req)+ , testProperty "set/get socket option (Sub)" (prop_set_get_socket_option ZMQ.Sub)+ , testProperty "set/get socket option (Pub)" (prop_set_get_socket_option ZMQ.Pub)+ , testProperty "set/get socket option (Pair)" (prop_set_get_socket_option ZMQ.Pair)+ , testProperty "set/get socket option (Down)" (prop_set_get_socket_option ZMQ.Down)+ , testProperty "set/get socket option (Up)" (prop_set_get_socket_option ZMQ.Up)++ , testProperty "(un-)subscribe" (prop_subscribe ZMQ.Sub)+ ]+ , testGroup "0MQ Messages" [+ testProperty "msg send == msg received (Req/Rep)" (prop_send_receive ZMQ.Req ZMQ.Rep)+ , testProperty "msg send == msg received (Push/Pull)" (prop_send_receive ZMQ.Push ZMQ.Pull)+ , testProperty "msg send == msg received (Pair/Pair)" (prop_send_receive ZMQ.Pair ZMQ.Pair)+ , testProperty "publish/subscribe (Pub/Sub)" (prop_pub_sub ZMQ.Pub ZMQ.Sub)+ ]+ ]++prop_get_socket_option :: ZMQ.SType a => a -> Property+prop_get_socket_option t = forAll readOnlyOptions canGetOption+ where+ canGetOption opt = monadicIO $ run $+ ZMQ.withContext 1 $ \c ->+ ZMQ.withSocket c t $ \s -> ZMQ.getOption s opt++prop_set_get_socket_option :: ZMQ.SType a => a -> ZMQ.SocketOption -> Property+prop_set_get_socket_option t opt = monadicIO $ do+ o <- run $ ZMQ.withContext 1 $ \c ->+ ZMQ.withSocket c t $ \s -> do+ ZMQ.setOption s opt+ ZMQ.getOption s opt+ assert (opt == o)++prop_subscribe :: (ZMQ.SubsType a, ZMQ.SType a) => a -> String -> Property+prop_subscribe t subs = monadicIO $ run $+ ZMQ.withContext 1 $ \c ->+ ZMQ.withSocket c t $ \s -> do+ ZMQ.subscribe s subs+ ZMQ.unsubscribe s subs++prop_send_receive :: (ZMQ.SType a, ZMQ.SType b) => a -> b -> ByteString -> Property+prop_send_receive a b msg = monadicIO $ do+ msg' <- run $ ZMQ.withContext 0 $ \c ->+ ZMQ.withSocket c a $ \sender ->+ ZMQ.withSocket c b $ \receiver -> do+ ZMQ.bind receiver "inproc://endpoint"+ ZMQ.connect sender "inproc://endpoint"+ ZMQ.send sender msg []+ ZMQ.receive receiver []+ assert (msg == msg')++prop_pub_sub :: (ZMQ.SType a, ZMQ.SubsType b, ZMQ.SType b) => a -> b -> ByteString -> Property+prop_pub_sub a b msg = monadicIO $ do+ msg' <- run $ ZMQ.withContext 0 $ \c ->+ ZMQ.withSocket c a $ \pub ->+ ZMQ.withSocket c b $ \sub -> do+ ZMQ.subscribe sub ""+ ZMQ.bind sub "inproc://endpoint"+ ZMQ.connect pub "inproc://endpoint"+ ZMQ.send pub msg []+ ZMQ.receive sub []+ assert (msg == msg')+instance Arbitrary ZMQ.SocketOption where+ arbitrary = oneof [+ ZMQ.Affinity . fromIntegral <$> (arbitrary :: Gen Word64)+ , ZMQ.Backlog . fromIntegral <$> (arbitrary :: Gen Int32)+ , ZMQ.Linger . fromIntegral <$> (arbitrary :: Gen Int32)+ , ZMQ.Rate . fromIntegral <$> (arbitrary :: Gen Word32)+ , ZMQ.ReceiveBuf . fromIntegral <$> (arbitrary :: Gen Word64)+ , ZMQ.ReconnectIVL . fromIntegral <$> (arbitrary :: Gen Int32) `suchThat` (>= 0)+ , ZMQ.ReconnectIVLMax . fromIntegral <$> (arbitrary :: Gen Int32) `suchThat` (>= 0)+ , ZMQ.RecoveryIVL . fromIntegral <$> (arbitrary :: Gen Word32)+ , ZMQ.RecoveryIVLMsec .fromIntegral <$> (arbitrary :: Gen Int32) `suchThat` (>= 0)+ , ZMQ.SendBuf . fromIntegral <$> (arbitrary :: Gen Word64)+ , ZMQ.HighWM . fromIntegral <$> (arbitrary :: Gen Word64)+ , ZMQ.McastLoop <$> (arbitrary :: Gen Bool)+ , ZMQ.Swap . fromIntegral <$> (arbitrary :: Gen Int64) `suchThat` (>= 0)+ , ZMQ.Identity . show <$> arbitrary `suchThat` (\s -> SB.length s > 0 && SB.length s < 255)+ ]++instance Arbitrary ByteString where+ arbitrary = CB.pack <$> arbitrary++readOnlyOptions :: Gen ZMQ.SocketOption+readOnlyOptions = elements [ZMQ.FD undefined, ZMQ.ReceiveMore undefined, ZMQ.Events undefined]+
+ tests/tests.hs view
@@ -0,0 +1,7 @@+{-# LANGUAGE CPP #-}+import Test.Framework (defaultMain)+import qualified System.ZMQ.Test.Properties as Properties++main :: IO ()+main = defaultMain Properties.tests+
zeromq-haskell.cabal view
@@ -1,7 +1,6 @@ name: zeromq-haskell-version: 0.7.1-synopsis: bindings to zeromq-description: Bindings to zeromq (http://zeromq.org)+version: 0.8.4+synopsis: Bindings to ZeroMQ 2.1.x category: System, FFI license: MIT license-file: LICENSE@@ -10,25 +9,61 @@ copyright: Copyright (c) 2011 zeromq-haskell authors homepage: http://github.com/twittner/zeromq-haskell/ stability: experimental-tested-With: GHC == 7.0.2-cabal-version: >= 1.6.0+tested-With: GHC == 7.4.1+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+ , tests/System/ZMQ/Test/*.hs+description:+ The 0MQ lightweight messaging kernel is a library which extends+ the standard socket interfaces with features traditionally provided+ by specialised messaging middleware products. 0MQ sockets provide+ an abstraction of asynchronous message queues, multiple messaging+ patterns, message filtering (subscriptions), seamless access to+ multiple transport protocols and more. + This library provides the Haskell language binding to 0MQ and+ supports 0MQ 2.1.x.+ library- exposed-modules: System.ZMQ- other-modules: System.ZMQ.Base, System.ZMQ.Internal- ghc-options: -Wall -O2- extensions: CPP,- ForeignFunctionInterface,- ExistentialQuantification- build-depends: base >= 3 && < 5,- containers,- bytestring- hs-source-dirs: src- includes: zmq.h- if os(freebsd)- extra-libraries: zmq, pthread- else- extra-libraries: zmq+ hs-source-dirs: src+ exposed-modules: System.ZMQ+ other-modules: System.ZMQ.Base+ , System.ZMQ.Internal+ includes: zmq.h+ ghc-options: -Wall -O2+ extensions: CPP+ , ForeignFunctionInterface+ , ExistentialQuantification+ build-depends: base >= 3 && < 5+ , containers+ , bytestring++ if os(freebsd)+ 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-quickcheck2 >= 0.2+ , QuickCheck >= 2.4+ ghc-options: -Wall -threaded -rtsopts+ cpp-options: -DZMQ2++source-repository head+ type: git+ location: https://github.com/twittner/zeromq-haskell+