packages feed

zeromq3-haskell 0.1.4 → 0.2

raw patch · 16 files changed

+498/−103 lines, 16 files

Files

AUTHORS view
@@ -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
README.md view
@@ -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
examples/display.hs view
@@ -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 <address> [<address>, ...]"         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
+ examples/echo.hs view
@@ -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 []
examples/perf/local_lat.hs view
@@ -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
examples/perf/local_thr.hs view
@@ -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
examples/perf/remote_lat.hs view
@@ -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
examples/perf/remote_thr.hs view
@@ -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
examples/prompt.hs view
@@ -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
+ examples/threads.hs view
@@ -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+
src/Data/Restricted.hs view
@@ -8,7 +8,7 @@ -- Module      : Data.Restricted -- Copyright   : (c) 2011-2012 Toralf Wittner -- License     : MIT--- Maintainer  : toralf.wittner@gmail.com+-- Maintainer  : Toralf Wittner <tw@dtex.org> -- Stability   : experimental -- Portability : non-portable --
src/System/ZMQ3.hs view
@@ -2,7 +2,7 @@ -- Module      : System.ZMQ3 -- Copyright   : (c) 2010-2012 Toralf Wittner -- License     : MIT--- Maintainer  : toralf.wittner@gmail.com+-- Maintainer  : Toralf Wittner <tw@dtex.org> -- 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
src/System/ZMQ3/Base.hsc view
@@ -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
src/System/ZMQ3/Error.hs view
@@ -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)
src/System/ZMQ3/Internal.hs view
@@ -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)
zeromq3-haskell.cabal view
@@ -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 <tw@dtex.org> 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