diff --git a/AUTHORS b/AUTHORS --- a/AUTHORS +++ b/AUTHORS @@ -5,4 +5,4 @@ Jeremy Fitzhardinge integrated with GHC's I/O manager Bryan O'Sullivan added resource wrappers, socket finalizer Ben Lever fixed and tweaked some of the test examples - +Alexander Vershilov added support for zmq_proxy diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -5,6 +5,11 @@ This software currently has *beta* status, i.e. it had seen limited testing. +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 + I/O threads as first argument. + Version 0.1.4 - Expose 'waitRead' and 'waitWrite'. Version 0.1.3 - Deprecated 'Xreq', 'XRep' in favour of 'Dealer' and 'Router' @@ -62,5 +67,5 @@ If you find any bugs or other shortcomings I would greatly appreciate a bug report, preferably via http://github.com/twittner/zeromq-haskell/issues or -e-mail to toralf.wittner@gmail.com +e-mail to tw@dtex.org diff --git a/examples/display.hs b/examples/display.hs --- a/examples/display.hs +++ b/examples/display.hs @@ -5,6 +5,7 @@ import Control.Exception import qualified System.ZMQ3 as ZMQ import qualified Data.ByteString as SB +import qualified Data.ByteString.Char8 as CS main :: IO () main = do @@ -12,12 +13,12 @@ when (length args < 1) $ do hPutStrLn stderr "usage: display
[
, ...]" exitFailure - ZMQ.withContext 1 $ \c -> + ZMQ.withContext $ \c -> ZMQ.withSocket c ZMQ.Sub $ \s -> do ZMQ.subscribe s "" mapM (ZMQ.connect s) args forever $ do line <- ZMQ.receive s - SB.putStrLn line + CS.putStrLn line hFlush stdout diff --git a/examples/echo.hs b/examples/echo.hs new file mode 100644 --- /dev/null +++ b/examples/echo.hs @@ -0,0 +1,22 @@ +{-# 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/local_lat.hs b/examples/perf/local_lat.hs --- a/examples/perf/local_lat.hs +++ b/examples/perf/local_lat.hs @@ -14,7 +14,7 @@ let bindTo = args !! 0 size = read $ args !! 1 rounds = read $ args !! 2 - ZMQ.withContext 1 $ \c -> + ZMQ.withContext $ \c -> ZMQ.withSocket c ZMQ.Rep $ \s -> do ZMQ.bind s bindTo loop s rounds size 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 @@ -16,7 +16,7 @@ let bindTo = args !! 0 size = read $ args !! 1 :: Int count = read $ args !! 2 :: Int - ZMQ.withContext 1 $ \c -> + ZMQ.withContext $ \c -> ZMQ.withSocket c ZMQ.Sub $ \s -> do ZMQ.subscribe s "" ZMQ.bind s bindTo 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 @@ -16,7 +16,7 @@ size = read $ args !! 1 rounds = read $ args !! 2 message = SB.replicate size 0x65 - ZMQ.withContext 1 $ \c -> + ZMQ.withContext $ \c -> ZMQ.withSocket c ZMQ.Req $ \s -> do ZMQ.connect s connTo start <- getCurrentTime 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 @@ -16,7 +16,7 @@ size = read $ args !! 1 count = read $ args !! 2 message = SB.replicate size 0x65 - ZMQ.withContext 1 $ \c -> + ZMQ.withContext $ \c -> ZMQ.withSocket c ZMQ.Pub $ \s -> do ZMQ.connect s connTo replicateM_ count $ ZMQ.send s [] message diff --git a/examples/prompt.hs b/examples/prompt.hs --- a/examples/prompt.hs +++ b/examples/prompt.hs @@ -16,7 +16,7 @@ exitFailure let addr = args !! 0 name = SB.append (SB.fromString $ args !! 1) ": " - ZMQ.withContext 1 $ \c -> + ZMQ.withContext $ \c -> ZMQ.withSocket c ZMQ.Pub $ \s -> do ZMQ.bind s addr forever $ do diff --git a/examples/threads.hs b/examples/threads.hs new file mode 100644 --- /dev/null +++ b/examples/threads.hs @@ -0,0 +1,56 @@ +{-# 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 @@ -8,7 +8,7 @@ -- Module : Data.Restricted -- Copyright : (c) 2011-2012 Toralf Wittner -- License : MIT --- Maintainer : toralf.wittner@gmail.com +-- Maintainer : Toralf Wittner -- Stability : experimental -- Portability : non-portable -- diff --git a/src/System/ZMQ3.hs b/src/System/ZMQ3.hs --- a/src/System/ZMQ3.hs +++ b/src/System/ZMQ3.hs @@ -2,7 +2,7 @@ -- Module : System.ZMQ3 -- Copyright : (c) 2010-2012 Toralf Wittner -- License : MIT --- Maintainer : toralf.wittner@gmail.com +-- Maintainer : Toralf Wittner -- Stability : experimental -- Portability : non-portable -- @@ -75,8 +75,11 @@ , Context , Socket , Flag (SendMore) + , Switch (..) , Timeout , Event (..) + , EventType (..) + , EventMsg (..) -- ** Type Classes , SocketType @@ -103,6 +106,7 @@ , withContext , withSocket , bind + , unbind , connect , send , send' @@ -110,17 +114,28 @@ , receive , receiveMulti , version + , monitor , System.ZMQ3.subscribe , System.ZMQ3.unsubscribe + -- * Context Options (Read) + , ioThreads + , maxSockets + + -- * Context Options (Write) + , setIoThreads + , setMaxSockets + -- * Socket Options (Read) , System.ZMQ3.affinity , System.ZMQ3.backlog + , System.ZMQ3.delayAttachOnConnect , System.ZMQ3.events , System.ZMQ3.fileDescriptor , System.ZMQ3.identity , System.ZMQ3.ipv4Only + , System.ZMQ3.lastEndpoint , System.ZMQ3.linger , System.ZMQ3.maxMessageSize , System.ZMQ3.mcastHops @@ -135,25 +150,37 @@ , System.ZMQ3.sendBuffer , System.ZMQ3.sendHighWM , System.ZMQ3.sendTimeout + , System.ZMQ3.tcpKeepAlive + , System.ZMQ3.tcpKeepAliveCount + , System.ZMQ3.tcpKeepAliveIdle + , System.ZMQ3.tcpKeepAliveInterval -- * Socket Options (Write) , setAffinity , setBacklog + , setDelayAttachOnConnect , setIdentity + , setIpv4Only , setLinger + , setMaxMessageSize + , setMcastHops , setRate , setReceiveBuffer + , setReceiveHighWM + , setReceiveTimeout , setReconnectInterval , setReconnectIntervalMax , setRecoveryInterval + , setRouterMandatory , setSendBuffer - , setIpv4Only - , setMcastHops - , setReceiveHighWM - , setReceiveTimeout , setSendHighWM , setSendTimeout - , setMaxMessageSize + , setTcpAcceptFilter + , setTcpKeepAlive + , setTcpKeepAliveCount + , setTcpKeepAliveIdle + , setTcpKeepAliveInterval + , setXPubVerbose -- * Restrictions , Data.Restricted.restrict @@ -168,21 +195,25 @@ -- * Low-level Functions , init , term + , context + , destroy , socket , close , waitRead , waitWrite + -- * Utils + , proxy ) where import Prelude hiding (init) import qualified Prelude as P import Control.Applicative import Control.Exception -import Control.Monad (unless, when) +import Control.Monad (unless, when, void) import Data.Restricted import Data.IORef (atomicModifyIORef) -import Foreign hiding (throwIf, throwIf_, throwIfNull) +import Foreign hiding (throwIf, throwIf_, throwIfNull, void) import Foreign.C.String import Foreign.C.Types (CInt) import qualified Data.ByteString as SB @@ -353,25 +384,34 @@ where tupleUp a b c = (fromIntegral a, fromIntegral b, fromIntegral c) --- | Initialize a 0MQ context (cf. zmq_init for details). You should --- normally prefer to use 'withContext' instead. init :: Size -> IO Context -init ioThreads = do - c <- throwIfNull "init" $ c_zmq_init (fromIntegral ioThreads) - return (Context c) +init n = do + c <- context + setIoThreads n c + return c +{-# DEPRECATED init "Use context" #-} --- | Terminate a 0MQ context (cf. zmq_term). You should normally --- prefer to use 'withContext' instead. +-- | Initialize a 0MQ context (cf. zmq_ctx_new for details). You should +-- normally prefer to use 'withContext' instead. +context :: IO Context +context = Context <$> throwIfNull "init" c_zmq_ctx_new + term :: Context -> IO () -term = throwIfMinus1Retry_ "term" . c_zmq_term . ctx +term = destroy +{-# DEPRECATED term "Use destroy" #-} +-- | 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 + -- | Run an action with a 0MQ context. The 'Context' supplied to your -- action will /not/ be valid after the action either returns or -- throws an exception. -withContext :: Size -> (Context -> IO a) -> IO a -withContext ioThreads act = - bracket (throwIfNull "withContext (init)" $ c_zmq_init (fromIntegral ioThreads)) - (throwIfMinus1Retry_ "withContext (term)" . c_zmq_term) +withContext :: (Context -> IO a) -> IO a +withContext act = + bracket (throwIfNull "withContext (new)" $ c_zmq_ctx_new) + (throwIfMinus1Retry_ "withContext (destroy)" . c_zmq_ctx_destroy) (act . Context) -- | Run an action with a 0MQ socket. The socket will be closed after running @@ -418,6 +458,14 @@ -- Read +-- | Cf. @zmq_ctx_get ZMQ_IO_THREADS@ +ioThreads :: Context -> IO Word +ioThreads = ctxIntOption "ioThreads" _ioThreads + +-- | Cf. @zmq_ctx_get ZMQ_MAX_SOCKETS@ +maxSockets :: Context -> IO Word +maxSockets = ctxIntOption "maxSockets" _maxSockets + -- | Cf. @zmq_getsockopt ZMQ_IDENTITY@ identity :: Socket a -> IO String identity s = getStrOpt s B.identity @@ -438,10 +486,18 @@ backlog :: Socket a -> IO Int backlog = getInt32Option B.backlog +-- | Cf. @zmq_getsockopt ZMQ_DELAY_ATTACH_ON_CONNECT@ +delayAttachOnConnect :: Socket a -> IO Bool +delayAttachOnConnect s = (== 1) <$> getInt32Option B.delayAttachOnConnect s + -- | Cf. @zmq_getsockopt ZMQ_LINGER@ linger :: Socket a -> IO Int linger = getInt32Option B.linger +-- | Cf. @zmq_getsockopt ZMQ_LAST_ENDPOINT@ +lastEndpoint :: Socket a -> IO String +lastEndpoint s = getStrOpt s B.lastEndpoint + -- | Cf. @zmq_getsockopt ZMQ_RATE@ rate :: Socket a -> IO Int rate = getInt32Option B.rate @@ -486,8 +542,35 @@ sendHighWM :: Socket a -> IO Int sendHighWM = getInt32Option B.sendHighWM +-- | Cf. @zmq_getsockopt ZMQ_TCP_KEEPALIVE@ +tcpKeepAlive :: Socket a -> IO Switch +tcpKeepAlive s = getInt32Option B.tcpKeepAlive s >>= convert . toSwitch + where + convert Nothing = throwError "Invalid value for ZMQ_TCP_KEEPALIVE" + convert (Just i) = return i + +-- | Cf. @zmq_getsockopt ZMQ_TCP_KEEPALIVE_CNT@ +tcpKeepAliveCount :: Socket a -> IO Int +tcpKeepAliveCount = getInt32Option B.tcpKeepAliveCount + +-- | Cf. @zmq_getsockopt ZMQ_TCP_KEEPALIVE_IDLE@ +tcpKeepAliveIdle :: Socket a -> IO Int +tcpKeepAliveIdle = getInt32Option B.tcpKeepAliveIdle + +-- | Cf. @zmq_getsockopt ZMQ_TCP_KEEPALIVE_INTVL@ +tcpKeepAliveInterval :: Socket a -> IO Int +tcpKeepAliveInterval = getInt32Option B.tcpKeepAliveInterval + -- Write +-- | Cf. @zmq_ctx_set ZMQ_IO_THREADS@ +setIoThreads :: Word -> Context -> IO () +setIoThreads n = setCtxIntOption "ioThreads" _ioThreads n + +-- | Cf. @zmq_ctx_set ZMQ_MAX_SOCKETS@ +setMaxSockets :: Word -> Context -> IO () +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) @@ -496,14 +579,17 @@ setAffinity :: Word64 -> Socket a -> IO () setAffinity x s = setIntOpt s B.affinity x +-- | Cf. @zmq_setsockopt ZMQ_DELAY_ATTACH_ON_CONNECT@ +setDelayAttachOnConnect :: Bool -> Socket a -> IO () +setDelayAttachOnConnect x s = setIntOpt s B.delayAttachOnConnect (bool2cint x) + -- | Cf. @zmq_setsockopt ZMQ_MAXMSGSIZE@ setMaxMessageSize :: Integral i => Restricted Nneg1 Int64 i -> Socket a -> IO () setMaxMessageSize x s = setIntOpt s B.maxMessageSize ((fromIntegral . rvalue $ x) :: Int64) -- | Cf. @zmq_setsockopt ZMQ_IPV4ONLY@ setIpv4Only :: Bool -> Socket a -> IO () -setIpv4Only True s = setIntOpt s B.ipv4Only (1 :: CInt) -setIpv4Only False s = setIntOpt s B.ipv4Only (0 :: CInt) +setIpv4Only x s = setIntOpt s B.ipv4Only (bool2cint x) -- | Cf. @zmq_setsockopt ZMQ_LINGER@ setLinger :: Integral i => Restricted Nneg1 Int32 i -> Socket a -> IO () @@ -513,6 +599,10 @@ setReceiveTimeout :: Integral i => Restricted Nneg1 Int32 i -> Socket a -> IO () setReceiveTimeout = setInt32OptFromRestricted B.receiveTimeout +-- | Cf. @zmq_setsockopt ZMQ_ROUTER_MANDATORY@ +setRouterMandatory :: Bool -> Socket Router -> IO () +setRouterMandatory x s = setIntOpt s B.routerMandatory (bool2cint x) + -- | Cf. @zmq_setsockopt ZMQ_SNDTIMEO@ setSendTimeout :: Integral i => Restricted Nneg1 Int32 i -> Socket a -> IO () setSendTimeout = setInt32OptFromRestricted B.sendTimeout @@ -557,11 +647,48 @@ setSendHighWM :: Integral i => Restricted N0 Int32 i -> Socket a -> IO () setSendHighWM = setInt32OptFromRestricted B.sendHighWM +-- | Cf. @zmq_setsockopt ZMQ_TCP_ACCEPT_FILTER@ +setTcpAcceptFilter :: Maybe String -> 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) + +-- | Cf. @zmq_setsockopt ZMQ_TCP_KEEPALIVE@ +setTcpKeepAlive :: Switch -> Socket a -> IO () +setTcpKeepAlive x s = setIntOpt s B.tcpKeepAlive (fromSwitch x :: CInt) + +-- | Cf. @zmq_setsockopt ZMQ_TCP_KEEPALIVE_CNT@ +setTcpKeepAliveCount :: Integral i => Restricted Nneg1 Int32 i -> Socket a -> IO () +setTcpKeepAliveCount = setInt32OptFromRestricted B.tcpKeepAliveCount + +-- | Cf. @zmq_setsockopt ZMQ_TCP_KEEPALIVE_IDLE@ +setTcpKeepAliveIdle :: Integral i => Restricted Nneg1 Int32 i -> Socket a -> IO () +setTcpKeepAliveIdle = setInt32OptFromRestricted B.tcpKeepAliveIdle + +-- | Cf. @zmq_setsockopt ZMQ_TCP_KEEPALIVE_INTVL@ +setTcpKeepAliveInterval :: Integral i => Restricted Nneg1 Int32 i -> Socket a -> IO () +setTcpKeepAliveInterval = setInt32OptFromRestricted B.tcpKeepAliveInterval + +-- | Cf. @zmq_setsockopt ZMQ_XPUB_VERBOSE@ +setXPubVerbose :: Bool -> Socket XPub -> IO () +setXPubVerbose x s = setIntOpt s B.xpubVerbose (bool2cint x) + -- | 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 +-- | 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 + -- | Connect the socket to the given address (cf. zmq_connect). connect :: Socket a -> String -> IO () connect sock str = onSocket "connect" sock $ @@ -629,6 +756,39 @@ next acc True = recvall acc next acc False = return (reverse acc) +-- | Setup socket monitoring, i.e. a 'Pair' socket which +-- sends monitoring events about the given 'Socket' to the +-- given address. +socketMonitor :: [EventType] -> String -> Socket a -> IO () +socketMonitor es addr soc = onSocket "socketMonitor" soc $ \s -> + withCString addr $ \a -> + throwIfMinus1_ "zmq_socket_monitor" $ + c_zmq_socket_monitor s a (events2cint es) + +-- | Monitor socket events. +-- This function returns a function which can be invoked to retrieve +-- the next socket event, potentially blocking until the next one becomes +-- available. When applied to 'False', monitoring will terminate, i.e. +-- internal monitoring resources will be disposed. Consequently after +-- 'monitor' has been invoked, the returned function must be applied +-- /once/ to 'False'. +monitor :: [EventType] -> Context -> Socket a -> IO (Bool -> IO (Maybe EventMsg)) +monitor es ctx sock = do + let addr = "inproc://" ++ show (_socket sock) + s <- socket ctx Pair + socketMonitor es addr sock + connect s addr + next s <$> messageInit + where + next soc m False = messageClose m `finally` close soc >> return Nothing + next soc m True = onSocket "recv" soc $ \s -> do + retry "recv" (waitRead soc) $ c_zmq_recvmsg s (msgPtr m) (flagVal dontWait) + ptr <- c_zmq_msg_data (msgPtr m) + str <- peekByteOff ptr zmqEventAddrOffset >>= SB.packCString + dat <- peekByteOff ptr zmqEventDataOffset :: IO CInt + 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 @@ -662,3 +822,18 @@ waitWrite :: Socket a -> IO () waitWrite = wait' threadWaitWrite pollOut +-- | Starts built-in 0MQ proxy. +-- +-- Proxy connects front to back socket +-- +-- Before calling proxy all sockets should be binded +-- +-- If the capture socket is not Nothing, the proxy shall send all +-- messages, received on both frontend and backend, to the capture socket. +proxy :: Socket a -> Socket b -> Maybe (Socket c) -> IO () +proxy front back capture = + onSocket "proxy-front" front $ \f -> + onSocket "proxy-back" back $ \b -> + void (c_zmq_proxy f b c) + where + c = maybe nullPtr _socket 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 @@ -1,13 +1,4 @@ {-# LANGUAGE CPP, ForeignFunctionInterface #-} --- | --- Module : System.ZMQ3.Base --- Copyright : (c) 2010-2012 Toralf Wittner --- License : MIT --- Maintainer : toralf.wittner@gmail.com --- Stability : experimental --- Portability : non-portable --- - module System.ZMQ3.Base where import Foreign @@ -21,7 +12,7 @@ #error *** INVALID 0MQ VERSION (must be 3.x) *** #endif -newtype ZMQMsg = ZMQMsg { content :: Ptr () } +newtype ZMQMsg = ZMQMsg { content :: Ptr () } deriving (Eq, Ord) instance Storable ZMQMsg where alignment _ = #{alignment zmq_msg_t} @@ -30,10 +21,10 @@ poke p (ZMQMsg c) = #{poke zmq_msg_t, _} p c data ZMQPoll = ZMQPoll - { pSocket :: ZMQSocket - , pFd :: CInt - , pEvents :: CShort - , pRevents :: CShort + { pSocket :: {-# UNPACK #-} !ZMQSocket + , pFd :: {-# UNPACK #-} !CInt + , pEvents :: {-# UNPACK #-} !CShort + , pRevents :: {-# UNPACK #-} !CShort } instance Storable ZMQPoll where @@ -61,46 +52,82 @@ newtype ZMQSocketType = ZMQSocketType { typeVal :: CInt } deriving (Eq, Ord) #{enum ZMQSocketType, ZMQSocketType - , pair = ZMQ_PAIR - , pub = ZMQ_PUB - , sub = ZMQ_SUB - , xpub = ZMQ_XPUB - , xsub = ZMQ_XSUB - , request = ZMQ_REQ - , response = ZMQ_REP - , dealer = ZMQ_DEALER - , router = ZMQ_ROUTER - , pull = ZMQ_PULL - , push = ZMQ_PUSH + , pair = ZMQ_PAIR + , pub = ZMQ_PUB + , sub = ZMQ_SUB + , xpub = ZMQ_XPUB + , xsub = ZMQ_XSUB + , request = ZMQ_REQ + , response = ZMQ_REP + , dealer = ZMQ_DEALER + , router = ZMQ_ROUTER + , pull = ZMQ_PULL + , push = ZMQ_PUSH } newtype ZMQOption = ZMQOption { optVal :: CInt } deriving (Eq, Ord) #{enum ZMQOption, ZMQOption - , affinity = ZMQ_AFFINITY - , backlog = ZMQ_BACKLOG - , events = ZMQ_EVENTS - , filedesc = ZMQ_FD - , identity = ZMQ_IDENTITY - , ipv4Only = ZMQ_IPV4ONLY - , linger = ZMQ_LINGER - , maxMessageSize = ZMQ_MAXMSGSIZE - , mcastHops = ZMQ_MULTICAST_HOPS - , rate = ZMQ_RATE - , receiveBuf = ZMQ_RCVBUF - , receiveHighWM = ZMQ_RCVHWM - , receiveMore = ZMQ_RCVMORE - , receiveTimeout = ZMQ_RCVTIMEO - , reconnectIVL = ZMQ_RECONNECT_IVL - , reconnectIVLMax = ZMQ_RECONNECT_IVL_MAX - , recoveryIVL = ZMQ_RECOVERY_IVL - , sendBuf = ZMQ_SNDBUF - , sendHighWM = ZMQ_SNDHWM - , sendTimeout = ZMQ_SNDTIMEO - , subscribe = ZMQ_SUBSCRIBE - , unsubscribe = ZMQ_UNSUBSCRIBE + , affinity = ZMQ_AFFINITY + , backlog = ZMQ_BACKLOG + , delayAttachOnConnect = ZMQ_DELAY_ATTACH_ON_CONNECT + , events = ZMQ_EVENTS + , filedesc = ZMQ_FD + , identity = ZMQ_IDENTITY + , ipv4Only = ZMQ_IPV4ONLY + , lastEndpoint = ZMQ_LAST_ENDPOINT + , linger = ZMQ_LINGER + , maxMessageSize = ZMQ_MAXMSGSIZE + , mcastHops = ZMQ_MULTICAST_HOPS + , rate = ZMQ_RATE + , receiveBuf = ZMQ_RCVBUF + , receiveHighWM = ZMQ_RCVHWM + , receiveMore = ZMQ_RCVMORE + , receiveTimeout = ZMQ_RCVTIMEO + , reconnectIVL = ZMQ_RECONNECT_IVL + , reconnectIVLMax = ZMQ_RECONNECT_IVL_MAX + , recoveryIVL = ZMQ_RECOVERY_IVL + , routerMandatory = ZMQ_ROUTER_MANDATORY + , sendBuf = ZMQ_SNDBUF + , sendHighWM = ZMQ_SNDHWM + , sendTimeout = ZMQ_SNDTIMEO + , subscribe = ZMQ_SUBSCRIBE + , tcpAcceptFilter = ZMQ_TCP_ACCEPT_FILTER + , tcpKeepAlive = ZMQ_TCP_KEEPALIVE + , tcpKeepAliveCount = ZMQ_TCP_KEEPALIVE_CNT + , tcpKeepAliveIdle = ZMQ_TCP_KEEPALIVE_IDLE + , tcpKeepAliveInterval = ZMQ_TCP_KEEPALIVE_INTVL + , unsubscribe = ZMQ_UNSUBSCRIBE + , xpubVerbose = ZMQ_XPUB_VERBOSE } +newtype ZMQCtxOption = ZMQCtxOption { ctxOptVal :: CInt } deriving (Eq, Ord) + +#{enum ZMQCtxOption, ZMQCtxOption + , _ioThreads = ZMQ_IO_THREADS + , _maxSockets = ZMQ_MAX_SOCKETS +} + +newtype ZMQEventType = ZMQEventType { eventTypeVal :: CInt } deriving (Eq, Ord) + +#{enum ZMQEventType, ZMQEventType + , connected = ZMQ_EVENT_CONNECTED + , connectDelayed = ZMQ_EVENT_CONNECT_DELAYED + , connectRetried = ZMQ_EVENT_CONNECT_RETRIED + , listening = ZMQ_EVENT_LISTENING + , bindFailed = ZMQ_EVENT_BIND_FAILED + , accepted = ZMQ_EVENT_ACCEPTED + , acceptFailed = ZMQ_EVENT_ACCEPT_FAILED + , closed = ZMQ_EVENT_CLOSED + , closeFailed = ZMQ_EVENT_CLOSE_FAILED + , disconnected = ZMQ_EVENT_DISCONNECTED + , allEvents = ZMQ_EVENT_ALL +} + +zmqEventAddrOffset, zmqEventDataOffset :: Int +zmqEventAddrOffset = #{offset zmq_event_t, data.connected.addr} +zmqEventDataOffset = #{offset zmq_event_t, data.connected.fd} + newtype ZMQMsgOption = ZMQMsgOption { msgOptVal :: CInt } deriving (Eq, Ord) #{enum ZMQMsgOption, ZMQMsgOption @@ -128,12 +155,18 @@ foreign import ccall unsafe "zmq.h zmq_version" c_zmq_version :: Ptr CInt -> Ptr CInt -> Ptr CInt -> IO () -foreign import ccall unsafe "zmq.h zmq_init" - c_zmq_init :: CInt -> IO ZMQCtx +foreign import ccall unsafe "zmq.h zmq_ctx_new" + c_zmq_ctx_new :: IO ZMQCtx -foreign import ccall unsafe "zmq.h zmq_term" - c_zmq_term :: ZMQCtx -> IO CInt +foreign import ccall unsafe "zmq.h zmq_ctx_destroy" + c_zmq_ctx_destroy :: ZMQCtx -> IO CInt +foreign import ccall unsafe "zmq.h zmq_ctx_get" + c_zmq_ctx_get :: ZMQCtx -> CInt -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_ctx_set" + c_zmq_ctx_set :: ZMQCtx -> CInt -> CInt -> IO CInt + -- zmq_msg_t related foreign import ccall unsafe "zmq.h zmq_msg_init" @@ -151,21 +184,11 @@ foreign import ccall unsafe "zmq.h zmq_msg_size" c_zmq_msg_size :: ZMQMsgPtr -> IO CSize -#if 0 foreign import ccall unsafe "zmq.h zmq_msg_get" - c_zmq_msg_get :: ZMQMsgPtr - -> CInt -- option - -> Ptr () -- option value - -> Ptr CSize -- option value size ptr - -> IO CInt + c_zmq_msg_get :: ZMQMsgPtr -> CInt -> IO CInt foreign import ccall unsafe "zmq.h zmq_msg_set" - c_zmq_msg_set :: ZMQMsgPtr - -> CInt -- option - -> Ptr () -- option value - -> Ptr CSize -- option value size ptr - -> IO CInt -#endif + c_zmq_msg_set :: ZMQMsgPtr -> CInt -> CInt -> IO CInt -- socket @@ -192,6 +215,9 @@ foreign import ccall unsafe "zmq.h zmq_bind" c_zmq_bind :: ZMQSocket -> CString -> IO CInt +foreign import ccall unsafe "zmq.h zmq_unbind" + c_zmq_unbind :: ZMQSocket -> CString -> IO CInt + foreign import ccall unsafe "zmq.h zmq_connect" c_zmq_connect :: ZMQSocket -> CString -> IO CInt @@ -201,8 +227,15 @@ foreign import ccall unsafe "zmq.h zmq_recvmsg" c_zmq_recvmsg :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt +foreign import ccall unsafe "zmq.h zmq_socket_monitor" + c_zmq_socket_monitor :: ZMQSocket -> CString -> CInt -> IO CInt + -- error messages foreign import ccall unsafe "zmq.h zmq_strerror" c_zmq_strerror :: CInt -> IO CString +-- proxy + +foreign import ccall unsafe "zmq.h zmq_proxy" + c_zmq_proxy :: ZMQSocket -> ZMQSocket -> ZMQSocket -> IO CInt diff --git a/src/System/ZMQ3/Error.hs b/src/System/ZMQ3/Error.hs --- a/src/System/ZMQ3/Error.hs +++ b/src/System/ZMQ3/Error.hs @@ -6,11 +6,12 @@ -- used by the standard throw* functions in 'Foreign.C.Error'. module System.ZMQ3.Error where +import Control.Monad import Control.Exception import Text.Printf import Data.Typeable (Typeable) -import Foreign hiding (throwIf, throwIf_) +import Foreign hiding (throwIf, throwIf_, void) import Foreign.C.Error import Foreign.C.String import Foreign.C.Types (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 @@ -5,6 +5,9 @@ , Flag(..) , Timeout , Size + , Switch (..) + , EventType (..) + , EventMsg (..) , messageOf , messageOfLazy @@ -17,12 +20,20 @@ , getStrOpt , getInt32Option , setInt32OptFromRestricted + , ctxIntOption + , setCtxIntOption , toZMQFlag , combine , mkSocket , onSocket + , bool2cint + , toSwitch + , fromSwitch + , events2cint + , eventMessage + ) where import Control.Applicative @@ -40,6 +51,7 @@ import Data.IORef (newIORef) import Data.Restricted +import System.Posix.Types (Fd(..)) import System.ZMQ3.Base import System.ZMQ3.Error @@ -47,12 +59,49 @@ type Size = Word -- | Flags to apply on send operations (cf. man zmq_send) -data Flag = DontWait -- ^ ZMQ_DONTWAIT - | SendMore -- ^ ZMQ_SNDMORE +data Flag = + DontWait -- ^ ZMQ_DONTWAIT + | SendMore -- ^ ZMQ_SNDMORE deriving (Eq, Ord, Show) +-- | Configuration switch +data Switch = + Default -- ^ Use default setting + | On -- ^ Activate setting + | Off -- ^ De-activate setting + deriving (Eq, Ord, Show) + +-- | Event types to monitor. +data EventType = + ConnectedEvent + | ConnectDelayedEvent + | ConnectRetriedEvent + | ListeningEvent + | BindFailedEvent + | AcceptedEvent + | AcceptFailedEvent + | ClosedEvent + | CloseFailedEvent + | DisconnectedEvent + | AllEvents + deriving (Eq, Ord, Show) + +-- | Event Message to receive when monitoring socket events. +data EventMsg = + Connected !SB.ByteString !Fd + | ConnectDelayed !SB.ByteString !Fd + | ConnectRetried !SB.ByteString !Int + | Listening !SB.ByteString !Fd + | BindFailed !SB.ByteString !Fd + | Accepted !SB.ByteString !Fd + | AcceptFailed !SB.ByteString !Int + | Closed !SB.ByteString !Fd + | CloseFailed !SB.ByteString !Int + | Disconnected !SB.ByteString !Int + deriving (Eq, Show) + -- | A 0MQ context representation. -newtype Context = Context { ctx :: ZMQCtx } +newtype Context = Context { _ctx :: ZMQCtx } -- | A 0MQ Socket. data Socket a = Socket { @@ -152,9 +201,62 @@ setInt32OptFromRestricted :: Integral i => ZMQOption -> Restricted l u i -> Socket b -> IO () setInt32OptFromRestricted o x s = setIntOpt s o ((fromIntegral . rvalue $ x) :: CInt) +ctxIntOption :: Integral i => String -> ZMQCtxOption -> Context -> IO i +ctxIntOption name opt ctx = fromIntegral <$> + (throwIfMinus1 name $ c_zmq_ctx_get (_ctx ctx) (ctxOptVal opt)) + +setCtxIntOption :: Integral i => String -> ZMQCtxOption -> i -> Context -> IO () +setCtxIntOption name opt val ctx = throwIfMinus1_ name $ + c_zmq_ctx_set (_ctx ctx) (ctxOptVal opt) (fromIntegral val) + toZMQFlag :: Flag -> ZMQFlag toZMQFlag DontWait = dontWait toZMQFlag SendMore = sndMore combine :: [Flag] -> CInt combine = fromIntegral . foldr ((.|.) . flagVal . toZMQFlag) 0 + +bool2cint :: Bool -> CInt +bool2cint True = 1 +bool2cint False = 0 + +toSwitch :: Integral a => a -> Maybe Switch +toSwitch (-1) = Just Default +toSwitch 0 = Just Off +toSwitch 1 = Just On +toSwitch _ = Nothing + +fromSwitch :: Integral a => Switch -> a +fromSwitch Default = -1 +fromSwitch Off = 0 +fromSwitch On = 1 + +toZMQEventType :: EventType -> ZMQEventType +toZMQEventType AllEvents = allEvents +toZMQEventType ConnectedEvent = connected +toZMQEventType ConnectDelayedEvent = connectDelayed +toZMQEventType ConnectRetriedEvent = connectRetried +toZMQEventType ListeningEvent = listening +toZMQEventType BindFailedEvent = bindFailed +toZMQEventType AcceptedEvent = accepted +toZMQEventType AcceptFailedEvent = acceptFailed +toZMQEventType ClosedEvent = closed +toZMQEventType CloseFailedEvent = closeFailed +toZMQEventType DisconnectedEvent = disconnected + +events2cint :: [EventType] -> CInt +events2cint = fromIntegral . foldr ((.|.) . eventTypeVal . toZMQEventType) 0 + +eventMessage :: Integral a => SB.ByteString -> a -> ZMQEventType -> EventMsg +eventMessage str dat tag + | tag == connected = Connected str (Fd . fromIntegral $ dat) + | tag == connectDelayed = ConnectDelayed str (Fd . fromIntegral $ dat) + | tag == connectRetried = ConnectRetried str (fromIntegral dat) + | tag == listening = Listening str (Fd . fromIntegral $ dat) + | tag == bindFailed = BindFailed str (Fd . fromIntegral $ dat) + | tag == accepted = Accepted str (Fd . fromIntegral $ dat) + | tag == acceptFailed = AcceptFailed str (fromIntegral dat) + | tag == closed = Closed str (Fd . fromIntegral $ dat) + | tag == closeFailed = CloseFailed str (fromIntegral dat) + | tag == disconnected = Disconnected str (fromIntegral dat) + | otherwise = error $ "unknown event type: " ++ (show . eventTypeVal $ tag) diff --git a/zeromq3-haskell.cabal b/zeromq3-haskell.cabal --- a/zeromq3-haskell.cabal +++ b/zeromq3-haskell.cabal @@ -1,15 +1,15 @@ name: zeromq3-haskell -version: 0.1.4 +version: 0.2 synopsis: Bindings to ZeroMQ 3.x category: System, FFI license: MIT license-file: LICENSE author: Toralf Wittner -maintainer: toralf.wittner@gmail.com +maintainer: Toralf Wittner copyright: Copyright (c) 2010 - 2012 zeromq-haskell authors homepage: http://github.com/twittner/zeromq-haskell/ stability: experimental -tested-With: GHC == 7.4.1 +tested-With: GHC == 7.6.1 cabal-version: >= 1.8 build-type: Simple extra-source-files: README.md @@ -28,7 +28,7 @@ patterns, message filtering (subscriptions), seamless access to multiple transport protocols and more. - This library provides the Haskell language binding to 0MQ >= 3.1.0 + This library provides the Haskell language binding to 0MQ >= 3.2.2 library hs-source-dirs: src