zeromq3-haskell 0.2 → 0.3.0
raw patch · 20 files changed
+843/−330 lines, 20 filesdep +MonadCatchIO-transformersdep +ansi-terminaldep +checkersdep −test-frameworkdep −test-framework-quickcheck2dep ~QuickCheck
Dependencies added: MonadCatchIO-transformers, ansi-terminal, checkers, transformers
Dependencies removed: test-framework, test-framework-quickcheck2
Dependency ranges changed: QuickCheck
Files
- README.md +6/−2
- examples/Makefile +2/−2
- examples/display.hs +10/−13
- examples/echo.hs +0/−22
- examples/perf/Makefile +2/−2
- examples/perf/local_lat.hs +7/−7
- examples/perf/local_thr.hs +18/−16
- examples/perf/remote_lat.hs +10/−10
- examples/perf/remote_thr.hs +6/−6
- examples/prompt.hs +11/−11
- examples/threads.hs +0/−56
- src/Data/Restricted.hs +10/−0
- src/System/ZMQ3.hs +83/−57
- src/System/ZMQ3/Base.hsc +6/−2
- src/System/ZMQ3/Internal.hs +56/−22
- src/System/ZMQ3/Monadic.hs +475/−0
- tests/System/ZMQ3/Test/Properties.hs +95/−94
- tests/Test/Tools.hs +35/−0
- tests/tests.hs +1/−2
- zeromq3-haskell.cabal +10/−6
README.md view
@@ -1,10 +1,15 @@-This library provides Haskell bindings to 0MQ 3.x (http://zeromq.org).+This library provides Haskell bindings to 0MQ 3.2.x (http://zeromq.org). Current status -------------- This software currently has *beta* status, i.e. it had seen limited testing. +Version 0.3.0 - Add monadic layer on top of System.ZMQ3 and substitute+ String for ByteString in a number of cases, where the 0MQ+ API speaks of "binary data", i.e. subscribe/unsubscribe,+ identity/setIdentity and setTcpAcceptFilter.+ Version 0.2 - Add additional functionality from 3.2 stable release, e.g. zmq_proxy support, new socket options, socket monitoring etc. *API Change*: withContext no longer accepts the number of@@ -68,4 +73,3 @@ 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 tw@dtex.org-
examples/Makefile view
@@ -6,6 +6,6 @@ .PHONY: clean clean:- @rm *.o *.hi- @rm display prompt+ -rm -f *.o *.hi+ -rm -f display prompt
examples/display.hs view
@@ -1,10 +1,9 @@+{-# LANGUAGE OverloadedStrings #-} import Control.Monad-import System.IO import System.Exit+import System.IO import System.Environment-import Control.Exception-import qualified System.ZMQ3 as ZMQ-import qualified Data.ByteString as SB+import System.ZMQ3.Monadic import qualified Data.ByteString.Char8 as CS main :: IO ()@@ -13,12 +12,10 @@ when (length args < 1) $ do hPutStrLn stderr "usage: display <address> [<address>, ...]" exitFailure- ZMQ.withContext $ \c ->- ZMQ.withSocket c ZMQ.Sub $ \s -> do- ZMQ.subscribe s ""- mapM (ZMQ.connect s) args- forever $ do- line <- ZMQ.receive s- CS.putStrLn line- hFlush stdout-+ runZMQ $ do+ sub <- socket Sub+ subscribe sub ""+ mapM_ (connect sub) args+ forever $ do+ receive sub >>= liftIO . CS.putStrLn+ liftIO $ hFlush stdout
− examples/echo.hs
@@ -1,22 +0,0 @@-{-# LANGUAGE DeriveDataTypeable #-}-module Main where--import Control.Monad-import System.ZMQ3-import System.Console.CmdArgs.Implicit--data Echo = Echo- { address :: String- } deriving (Show, Data, Typeable)--opts = Echo { address = def &= help "address to connect"- }- &= summary "Echo"--main :: IO ()-main = do- o <- cmdArgs opts- withContext 2 $ \ctx ->- withSocket ctx Rep $ \sck -> do- bind sck (address o)- forever $ receive sck >>= send sck []
examples/perf/Makefile view
@@ -10,6 +10,6 @@ .PHONY: clean clean:- @rm *.o *.hi- @rm local_lat remote_lat local_thr remote_thr+ -rm -f *.o *.hi+ -rm -f local_lat remote_lat local_thr remote_thr
examples/perf/local_lat.hs view
@@ -2,7 +2,7 @@ import System.IO import System.Exit import System.Environment-import qualified System.ZMQ3 as ZMQ+import System.ZMQ3.Monadic import qualified Data.ByteString as SB main :: IO ()@@ -14,16 +14,16 @@ let bindTo = args !! 0 size = read $ args !! 1 rounds = read $ args !! 2- ZMQ.withContext $ \c ->- ZMQ.withSocket c ZMQ.Rep $ \s -> do- ZMQ.bind s bindTo- loop s rounds size+ runZMQ $ do+ s <- socket Rep+ bind s bindTo+ loop s rounds size where loop s r sz = unless (r <= 0) $ do- msg <- ZMQ.receive s+ msg <- receive s when (SB.length msg /= sz) $ error "message of incorrect size received"- ZMQ.send s [] msg+ send s [] msg loop s (r - 1) sz usage :: String
examples/perf/local_thr.hs view
@@ -1,9 +1,11 @@+{-# LANGUAGE OverloadedStrings #-}+import Control.Concurrent import Control.Monad import System.IO import System.Exit import System.Environment import Data.Time.Clock-import qualified System.ZMQ3 as ZMQ+import System.ZMQ3.Monadic import qualified Data.ByteString as SB import Text.Printf @@ -14,25 +16,25 @@ hPutStrLn stderr usage exitFailure let bindTo = args !! 0- size = read $ args !! 1 :: Int- count = read $ args !! 2 :: Int- ZMQ.withContext $ \c ->- ZMQ.withSocket c ZMQ.Sub $ \s -> do- ZMQ.subscribe s ""- ZMQ.bind s bindTo- receive s size- start <- getCurrentTime- loop s (count - 1) size- end <- getCurrentTime- printStat start end size count+ size = read $ args !! 1+ count = read $ args !! 2+ runZMQ $ do+ s <- socket Sub+ subscribe s ""+ bind s bindTo+ receive' s size+ start <- liftIO $ getCurrentTime+ loop s count size+ end <- liftIO $ getCurrentTime+ liftIO $ printStat start end size count where- receive s sz = do- msg <- ZMQ.receive s+ receive' s sz = do+ msg <- receive s when (SB.length msg /= sz) $ error "message of incorrect size received" - loop s c sz = unless (c <= 0) $ do- receive s sz+ loop s c sz = unless (c < 0) $ do+ receive' s sz loop s (c - 1) sz printStat :: UTCTime -> UTCTime -> Int -> Int -> IO ()
examples/perf/remote_lat.hs view
@@ -3,7 +3,7 @@ import System.Exit import System.Environment import Data.Time.Clock-import qualified System.ZMQ3 as ZMQ+import System.ZMQ3.Monadic import qualified Data.ByteString as SB main :: IO ()@@ -16,17 +16,17 @@ size = read $ args !! 1 rounds = read $ args !! 2 message = SB.replicate size 0x65- ZMQ.withContext $ \c ->- ZMQ.withSocket c ZMQ.Req $ \s -> do- ZMQ.connect s connTo- start <- getCurrentTime- loop s rounds message- end <- getCurrentTime- print (diffUTCTime end start)+ runZMQ $ do+ s <- socket Req+ connect s connTo+ start <- liftIO $ getCurrentTime+ loop s rounds message+ end <- liftIO $ getCurrentTime+ liftIO $ print (diffUTCTime end start) where loop s r msg = unless (r <= 0) $ do- ZMQ.send s [] msg- msg' <- ZMQ.receive s+ send s [] msg+ msg' <- receive s when (SB.length msg' /= SB.length msg) $ error "message of incorrect size received" loop s (r - 1) msg
examples/perf/remote_thr.hs view
@@ -3,7 +3,7 @@ import System.IO import System.Exit import System.Environment-import qualified System.ZMQ3 as ZMQ+import System.ZMQ3.Monadic import qualified Data.ByteString as SB main :: IO ()@@ -16,11 +16,11 @@ size = read $ args !! 1 count = read $ args !! 2 message = SB.replicate size 0x65- ZMQ.withContext $ \c ->- ZMQ.withSocket c ZMQ.Pub $ \s -> do- ZMQ.connect s connTo- replicateM_ count $ ZMQ.send s [] message- threadDelay 10000000+ runZMQ $ do+ s <- socket Pub+ connect s connTo+ replicateM_ count $ send s [] message+ liftIO $ threadDelay 10000000 usage :: String usage = "usage: remote_thr <connect-to> <message-size> <message-count>"
examples/prompt.hs view
@@ -1,12 +1,13 @@ {-# LANGUAGE OverloadedStrings #-} import Control.Applicative import Control.Monad+import Data.Monoid+import Data.String import System.IO import System.Exit import System.Environment-import qualified System.ZMQ3 as ZMQ+import System.ZMQ3.Monadic import qualified Data.ByteString.UTF8 as SB-import qualified Data.ByteString.Char8 as SB main :: IO () main = do@@ -14,12 +15,11 @@ when (length args /= 2) $ do hPutStrLn stderr "usage: prompt <address> <username>" exitFailure- let addr = args !! 0- name = SB.append (SB.fromString $ args !! 1) ": "- ZMQ.withContext $ \c ->- ZMQ.withSocket c ZMQ.Pub $ \s -> do- ZMQ.bind s addr- forever $ do- line <- SB.fromString <$> getLine- ZMQ.send s [] (SB.append name line)-+ let addr = head args+ name = fromString (args !! 1) <> ": "+ runZMQ $ do+ pub <- socket Pub+ bind pub addr+ forever $ do+ line <- liftIO $ SB.fromString <$> getLine+ send pub [] (name <> line)
− examples/threads.hs
@@ -1,56 +0,0 @@-{-# LANGUAGE DeriveDataTypeable #-}-module Main where--import Control.Concurrent (threadDelay)-import Control.Monad-import Control.Monad.IO.Class-import Control.Monad.CatchIO-import Control.Scope-import Data.ByteString.Char8 (pack, unpack)-import System.ZMQ3.Safe-import System.Random-import System.Console.CmdArgs.Implicit-import Control.Concurrent.Thread.Group (ThreadGroup)-import qualified Control.Concurrent.Thread.Group as TG--runner :: (InScope m1 m2, MonadIO m2) => StdGen -> Int -> Socket m1 Req -> m2 ()-runner gen runs s = go 0- where- go ctr = unless (ctr == runs) $ do- let (r, g) = randomR (100, 100000) gen- liftIO $ setStdGen g- send s [] (pack . show $ ctr)- ret <- receive s- liftIO $ print ctr- if ((read . unpack $ ret) /= ctr)- then error "unexpected return value"- else (liftIO $ threadDelay r) >> go (ctr + 1)--data Threads = Threads- { address :: String- , threads :: Int- , runs :: Int- } deriving (Show, Data, Typeable)--opts :: Threads-opts = Threads { address = def &= help "address to connect"- , threads = def &= help "number of client threads"- , runs = def &= help "number of runs" &= opt (1000 :: Int)- }- &= summary "Threads Test"--setup :: (InScope m1 m2, MonadCatchIO m2) => Threads -> Context m1 -> m2 ()-setup options ctx = do- rnd <- liftIO $ getStdGen- scope $ socket ctx Req >>= \s -> do- connect s (address options)- runner rnd (runs options) s- return ()--main :: IO ()-main = do- o <- cmdArgs opts- g <- TG.new- scope $ context 2 >>= \c -> do- setup o c-
src/Data/Restricted.hs view
@@ -3,6 +3,7 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE OverloadedStrings #-} -- | -- Module : Data.Restricted@@ -36,6 +37,8 @@ ) where import Data.Int+import Data.ByteString (ByteString)+import qualified Data.ByteString as B -- | Type level restriction. data Restricted l u v = Restricted !v deriving Show@@ -129,6 +132,13 @@ restrict s | length s < 1 = Restricted " " | otherwise = Restricted (take 254 s)++instance Restriction N1 N254 ByteString where+ toRestricted s | check (1, 254) (B.length s) = Just $ Restricted s+ | otherwise = Nothing++ restrict s | B.length s < 1 = Restricted (B.singleton 0x20)+ | otherwise = Restricted (B.take 254 s) -- Helpers
src/System/ZMQ3.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE GADTs #-} -- | -- Module : System.ZMQ3 -- Copyright : (c) 2010-2012 Toralf Wittner@@ -53,14 +54,6 @@ -- Devices are no longer present in 0MQ 3.x and consequently have been -- removed form this binding as well. ----- /Poll/------ Removed support for polling. This should not be necessary as 'send' and--- 'receive' are internally non-blocking and use GHC's I/O manager to block--- calling threads when send or receive would yield EAGAIN. This combined with--- GHC's scalable threading model should relieve client code from the burden--- to do it's own polling.--- -- /Error Handling/ -- -- The type 'ZMQError' is introduced, together with inspection functions 'errno',@@ -80,6 +73,7 @@ , Event (..) , EventType (..) , EventMsg (..)+ , Poll (..) -- ** Type Classes , SocketType@@ -115,6 +109,7 @@ , receiveMulti , version , monitor+ , poll , System.ZMQ3.subscribe , System.ZMQ3.unsubscribe@@ -210,12 +205,13 @@ import qualified Prelude as P import Control.Applicative import Control.Exception-import Control.Monad (unless, when, void)+import Control.Monad (unless, void)+import Control.Monad.IO.Class+import Data.List (intersect, foldl') import Data.Restricted-import Data.IORef (atomicModifyIORef) import Foreign hiding (throwIf, throwIf_, throwIfNull, void) import Foreign.C.String-import Foreign.C.Types (CInt)+import Foreign.C.Types (CInt, CShort) import qualified Data.ByteString as SB import qualified Data.ByteString.Lazy as LB import System.Posix.Types (Fd(..))@@ -309,10 +305,6 @@ -- node becomes available for sending; messages are not discarded. data Push = Push --- | Socket types.-class SocketType a where- zmqSocketType :: a -> ZMQSocketType- -- | Sockets which can 'subscribe'. class Subscriber a @@ -367,11 +359,16 @@ data Event = In -- ^ ZMQ_POLLIN (incoming messages) | Out -- ^ ZMQ_POLLOUT (outgoing messages, i.e. at least 1 byte can be written)- | InOut -- ^ ZMQ_POLLIN | ZMQ_POLLOUT- | Native -- ^ ZMQ_POLLERR- | None- deriving (Eq, Ord, Show)+ | Err -- ^ ZMQ_POLLERR+ deriving (Eq, Ord, Read, Show) +-- | Type representing a descriptor, poll is waiting for+-- (either a 0MQ socket or a file descriptor) plus the type+-- of event to wait for.+data Poll m where+ Sock :: Socket s -> [Event] -> Maybe ([Event] -> m ()) -> Poll m+ File :: Fd -> [Event] -> Maybe ([Event] -> m ()) -> Poll m+ -- | Return the runtime version of the underlying 0MQ library as a -- (major, minor, patch) triple. version :: IO (Int, Int, Int)@@ -403,7 +400,7 @@ -- | Terminate a 0MQ context (cf. zmq_ctx_destroy). You should normally -- prefer to use 'withContext' instead. destroy :: Context -> IO ()-destroy = throwIfMinus1Retry_ "term" . c_zmq_ctx_destroy . _ctx+destroy c = throwIfMinus1Retry_ "term" . c_zmq_ctx_destroy . _ctx $ c -- | Run an action with a 0MQ context. The 'Context' supplied to your -- action will /not/ be valid after the action either returns or@@ -423,30 +420,26 @@ -- | Create a new 0MQ socket within the given context. 'withSocket' provides -- automatic socket closing and may be safer to use. socket :: SocketType a => Context -> a -> IO (Socket a)-socket (Context c) t = do- let zt = typeVal . zmqSocketType $ t- throwIfNull "socket" (c_zmq_socket c zt) >>= mkSocket+socket c t = Socket <$> mkSocketRepr t c -- | Close a 0MQ socket. 'withSocket' provides automatic socket closing and may -- be safer to use. close :: Socket a -> IO ()-close sock@(Socket _ status) = onSocket "close" sock $ \s -> do- alive <- atomicModifyIORef status (\b -> (False, b))- when alive $ throwIfMinus1_ "close" . c_zmq_close $ s+close = closeSock . _socketRepr -- | Subscribe Socket to given subscription.-subscribe :: Subscriber a => Socket a -> String -> IO ()-subscribe s = setStrOpt s B.subscribe+subscribe :: Subscriber a => Socket a -> SB.ByteString -> IO ()+subscribe s = setByteStringOpt s B.subscribe -- | Unsubscribe Socket from given subscription.-unsubscribe :: Subscriber a => Socket a -> String -> IO ()-unsubscribe s = setStrOpt s B.unsubscribe+unsubscribe :: Subscriber a => Socket a -> SB.ByteString -> IO ()+unsubscribe s = setByteStringOpt s B.unsubscribe -- Read Only -- | Cf. @zmq_getsockopt ZMQ_EVENTS@-events :: Socket a -> IO Event-events s = toEvent <$> getIntOpt s B.events 0+events :: Socket a -> IO [Event]+events s = toEvents <$> getIntOpt s B.events 0 -- | Cf. @zmq_getsockopt ZMQ_FD@ fileDescriptor :: Socket a -> IO Fd@@ -467,8 +460,8 @@ maxSockets = ctxIntOption "maxSockets" _maxSockets -- | Cf. @zmq_getsockopt ZMQ_IDENTITY@-identity :: Socket a -> IO String-identity s = getStrOpt s B.identity+identity :: Socket a -> IO SB.ByteString+identity s = getByteStringOpt s B.identity -- | Cf. @zmq_getsockopt ZMQ_AFFINITY@ affinity :: Socket a -> IO Word64@@ -572,8 +565,8 @@ setMaxSockets n = setCtxIntOption "maxSockets" _maxSockets n -- | Cf. @zmq_setsockopt ZMQ_IDENTITY@-setIdentity :: Restricted N1 N254 String -> Socket a -> IO ()-setIdentity x s = setStrOpt s B.identity (rvalue x)+setIdentity :: Restricted N1 N254 SB.ByteString -> Socket a -> IO ()+setIdentity x s = setByteStringOpt s B.identity (rvalue x) -- | Cf. @zmq_setsockopt ZMQ_AFFINITY@ setAffinity :: Word64 -> Socket a -> IO ()@@ -648,16 +641,11 @@ setSendHighWM = setInt32OptFromRestricted B.sendHighWM -- | Cf. @zmq_setsockopt ZMQ_TCP_ACCEPT_FILTER@-setTcpAcceptFilter :: Maybe String -> Socket a -> IO ()+setTcpAcceptFilter :: Maybe SB.ByteString -> Socket a -> IO () setTcpAcceptFilter Nothing sock = onSocket "setTcpAcceptFilter" sock $ \s -> throwIfMinus1Retry_ "setStrOpt" $ c_zmq_setsockopt s (optVal tcpAcceptFilter) nullPtr 0-setTcpAcceptFilter (Just dat) sock = onSocket "setTcpAcceptFilter" sock $ \s ->- throwIfMinus1Retry_ "setStrOpt" $- withCStringLen dat $ \(ptr, len) ->- c_zmq_setsockopt s (optVal tcpAcceptFilter)- (castPtr ptr)- (fromIntegral len)+setTcpAcceptFilter (Just dat) sock = setByteStringOpt sock tcpAcceptFilter dat -- | Cf. @zmq_setsockopt ZMQ_TCP_KEEPALIVE@ setTcpKeepAlive :: Switch -> Socket a -> IO ()@@ -682,17 +670,17 @@ -- | Bind the socket to the given address (cf. zmq_bind) bind :: Socket a -> String -> IO () bind sock str = onSocket "bind" sock $- throwIfMinus1_ "bind" . withCString str . c_zmq_bind+ throwIfMinus1Retry_ "bind" . withCString str . c_zmq_bind -- | Unbind the socket from the given address (cf. zmq_unbind) unbind :: Socket a -> String -> IO () unbind sock str = onSocket "unbind" sock $- throwIfMinus1_ "unbind" . withCString str . c_zmq_unbind+ throwIfMinus1Retry_ "unbind" . withCString str . c_zmq_unbind -- | Connect the socket to the given address (cf. zmq_connect). connect :: Socket a -> String -> IO () connect sock str = onSocket "connect" sock $- throwIfMinus1_ "connect" . withCString str . c_zmq_connect+ throwIfMinus1Retry_ "connect" . withCString str . c_zmq_connect -- | Send the given 'SB.ByteString' over the socket (cf. zmq_sendmsg). --@@ -704,7 +692,7 @@ send sock fls val = bracket (messageOf val) messageClose $ \m -> onSocket "send" sock $ \s -> retry "send" (waitWrite sock) $- c_zmq_sendmsg s (msgPtr m) (combine (DontWait : fls))+ c_zmq_sendmsg s (msgPtr m) (combineFlags (DontWait : fls)) -- | Send the given 'LB.ByteString' over the socket (cf. zmq_sendmsg). -- This is operationally identical to @send socket (Strict.concat@@ -718,7 +706,7 @@ send' sock fls val = bracket (messageOfLazy val) messageClose $ \m -> onSocket "send'" sock $ \s -> retry "send'" (waitWrite sock) $- c_zmq_sendmsg s (msgPtr m) (combine (DontWait : fls))+ c_zmq_sendmsg s (msgPtr m) (combineFlags (DontWait : fls)) -- | Send a multi-part message. -- This function applies the 'SendMore' 'Flag' between all message parts.@@ -774,7 +762,7 @@ -- /once/ to 'False'. monitor :: [EventType] -> Context -> Socket a -> IO (Bool -> IO (Maybe EventMsg)) monitor es ctx sock = do- let addr = "inproc://" ++ show (_socket sock)+ let addr = "inproc://" ++ show (_socket . _socketRepr $ sock) s <- socket ctx Pair socketMonitor es addr sock connect s addr@@ -789,14 +777,52 @@ tag <- peek ptr :: IO CInt return . Just $ eventMessage str dat (ZMQEventType tag) --- Convert bit-masked word into Event.-toEvent :: Word32 -> Event-toEvent e | e == (fromIntegral . pollVal $ pollIn) = In- | e == (fromIntegral . pollVal $ pollOut) = Out- | e == (fromIntegral . pollVal $ pollInOut) = InOut- | e == (fromIntegral . pollVal $ pollerr) = Native- | otherwise = None+-- | Polls for events on the given 'Poll' descriptors. Returns the+-- same list of 'Poll' descriptors with an "updated" 'PollEvent' field+-- (cf. zmq_poll). Sockets which have seen no activity have 'None' in+-- their 'PollEvent' field.+poll :: MonadIO m => Timeout -> [Poll m] -> m [[Event]]+poll to desc = do+ let len = length desc+ let ps = map toZMQPoll desc+ ps' <- liftIO $ withArray ps $ \ptr -> do+ throwIfMinus1Retry_ "poll" $+ c_zmq_poll ptr (fromIntegral len) (fromIntegral to)+ peekArray len ptr+ mapM fromZMQPoll (zip desc ps')+ where+ toZMQPoll :: MonadIO m => Poll m -> ZMQPoll+ toZMQPoll (Sock (Socket (SocketRepr s _)) e _) =+ ZMQPoll s 0 (combine (map fromEvent e)) 0 + toZMQPoll (File (Fd s) e _) =+ ZMQPoll nullPtr (fromIntegral s) (combine (map fromEvent e)) 0++ fromZMQPoll :: MonadIO m => (Poll m, ZMQPoll) -> m [Event]+ fromZMQPoll (p, zp) = do+ let e = toEvents . fromIntegral . pRevents $ zp+ let (e', f) = case p of+ (Sock _ x g) -> (x, g)+ (File _ x g) -> (x, g)+ unless (null (e `intersect` e')) $+ maybe (return ()) ($ e) f+ return e++ fromEvent :: Event -> CShort+ fromEvent In = fromIntegral . pollVal $ pollIn+ fromEvent Out = fromIntegral . pollVal $ pollOut+ fromEvent Err = fromIntegral . pollVal $ pollerr++-- Convert bit-masked word into Event list.+toEvents :: Word32 -> [Event]+toEvents e = foldl' (\es f -> f e es) [] tests+ where+ tests =+ [ \i xs -> if i .&. (fromIntegral . pollVal $ pollIn) /= 0 then In:xs else xs+ , \i xs -> if i .&. (fromIntegral . pollVal $ pollOut) /= 0 then Out:xs else xs+ , \i xs -> if i .&. (fromIntegral . pollVal $ pollerr) /= 0 then Err:xs else xs+ ]+ retry :: String -> IO () -> IO CInt -> IO () retry msg wait act = throwIfMinus1RetryMayBlock_ msg act wait @@ -836,4 +862,4 @@ onSocket "proxy-back" back $ \b -> void (c_zmq_proxy f b c) where- c = maybe nullPtr _socket capture+ c = maybe nullPtr (_socket . _socketRepr) capture
src/System/ZMQ3/Base.hsc view
@@ -146,8 +146,7 @@ #{enum ZMQPollEvent, ZMQPollEvent, pollIn = ZMQ_POLLIN, pollOut = ZMQ_POLLOUT,- pollerr = ZMQ_POLLERR,- pollInOut = ZMQ_POLLIN | ZMQ_POLLOUT+ pollerr = ZMQ_POLLERR } -- general initialization@@ -239,3 +238,8 @@ foreign import ccall unsafe "zmq.h zmq_proxy" c_zmq_proxy :: ZMQSocket -> ZMQSocket -> ZMQSocket -> IO CInt++-- poll++foreign import ccall safe "zmq.h zmq_poll"+ c_zmq_poll :: ZMQPollPtr -> CInt -> CLong -> IO CInt
src/System/ZMQ3/Internal.hs view
@@ -1,6 +1,8 @@ module System.ZMQ3.Internal ( Context(..) , Socket(..)+ , SocketRepr(..)+ , SocketType(..) , Message(..) , Flag(..) , Timeout@@ -22,10 +24,14 @@ , setInt32OptFromRestricted , ctxIntOption , setCtxIntOption+ , getByteStringOpt+ , setByteStringOpt , toZMQFlag , combine- , mkSocket+ , combineFlags+ , mkSocketRepr+ , closeSock , onSocket , bool2cint@@ -39,9 +45,9 @@ import Control.Applicative import Control.Monad (foldM_, when) import Control.Exception-import Data.IORef (IORef, mkWeakIORef, readIORef)+import Data.IORef (IORef, mkWeakIORef, readIORef, atomicModifyIORef) -import Foreign+import Foreign hiding (throwIfNull) import Foreign.C.String import Foreign.C.Types (CInt, CSize) @@ -104,30 +110,44 @@ newtype Context = Context { _ctx :: ZMQCtx } -- | A 0MQ Socket.-data Socket a = Socket {- _socket :: ZMQSocket- , _sockLive :: IORef Bool- }+newtype Socket a = Socket+ { _socketRepr :: SocketRepr } +data SocketRepr = SocketRepr+ { _socket :: ZMQSocket+ , _sockLive :: IORef Bool+ }++-- | Socket types.+class SocketType a where+ zmqSocketType :: a -> ZMQSocketType+ -- A 0MQ Message representation. newtype Message = Message { msgPtr :: ZMQMsgPtr } -- internal helpers: onSocket :: String -> Socket a -> (ZMQSocket -> IO b) -> IO b-onSocket _func (Socket sock _state) act = act sock+onSocket _func (Socket (SocketRepr sock _state)) act = act sock {-# INLINE onSocket #-} -mkSocket :: ZMQSocket -> IO (Socket a)-mkSocket s = do+mkSocketRepr :: SocketType t => t -> Context -> IO SocketRepr+mkSocketRepr t c = do+ let ty = typeVal (zmqSocketType t)+ s <- throwIfNull "mkSocketRepr" (c_zmq_socket (_ctx c) ty) ref <- newIORef True addFinalizer ref $ do alive <- readIORef ref when alive $ c_zmq_close s >> return ()- return (Socket s ref)+ return (SocketRepr s ref) where addFinalizer r f = mkWeakIORef r f >> return () +closeSock :: SocketRepr -> IO ()+closeSock (SocketRepr s status) = do+ alive <- atomicModifyIORef status (\b -> (False, b))+ when alive $ throwIfMinus1_ "close" . c_zmq_close $ s+ messageOf :: SB.ByteString -> IO Message messageOf b = UB.unsafeUseAsCStringLen b $ \(cstr, len) -> do msg <- messageInitSize (fromIntegral len)@@ -172,12 +192,17 @@ (castPtr ptr) (fromIntegral . sizeOf $ i) +setCStrOpt :: ZMQSocket -> ZMQOption -> CStringLen -> IO CInt+setCStrOpt s (ZMQOption o) (cstr, len) =+ c_zmq_setsockopt s (fromIntegral o) (castPtr cstr) (fromIntegral len)++setByteStringOpt :: Socket a -> ZMQOption -> SB.ByteString -> IO ()+setByteStringOpt sock opt str = onSocket "setByteStringOpt" sock $ \s ->+ throwIfMinus1Retry_ "setByteStringOpt" . UB.unsafeUseAsCStringLen str $ setCStrOpt s opt+ setStrOpt :: Socket a -> ZMQOption -> String -> IO ()-setStrOpt sock (ZMQOption o) str = onSocket "setStrOpt" sock $ \s ->- throwIfMinus1Retry_ "setStrOpt" $ withCStringLen str $ \(cstr, len) ->- c_zmq_setsockopt s (fromIntegral o)- (castPtr cstr)- (fromIntegral len)+setStrOpt sock opt str = onSocket "setStrOpt" sock $ \s ->+ throwIfMinus1Retry_ "setStrOpt" . withCStringLen str $ setCStrOpt s opt getIntOpt :: (Storable b, Integral b) => Socket a -> ZMQOption -> b -> IO b getIntOpt sock (ZMQOption o) i = onSocket "getIntOpt" sock $ \s -> do@@ -187,14 +212,20 @@ c_zmq_getsockopt s (fromIntegral o) (castPtr iptr) jptr peek iptr -getStrOpt :: Socket a -> ZMQOption -> IO String-getStrOpt sock (ZMQOption o) = onSocket "getStrOpt" sock $ \s ->+getCStrOpt :: (CStringLen -> IO s) -> Socket a -> ZMQOption -> IO s+getCStrOpt peekA sock (ZMQOption o) = onSocket "getCStrOpt" sock $ \s -> bracket (mallocBytes 255) free $ \bPtr -> bracket (new (255 :: CSize)) free $ \sPtr -> do- throwIfMinus1Retry_ "getStrOpt" $+ throwIfMinus1Retry_ "getCStrOpt" $ c_zmq_getsockopt s (fromIntegral o) (castPtr bPtr) sPtr- peek sPtr >>= \len -> peekCStringLen (bPtr, fromIntegral len)+ peek sPtr >>= \len -> peekA (bPtr, fromIntegral len) +getStrOpt :: Socket a -> ZMQOption -> IO String+getStrOpt = getCStrOpt peekCStringLen++getByteStringOpt :: Socket a -> ZMQOption -> IO SB.ByteString+getByteStringOpt = getCStrOpt SB.packCStringLen+ getInt32Option :: ZMQOption -> Socket a -> IO Int getInt32Option o s = fromIntegral <$> getIntOpt s o (0 :: CInt) @@ -213,8 +244,11 @@ toZMQFlag DontWait = dontWait toZMQFlag SendMore = sndMore -combine :: [Flag] -> CInt-combine = fromIntegral . foldr ((.|.) . flagVal . toZMQFlag) 0+combineFlags :: [Flag] -> CInt+combineFlags = fromIntegral . combine . map (flagVal . toZMQFlag)++combine :: (Integral i, Bits i) => [i] -> i+combine = foldr (.|.) 0 bool2cint :: Bool -> CInt bool2cint True = 1
+ src/System/ZMQ3/Monadic.hs view
@@ -0,0 +1,475 @@+{-# LANGUAGE RankNTypes #-}+-- |+-- Module : System.ZMQ3.Monadic+-- Copyright : (c) 2013 Toralf Wittner+-- License : MIT+-- Maintainer : Toralf Wittner <tw@dtex.org>+-- Stability : experimental+-- Portability : non-portable+--+-- This modules exposes a monadic interface of 'System.ZMQ3'. Actions run+-- inside a 'ZMQ' monad and 'Socket's are guaranteed not to leak outside+-- their corresponding 'runZMQ' scope. Running 'ZMQ' computations+-- asynchronously is directly supported through 'async'.+module System.ZMQ3.Monadic+ ( -- * Type Definitions+ ZMQ+ , Socket+ , Z.Flag (SendMore)+ , Z.Switch (..)+ , Z.Timeout+ , Z.Event (..)+ , Z.EventType (..)+ , Z.EventMsg (..)+ , Z.Poll (..)++ -- ** Type Classes+ , Z.SocketType+ , Z.Sender+ , Z.Receiver+ , Z.Subscriber++ -- ** Socket Types+ , Z.Pair(..)+ , Z.Pub(..)+ , Z.Sub(..)+ , Z.XPub(..)+ , Z.XSub(..)+ , Z.Req(..)+ , Z.Rep(..)+ , Z.Dealer(..)+ , Z.Router(..)+ , Z.Pull(..)+ , Z.Push(..)++ -- * General Operations+ , version+ , runZMQ+ , async+ , socket++ -- * ZMQ Options (Read)+ , ioThreads+ , maxSockets++ -- * ZMQ Options (Write)+ , setIoThreads+ , setMaxSockets++ -- * Socket operations+ , close+ , bind+ , unbind+ , connect+ , send+ , send'+ , sendMulti+ , receive+ , receiveMulti+ , subscribe+ , unsubscribe+ , proxy+ , monitor+ , Z.poll++ -- * Socket Options (Read)+ , affinity+ , backlog+ , delayAttachOnConnect+ , events+ , fileDescriptor+ , identity+ , ipv4Only+ , lastEndpoint+ , linger+ , maxMessageSize+ , mcastHops+ , moreToReceive+ , rate+ , receiveBuffer+ , receiveHighWM+ , receiveTimeout+ , reconnectInterval+ , reconnectIntervalMax+ , recoveryInterval+ , sendBuffer+ , sendHighWM+ , sendTimeout+ , tcpKeepAlive+ , tcpKeepAliveCount+ , tcpKeepAliveIdle+ , tcpKeepAliveInterval++ -- * Socket Options (Write)+ , setAffinity+ , setBacklog+ , setDelayAttachOnConnect+ , setIdentity+ , setIpv4Only+ , setLinger+ , setMaxMessageSize+ , setMcastHops+ , setRate+ , setReceiveBuffer+ , setReceiveHighWM+ , setReceiveTimeout+ , setReconnectInterval+ , setReconnectIntervalMax+ , setRecoveryInterval+ , setRouterMandatory+ , setSendBuffer+ , setSendHighWM+ , setSendTimeout+ , setTcpAcceptFilter+ , setTcpKeepAlive+ , setTcpKeepAliveCount+ , setTcpKeepAliveIdle+ , setTcpKeepAliveInterval+ , setXPubVerbose++ -- * Error Handling+ , Z.ZMQError+ , Z.errno+ , Z.source+ , Z.message++ -- * Re-exports+ , Control.Monad.IO.Class.liftIO+ , Data.Restricted.restrict+ , Data.Restricted.toRestricted++ -- * Low-level Functions+ , waitRead+ , waitWrite+ )+where++import Control.Applicative+import Control.Concurrent (forkIO)+import Control.Monad+import Control.Monad.Trans.Reader+import Control.Monad.IO.Class+import Control.Monad.CatchIO+import Data.Int+import Data.IORef+import Data.Restricted+import Data.Word+import Data.ByteString (ByteString)+import System.Posix.Types (Fd)+import qualified Data.ByteString.Lazy as Lazy+import qualified Control.Exception as E+import qualified System.ZMQ3 as Z+import qualified System.ZMQ3.Internal as I+import qualified Control.Monad.CatchIO as M++data ZMQEnv = ZMQEnv+ { _refcount :: !(IORef Word)+ , _context :: !Z.Context+ , _sockets :: !(IORef [I.SocketRepr])+ }++-- | The ZMQ monad is modeled after 'Control.Monad.ST' and encapsulates+-- a 'System.ZMQ3.Context'. It uses the uninstantiated type variable 'z' to+-- distinguish different invoctions of 'runZMQ' and to prevent+-- unintented use of 'Socket's outside their scope. Cf. the paper+-- of John Launchbury and Simon Peyton Jones /Lazy Functional State Threads/.+newtype ZMQ z a = ZMQ { _unzmq :: ReaderT ZMQEnv IO a }++-- | The ZMQ socket, parameterised by 'SocketType' and belonging to+-- a particular 'ZMQ' thread.+newtype Socket z t = Socket { _unsocket :: Z.Socket t }++instance Monad (ZMQ z) where+ return = ZMQ . return+ (ZMQ m) >>= f = ZMQ $! m >>= _unzmq . f++instance MonadIO (ZMQ z) where+ liftIO m = ZMQ $! liftIO m++instance MonadCatchIO (ZMQ z) where+ catch (ZMQ m) f = ZMQ $! m `M.catch` (_unzmq . f)+ block (ZMQ m) = ZMQ $! block m+ unblock (ZMQ m) = ZMQ $! unblock m++instance Functor (ZMQ z) where+ fmap = liftM++instance Applicative (ZMQ z) where+ pure = return+ (<*>) = ap++-- | Return the value computed by the given 'ZMQ' monad. Rank-2+-- polymorphism is used to prevent leaking of 'z'.+-- An invocation of 'runZMQ' will internally create a 'System.ZMQ3.Context'+-- and all actions are executed relative to this context. On finish the+-- context will be disposed, but see 'async'.+runZMQ :: MonadIO m => (forall z. ZMQ z a) -> m a+runZMQ z = liftIO $ E.bracket make destroy (runReaderT (_unzmq z))+ where+ make = ZMQEnv <$> newIORef 1 <*> Z.context <*> newIORef []++-- | Run the given 'ZMQ' computation asynchronously, i.e. this function+-- runs the computation in a new thread using 'forkIO'.+-- /N.B./ reference counting is used to prolong the lifetime of the+-- 'System.ZMQ.Context' encapsulated in 'ZMQ' as necessary, e.g.:+--+-- @+-- runZMQ $ do+-- s <- socket Pair+-- async $ do+-- liftIO (threadDelay 10000000)+-- identity s >>= liftIO . print+-- @+--+-- Here, 'runZMQ' will finish before the code section in 'async', but due to+-- reference counting, the 'System.ZMQ3.Context' will only be disposed after+-- 'async' finishes as well.+async :: ZMQ z a -> ZMQ z ()+async z = ZMQ $ do+ e <- ask+ liftIO $ atomicModifyIORef (_refcount e) $ \n -> (succ n, ())+ _ <- liftIO . forkIO $ (runReaderT (_unzmq z) e >> return ()) `E.finally` destroy e+ return ()++ioThreads :: ZMQ z Word+ioThreads = onContext Z.ioThreads++setIoThreads :: Word -> ZMQ z ()+setIoThreads = onContext . Z.setIoThreads++maxSockets :: ZMQ z Word+maxSockets = onContext Z.maxSockets++setMaxSockets :: Word -> ZMQ z ()+setMaxSockets = onContext . Z.setMaxSockets++socket :: Z.SocketType t => t -> ZMQ z (Socket z t)+socket t = ZMQ $ do+ c <- asks _context+ s <- asks _sockets+ x <- liftIO $ I.mkSocketRepr t c+ liftIO $ atomicModifyIORef s $ \ss -> (x:ss, ())+ return (Socket (I.Socket x))++version :: ZMQ z (Int, Int, Int)+version = liftIO $! Z.version++-- * Socket operations++close :: Socket z t -> ZMQ z ()+close = liftIO . Z.close . _unsocket++bind :: Socket z t -> String -> ZMQ z ()+bind s = liftIO . Z.bind (_unsocket s)++unbind :: Socket z t -> String -> ZMQ z ()+unbind s = liftIO . Z.unbind (_unsocket s)++connect :: Socket z t -> String -> ZMQ z ()+connect s = liftIO . Z.connect (_unsocket s)++send :: Z.Sender t => Socket z t -> [Z.Flag] -> ByteString -> ZMQ z ()+send s f = liftIO . Z.send (_unsocket s) f++send' :: Z.Sender t => Socket z t -> [Z.Flag] -> Lazy.ByteString -> ZMQ z ()+send' s f = liftIO . Z.send' (_unsocket s) f++sendMulti :: Z.Sender t => Socket z t -> [ByteString] -> ZMQ z ()+sendMulti s = liftIO . Z.sendMulti (_unsocket s)++receive :: Z.Receiver t => Socket z t -> ZMQ z ByteString+receive = liftIO . Z.receive . _unsocket++receiveMulti :: Z.Receiver t => Socket z t -> ZMQ z [ByteString]+receiveMulti = liftIO . Z.receiveMulti . _unsocket++subscribe :: Z.Subscriber t => Socket z t -> ByteString -> ZMQ z ()+subscribe s = liftIO . Z.subscribe (_unsocket s)++unsubscribe :: Z.Subscriber t => Socket z t -> ByteString -> ZMQ z ()+unsubscribe s = liftIO . Z.unsubscribe (_unsocket s)++proxy :: Socket z a -> Socket z b -> Maybe (Socket z c) -> ZMQ z ()+proxy a b c = liftIO $ Z.proxy (_unsocket a) (_unsocket b) (_unsocket <$> c)++monitor :: [Z.EventType] -> Socket z t -> ZMQ z (Bool -> IO (Maybe Z.EventMsg))+monitor es s = onContext $ \ctx -> Z.monitor es ctx (_unsocket s)++-- * Socket Options (Read)++affinity :: Socket z t -> ZMQ z Word64+affinity = liftIO . Z.affinity . _unsocket++backlog :: Socket z t -> ZMQ z Int+backlog = liftIO . Z.backlog . _unsocket++delayAttachOnConnect :: Socket z t -> ZMQ z Bool+delayAttachOnConnect = liftIO . Z.delayAttachOnConnect . _unsocket++events :: Socket z t -> ZMQ z [Z.Event]+events = liftIO . Z.events . _unsocket++fileDescriptor :: Socket z t -> ZMQ z Fd+fileDescriptor = liftIO . Z.fileDescriptor . _unsocket++identity :: Socket z t -> ZMQ z ByteString+identity = liftIO . Z.identity . _unsocket++ipv4Only :: Socket z t -> ZMQ z Bool+ipv4Only = liftIO . Z.ipv4Only . _unsocket++lastEndpoint :: Socket z t -> ZMQ z String+lastEndpoint = liftIO . Z.lastEndpoint . _unsocket++linger :: Socket z t -> ZMQ z Int+linger = liftIO . Z.linger . _unsocket++maxMessageSize :: Socket z t -> ZMQ z Int64+maxMessageSize = liftIO . Z.maxMessageSize . _unsocket++mcastHops :: Socket z t -> ZMQ z Int+mcastHops = liftIO . Z.mcastHops . _unsocket++moreToReceive :: Socket z t -> ZMQ z Bool+moreToReceive = liftIO . Z.moreToReceive . _unsocket++rate :: Socket z t -> ZMQ z Int+rate = liftIO . Z.rate . _unsocket++receiveBuffer :: Socket z t -> ZMQ z Int+receiveBuffer = liftIO . Z.receiveBuffer . _unsocket++receiveHighWM :: Socket z t -> ZMQ z Int+receiveHighWM = liftIO . Z.receiveHighWM . _unsocket++receiveTimeout :: Socket z t -> ZMQ z Int+receiveTimeout = liftIO . Z.receiveTimeout . _unsocket++reconnectInterval :: Socket z t -> ZMQ z Int+reconnectInterval = liftIO . Z.reconnectInterval . _unsocket++reconnectIntervalMax :: Socket z t -> ZMQ z Int+reconnectIntervalMax = liftIO . Z.reconnectIntervalMax . _unsocket++recoveryInterval :: Socket z t -> ZMQ z Int+recoveryInterval = liftIO . Z.recoveryInterval . _unsocket++sendBuffer :: Socket z t -> ZMQ z Int+sendBuffer = liftIO . Z.sendBuffer . _unsocket++sendHighWM :: Socket z t -> ZMQ z Int+sendHighWM = liftIO . Z.sendHighWM . _unsocket++sendTimeout :: Socket z t -> ZMQ z Int+sendTimeout = liftIO . Z.sendTimeout . _unsocket++tcpKeepAlive :: Socket z t -> ZMQ z Z.Switch+tcpKeepAlive = liftIO . Z.tcpKeepAlive . _unsocket++tcpKeepAliveCount :: Socket z t -> ZMQ z Int+tcpKeepAliveCount = liftIO . Z.tcpKeepAliveCount . _unsocket++tcpKeepAliveIdle :: Socket z t -> ZMQ z Int+tcpKeepAliveIdle = liftIO . Z.tcpKeepAliveIdle . _unsocket++tcpKeepAliveInterval :: Socket z t -> ZMQ z Int+tcpKeepAliveInterval = liftIO . Z.tcpKeepAliveInterval . _unsocket++-- * Socket Options (Write)++setAffinity :: Word64 -> Socket z t -> ZMQ z ()+setAffinity a = liftIO . Z.setAffinity a . _unsocket++setBacklog :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()+setBacklog b = liftIO . Z.setBacklog b . _unsocket++setDelayAttachOnConnect :: Bool -> Socket z t -> ZMQ z ()+setDelayAttachOnConnect d = liftIO . Z.setDelayAttachOnConnect d . _unsocket++setIdentity :: Restricted N1 N254 ByteString -> Socket z t -> ZMQ z ()+setIdentity i = liftIO . Z.setIdentity i . _unsocket++setIpv4Only :: Bool -> Socket z t -> ZMQ z ()+setIpv4Only i = liftIO . Z.setIpv4Only i . _unsocket++setLinger :: Integral i => Restricted Nneg1 Int32 i -> Socket z t -> ZMQ z ()+setLinger l = liftIO . Z.setLinger l . _unsocket++setMaxMessageSize :: Integral i => Restricted Nneg1 Int64 i -> Socket z t -> ZMQ z ()+setMaxMessageSize s = liftIO . Z.setMaxMessageSize s . _unsocket++setMcastHops :: Integral i => Restricted N1 Int32 i -> Socket z t -> ZMQ z ()+setMcastHops k = liftIO . Z.setMcastHops k . _unsocket++setRate :: Integral i => Restricted N1 Int32 i -> Socket z t -> ZMQ z ()+setRate r = liftIO . Z.setRate r . _unsocket++setReceiveBuffer :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()+setReceiveBuffer k = liftIO . Z.setReceiveBuffer k . _unsocket++setReceiveHighWM :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()+setReceiveHighWM k = liftIO . Z.setReceiveHighWM k . _unsocket++setReceiveTimeout :: Integral i => Restricted Nneg1 Int32 i -> Socket z t -> ZMQ z ()+setReceiveTimeout t = liftIO . Z.setReceiveTimeout t . _unsocket++setReconnectInterval :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()+setReconnectInterval i = liftIO . Z.setReconnectInterval i . _unsocket++setReconnectIntervalMax :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()+setReconnectIntervalMax i = liftIO . Z.setReconnectIntervalMax i . _unsocket++setRecoveryInterval :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()+setRecoveryInterval i = liftIO . Z.setRecoveryInterval i . _unsocket++setRouterMandatory :: Bool -> Socket z Z.Router -> ZMQ z ()+setRouterMandatory b = liftIO . Z.setRouterMandatory b . _unsocket++setSendBuffer :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()+setSendBuffer i = liftIO . Z.setSendBuffer i . _unsocket++setSendHighWM :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()+setSendHighWM i = liftIO . Z.setSendHighWM i . _unsocket++setSendTimeout :: Integral i => Restricted Nneg1 Int32 i -> Socket z t -> ZMQ z ()+setSendTimeout i = liftIO . Z.setSendTimeout i . _unsocket++setTcpAcceptFilter :: Maybe ByteString -> Socket z t -> ZMQ z ()+setTcpAcceptFilter s = liftIO . Z.setTcpAcceptFilter s . _unsocket++setTcpKeepAlive :: Z.Switch -> Socket z t -> ZMQ z ()+setTcpKeepAlive s = liftIO . Z.setTcpKeepAlive s . _unsocket++setTcpKeepAliveCount :: Integral i => Restricted Nneg1 Int32 i -> Socket z t -> ZMQ z ()+setTcpKeepAliveCount c = liftIO . Z.setTcpKeepAliveCount c . _unsocket++setTcpKeepAliveIdle :: Integral i => Restricted Nneg1 Int32 i -> Socket z t -> ZMQ z ()+setTcpKeepAliveIdle i = liftIO . Z.setTcpKeepAliveIdle i . _unsocket++setTcpKeepAliveInterval :: Integral i => Restricted Nneg1 Int32 i -> Socket z t -> ZMQ z ()+setTcpKeepAliveInterval i = liftIO . Z.setTcpKeepAliveInterval i . _unsocket++setXPubVerbose :: Bool -> Socket z Z.XPub -> ZMQ z ()+setXPubVerbose b = liftIO . Z.setXPubVerbose b . _unsocket++-- * Low Level Functions++waitRead :: Socket z t -> ZMQ z ()+waitRead = liftIO . Z.waitRead . _unsocket++waitWrite :: Socket z t -> ZMQ z ()+waitWrite = liftIO . Z.waitWrite . _unsocket++-- * Internal++onContext :: (Z.Context -> IO a) -> ZMQ z a+onContext f = ZMQ $! asks _context >>= liftIO . f++destroy :: ZMQEnv -> IO ()+destroy env = do+ n <- atomicModifyIORef (_refcount env) $ \n -> (pred n, n)+ when (n == 1) $ do+ readIORef (_sockets env) >>= mapM_ close'+ Z.destroy (_context env)+ where+ close' s = I.closeSock s `E.catch` (\e -> print (e :: E.SomeException))
tests/System/ZMQ3/Test/Properties.hs view
@@ -1,130 +1,131 @@-{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -fno-warn-orphans #-} module System.ZMQ3.Test.Properties where -import Control.Applicative-import Test.Framework (Test, testGroup)-import Test.Framework.Providers.QuickCheck2 import Test.QuickCheck import Test.QuickCheck.Monadic+import Test.Tools +import Control.Applicative import Data.Int import Data.Word import Data.Restricted import Data.Maybe (fromJust) import Data.ByteString (ByteString) import Control.Concurrent-import Control.Concurrent.MVar-import System.ZMQ3+import System.ZMQ3.Monadic import System.Posix.Types (Fd(..)) import qualified Data.ByteString as SB import qualified Data.ByteString.Char8 as CB -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 (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)- , testProperty "msg send == msg received (Push/Pull)" (prop_send_receive Push Pull)- , testProperty "msg send == msg received (Pair/Pair)" (prop_send_receive Pair Pair)- , testProperty "publish/subscribe (Pub/Sub)" (prop_pub_sub Pub Sub)- ]- ]+tests :: IO ()+tests = do+ quickBatch' ("0MQ Socket Properties"+ , [ ("get socket option (Pair)", property $ prop_get_socket_option Pair)+ , ("get socket option (Pub)", property $ prop_get_socket_option Pub)+ , ("get socket option (Sub)", property $ prop_get_socket_option Sub)+ , ("get socket option (XPub)", property $ prop_get_socket_option XPub)+ , ("get socket option (XSub)", property $ prop_get_socket_option XSub)+ , ("get socket option (Req)", property $ prop_get_socket_option Req)+ , ("get socket option (Rep)", property $ prop_get_socket_option Rep)+ , ("get socket option (Dealer)", property $ prop_get_socket_option Dealer)+ , ("get socket option (Router)", property $ prop_get_socket_option Router)+ , ("get socket option (Pull)", property $ prop_get_socket_option Pull)+ , ("get socket option (Push)", property $ prop_get_socket_option Push)+ , ("set;get socket option (Pair)", property $ prop_set_get_socket_option Pair)+ , ("set;get socket option (Pub)", property $ prop_set_get_socket_option Pub)+ , ("set;get socket option (Sub)", property $ prop_set_get_socket_option Sub)+ , ("set;get socket option (XPub)", property $ prop_set_get_socket_option XPub)+ , ("set;get socket option (XSub)", property $ prop_set_get_socket_option XSub)+ , ("set;get socket option (Req)", property $ prop_set_get_socket_option Req)+ , ("set;get socket option (Rep)", property $ prop_set_get_socket_option Rep)+ , ("set;get socket option (Dealer)", property $ prop_set_get_socket_option Dealer)+ , ("set;get socket option (Router)", property $ prop_set_get_socket_option Router)+ , ("set;get socket option (Pull)", property $ prop_set_get_socket_option Pull)+ , ("set;get socket option (Push)", property $ prop_set_get_socket_option Push)+ , ("(un-)subscribe", property $ prop_subscribe Sub)+ ]) + quickBatch' ("0MQ Messages"+ , [ ("msg send == msg received (Req/Rep)", property $ prop_send_receive Req Rep)+ , ("msg send == msg received (Push/Pull)", property $ prop_send_receive Push Pull)+ , ("msg send == msg received (Pair/Pair)", property $ prop_send_receive Pair Pair)++ -- , ("publish/subscribe", property $ prop_pub_sub Pub Sub)+ -- (disabled due to LIBZMQ-270 [https://zeromq.jira.com/browse/LIBZMQ-270])+ ])+ prop_get_socket_option :: SocketType t => t -> GetOpt -> Property prop_get_socket_option t opt = monadicIO $ run $ do- withContext 1 $ \c ->- withSocket c t $ \s ->- case opt of- Events _ -> events s >> return ()- Filedesc _ -> fileDescriptor s >> return ()- ReceiveMore _ -> moreToReceive s >> return ()+ runZMQ $ do+ s <- socket t+ case opt of+ Events _ -> events s >> return ()+ Filedesc _ -> fileDescriptor s >> return ()+ ReceiveMore _ -> moreToReceive s >> return () prop_set_get_socket_option :: SocketType t => t -> SetOpt -> Property prop_set_get_socket_option t opt = monadicIO $ do- r <- run $- withContext 1 $ \c ->- withSocket c t $ \s ->- case opt of- Identity val -> (== (rvalue val)) <$> (setIdentity val s >> identity s)- Ipv4Only val -> (== val) <$> (setIpv4Only val s >> ipv4Only s)- Affinity val -> (eq val) <$> (setAffinity val s >> affinity s)- Backlog val -> (eq (rvalue val)) <$> (setBacklog val s >> backlog s)- Linger val -> (eq (rvalue val)) <$> (setLinger val s >> linger s)- Rate val -> (eq (rvalue val)) <$> (setRate val s >> rate s)- ReceiveBuf val -> (eq (rvalue val)) <$> (setReceiveBuffer val s >> receiveBuffer s)- ReconnectIVL val -> (eq (rvalue val)) <$> (setReconnectInterval val s >> reconnectInterval s)- ReconnectIVLMax val -> (eq (rvalue val)) <$> (setReconnectIntervalMax val s >> reconnectIntervalMax s)- RecoveryIVL val -> (eq (rvalue val)) <$> (setRecoveryInterval val s >> recoveryInterval s)- SendBuf val -> (eq (rvalue val)) <$> (setSendBuffer val s >> sendBuffer s)- MaxMessageSize val -> (eq (rvalue val)) <$> (setMaxMessageSize val s >> maxMessageSize s)- McastHops val -> (eq (rvalue val)) <$> (setMcastHops val s >> mcastHops s)- ReceiveHighWM val -> (eq (rvalue val)) <$> (setReceiveHighWM val s >> receiveHighWM s)- ReceiveTimeout val -> (eq (rvalue val)) <$> (setReceiveTimeout val s >> receiveTimeout s)- SendHighWM val -> (eq (rvalue val)) <$> (setSendHighWM val s >> sendHighWM s)- SendTimeout val -> (eq (rvalue val)) <$> (setSendTimeout val s >> sendTimeout s)+ r <- run $ runZMQ $ do+ s <- socket t+ case opt of+ Identity val -> (== (rvalue val)) <$> (setIdentity val s >> identity s)+ Ipv4Only val -> (== val) <$> (setIpv4Only val s >> ipv4Only s)+ Affinity val -> (ieq val) <$> (setAffinity val s >> affinity s)+ Backlog val -> (ieq (rvalue val)) <$> (setBacklog val s >> backlog s)+ Linger val -> (ieq (rvalue val)) <$> (setLinger val s >> linger s)+ Rate val -> (ieq (rvalue val)) <$> (setRate val s >> rate s)+ ReceiveBuf val -> (ieq (rvalue val)) <$> (setReceiveBuffer val s >> receiveBuffer s)+ ReconnectIVL val -> (ieq (rvalue val)) <$> (setReconnectInterval val s >> reconnectInterval s)+ ReconnectIVLMax val -> (ieq (rvalue val)) <$> (setReconnectIntervalMax val s >> reconnectIntervalMax s)+ RecoveryIVL val -> (ieq (rvalue val)) <$> (setRecoveryInterval val s >> recoveryInterval s)+ SendBuf val -> (ieq (rvalue val)) <$> (setSendBuffer val s >> sendBuffer s)+ MaxMessageSize val -> (ieq (rvalue val)) <$> (setMaxMessageSize val s >> maxMessageSize s)+ McastHops val -> (ieq (rvalue val)) <$> (setMcastHops val s >> mcastHops s)+ ReceiveHighWM val -> (ieq (rvalue val)) <$> (setReceiveHighWM val s >> receiveHighWM s)+ ReceiveTimeout val -> (ieq (rvalue val)) <$> (setReceiveTimeout val s >> receiveTimeout s)+ SendHighWM val -> (ieq (rvalue val)) <$> (setSendHighWM val s >> sendHighWM s)+ SendTimeout val -> (ieq (rvalue val)) <$> (setSendTimeout val s >> sendTimeout s) assert r where- eq :: (Integral i, Integral k) => i -> k -> Bool- eq i k = fromIntegral i == fromIntegral k+ ieq :: (Integral i, Integral k) => i -> k -> Bool+ ieq i k = (fromIntegral i :: Int) == (fromIntegral k :: Int) -prop_subscribe :: (Subscriber a, SocketType a) => a -> String -> Property+prop_subscribe :: (Subscriber a, SocketType a) => a -> ByteString -> Property prop_subscribe t subs = monadicIO $ run $- withContext 1 $ \c ->- withSocket c t $ \s -> do- subscribe s subs- unsubscribe s subs+ runZMQ $ do+ s <- socket t+ subscribe s subs+ unsubscribe s subs prop_send_receive :: (SocketType a, SocketType b, Receiver b, Sender a) => a -> b -> ByteString -> Property prop_send_receive a b msg = monadicIO $ do- msg' <- run $ withContext 0 $ \c ->- withSocket c a $ \sender ->- withSocket c b $ \receiver -> do- sync <- newEmptyMVar :: IO (MVar ByteString)- bind receiver "inproc://endpoint"- forkIO $ receive receiver >>= putMVar sync- connect sender "inproc://endpoint"- send sender [] msg- takeMVar sync+ msg' <- run $ runZMQ $ do+ sender <- socket a+ receiver <- socket b+ sync <- liftIO newEmptyMVar+ bind receiver "inproc://endpoint"+ async $ receive receiver >>= liftIO . putMVar sync+ connect sender "inproc://endpoint"+ send sender [] msg+ liftIO $ takeMVar sync assert (msg == msg') prop_pub_sub :: (SocketType a, Subscriber b, SocketType b, Sender a, Receiver b) => a -> b -> ByteString -> Property prop_pub_sub a b msg = monadicIO $ do- msg' <- run $ withContext 0 $ \c ->- withSocket c a $ \pub ->- withSocket c b $ \sub -> do- subscribe sub ""- bind sub "inproc://endpoint"- connect pub "inproc://endpoint"- send pub [] msg- receive sub+ msg' <- run $ runZMQ $ do+ pub <- socket a+ sub <- socket b+ subscribe sub ""+ bind sub "inproc://endpoint"+ connect pub "inproc://endpoint"+ send pub [] msg+ receive sub assert (msg == msg') instance Arbitrary ByteString where- arbitrary = CB.pack <$> arbitrary+ arbitrary = CB.pack . filter (/= '\0') <$> arbitrary data GetOpt = Events Int@@ -135,7 +136,7 @@ data SetOpt = Affinity Word64 | Backlog (Restricted N0 Int32 Int)- | Identity (Restricted N1 N254 String)+ | Identity (Restricted N1 N254 ByteString) | Ipv4Only Bool | Linger (Restricted Nneg1 Int32 Int) | MaxMessageSize (Restricted Nneg1 Int64 Int64)@@ -177,7 +178,7 @@ , SendHighWM . toR0 <$> (arbitrary :: Gen Int32) `suchThat` (>= 0) , SendTimeout . toRneg1 <$> (arbitrary :: Gen Int32) `suchThat` (>= -1) , MaxMessageSize . toRneg1' <$> (arbitrary :: Gen Int64) `suchThat` (>= -1)- , Identity . fromJust . toRestricted . show <$> arbitrary `suchThat` (\s -> SB.length s > 0 && SB.length s < 255)+ , Identity . fromJust . toRestricted <$> arbitrary `suchThat` (\s -> SB.length s > 0 && SB.length s < 255) ] toR1 :: Int32 -> Restricted N1 Int32 Int
+ tests/Test/Tools.hs view
@@ -0,0 +1,35 @@+module Test.Tools (quickBatch', checkBatch') where++import Control.Monad+import System.Console.ANSI+import System.Exit+import Test.QuickCheck+import Test.QuickCheck.Checkers+import qualified Control.Exception as E++quickBatch' :: TestBatch -> IO ()+quickBatch' = checkBatch' (stdArgs { maxSuccess = 500 })++checkBatch' :: Args -> TestBatch -> IO ()+checkBatch' args (name, tsts) = do+ writeLn Cyan name+ forM_ tsts $ \(s, p) -> do+ write White (" " ++ s ++ ": ")+ r <- quickCheckWithResult (args { chatty = False}) p+ `E.catch`+ ((\e -> write Red (show e) >> exitFailure) :: E.SomeException -> IO a)+ case r of+ Success _ _ m -> write Green m+ GaveUp _ _ m -> write Magenta m >> exitFailure+ Failure _ _ _ _ _ _ _ m -> write Red m >> exitFailure+ NoExpectedFailure _ _ m -> write Red m >> exitFailure++write, writeLn :: Color -> String -> IO ()+write c = withColour c . putStr+writeLn c = withColour c . putStrLn++withColour :: Color -> IO () -> IO ()+withColour c a = do+ setSGR [Reset, SetColor Foreground Vivid c]+ a+ setSGR [Reset]
tests/tests.hs view
@@ -1,6 +1,5 @@-import Test.Framework (defaultMain) import qualified System.ZMQ3.Test.Properties as Properties main :: IO ()-main = defaultMain Properties.tests+main = Properties.tests
zeromq3-haskell.cabal view
@@ -1,5 +1,5 @@ name: zeromq3-haskell-version: 0.2+version: 0.3.0 synopsis: Bindings to ZeroMQ 3.x category: System, FFI license: MIT@@ -32,7 +32,7 @@ library hs-source-dirs: src- exposed-modules: System.ZMQ3, Data.Restricted+ exposed-modules: System.ZMQ3, System.ZMQ3.Monadic, Data.Restricted other-modules: System.ZMQ3.Base , System.ZMQ3.Internal , System.ZMQ3.Error@@ -43,7 +43,8 @@ build-depends: base >= 3 && < 5 , containers , bytestring-+ , transformers+ , MonadCatchIO-transformers if os(freebsd) extra-libraries: zmq, pthread else@@ -53,13 +54,16 @@ type: exitcode-stdio-1.0 hs-source-dirs: tests main-is: tests.hs+ other-modules: Test.Tools build-depends: zeromq3-haskell , base >= 3 && < 5 , containers , bytestring- , test-framework >= 0.4- , test-framework-quickcheck2 >= 0.2- , QuickCheck >= 2.4+ , transformers+ , MonadCatchIO-transformers+ , QuickCheck >= 2.6+ , checkers >= 0.3+ , ansi-terminal >= 0.6 ghc-options: -Wall -threaded -rtsopts source-repository head