diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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 - diff --git a/examples/Makefile b/examples/Makefile --- a/examples/Makefile +++ b/examples/Makefile @@ -6,6 +6,6 @@ .PHONY: clean clean: - @rm *.o *.hi - @rm display prompt + -rm -f *.o *.hi + -rm -f display prompt diff --git a/examples/display.hs b/examples/display.hs --- a/examples/display.hs +++ b/examples/display.hs @@ -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
[
, ...]" 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 diff --git a/examples/echo.hs b/examples/echo.hs deleted file mode 100644 --- a/examples/echo.hs +++ /dev/null @@ -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 [] diff --git a/examples/perf/Makefile b/examples/perf/Makefile --- a/examples/perf/Makefile +++ b/examples/perf/Makefile @@ -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 diff --git a/examples/perf/local_lat.hs b/examples/perf/local_lat.hs --- a/examples/perf/local_lat.hs +++ b/examples/perf/local_lat.hs @@ -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 diff --git a/examples/perf/local_thr.hs b/examples/perf/local_thr.hs --- a/examples/perf/local_thr.hs +++ b/examples/perf/local_thr.hs @@ -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 () diff --git a/examples/perf/remote_lat.hs b/examples/perf/remote_lat.hs --- a/examples/perf/remote_lat.hs +++ b/examples/perf/remote_lat.hs @@ -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 diff --git a/examples/perf/remote_thr.hs b/examples/perf/remote_thr.hs --- a/examples/perf/remote_thr.hs +++ b/examples/perf/remote_thr.hs @@ -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 " diff --git a/examples/prompt.hs b/examples/prompt.hs --- a/examples/prompt.hs +++ b/examples/prompt.hs @@ -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
" 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) diff --git a/examples/threads.hs b/examples/threads.hs deleted file mode 100644 --- a/examples/threads.hs +++ /dev/null @@ -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 - diff --git a/src/Data/Restricted.hs b/src/Data/Restricted.hs --- a/src/Data/Restricted.hs +++ b/src/Data/Restricted.hs @@ -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 diff --git a/src/System/ZMQ3.hs b/src/System/ZMQ3.hs --- a/src/System/ZMQ3.hs +++ b/src/System/ZMQ3.hs @@ -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 diff --git a/src/System/ZMQ3/Base.hsc b/src/System/ZMQ3/Base.hsc --- a/src/System/ZMQ3/Base.hsc +++ b/src/System/ZMQ3/Base.hsc @@ -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 diff --git a/src/System/ZMQ3/Internal.hs b/src/System/ZMQ3/Internal.hs --- a/src/System/ZMQ3/Internal.hs +++ b/src/System/ZMQ3/Internal.hs @@ -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 diff --git a/src/System/ZMQ3/Monadic.hs b/src/System/ZMQ3/Monadic.hs new file mode 100644 --- /dev/null +++ b/src/System/ZMQ3/Monadic.hs @@ -0,0 +1,475 @@ +{-# LANGUAGE RankNTypes #-} +-- | +-- Module : System.ZMQ3.Monadic +-- Copyright : (c) 2013 Toralf Wittner +-- License : MIT +-- Maintainer : Toralf Wittner +-- 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)) diff --git a/tests/System/ZMQ3/Test/Properties.hs b/tests/System/ZMQ3/Test/Properties.hs --- a/tests/System/ZMQ3/Test/Properties.hs +++ b/tests/System/ZMQ3/Test/Properties.hs @@ -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 diff --git a/tests/Test/Tools.hs b/tests/Test/Tools.hs new file mode 100644 --- /dev/null +++ b/tests/Test/Tools.hs @@ -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] diff --git a/tests/tests.hs b/tests/tests.hs --- a/tests/tests.hs +++ b/tests/tests.hs @@ -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 diff --git a/zeromq3-haskell.cabal b/zeromq3-haskell.cabal --- a/zeromq3-haskell.cabal +++ b/zeromq3-haskell.cabal @@ -1,5 +1,5 @@ name: zeromq3-haskell -version: 0.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