diff --git a/AUTHORS b/AUTHORS --- a/AUTHORS +++ b/AUTHORS @@ -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 diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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 diff --git a/src/System/ZMQ.hs b/src/System/ZMQ.hs --- a/src/System/ZMQ.hs +++ b/src/System/ZMQ.hs @@ -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 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 @@ -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" diff --git a/src/System/ZMQ/Internal.hs b/src/System/ZMQ/Internal.hs new file mode 100644 --- /dev/null +++ b/src/System/ZMQ/Internal.hs @@ -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 diff --git a/test/display.hs b/test/display.hs --- a/test/display.hs +++ b/test/display.hs @@ -12,11 +12,11 @@ hPutStrLn stderr "usage: display
" 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 diff --git a/test/perf/local_lat.hs b/test/perf/local_lat.hs --- a/test/perf/local_lat.hs +++ b/test/perf/local_lat.hs @@ -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 [] diff --git a/test/perf/local_thr.hs b/test/perf/local_thr.hs --- a/test/perf/local_thr.hs +++ b/test/perf/local_thr.hs @@ -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 [] diff --git a/test/perf/remote_lat.hs b/test/perf/remote_lat.hs --- a/test/perf/remote_lat.hs +++ b/test/perf/remote_lat.hs @@ -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 [] diff --git a/test/perf/remote_thr.hs b/test/perf/remote_thr.hs --- a/test/perf/remote_thr.hs +++ b/test/perf/remote_thr.hs @@ -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 " diff --git a/test/poll.hs b/test/poll.hs --- a/test/poll.hs +++ b/test/poll.hs @@ -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 () diff --git a/test/prompt.hs b/test/prompt.hs --- a/test/prompt.hs +++ b/test/prompt.hs @@ -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) [] diff --git a/zeromq-haskell.cabal b/zeromq-haskell.cabal --- a/zeromq-haskell.cabal +++ b/zeromq-haskell.cabal @@ -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,