zeromq-haskell 0.4.2 → 0.5.0

raw patch · 13 files changed

+300/−214 lines, 13 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• AUTHORS +2/−0
• README.md +10/−28
• src/System/ZMQ.hs +86/−134
• src/System/ZMQ/Base.hsc +3/−0
• src/System/ZMQ/Internal.hs +151/−0
• test/display.hs +5/−5
• test/perf/local_lat.hs +5/−6
• test/perf/local_thr.hs +10/−11
• test/perf/remote_lat.hs +8/−9
• test/perf/remote_thr.hs +6/−7
• test/poll.hs +6/−6
• test/prompt.hs +4/−4
• zeromq-haskell.cabal +4/−4

@@ -2,4 +2,6 @@ David Himmelstrup   added send' Nicolas Trangez     added support for zmq_device and "queue" test app Ville Tirronen      added support for ZMG_SNDMORE+Jeremy Fitzhardinge integrated with GHC's I/O manager+Bryan O'Sullivan    added resource wrappers, socket finalizer 

@@ -3,11 +3,10 @@ Current status -------------- -Version 0.4.1 - This software currently has *beta* status, i.e. it had+Version 0.5.0 - This software currently has *beta* status, i.e. it had seen limited testing. Changes to its API may still happen. -This software was developed and tested on Linux 2.6.35 with GHC-6.12.3-using zeromq-2.0.9.+This software requires zeromq version 2.1.0.  Installation ------------@@ -15,32 +14,15 @@ 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. -Usage+Notes ----- -The API mostly follows 0MQ's. Public functions are:--- `init`-- `term`-- `socket`-- `close`-- `setOption`-- `getOption`-- `subscribe`-- `unsubscribe`-- `bind`-- `connect`-- `send`-- `send'`-- `receive`-- `moreToReceive`-- `poll`--One difference to 0MQ's API is that sockets are parameterized types, i.e. there-is not one socket type but when creating a socket the desired socket type has-to be specified, e.g. `Pair` and the resulting socket is of type `Socket Pair`.+zeromq-haskell mostly follows 0MQ's API. One difference though is that sockets+are parameterized types, i.e. there is not one socket type but when creating a+socket the desired socket type has to be specified, e.g. `Pair` and the+resulting socket is of type `Socket Pair`. This additional type information is used to ensure that only options applicable-to the socket type can be set, hence `ZMQ_SUBSCRIBE` and `ZMQ_UNSUBSCRIBE` which +to the socket type can be set, hence `ZMQ_SUBSCRIBE` and `ZMQ_UNSUBSCRIBE` which only apply to `ZMQ_SUB` sockets have their own functions (`subscribe` and `unsubscribe`) which can only be used with sockets of type `Socket Sub`. @@ -58,6 +40,6 @@ ----  If you find any bugs or other shortcomings I would greatly appreciate a bug-report, preferably via http://github.com/twittner/zeromq-haskell/issues or e-mail-to toralf.wittner@gmail.com+report, preferably via http://github.com/twittner/zeromq-haskell/issues or+e-mail to toralf.wittner@gmail.com 

@@ -20,6 +20,7 @@     Flag(..),     SocketOption(..),     Poll(..),+    Timeout,     PollEvent(..),     Device(..), @@ -37,8 +38,7 @@     Up(..),     Down(..), -    init,-    term,+    with,     socket,     close,     setOption,@@ -52,38 +52,34 @@     receive,     moreToReceive,     poll,-    device+    device, +    -- * Low-level functions+    init,+    term+ ) where  import Prelude hiding (init) import Control.Applicative-import Control.Monad (foldM_) import Control.Exception+import Control.Monad (unless, when)+import Data.IORef (readIORef, writeIORef) import Data.Int import Data.Maybe import System.ZMQ.Base import qualified System.ZMQ.Base as B-import Foreign+import System.ZMQ.Internal+import Foreign hiding (with) import Foreign.C.Error import Foreign.C.String-import Foreign.C.Types (CInt, CShort, CSize)+import Foreign.C.Types (CInt, CShort) import qualified Data.ByteString as SB import qualified Data.ByteString.Lazy as LB-import qualified Data.ByteString.Unsafe as UB+import System.Mem.Weak (addFinalizer) import System.Posix.Types (Fd(..)) --- | A 0MQ context representation.-newtype Context = Context { ctx :: ZMQCtx }---- | A 0MQ Socket.-newtype Socket a = Socket { sock :: ZMQSocket }---- A 0MQ Message representation.-newtype Message = Message { msgPtr :: ZMQMsgPtr }--type Timeout = Int64-type Size    = Word+import GHC.Conc (threadWaitRead, threadWaitWrite)  -- Socket types: @@ -275,15 +271,6 @@   | Backlog         CInt   -- ^ ZMQ_BACKLOG   deriving (Eq, Ord, Show) --- | Flags to apply on send operations (cf. man zmq_send)------ [@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-          | SndMore -- ^ ZMQ_SNDMORE-  deriving (Eq, Ord, Show)- -- | The events to wait for in poll (cf. man zmq_poll) data PollEvent =     In     -- ^ ZMQ_POLLIN (incoming messages)@@ -306,25 +293,45 @@   | Queue     -- ^ ZMQ_QUEUE   deriving (Eq, Ord, Show) --- | Initialize a 0MQ context (cf. zmq_init for details).+-- | Initialize a 0MQ context (cf. zmq_init for details).  You should+-- normally prefer to use 'with' instead. init :: Size -> IO Context init ioThreads = do     c <- throwErrnoIfNull "init" $ c_zmq_init (fromIntegral ioThreads)     return (Context c) --- | Terminate 0MQ context (cf. zmq_term).+-- | 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 +-- | Run an action with a 0MQ context.  The 'Context' supplied to your+-- action will /not/ be valid after the action either returns or+-- throws an exception.+with :: Size -> (Context -> IO a) -> IO a+with ioThreads act =+  bracket (throwErrnoIfNull "c_zmq_init" $ c_zmq_init (fromIntegral ioThreads))+          (throwErrnoIfMinus1_ "c_zmq_term" . c_zmq_term)+          (act . Context)+ -- | Create a new 0MQ socket within the given context. socket :: SType a => Context -> a -> IO (Socket a)-socket (Context c) t =-    let zt = typeVal . zmqSocketType $ t-    in  Socket <$> throwErrnoIfNull "socket" (c_zmq_socket c zt)+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 <- readIORef status+    unless alive $ c_zmq_close s >> return ()+  return sock  -- | Close a 0MQ socket. close :: Socket a -> IO ()-close = throwErrnoIfMinus1_ "close" . c_zmq_close . sock+close sock@(Socket _ status) = withSocket "close" sock $ \s -> do+  alive <- readIORef status+  when alive $ do+    writeIORef status False+    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@@ -386,30 +393,36 @@  -- | Bind the socket to the given address (zmq_bind) bind :: Socket a -> String -> IO ()-bind (Socket s) str = throwErrnoIfMinus1_ "bind" $-    withCString str (c_zmq_bind s)+bind sock str = withSocket "bind" sock $+    throwErrnoIfMinus1_ "bind" . withCString str . c_zmq_bind  -- | Connect the socket to the given address (zmq_connect). connect :: Socket a -> String -> IO ()-connect (Socket s) str = throwErrnoIfMinus1_ "connect" $-    withCString str (c_zmq_connect s)+connect sock str = withSocket "connect" sock $+    throwErrnoIfMinus1_ "connect" . withCString str . c_zmq_connect  -- | Send the given 'SB.ByteString' over the socket (zmq_send). send :: Socket a -> SB.ByteString -> [Flag] -> IO ()-send (Socket s) val fls = bracket (messageOf val) messageClose $ \m ->-    throwErrnoIfMinus1_ "send" $ c_zmq_send s (msgPtr m) (combine fls)+send sock val fls = bracket (messageOf val) messageClose $ \m ->+  withSocket "send" sock $ \s ->+    retry "send" (waitWrite sock) $+          c_zmq_send s (msgPtr m) (combine (NoBlock : fls))  -- | Send the given 'LB.ByteString' over the socket (zmq_send). --   This is operationally identical to @send socket (Strict.concat --   (Lazy.toChunks lbs)) flags@ but may be more efficient. send' :: Socket a -> LB.ByteString -> [Flag] -> IO ()-send' (Socket s) val fls = bracket (messageOfLazy val) messageClose $ \m ->-    throwErrnoIfMinus1_ "send'" $ c_zmq_send s (msgPtr m) (combine fls)+send' sock val fls = bracket (messageOfLazy val) messageClose $ \m ->+  withSocket "send'" sock $ \s ->+    retry "send'" (waitWrite sock) $+          c_zmq_send s (msgPtr m) (combine (NoBlock : fls))  -- | Receive a 'ByteString' from socket (zmq_recv). receive :: Socket a -> [Flag] -> IO (SB.ByteString)-receive (Socket s) fls = bracket messageInit messageClose $ \m -> do-    throwErrnoIfMinus1Retry_ "receive" $ c_zmq_recv s (msgPtr m) (combine fls)+receive sock fls = bracket messageInit messageClose $ \m ->+  withSocket "receive" sock $ \s -> do+    retry "receive" (waitRead sock) $+          c_zmq_recv_unsafe 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)@@ -427,24 +440,28 @@         createPoll ps' []  where     createZMQPoll :: Poll -> ZMQPoll-    createZMQPoll (S (Socket s) e) =+    createZMQPoll (S (Socket s _) e) =         ZMQPoll s 0 (fromEvent e) 0     createZMQPoll (F (Fd s) e) =         ZMQPoll nullPtr (fromIntegral s) (fromEvent e) 0      createPoll :: [ZMQPoll] -> [Poll] -> IO [Poll]-    createPoll []     fd = return fd-    createPoll (p:pp) fd = do+    createPoll []     pfds = return pfds+    createPoll (p:pp) pfds = do         let s = pSocket p;             f = pFd p;             r = toEvent $ pRevents p         if isJust r-            then createPoll pp (newPoll s f r:fd)-            else createPoll pp fd+            then do+              pfd <- newPoll s f r+              createPoll pp (pfd:pfds)+            else createPoll pp pfds -    newPoll :: ZMQSocket -> CInt -> Maybe PollEvent -> Poll-    newPoll s 0 r = S (Socket s) (fromJust r)-    newPoll _ f r = F (Fd f) (fromJust r)+    newPoll :: ZMQSocket -> CInt -> Maybe PollEvent -> IO Poll+    newPoll s 0 r = do+                sock <- mkSocket s+                return $ S sock (fromJust r)+    newPoll _ f r = return $ F (Fd f) (fromJust r)      fromEvent :: PollEvent -> CShort     fromEvent In     = fromIntegral . pollVal $ pollIn@@ -459,97 +476,32 @@               | e == (fromIntegral . pollVal $ pollerr)   = Just Native               | otherwise                                 = Nothing +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++waitRead, waitWrite :: Socket a -> IO ()+waitRead = wait' threadWaitRead pollIn+waitWrite = wait' threadWaitWrite pollOut+ -- | Launch a ZeroMQ device (zmq_device). -- -- Please note that this call never returns. device :: Device -> Socket a -> Socket b -> IO ()-device device' (Socket insocket) (Socket outsocket) =-    throwErrnoIfMinus1_ "device" $+device device' insock outsock =+  withSocket "device" insock $ \insocket ->+  withSocket "device" outsock $ \outsocket ->+    throwErrnoIfMinus1Retry_ "device" $         c_zmq_device (fromDevice device') insocket outsocket  where     fromDevice :: Device -> CInt     fromDevice Streamer  = fromIntegral . deviceType $ deviceStreamer     fromDevice Forwarder = fromIntegral . deviceType $ deviceForwarder     fromDevice Queue     = fromIntegral . deviceType $ deviceQueue----- internal helpers:--messageOf :: SB.ByteString -> IO Message-messageOf b = UB.unsafeUseAsCStringLen b $ \(cstr, len) -> do-    msg <- messageInitSize (fromIntegral len)-    data_ptr <- c_zmq_msg_data (msgPtr msg)-    copyBytes data_ptr cstr len-    return msg--messageOfLazy :: LB.ByteString -> IO Message-messageOfLazy lbs = do-    msg <- messageInitSize (fromIntegral len)-    data_ptr <- c_zmq_msg_data (msgPtr msg)-    let fn offset bs = UB.unsafeUseAsCStringLen bs $ \(cstr, str_len) -> do-        copyBytes (data_ptr `plusPtr` offset) cstr str_len-        return (offset + str_len)-    foldM_ fn 0 (LB.toChunks lbs)-    return msg- where-    len = LB.length lbs--messageClose :: Message -> IO ()-messageClose (Message ptr) = do-    throwErrnoIfMinus1_ "messageClose" $ c_zmq_msg_close ptr-    free ptr--messageInit :: IO Message-messageInit = do-    ptr <- new (ZMQMsg nullPtr)-    throwErrnoIfMinus1_ "messageInit" $ c_zmq_msg_init ptr-    return (Message ptr)--messageInitSize :: Size -> IO Message-messageInitSize s = do-    ptr <- new (ZMQMsg nullPtr)-    throwErrnoIfMinus1_ "messageInitSize" $-        c_zmq_msg_init_size ptr (fromIntegral s)-    return (Message ptr)--setIntOpt :: (Storable b, Integral b) => Socket a -> ZMQOption -> b -> IO ()-setIntOpt (Socket s) (ZMQOption o) i =-    throwErrnoIfMinus1_ "setIntOpt" $ with i $ \ptr ->-        c_zmq_setsockopt s (fromIntegral o)-                           (castPtr ptr)-                           (fromIntegral . sizeOf $ i)--setStrOpt :: Socket a -> ZMQOption -> String -> IO ()-setStrOpt (Socket s) (ZMQOption o) str = throwErrnoIfMinus1_ "setStrOpt" $-    withCStringLen str $ \(cstr, len) ->-        c_zmq_setsockopt s (fromIntegral o)-                           (castPtr cstr)-                           (fromIntegral len)--getBoolOpt :: Socket a -> ZMQOption -> IO Bool-getBoolOpt s o = ((1 :: Int64) ==) <$> getIntOpt s o--getIntOpt :: (Storable b, Integral b) => Socket a -> ZMQOption -> IO b-getIntOpt (Socket s) (ZMQOption o) = do-    let i = 0-    bracket (new i) free $ \iptr ->-        bracket (new (fromIntegral . sizeOf $ i :: CSize)) free $ \jptr -> do-            throwErrnoIfMinus1_ "integralOpt" $-                c_zmq_getsockopt s (fromIntegral o) (castPtr iptr) jptr-            peek iptr--getStrOpt :: Socket a -> ZMQOption -> IO String-getStrOpt (Socket s) (ZMQOption o) =-    bracket (mallocBytes 255) free $ \bPtr ->-    bracket (new (255 :: CSize)) free $ \sPtr -> do-        throwErrnoIfMinus1_ "getStrOpt" $-            c_zmq_getsockopt s (fromIntegral o) (castPtr bPtr) sPtr-        peek sPtr >>= \len -> peekCStringLen (bPtr, fromIntegral len)--toZMQFlag :: Flag -> ZMQFlag-toZMQFlag NoBlock = noBlock-toZMQFlag SndMore = sndMore--combine :: [Flag] -> CInt-combine = fromIntegral . foldr ((.|.) . flagVal . toZMQFlag) 0 

@@ -178,6 +178,9 @@ 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+ -- poll  foreign import ccall safe "zmq.h zmq_poll"

@@ -0,0 +1,151 @@+module System.ZMQ.Internal+    ( Context(..)+    , Socket(..)+    , Message(..)+    , Flag(..)+    , Timeout+    , Size++    , messageOf+    , messageOfLazy+    , messageClose+    , messageInit+    , messageInitSize+    , setIntOpt+    , setStrOpt+    , getBoolOpt+    , getIntOpt+    , getStrOpt+    , toZMQFlag+    , combine+    , mkSocket+    , withSocket+    ) where++import Control.Applicative+import Control.Monad (foldM_)+import Control.Exception+import Data.IORef (IORef)++import Foreign+import Foreign.C.Error+import Foreign.C.String+import Foreign.C.Types (CInt, CSize)++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 System.ZMQ.Base++type Timeout = Int64+type Size    = Word++-- | Flags to apply on send operations (cf. man zmq_send)+--+-- [@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+          | SndMore -- ^ ZMQ_SNDMORE+  deriving (Eq, Ord, Show)++-- | A 0MQ context representation.+newtype Context = Context { ctx :: ZMQCtx }++-- | A 0MQ Socket.+data Socket a = Socket {+      _socket :: ZMQSocket+    , _sockLive :: IORef Bool+    }++-- A 0MQ Message representation.+newtype Message = Message { msgPtr :: ZMQMsgPtr }++-- internal helpers:++withSocket :: String -> Socket a -> (ZMQSocket -> IO b) -> IO b+withSocket _func (Socket sock _state) act = act sock+{-# INLINE withSocket #-}++mkSocket :: ZMQSocket -> IO (Socket a)+mkSocket s = Socket s <$> newIORef True++messageOf :: SB.ByteString -> IO Message+messageOf b = UB.unsafeUseAsCStringLen b $ \(cstr, len) -> do+    msg <- messageInitSize (fromIntegral len)+    data_ptr <- c_zmq_msg_data (msgPtr msg)+    copyBytes data_ptr cstr len+    return msg++messageOfLazy :: LB.ByteString -> IO Message+messageOfLazy lbs = do+    msg <- messageInitSize (fromIntegral len)+    data_ptr <- c_zmq_msg_data (msgPtr msg)+    let fn offset bs = UB.unsafeUseAsCStringLen bs $ \(cstr, str_len) -> do+        copyBytes (data_ptr `plusPtr` offset) cstr str_len+        return (offset + str_len)+    foldM_ fn 0 (LB.toChunks lbs)+    return msg+ where+    len = LB.length lbs++messageClose :: Message -> IO ()+messageClose (Message ptr) = do+    throwErrnoIfMinus1_ "messageClose" $ c_zmq_msg_close ptr+    free ptr++messageInit :: IO Message+messageInit = do+    ptr <- new (ZMQMsg nullPtr)+    throwErrnoIfMinus1_ "messageInit" $ c_zmq_msg_init ptr+    return (Message ptr)++messageInitSize :: Size -> IO Message+messageInitSize s = do+    ptr <- new (ZMQMsg nullPtr)+    throwErrnoIfMinus1_ "messageInitSize" $+        c_zmq_msg_init_size ptr (fromIntegral s)+    return (Message ptr)++setIntOpt :: (Storable b, Integral b) => Socket a -> ZMQOption -> b -> IO ()+setIntOpt sock (ZMQOption o) i = withSocket "setIntOpt" sock $ \s ->+    throwErrnoIfMinus1_ "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 = withSocket "setStrOpt" sock $ \s ->+  throwErrnoIfMinus1_ "setStrOpt" $ withCStringLen str $ \(cstr, len) ->+        c_zmq_setsockopt s (fromIntegral o)+                           (castPtr cstr)+                           (fromIntegral len)++getBoolOpt :: Socket a -> ZMQOption -> IO Bool+getBoolOpt s o = ((1 :: Int64) ==) <$> getIntOpt s o++getIntOpt :: (Storable b, Integral b) => Socket a -> ZMQOption -> IO b+getIntOpt sock (ZMQOption o) = withSocket "getIntOpt" sock $ \s -> do+    let i = 0+    bracket (new i) free $ \iptr ->+        bracket (new (fromIntegral . sizeOf $ i :: CSize)) free $ \jptr -> do+            throwErrnoIfMinus1_ "integralOpt" $+                c_zmq_getsockopt s (fromIntegral o) (castPtr iptr) jptr+            peek iptr++getStrOpt :: Socket a -> ZMQOption -> IO String+getStrOpt sock (ZMQOption o) = withSocket "getStrOpt" sock $ \s ->+    bracket (mallocBytes 255) free $ \bPtr ->+    bracket (new (255 :: CSize)) free $ \sPtr -> do+        throwErrnoIfMinus1_ "getStrOpt" $+            c_zmq_getsockopt s (fromIntegral o) (castPtr bPtr) sPtr+        peek sPtr >>= \len -> peekCStringLen (bPtr, fromIntegral len)++toZMQFlag :: Flag -> ZMQFlag+toZMQFlag NoBlock = noBlock+toZMQFlag SndMore = sndMore++combine :: [Flag] -> CInt+combine = fromIntegral . foldr ((.|.) . flagVal . toZMQFlag) 0

@@ -12,11 +12,11 @@         hPutStrLn stderr "usage: display <address>"         exitFailure     let addr = head args-    c <- ZMQ.init 1-    s <- ZMQ.socket c ZMQ.Sub-    ZMQ.subscribe s ""-    ZMQ.connect s addr-    forever $ do+    ZMQ.with 1 $ \c -> do+      s <- ZMQ.socket c ZMQ.Sub+      ZMQ.subscribe s ""+      ZMQ.connect s addr+      forever $ do         line <- ZMQ.receive s []         SB.putStrLn line         hFlush stdout

@@ -14,12 +14,11 @@     let bindTo = args !! 0         size   = read $ args !! 1         rounds = read $ args !! 2-    c <- ZMQ.init 1-    s <- ZMQ.socket c ZMQ.Rep-    ZMQ.bind s bindTo-    loop s rounds size-    ZMQ.close s-    ZMQ.term c+    ZMQ.with 1 $ \c -> do+      s <- ZMQ.socket c ZMQ.Rep+      ZMQ.bind s bindTo+      loop s rounds size+      ZMQ.close s  where     loop s r sz = unless (r <= 0) $ do         msg <- ZMQ.receive s []

@@ -16,17 +16,16 @@     let bindTo = args !! 0         size   = read $ args !! 1 :: Int         count  = read $ args !! 2 :: Int-    c <- ZMQ.init 1-    s <- ZMQ.socket c ZMQ.Sub-    ZMQ.subscribe s ""-    ZMQ.bind s bindTo-    receive s size-    start <- getCurrentTime-    loop s (count - 1) size-    end <- getCurrentTime-    printStat start end size count-    ZMQ.close s-    ZMQ.term c+    ZMQ.with 1 $ \c -> do+      s <- ZMQ.socket c ZMQ.Sub+      ZMQ.subscribe s ""+      ZMQ.bind s bindTo+      receive s size+      start <- getCurrentTime+      loop s (count - 1) size+      end <- getCurrentTime+      printStat start end size count+      ZMQ.close s  where     receive s sz = do         msg <- ZMQ.receive s []

@@ -16,15 +16,14 @@         size    = read $ args !! 1         rounds  = read $ args !! 2         message = SB.replicate size 0x65-    c <- ZMQ.init 1-    s <- ZMQ.socket c ZMQ.Req-    ZMQ.connect s connTo-    start <- getCurrentTime-    loop s rounds message-    end <- getCurrentTime-    print (diffUTCTime end start)-    ZMQ.close s-    ZMQ.term c+    ZMQ.with 1 $ \c -> do+      s <- ZMQ.socket c ZMQ.Req+      ZMQ.connect s connTo+      start <- getCurrentTime+      loop s rounds message+      end <- getCurrentTime+      print (diffUTCTime end start)+      ZMQ.close s  where     loop s r msg = unless (r <= 0) $ do         ZMQ.send s msg []

@@ -16,13 +16,12 @@         size    = read $ args !! 1         count   = read $ args !! 2         message = SB.replicate size 0x65-    c <- ZMQ.init 1-    s <- ZMQ.socket c ZMQ.Pub-    ZMQ.connect s connTo-    replicateM_ count $ ZMQ.send s message []-    threadDelay 10000000-    ZMQ.close s-    ZMQ.term c+    ZMQ.with 1 $ \c -> do+      s <- ZMQ.socket c ZMQ.Pub+      ZMQ.connect s connTo+      replicateM_ count $ ZMQ.send s message []+      threadDelay 10000000+      ZMQ.close s  usage :: String usage = "usage: remote_thr <connect-to> <message-size> <message-count>"

@@ -12,12 +12,12 @@     when (length args /= 1) $ do         hPutStrLn stderr usage         exitFailure-    c <- ZMQ.init 1-    s <- ZMQ.socket c ZMQ.Rep-    let bindTo = head args-        toPoll = [ZMQ.S s ZMQ.In]-    ZMQ.bind s bindTo-    forever $+    ZMQ.with 1 $ \c -> do+      s <- ZMQ.socket c ZMQ.Rep+      let bindTo = head args+          toPoll = [ZMQ.S s ZMQ.In]+      ZMQ.bind s bindTo+      forever $         ZMQ.poll toPoll 1000000 >>= receive  where     receive []               = return ()

@@ -16,10 +16,10 @@         exitFailure     let addr = args !! 0         name = SB.append (SB.fromString $ args !! 1) ": "-    c <- ZMQ.init 1-    s <- ZMQ.socket c ZMQ.Pub-    ZMQ.connect s addr-    forever $ do+    ZMQ.with 1 $ \c -> do+      s <- ZMQ.socket c ZMQ.Pub+      ZMQ.connect s addr+      forever $ do         line <- SB.fromString <$> getLine         ZMQ.send s (SB.append name line) [] 

@@ -1,5 +1,5 @@ name:               zeromq-haskell-version:            0.4.2+version:            0.5.0 synopsis:           bindings to zeromq description:        Bindings to zeromq (http://zeromq.org) category:           System, FFI@@ -7,17 +7,17 @@ license-file:       LICENSE author:             Toralf Wittner maintainer:         toralf.wittner@gmail.com-copyright:          Copyright (c) 2010 zeromq-haskell authors+copyright:          Copyright (c) 2011 zeromq-haskell authors homepage:           http://github.com/twittner/zeromq-haskell/ stability:          experimental-tested-With:        GHC == 6.12.3+tested-With:        GHC == 7.0.2 cabal-version:      >= 1.6.0 build-type:         Simple extra-source-files: README.md, AUTHORS, test/*.hs, test/perf/*.hs  library   exposed-modules:  System.ZMQ-  other-modules:    System.ZMQ.Base+  other-modules:    System.ZMQ.Base, System.ZMQ.Internal   ghc-options:      -Wall -O2   extensions:       CPP,                     ForeignFunctionInterface,