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.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 ----- diff --git a/src/System/ZMQ.hs b/src/System/ZMQ.hs --- a/src/System/ZMQ.hs +++ b/src/System/ZMQ.hs @@ -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 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 @@ -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 +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 diff --git a/src/System/ZMQ/Internal.hs b/src/System/ZMQ/Internal.hs --- a/src/System/ZMQ/Internal.hs +++ b/src/System/ZMQ/Internal.hs @@ -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 diff --git a/zeromq-haskell.cabal b/zeromq-haskell.cabal --- a/zeromq-haskell.cabal +++ b/zeromq-haskell.cabal @@ -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