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.1.3 - Deprecated 'Xreq', 'XRep' in favour of 'Dealer' and 'Router' + as in libzmq. Fixes to compile and run with GHC 7.4.1. + Version 0.1.2 - Add 'sendMulti' and 'receiveMulti'. Rename 'SndMore' to 'SendMore'. diff --git a/examples/perf/local_thr.hs b/examples/perf/local_thr.hs --- a/examples/perf/local_thr.hs +++ b/examples/perf/local_thr.hs @@ -35,6 +35,7 @@ receive s sz loop s (c - 1) sz + printStat :: UTCTime -> UTCTime -> Int -> Int -> IO () printStat start end size count = do let elapsed = fromRational . toRational $ diffUTCTime end start :: Double through = fromIntegral count / elapsed diff --git a/src/System/ZMQ3.hs b/src/System/ZMQ3.hs --- a/src/System/ZMQ3.hs +++ b/src/System/ZMQ3.hs @@ -18,8 +18,9 @@ -- -- * 'System.ZMQ.Up' and 'System.ZMQ.Down' no longer exist. -- --- * Some 0MQ tutorials mention 'ZMQ_DEALER' and 'ZMQ_ROUTER'. These --- are aliases of 'XRep' and 'XReq'. +-- * 'XReq' is renamed to 'Dealer' and 'XRep' is renamed to 'Router' +-- (in accordance with libzmq). 'XReq' and 'XRep' are available as +-- deprecated aliases. -- -- * Renamed type-classes: -- @'SType' -\> 'SocketType'@, @'SubsType' -\> 'Subscriber'@. @@ -91,8 +92,10 @@ , XSub(..) , Req(..) , Rep(..) - , XReq(..) - , XRep(..) + , Dealer(..) + , Router(..) + , XReq + , XRep , Pull(..) , Push(..) @@ -182,7 +185,6 @@ import Foreign.C.Types (CInt) 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.ZMQ3.Base import qualified System.ZMQ3.Base as B @@ -223,31 +225,39 @@ -- | 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. --- /Compatible peer sockets/: 'Rep', 'Xrep'. +-- /Compatible peer sockets/: 'Rep', 'Router'. data Req = Req -- | Socket to receive requests and send replies. This socket type -- allows only an alternated sequence of receive's and send's. Each -- send is routed to the peer that issued the last received request. --- /Compatible peer sockets/: 'Req', 'XReq'. +-- /Compatible peer sockets/: 'Req', 'Dealer'. data Rep = Rep --- | Special socket type to be used in request/reply middleboxes --- such as zmq_queue(7). Requests forwarded using this socket type --- should be tagged by a proper prefix identifying the original requester. --- 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 +-- | Each message sent is round-robined among all connected peers, +-- and each message received is fair-queued from all connected peers. +-- /Compatible peer sockets/: 'Router', 'Req', 'Rep'. +data Dealer = 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 +-- | /Deprecated Alias/ +type XReq = Dealer +{-# DEPRECATED XReq "Use Dealer" #-} +-- | When receiving messages a Router socket shall prepend a message +-- part containing the identity of the originating peer to +-- the message before passing it to the application. Messages +-- received are fair-queued from among all connected peers. When +-- sending messages a Router socket shall remove the first part of +-- the message and use it to determine the identity of the peer the +-- message shall be routed to. If the peer does not exist anymore +-- the message shall be silently discarded. +-- /Compatible peer sockets/: 'Dealer', 'Req', 'Rep'. +data Router = Router + +-- | /Deprecated Alias/ +type XRep = Router +{-# DEPRECATED XRep "Use 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 @@ -306,13 +316,13 @@ instance Sender Rep instance Receiver Rep -instance SocketType XReq where zmqSocketType = const xrequest -instance Sender XReq -instance Receiver XReq +instance SocketType Dealer where zmqSocketType = const dealer +instance Sender Dealer +instance Receiver Dealer -instance SocketType XRep where zmqSocketType = const xresponse -instance Sender XRep -instance Receiver XRep +instance SocketType Router where zmqSocketType = const router +instance Sender Router +instance Receiver Router instance SocketType Pull where zmqSocketType = const pull instance Receiver Pull @@ -373,12 +383,7 @@ socket :: SocketType a => Context -> a -> IO (Socket a) socket (Context c) t = do let zt = typeVal . zmqSocketType $ t - s <- throwIfNull "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 + throwIfNull "socket" (c_zmq_socket c zt) >>= mkSocket -- | Close a 0MQ socket. 'withSocket' provides automatic socket closing and may -- be safer to use. @@ -610,7 +615,7 @@ size <- c_zmq_msg_size (msgPtr m) SB.packCStringLen (data_ptr, fromIntegral size) --- Receive a multi-part message. +-- | Receive a multi-part message. -- This function collects all message parts send via 'sendMulti'. receiveMulti :: Receiver a => Socket a -> IO [SB.ByteString] receiveMulti sock = recvall [] diff --git a/src/System/ZMQ3/Base.hsc b/src/System/ZMQ3/Base.hsc --- a/src/System/ZMQ3/Base.hsc +++ b/src/System/ZMQ3/Base.hsc @@ -68,8 +68,8 @@ , 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 } @@ -151,6 +151,22 @@ foreign import ccall unsafe "zmq.h zmq_msg_size" c_zmq_msg_size :: ZMQMsgPtr -> IO CSize +#if 0 +foreign import ccall unsafe "zmq.h zmq_msg_get" + c_zmq_msg_get :: ZMQMsgPtr + -> CInt -- option + -> Ptr () -- option value + -> Ptr CSize -- option value size ptr + -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_msg_set" + c_zmq_msg_set :: ZMQMsgPtr + -> CInt -- option + -> Ptr () -- option value + -> Ptr CSize -- option value size ptr + -> IO CInt +#endif + -- socket foreign import ccall unsafe "zmq.h zmq_socket" @@ -185,12 +201,6 @@ foreign import ccall unsafe "zmq.h zmq_recvmsg" c_zmq_recvmsg :: 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 -- error messages foreign import ccall unsafe "zmq.h zmq_strerror" diff --git a/src/System/ZMQ3/Error.hs b/src/System/ZMQ3/Error.hs --- a/src/System/ZMQ3/Error.hs +++ b/src/System/ZMQ3/Error.hs @@ -56,19 +56,19 @@ throwIfRetry_ :: (a -> Bool) -> String -> IO a -> IO () throwIfRetry_ p src act = void $ throwIfRetry p src act -throwIfMinus1 :: Num a => String -> IO a -> IO a +throwIfMinus1 :: (Eq a, Num a) => String -> IO a -> IO a throwIfMinus1 = throwIf (== -1) -throwIfMinus1_ :: Num a => String -> IO a -> IO () +throwIfMinus1_ :: (Eq a, Num a) => String -> IO a -> IO () throwIfMinus1_ = throwIf_ (== -1) throwIfNull :: String -> IO (Ptr a) -> IO (Ptr a) throwIfNull = throwIf (== nullPtr) -throwIfMinus1Retry :: Num a => String -> IO a -> IO a +throwIfMinus1Retry :: (Eq a, Num a) => String -> IO a -> IO a throwIfMinus1Retry = throwIfRetry (== -1) -throwIfMinus1Retry_ :: Num a => String -> IO a -> IO () +throwIfMinus1Retry_ :: (Eq a, Num a) => String -> IO a -> IO () throwIfMinus1Retry_ = throwIfRetry_ (== -1) throwIfRetryMayBlock :: (a -> Bool) -> String -> IO a -> IO b -> IO a @@ -83,10 +83,10 @@ throwIfRetryMayBlock_ :: (a -> Bool) -> String -> IO a -> IO b -> IO () throwIfRetryMayBlock_ p src f on_block = void $ throwIfRetryMayBlock p src f on_block -throwIfMinus1RetryMayBlock :: Num a => String -> IO a -> IO b -> IO a +throwIfMinus1RetryMayBlock :: (Eq a, Num a) => String -> IO a -> IO b -> IO a throwIfMinus1RetryMayBlock = throwIfRetryMayBlock (== -1) -throwIfMinus1RetryMayBlock_ :: Num a => String -> IO a -> IO b -> IO () +throwIfMinus1RetryMayBlock_ :: (Eq a, Num a) => String -> IO a -> IO b -> IO () throwIfMinus1RetryMayBlock_ = throwIfRetryMayBlock_ (== -1) zmqErrnoMessage :: CInt -> IO String diff --git a/src/System/ZMQ3/Internal.hs b/src/System/ZMQ3/Internal.hs --- a/src/System/ZMQ3/Internal.hs +++ b/src/System/ZMQ3/Internal.hs @@ -15,7 +15,6 @@ , setStrOpt , getIntOpt , getStrOpt - , getIntMsgOpt , getInt32Option , setInt32OptFromRestricted @@ -27,9 +26,9 @@ ) where import Control.Applicative -import Control.Monad (foldM_) +import Control.Monad (foldM_, when) import Control.Exception -import Data.IORef (IORef) +import Data.IORef (IORef, mkWeakIORef, readIORef) import Foreign import Foreign.C.String @@ -71,7 +70,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 @@ -139,14 +145,6 @@ throwIfMinus1Retry_ "getStrOpt" $ c_zmq_getsockopt s (fromIntegral o) (castPtr bPtr) sPtr peek sPtr >>= \len -> peekCStringLen (bPtr, fromIntegral len) - -getIntMsgOpt :: (Storable a, Integral a) => Message -> ZMQMsgOption -> a -> IO a -getIntMsgOpt (Message m) (ZMQMsgOption o) i = do - bracket (new i) free $ \iptr -> - bracket (new (fromIntegral . sizeOf $ i :: CSize)) free $ \jptr -> do - throwIfMinus1Retry_ "getIntMsgOpt" $ - c_zmq_getmsgopt m (fromIntegral o) (castPtr iptr) jptr - peek iptr getInt32Option :: ZMQOption -> Socket a -> IO Int getInt32Option o s = fromIntegral <$> getIntOpt s o (0 :: CInt) diff --git a/tests/System/ZMQ3/Test/Properties.hs b/tests/System/ZMQ3/Test/Properties.hs --- a/tests/System/ZMQ3/Test/Properties.hs +++ b/tests/System/ZMQ3/Test/Properties.hs @@ -22,29 +22,29 @@ tests :: [Test] tests = [ testGroup "0MQ Socket Properties" [ - testProperty "get socket option (Pair)" (prop_get_socket_option Pair) - , testProperty "get socket option (Pub)" (prop_get_socket_option Pub) - , testProperty "get socket option (Sub)" (prop_get_socket_option Sub) - , testProperty "get socket option (XPub)" (prop_get_socket_option XPub) - , testProperty "get socket option (XSub)" (prop_get_socket_option XSub) - , testProperty "get socket option (Req)" (prop_get_socket_option Req) - , testProperty "get socket option (Rep)" (prop_get_socket_option Rep) - , testProperty "get socket option (XReq)" (prop_get_socket_option XReq) - , testProperty "get socket option (XRep)" (prop_get_socket_option XRep) - , testProperty "get socket option (Pull)" (prop_get_socket_option Pull) - , testProperty "get socket option (Push)" (prop_get_socket_option Push) - , testProperty "set/get socket option (Pair)" (prop_set_get_socket_option Pair) - , testProperty "set/get socket option (Pub)" (prop_set_get_socket_option Pub) - , testProperty "set/get socket option (Sub)" (prop_set_get_socket_option Sub) - , testProperty "set/get socket option (XPub)" (prop_set_get_socket_option XPub) - , testProperty "set/get socket option (XSub)" (prop_set_get_socket_option XSub) - , testProperty "set/get socket option (Req)" (prop_set_get_socket_option Req) - , testProperty "set/get socket option (Rep)" (prop_set_get_socket_option Rep) - , testProperty "set/get socket option (XReq)" (prop_set_get_socket_option XReq) - , testProperty "set/get socket option (XRep)" (prop_set_get_socket_option XRep) - , testProperty "set/get socket option (Pull)" (prop_set_get_socket_option Pull) - , testProperty "set/get socket option (Push)" (prop_set_get_socket_option Push) - , testProperty "(un-)subscribe" (prop_subscribe Sub) + testProperty "get socket option (Pair)" (prop_get_socket_option Pair) + , testProperty "get socket option (Pub)" (prop_get_socket_option Pub) + , testProperty "get socket option (Sub)" (prop_get_socket_option Sub) + , testProperty "get socket option (XPub)" (prop_get_socket_option XPub) + , testProperty "get socket option (XSub)" (prop_get_socket_option XSub) + , testProperty "get socket option (Req)" (prop_get_socket_option Req) + , testProperty "get socket option (Rep)" (prop_get_socket_option Rep) + , testProperty "get socket option (Dealer)" (prop_get_socket_option Dealer) + , testProperty "get socket option (Router)" (prop_get_socket_option Router) + , testProperty "get socket option (Pull)" (prop_get_socket_option Pull) + , testProperty "get socket option (Push)" (prop_get_socket_option Push) + , testProperty "set/get socket option (Pair)" (prop_set_get_socket_option Pair) + , testProperty "set/get socket option (Pub)" (prop_set_get_socket_option Pub) + , testProperty "set/get socket option (Sub)" (prop_set_get_socket_option Sub) + , testProperty "set/get socket option (XPub)" (prop_set_get_socket_option XPub) + , testProperty "set/get socket option (XSub)" (prop_set_get_socket_option XSub) + , testProperty "set/get socket option (Req)" (prop_set_get_socket_option Req) + , testProperty "set/get socket option (Rep)" (prop_set_get_socket_option Rep) + , testProperty "set/get socket option (Dealer)" (prop_set_get_socket_option Dealer) + , testProperty "set/get socket option (Router)" (prop_set_get_socket_option Router) + , testProperty "set/get socket option (Pull)" (prop_set_get_socket_option Pull) + , testProperty "set/get socket option (Push)" (prop_set_get_socket_option Push) + , testProperty "(un-)subscribe" (prop_subscribe Sub) ] , testGroup "0MQ Messages" [ testProperty "msg send == msg received (Req/Rep)" (prop_send_receive Req Rep) diff --git a/zeromq3-haskell.cabal b/zeromq3-haskell.cabal --- a/zeromq3-haskell.cabal +++ b/zeromq3-haskell.cabal @@ -1,5 +1,5 @@ name: zeromq3-haskell -version: 0.1.2 +version: 0.1.3 synopsis: Bindings to ZeroMQ 3.x category: System, FFI license: MIT @@ -9,7 +9,7 @@ copyright: Copyright (c) 2010 - 2012 zeromq-haskell authors homepage: http://github.com/twittner/zeromq-haskell/ stability: experimental -tested-With: GHC == 7.0.3 +tested-With: GHC == 7.4.1 cabal-version: >= 1.8 build-type: Simple extra-source-files: README.md