zeromq3-haskell 0.1.2 → 0.1.3
raw patch · 8 files changed
+102/−85 lines, 8 files
Files
- README.md +3/−0
- examples/perf/local_thr.hs +1/−0
- src/System/ZMQ3.hs +39/−34
- src/System/ZMQ3/Base.hsc +18/−8
- src/System/ZMQ3/Error.hs +6/−6
- src/System/ZMQ3/Internal.hs +10/−12
- tests/System/ZMQ3/Test/Properties.hs +23/−23
- zeromq3-haskell.cabal +2/−2
README.md view
@@ -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'.
examples/perf/local_thr.hs view
@@ -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
src/System/ZMQ3.hs view
@@ -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 []
src/System/ZMQ3/Base.hsc view
@@ -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"
src/System/ZMQ3/Error.hs view
@@ -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
src/System/ZMQ3/Internal.hs view
@@ -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)
tests/System/ZMQ3/Test/Properties.hs view
@@ -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)
zeromq3-haskell.cabal view
@@ -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