zeromq-haskell 0.7.1 → 0.8

raw patch · 5 files changed

+400/−267 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• README.md +7/−0
• src/System/ZMQ.hs +224/−201
• src/System/ZMQ/Base.hsc +101/−45
• src/System/ZMQ/Internal.hs +24/−3
• zeromq-haskell.cabal +44/−18

@@ -5,6 +5,9 @@  This software currently has *beta* status, i.e. it had seen limited testing. +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@@ -23,6 +26,10 @@  As usual for Haskell packages this software is installed best via Cabal (http://www.haskell.org/cabal). In addition to GHC it depends on 0MQ of course.+Please note that by default zeromq-haskell attempts to compile against 0MQ 2.x.+If instead you want to use 0MQ 3.x, then provide the following flags to Cabal:++    --flags="-zmq2 zmq3"  Notes -----

@@ -1,7 +1,7 @@-{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE CPP, 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,71 @@ -- -- 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(..)+  , XPub(..)+  , XSub(..)+  , Req(..)+  , Rep(..)+  , XReq(..)+  , XRep(..)+  , Pull(..)+  , Push(..) -    withContext,-    withSocket,-    setOption,-    getOption,-    System.ZMQ.subscribe,-    System.ZMQ.unsubscribe,-    bind,-    connect,-    send,-    send',-    receive,-    moreToReceive,-    poll,-    device,+#ifdef ZMQ2+  , Up(..)+  , Down(..)+#endif +  , withContext+  , withSocket+  , setOption+  , getOption++#ifdef ZMQ3+  , getMsgOption+#endif++  , System.ZMQ.subscribe+  , System.ZMQ.unsubscribe+  , bind+  , connect+  , send+  , send'+  , receive+  , moreToReceive+  , poll+  , System.ZMQ.zmqVersion+     -- * Low-level functions-    init,-    term,-    socket,-    close,+  , init+  , term+  , socket+  , close +#ifdef ZMQ2+  , Device(..)+  , device+#endif+ ) where  import Prelude hiding (init)@@ -66,10 +81,6 @@ 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.C.Error import Foreign.C.String@@ -78,11 +89,13 @@ 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 @@ -108,6 +121,23 @@ instance SType Sub where     zmqSocketType = const sub +-- | 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+-- subscription body.+-- /Compatible peer sockets/: 'Sub', 'XSub'.+data XPub = XPub+instance SType XPub where+    zmqSocketType = const xpub++-- | Same as 'Sub' except that you subscribe by sending subscription+-- messages to the socket. Subscription message is a byte 1 (for subscriptions)+-- or byte 0 (for unsubscriptions) followed by the subscription body.+-- /Compatible peer sockets/: 'Pub', 'XPub'.+data XSub = XSub+instance SType XSub where+    zmqSocketType = const xsub+ -- | Socket to send requests and receive replies. Requests are -- load-balanced among all the peers. This socket type allows only an -- alternated sequence of send's and recv's.@@ -144,7 +174,7 @@ instance SType XRep where     zmqSocketType = const xresponse --- | A socket of type ZMQ_PULL is used by a pipeline node to receive+-- | 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 +182,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@@ -166,6 +196,7 @@ instance SType Push where     zmqSocketType = const push +#ifdef ZMQ2 {-# DEPRECATED Up "Use Pull instead." #-} -- | Socket to receive messages from up the stream. Messages are -- fair-queued from among all the connected peers. Send function is not@@ -181,96 +212,52 @@ data Down = Down instance SType Down where     zmqSocketType = const downstream---- Subscribable:+#endif +-- | Subscribable. class SubsType a+ instance SubsType Sub+instance SubsType XSub --- | 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+    Affinity        Word64    -- ^ ZMQ_AFFINITY+  | Backlog         Int       -- ^ ZMQ_BACKLOG+  | Events          PollEvent -- ^ ZMQ_EVENTS+  | FD              Int       -- ^ ZMQ_FD+  | Identity        String    -- ^ ZMQ_IDENTITY+  | Linger          Int       -- ^ 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+  | RecoveryIVL     Int64     -- ^ ZMQ_RECOVERY_IVL+  | SendBuf         Word64    -- ^ ZMQ_SNDBUF+#ifdef ZMQ2+  | HighWM          Word64    -- ^ ZMQ_HWM+  | McastLoop       Bool      -- ^ ZMQ_MCAST_LOOP+  | RecoveryIVLMsec Int64     -- ^ ZMQ_RECOVERY_IVL_MSEC+  | Swap            Int64     -- ^ ZMQ_SWAP+#endif+#ifdef ZMQ3+  | IPv4Only        Bool      -- ^ ZMQ_IPV4ONLY+  | MaxMsgSize      Int64     -- ^ ZMQ_MAXMSGSIZE+  | McastHops       Int       -- ^ ZMQ_MULTICAST_HOPS+  | ReceiveHighWM   Int       -- ^ ZMQ_RCVHWM+  | ReceiveTimeout  Int       -- ^ ZMQ_RCVTIMEO+  | SendHighWM      Int       -- ^ ZMQ_SNDHWM+  | SendTimeout     Int       -- ^ ZMQ_SNDTIMEO+#endif   deriving (Eq, Ord, Show) +#ifdef ZMQ3+data MessageOption = MoreMsgParts CInt -- ^ ZMQ_MORE+  deriving (Eq, Ord, Show)+#endif+ -- | The events to wait for in poll (cf. man zmq_poll) data PollEvent =     In     -- ^ ZMQ_POLLIN (incoming messages)@@ -287,13 +274,82 @@     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+#ifdef ZMQ2+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+#endif+#ifdef ZMQ3+setOption s (IPv4Only o)        = setBoolOpt s ipv4Only o+setOption s (MaxMsgSize o)      = setIntOpt s maxMessageSize o+setOption s (McastHops o)       = setIntOpt s mcastHops o+setOption s (ReceiveHighWM o)   = setIntOpt s receiveHighWM o+setOption s (ReceiveTimeout o)  = setIntOpt s receiveTimeout o+setOption s (SendHighWM o)      = setIntOpt s sendHighWM o+setOption s (SendTimeout o)     = setIntOpt s sendTimeout o+#endif +-- | 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+#ifdef ZMQ2+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+#endif+#ifdef ZMQ3+getOption s (IPv4Only _)        = IPv4Only <$> getBoolOpt s ipv4Only+getOption s (MaxMsgSize _)      = MaxMsgSize <$> getIntOpt s maxMessageSize+getOption s (McastHops _)       = McastHops <$> getIntOpt s mcastHops+getOption s (ReceiveHighWM _)   = ReceiveHighWM <$> getIntOpt s receiveHighWM+getOption s (ReceiveTimeout _)  = ReceiveTimeout <$> getIntOpt s receiveTimeout+getOption s (SendHighWM _)      = SendHighWM <$> getIntOpt s sendHighWM+getOption s (SendTimeout _)     = SendTimeout <$> getIntOpt s sendTimeout++getMsgOption :: Message -> MessageOption -> IO MessageOption+getMsgOption m (MoreMsgParts _) = MoreMsgParts <$> getIntMsgOpt m more+#endif++zmqVersion :: (Int, Int, Int)+zmqVersion = B.zmqVersion+ -- | Initialize a 0MQ context (cf. zmq_init for details).  You should -- normally prefer to use 'with' instead. init :: Size -> IO Context@@ -340,50 +396,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 +441,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 +468,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,27 +479,38 @@     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 +#ifdef ZMQ2+-- | 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). -- -- Please note that this call never returns.@@ -502,4 +525,4 @@     fromDevice Streamer  = fromIntegral . deviceType $ deviceStreamer     fromDevice Forwarder = fromIntegral . deviceType $ deviceForwarder     fromDevice Queue     = fromIntegral . deviceType $ deviceQueue-+#endif

@@ -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,20 +10,31 @@  module System.ZMQ.Base where -import Control.Applicative import Foreign import Foreign.C.Types import Foreign.C.String+import Control.Applicative  #include <zmq.h> +zmqVersion :: (Int, Int, Int)+zmqVersion = ( #const ZMQ_VERSION_MAJOR+             , #const ZMQ_VERSION_MINOR+             , #const ZMQ_VERSION_PATCH )+ newtype ZMQMsg = ZMQMsg { content :: Ptr () }  instance Storable ZMQMsg where     alignment _        = #{alignment zmq_msg_t}     sizeOf    _        = #{size zmq_msg_t}+#ifdef ZMQ2     peek p             = ZMQMsg <$> #{peek zmq_msg_t, content} p     poke p (ZMQMsg c)  = #{poke zmq_msg_t, content} p c+#endif+#ifdef ZMQ3+    peek p             = ZMQMsg <$> #{peek zmq_msg_t, _} p+    poke p (ZMQMsg c)  = #{poke zmq_msg_t, _} p c+#endif  data ZMQPoll = ZMQPoll     { pSocket  :: ZMQSocket@@ -56,51 +67,81 @@  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+  , pull       = ZMQ_PULL+  , push       = ZMQ_PUSH+#ifdef ZMQ2+  , upstream   = ZMQ_UPSTREAM+  , downstream = ZMQ_DOWNSTREAM+#endif }  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+#ifdef ZMQ2+  , highWM          = ZMQ_HWM+  , mcastLoop       = ZMQ_MCAST_LOOP+  , recoveryIVLMsec = ZMQ_RECOVERY_IVL_MSEC+  , swap            = ZMQ_SWAP+#endif+#ifdef ZMQ3+  , ipv4Only        = ZMQ_IPV4ONLY+  , maxMessageSize  = ZMQ_MAXMSGSIZE+  , mcastHops       = ZMQ_MULTICAST_HOPS+  , receiveHighWM   = ZMQ_RCVHWM+  , receiveTimeout  = ZMQ_RCVTIMEO+  , sendHighWM      = ZMQ_SNDHWM+  , sendTimeout     = ZMQ_SNDTIMEO+#endif } +#ifdef ZMQ3+newtype ZMQMsgOption = ZMQMsgOption { msgOptVal :: CInt } deriving (Eq, Ord)++#{enum ZMQMsgOption, ZMQMsgOption+  , more = ZMQ_MORE+}+#endif+ newtype ZMQFlag = ZMQFlag { flagVal :: CInt } deriving (Eq, Ord) -#{enum ZMQFlag, ZMQFlag,-    noBlock = ZMQ_NOBLOCK+#ifdef ZMQ2+#{enum ZMQFlag, ZMQFlag+  , noBlock = ZMQ_NOBLOCK   , sndMore = ZMQ_SNDMORE }+#endif+#ifdef ZMQ3+#{enum ZMQFlag, ZMQFlag+  , noBlock = ZMQ_DONTWAIT+  , sndMore = ZMQ_SNDMORE+}+#endif  newtype ZMQPollEvent = ZMQPollEvent { pollVal :: CShort } deriving (Eq, Ord) @@ -111,6 +152,8 @@     pollInOut = ZMQ_POLLIN | ZMQ_POLLOUT } ++#ifdef ZMQ2 newtype ZMQDevice = ZMQDevice { deviceType :: CInt } deriving (Eq, Ord)  #{enum ZMQDevice, ZMQDevice,@@ -119,6 +162,10 @@     deviceQueue     = ZMQ_QUEUE } +foreign import ccall safe "zmq.h zmq_device"+    c_zmq_device :: CInt -> ZMQSocket -> ZMQSocket -> IO CInt+#endif+ -- general initialization  foreign import ccall unsafe "zmq.h zmq_init"@@ -172,22 +219,31 @@ foreign import ccall unsafe "zmq.h zmq_connect"     c_zmq_connect :: ZMQSocket -> CString -> IO CInt ++#ifdef ZMQ2 foreign import ccall unsafe "zmq.h zmq_send"     c_zmq_send :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt -foreign import ccall safe "zmq.h zmq_recv"+foreign import ccall unsafe "zmq.h zmq_recv"     c_zmq_recv :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt+#endif+#ifdef ZMQ3+foreign import ccall unsafe "zmq.h zmq_sendmsg"+    c_zmq_send :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt -foreign import ccall unsafe "zmq.h zmq_recv"-    c_zmq_recv_unsafe :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt+foreign import ccall unsafe "zmq.h zmq_recvmsg"+    c_zmq_recv :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt +foreign import ccall unsafe "zmq.h zmq_getmsgopt"+    c_zmq_getmsgopt :: ZMQMsgPtr+                    -> CInt      -- option+                    -> Ptr ()    -- option value+                    -> Ptr CSize -- option value size ptr+                    -> IO CInt+#endif+ -- 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 

@@ -11,11 +11,17 @@     , messageClose     , messageInit     , messageInitSize+    , setBoolOpt     , setIntOpt     , setStrOpt     , getBoolOpt     , getIntOpt     , getStrOpt++#ifdef ZMQ3+    , getIntMsgOpt+#endif+     , toZMQFlag     , combine     , mkSocket@@ -47,7 +53,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 +62,7 @@  -- | A 0MQ Socket. data Socket a = Socket {-      _socket :: ZMQSocket+      _socket   :: ZMQSocket     , _sockLive :: IORef Bool     } @@ -109,6 +115,10 @@         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 ->@@ -131,7 +141,7 @@     let i = 0     bracket (new i) free $ \iptr ->         bracket (new (fromIntegral . sizeOf $ i :: CSize)) free $ \jptr -> do-            throwErrnoIfMinus1_ "integralOpt" $+            throwErrnoIfMinus1_ "getIntOpt" $                 c_zmq_getsockopt s (fromIntegral o) (castPtr iptr) jptr             peek iptr @@ -142,6 +152,17 @@         throwErrnoIfMinus1_ "getStrOpt" $             c_zmq_getsockopt s (fromIntegral o) (castPtr bPtr) sPtr         peek sPtr >>= \len -> peekCStringLen (bPtr, fromIntegral len)++#ifdef ZMQ3+getIntMsgOpt :: (Storable a, Integral a) => Message -> ZMQMsgOption -> IO a+getIntMsgOpt (Message m) (ZMQMsgOption o) = do+    let i = 0+    bracket (new i) free $ \iptr ->+        bracket (new (fromIntegral . sizeOf $ i :: CSize)) free $ \jptr -> do+            throwErrnoIfMinus1_ "getIntMsgOpt" $+                c_zmq_getmsgopt m (fromIntegral o) (castPtr iptr) jptr+            peek iptr+#endif  toZMQFlag :: Flag -> ZMQFlag toZMQFlag NoBlock = noBlock

@@ -1,7 +1,6 @@ name:               zeromq-haskell-version:            0.7.1+version:            0.8 synopsis:           bindings to zeromq-description:        Bindings to zeromq (http://zeromq.org) category:           System, FFI license:            MIT license-file:       LICENSE@@ -10,25 +9,52 @@ copyright:          Copyright (c) 2011 zeromq-haskell authors homepage:           http://github.com/twittner/zeromq-haskell/ stability:          experimental-tested-With:        GHC == 7.0.2+tested-With:        GHC == 7.0.3 cabal-version:      >= 1.6.0 build-type:         Simple extra-source-files: README.md, AUTHORS, test/*.hs, test/perf/*.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+    support 0MQ 2.x as well as 3.x (use -fzmq2 or -fzmq3 to select+    between the two).++flag zmq2+    description: Compile with 0MQ 2.x+    default: True++flag zmq3+    description: Compile with 0MQ 3.x+    default: False+ 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 flag(zmq2)+        cpp-options:      -DZMQ2++    if flag(zmq3)+        cpp-options:      -DZMQ3++    if os(freebsd)+        extra-libraries:  zmq, pthread+    else+        extra-libraries:  zmq