packages feed

zeromq4-haskell (empty) → 0.1

raw patch · 22 files changed

+3196/−0 lines, 22 filesdep +MonadCatchIO-transformersdep +QuickCheckdep +ansi-terminalsetup-changed

Dependencies added: MonadCatchIO-transformers, QuickCheck, ansi-terminal, async, base, bytestring, checkers, containers, exceptions, semigroups, transformers, zeromq4-haskell

Files

+ AUTHORS view
@@ -0,0 +1,8 @@+Toralf Wittner      original implementation+David Himmelstrup   added send'+Nicolas Trangez     added support for zmq_device and "queue" test app+Ville Tirronen      added support for ZMG_SNDMORE+Jeremy Fitzhardinge integrated with GHC's I/O manager+Bryan O'Sullivan    added resource wrappers, socket finalizer+Ben Lever           fixed and tweaked some of the test examples+Alexander Vershilov added support for zmq_proxy
+ LICENSE view
@@ -0,0 +1,22 @@+Copyright (c) 2010 zeromq-haskell authors++Permission is hereby granted, free of charge, to any person+obtaining a copy of this software and associated documentation+files (the "Software"), to deal in the Software without+restriction, including without limitation the rights to use,+copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the+Software is furnished to do so, subject to the following+conditions:++The above copyright notice and this permission notice shall be+included in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES+OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT+HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,+WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR+OTHER DEALINGS IN THE SOFTWARE.
+ README.md view
@@ -0,0 +1,4 @@+zeromq4-haskell+---------------++This library provides Haskell bindings to [0MQ 4.0](http://zeromq.org).
+ Setup.hs view
@@ -0,0 +1,6 @@+module Main where++import Distribution.Simple++main = defaultMain+
+ examples/Makefile view
@@ -0,0 +1,11 @@+all: chat++chat: display.hs prompt.hs+	ghc --make -threaded display.hs+	ghc --make -threaded prompt.hs++.PHONY: clean+clean:+	-rm -f *.o *.hi+	-rm -f display prompt+
+ examples/display.hs view
@@ -0,0 +1,21 @@+{-# LANGUAGE OverloadedStrings #-}+import Control.Monad+import System.Exit+import System.IO+import System.Environment+import System.ZMQ4.Monadic+import qualified Data.ByteString.Char8 as CS++main :: IO ()+main = do+    args <- getArgs+    when (length args < 1) $ do+        hPutStrLn stderr "usage: display <address> [<address>, ...]"+        exitFailure+    runZMQ $ do+        sub <- socket Sub+        subscribe sub ""+        mapM_ (connect sub) args+        forever $ do+            receive sub >>= liftIO . CS.putStrLn+            liftIO $ hFlush stdout
+ examples/perf/Makefile view
@@ -0,0 +1,15 @@+all: lat thr++lat: local_lat.hs remote_lat.hs+	ghc --make -threaded local_lat.hs+	ghc --make -threaded remote_lat.hs++thr: local_thr.hs remote_thr.hs+	ghc --make -threaded local_thr.hs+	ghc --make -threaded remote_thr.hs++.PHONY: clean+clean:+	-rm -f *.o *.hi+	-rm -f local_lat remote_lat local_thr remote_thr+
+ examples/perf/local_lat.hs view
@@ -0,0 +1,31 @@+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import System.ZMQ4.Monadic+import qualified Data.ByteString as SB++main :: IO ()+main = do+    args <- getArgs+    when (length args /= 3) $ do+        hPutStrLn stderr usage+        exitFailure+    let bindTo = args !! 0+        size   = read $ args !! 1+        rounds = read $ args !! 2+    runZMQ $ do+        s <- socket Rep+        bind s bindTo+        loop s rounds size+  where+    loop s r sz = unless (r <= 0) $ do+        msg <- receive s+        when (SB.length msg /= sz) $+            error "message of incorrect size received"+        send s [] msg+        loop s (r - 1) sz++usage :: String+usage = "usage: local_lat <bind-to> <message-size> <roundtrip-count>"+
+ examples/perf/local_thr.hs view
@@ -0,0 +1,52 @@+{-# LANGUAGE OverloadedStrings #-}+import Control.Concurrent+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import Data.Time.Clock+import System.ZMQ4.Monadic+import qualified Data.ByteString as SB+import Text.Printf++main :: IO ()+main = do+    args <- getArgs+    when (length args /= 3) $ do+        hPutStrLn stderr usage+        exitFailure+    let bindTo = args !! 0+        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 <- 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 - 1) sz++    printStat :: UTCTime -> UTCTime -> Int -> Int -> IO ()+    printStat start end size count = do+        let elapsed = fromRational . toRational $ diffUTCTime end start :: Double+            through = fromIntegral count / elapsed+            mbits   = (through * fromIntegral size * 8) / 1000000+        printf "message size: %d [B]\n" size+        printf "message count: %d\n" count+        printf "mean throughput: %.3f [msg/s]\n" through+        printf "mean throughput: %.3f [Mb/s]\n" mbits++usage :: String+usage = "usage: local_thr <bind-to> <message-size> <message-count>"+
+ examples/perf/remote_lat.hs view
@@ -0,0 +1,36 @@+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import Data.Time.Clock+import System.ZMQ4.Monadic+import qualified Data.ByteString as SB++main :: IO ()+main = do+    args <- getArgs+    when (length args /= 3) $ do+        hPutStrLn stderr usage+        exitFailure+    let connTo  = args !! 0+        size    = read $ args !! 1+        rounds  = read $ args !! 2+        message = SB.replicate size 0x65+    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+        send s [] msg+        msg' <- receive s+        when (SB.length msg' /= SB.length msg) $+            error "message of incorrect size received"+        loop s (r - 1) msg++usage :: String+usage = "usage: remote_lat <connect-to> <message-size> <roundtrip-count>"+
+ examples/perf/remote_thr.hs view
@@ -0,0 +1,27 @@+import Control.Monad+import Control.Concurrent+import System.IO+import System.Exit+import System.Environment+import System.ZMQ4.Monadic+import qualified Data.ByteString as SB++main :: IO ()+main = do+    args <- getArgs+    when (length args /= 3) $ do+        hPutStrLn stderr usage+        exitFailure+    let connTo  = args !! 0+        size    = read $ args !! 1+        count   = read $ args !! 2+        message = SB.replicate size 0x65+    runZMQ $ do+        s <- socket Pub+        connect s connTo+        replicateM_ count $ send s [] message+        liftIO $ threadDelay 10000000++usage :: String+usage = "usage: remote_thr <connect-to> <message-size> <message-count>"+
+ examples/prompt.hs view
@@ -0,0 +1,25 @@+{-# LANGUAGE OverloadedStrings #-}+import Control.Applicative+import Control.Monad+import Data.Monoid+import Data.String+import System.IO+import System.Exit+import System.Environment+import System.ZMQ4.Monadic+import qualified Data.ByteString.UTF8 as SB++main :: IO ()+main = do+    args <- getArgs+    when (length args /= 2) $ do+        hPutStrLn stderr "usage: prompt <address> <username>"+        exitFailure+    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)
+ src/Data/Restricted.hs view
@@ -0,0 +1,211 @@+{-# LANGUAGE EmptyDataDecls         #-}+{-# LANGUAGE MultiParamTypeClasses  #-}+{-# LANGUAGE FlexibleInstances      #-}+{-# LANGUAGE UndecidableInstances   #-}+{-# LANGUAGE TypeSynonymInstances   #-}+{-# LANGUAGE OverloadedStrings      #-}++-- |+-- Module      : Data.Restricted+-- Copyright   : (c) 2011-2013 Toralf Wittner+-- License     : MIT+-- Maintainer  : Toralf Wittner <tw@dtex.org>+-- Stability   : experimental+-- Portability : non-portable+--+-- Type-level restricted data.+-- This module allows for type declarations which embed certain restrictions,+-- such as value bounds. E.g. @Restricted N0 N1 Int@ denotes an 'Int' which can+-- only have values [0 .. 1]. When creating such a value, the constructor functions+-- 'restrict' or 'toRestricted' ensure that the restrictions are obeyed. Code+-- that consumes restricted types does not need to check the constraints.+--+-- /N.B./ This module is more or less tailored to be used within 'System.ZMQ3'.+-- Therefore the provided type level restrictions are limited.+module Data.Restricted (++    Restricted+  , Restriction (..)+  , rvalue++  , Nneg1+  , N1+  , N0+  , N254+  , Inf+  , Div4+  , Div5++) where++import Data.Int+import Data.ByteString (ByteString)+import qualified Data.ByteString as B++-- | Type level restriction.+newtype Restricted r v = Restricted v deriving Show++-- | A uniform way to restrict values.+class Restriction r v where++    -- | Create a restricted value. Returns 'Nothing' if+    -- the given value does not satisfy all restrictions.+    toRestricted :: v -> Maybe (Restricted r v)++    -- | Create a restricted value. If the given value+    -- does not satisfy the restrictions, a modified+    -- variant is used instead, e.g. if an integer is+    -- larger than the upper bound, the upper bound+    -- value is used.+    restrict :: v -> Restricted r v++-- | Get the actual value.+rvalue :: Restricted r v -> v+rvalue (Restricted v) = v++-- | type level -1+data Nneg1++-- | type-level   0+data N0++-- | type-level   1+data N1++-- | type-level 254+data N254++-- | type-level infinity+data Inf++-- | divisable by 4+data Div4++-- | divisable by 5+data Div5++instance Show Nneg1 where show _ = "Nneg1"+instance Show N0    where show _ = "N0"+instance Show N1    where show _ = "N1"+instance Show N254  where show _ = "N254"+instance Show Inf   where show _ = "Inf"+instance Show Div4  where show _ = "Div4"+instance Show Div5  where show _ = "Div5"++-- Natural numbers++instance (Integral a) => Restriction (N0, Inf) a where+    toRestricted = toIntRLB 0+    restrict     = intRLB   0++instance (Integral a) => Restriction (N0, Int32) a where+    toRestricted = toIntR 0 (maxBound :: Int32)+    restrict     = intR   0 (maxBound :: Int32)++instance (Integral a) => Restriction (N0, Int64) a where+    toRestricted = toIntR 0 (maxBound :: Int64)+    restrict     = intR   0 (maxBound :: Int64)++-- Positive natural numbers++instance (Integral a) => Restriction (N1, Inf) a where+    toRestricted = toIntRLB 1+    restrict     = intRLB   1++instance (Integral a) => Restriction (N1, Int32) a where+    toRestricted = toIntR 1 (maxBound :: Int32)+    restrict     = intR   1 (maxBound :: Int32)++instance (Integral a) => Restriction (N1, Int64) a where+    toRestricted = toIntR 1 (maxBound :: Int64)+    restrict     = intR   1 (maxBound :: Int64)++-- From -1 ranges++instance (Integral a) => Restriction (Nneg1, Inf) a where+    toRestricted = toIntRLB (-1)+    restrict     = intRLB   (-1)++instance (Integral a) => Restriction (Nneg1, Int32) a where+    toRestricted = toIntR (-1) (maxBound :: Int32)+    restrict     = intR   (-1) (maxBound :: Int32)++instance (Integral a) => Restriction (Nneg1, Int64) a where+    toRestricted = toIntR (-1) (maxBound :: Int64)+    restrict     = intR   (-1) (maxBound :: Int64)++-- Other ranges++instance Restriction (N1, N254) String where+    toRestricted s | check (1, 254) (length s) = Just $ Restricted s+                   | otherwise                 = Nothing++    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)++-- Other constraints++instance Restriction Div4 ByteString where+    toRestricted s | B.length s `mod` 4 == 0 = Just $ Restricted s+                   | otherwise               = Nothing++    restrict = fitByRem 4++instance Restriction Div5 ByteString where+    toRestricted s | B.length s `mod` 5 == 0 = Just $ Restricted s+                   | otherwise               = Nothing++    restrict = fitByRem 5++-- Helpers++toIntR :: (Integral i, Integral j) => i -> j -> i -> Maybe (Restricted (a, b) i)+toIntR lb ub i | check (lb, fromIntegral ub) i = Just $ Restricted i+               | otherwise                     = Nothing++intR :: (Integral i, Integral j) => i -> j -> i -> Restricted (a, b) i+intR lb ub = Restricted . lbfit lb . ubfit (fromIntegral ub)++toIntRLB :: Integral i => i -> i -> Maybe (Restricted (a, b) i)+toIntRLB lb i | lbcheck lb i = Just $ Restricted i+              | otherwise    = Nothing++intRLB :: Integral i => i -> i -> Restricted (a, b) i+intRLB lb = Restricted . lbfit lb++-- Bounds checks++lbcheck :: Ord a => a -> a -> Bool+lbcheck lb a = a >= lb++ubcheck :: Ord a => a -> a -> Bool+ubcheck ub a = a <= ub++check :: Ord a => (a, a) -> a -> Bool+check (lb, ub) a = lbcheck lb a && ubcheck ub a++-- Fit++lbfit :: Integral a => a -> a -> a+lbfit lb a | a >= lb   = a+           | otherwise = lb++ubfit :: Integral a => a -> a -> a+ubfit ub a | a <= ub   = a+           | otherwise = ub++fitByRem :: Int -> ByteString -> Restricted r ByteString+fitByRem r s =+    let len = B.length s+        x   = len `mod` r+    in if x == 0+        then Restricted s+        else Restricted (B.take (len - x) s)+
+ src/System/ZMQ4.hs view
@@ -0,0 +1,972 @@+{-# LANGUAGE GADTs #-}++-- |+-- Module      : System.ZMQ4+-- Copyright   : (c) 2010-2013 Toralf Wittner+-- License     : MIT+-- Maintainer  : Toralf Wittner <tw@dtex.org>+-- Stability   : experimental+-- Portability : non-portable+--+-- 0MQ haskell binding. The API closely follows the C-API of 0MQ with+-- the main difference being that sockets are typed.+--+-- /Notes/+--+-- Many option settings use a 'Restriction' to further constrain the+-- range of possible values of their integral types. For example+-- the maximum message size can be given as -1, which means no limit+-- or by greater values, which denote the message size in bytes. The+-- type of 'setMaxMessageSize' is therefore:+--+-- @setMaxMessageSize :: Integral i+--                    => Restricted (Nneg1, Int64) i+--                    -> Socket a+--                    -> IO ()@+--+-- which means any integral value in the range of @-1@ to+-- (@maxBound :: Int64@) can be given. To create a restricted+-- value from plain value, use 'toRestricted' or 'restrict'.++module System.ZMQ4+  ( -- * Type Definitions+    -- ** Socket Types+    Pair   (..)+  , Pub    (..)+  , Sub    (..)+  , XPub   (..)+  , XSub   (..)+  , Req    (..)+  , Rep    (..)+  , Dealer (..)+  , Router (..)+  , XReq+  , XRep+  , Pull   (..)+  , Push   (..)+  , Stream (..)++    -- ** Socket type-classes+  , SocketType+  , Sender+  , Receiver+  , Subscriber+  , SocketLike+  , Conflatable+  , SendProbe++    -- ** Various type definitions+  , Size+  , Context+  , Socket+  , Flag              (SendMore)+  , Switch            (..)+  , Timeout+  , Event             (..)+  , EventType         (..)+  , EventMsg          (..)+  , Poll              (..)+  , KeyFormat         (..)+  , SecurityMechanism (..)++    -- * General Operations+  , withContext+  , withSocket+  , bind+  , unbind+  , connect+  , send+  , send'+  , sendMulti+  , receive+  , receiveMulti+  , version+  , monitor+  , poll++  , System.ZMQ4.subscribe+  , System.ZMQ4.unsubscribe++    -- * Context Options (Read)+  , ioThreads+  , maxSockets++    -- * Context Options (Write)+  , setIoThreads+  , setMaxSockets++    -- * Socket Options (Read)+  , System.ZMQ4.affinity+  , System.ZMQ4.backlog+  , System.ZMQ4.conflate+  , System.ZMQ4.curvePublicKey+  , System.ZMQ4.curveSecretKey+  , System.ZMQ4.curveServerKey+  , System.ZMQ4.delayAttachOnConnect+  , System.ZMQ4.events+  , System.ZMQ4.fileDescriptor+  , System.ZMQ4.identity+  , System.ZMQ4.immediate+  , System.ZMQ4.ipv4Only+  , System.ZMQ4.ipv6+  , System.ZMQ4.lastEndpoint+  , System.ZMQ4.linger+  , System.ZMQ4.maxMessageSize+  , System.ZMQ4.mcastHops+  , System.ZMQ4.mechanism+  , System.ZMQ4.moreToReceive+  , System.ZMQ4.plainServer+  , System.ZMQ4.plainPassword+  , System.ZMQ4.plainUserName+  , System.ZMQ4.rate+  , System.ZMQ4.receiveBuffer+  , System.ZMQ4.receiveHighWM+  , System.ZMQ4.receiveTimeout+  , System.ZMQ4.reconnectInterval+  , System.ZMQ4.reconnectIntervalMax+  , System.ZMQ4.recoveryInterval+  , System.ZMQ4.sendBuffer+  , System.ZMQ4.sendHighWM+  , System.ZMQ4.sendTimeout+  , System.ZMQ4.tcpKeepAlive+  , System.ZMQ4.tcpKeepAliveCount+  , System.ZMQ4.tcpKeepAliveIdle+  , System.ZMQ4.tcpKeepAliveInterval+  , System.ZMQ4.zapDomain++    -- * Socket Options (Write)+  , setAffinity+  , setBacklog+  , setConflate+  , setCurveServer+  , setCurvePublicKey+  , setCurveSecretKey+  , setCurveServerKey+  , setDelayAttachOnConnect+  , setIdentity+  , setImmediate+  , setIpv4Only+  , setIpv6+  , setLinger+  , setMaxMessageSize+  , setMcastHops+  , setPlainServer+  , setPlainPassword+  , setPlainUserName+  , setProbeRouter+  , setRate+  , setReceiveBuffer+  , setReceiveHighWM+  , setReceiveTimeout+  , setReconnectInterval+  , setReconnectIntervalMax+  , setRecoveryInterval+  , setReqCorrelate+  , setReqRelaxed+  , setRouterMandatory+  , setSendBuffer+  , setSendHighWM+  , setSendTimeout+  , setTcpAcceptFilter+  , setTcpKeepAlive+  , setTcpKeepAliveCount+  , setTcpKeepAliveIdle+  , setTcpKeepAliveInterval+  , setXPubVerbose++    -- * Restrictions+  , Data.Restricted.restrict+  , Data.Restricted.toRestricted++    -- * Error Handling+  , ZMQError+  , errno+  , source+  , message++    -- * Low-level Functions+  , init+  , term+  , shutdown+  , context+  , socket+  , close+  , waitRead+  , waitWrite+  , z85Encode+  , z85Decode++    -- * Utils+  , proxy+  , curveKeyPair+  ) where++import Prelude hiding (init)+import Control.Applicative+import Control.Exception+import Control.Monad (unless)+import Control.Monad.IO.Class+import Data.List (intersect, foldl')+import Data.List.NonEmpty (NonEmpty)+import Data.Restricted+import Data.Traversable (forM)+import Foreign hiding (throwIf, throwIf_, throwIfNull, void)+import Foreign.C.String+import Foreign.C.Types (CInt, CShort)+import System.Posix.Types (Fd(..))+import System.ZMQ4.Base+import System.ZMQ4.Internal+import System.ZMQ4.Error++import qualified Data.ByteString      as SB+import qualified Data.ByteString.Lazy as LB+import qualified Data.List.NonEmpty   as S+import qualified Prelude              as P+import qualified System.ZMQ4.Base     as B++import GHC.Conc (threadWaitRead, threadWaitWrite)++-----------------------------------------------------------------------------+-- Socket Types++-- | <http://api.zeromq.org/4-0:zmq-socket ZMQ_PAIR>+data Pair = Pair++-- | <http://api.zeromq.org/4-0:zmq-socket ZMQ_PUB>+data Pub = Pub++-- | <http://api.zeromq.org/4-0:zmq-socket ZMQ_SUB>+data Sub = Sub++-- | <http://api.zeromq.org/4-0:zmq-socket ZMQ_XPUB>+data XPub = XPub++-- | <http://api.zeromq.org/4-0:zmq-socket ZMQ_XSUB>+data XSub = XSub++-- | <http://api.zeromq.org/4-0:zmq-socket ZMQ_REQ>+data Req = Req++-- | <http://api.zeromq.org/4-0:zmq-socket ZMQ_REP>+data Rep = Rep++-- | <http://api.zeromq.org/4-0:zmq-socket ZMQ_DEALER>+data Dealer = Dealer++-- | <http://api.zeromq.org/4-0:zmq-socket ZMQ_ROUTER>+data Router = Router++-- | <http://api.zeromq.org/4-0:zmq-socket ZMQ_PULL>+data Pull = Pull++-- | <http://api.zeromq.org/4-0:zmq-socket ZMQ_PUSH>+data Push = Push++-- | <http://api.zeromq.org/4-0:zmq-socket ZMQ_STREAM>+data Stream = Stream++type XReq = Dealer+{-# DEPRECATED XReq "Use Dealer" #-}++type XRep = Router+{-# DEPRECATED XRep "Use Router" #-}++-----------------------------------------------------------------------------+-- Socket Type Classifications++-- | Sockets which can 'subscribe'.+class Subscriber a++-- | Sockets which can 'send'.+class Sender a++-- | Sockets which can 'receive'.+class Receiver a++-- | Sockets which can be 'conflate'd.+class Conflatable a++-- | Sockets which can send probes (cf. 'setProbeRouter').+class SendProbe a++instance SocketType Pair where zmqSocketType = const pair+instance Sender     Pair+instance Receiver   Pair++instance SocketType  Pub where zmqSocketType = const pub+instance Sender      Pub+instance Conflatable Pub++instance SocketType  Sub where zmqSocketType = const sub+instance Subscriber  Sub+instance Receiver    Sub+instance Conflatable Sub++instance SocketType XPub where zmqSocketType = const xpub+instance Sender     XPub+instance Receiver   XPub++instance SocketType XSub where zmqSocketType = const xsub+instance Sender     XSub+instance Receiver   XSub++instance SocketType Req where zmqSocketType = const request+instance Sender     Req+instance Receiver   Req+instance SendProbe  Req++instance SocketType Rep where zmqSocketType = const response+instance Sender     Rep+instance Receiver   Rep++instance SocketType  Dealer where zmqSocketType = const dealer+instance Sender      Dealer+instance Receiver    Dealer+instance Conflatable Dealer+instance SendProbe   Dealer++instance SocketType Router where zmqSocketType = const router+instance Sender     Router+instance Receiver   Router+instance SendProbe  Router++instance SocketType  Pull where zmqSocketType = const pull+instance Receiver    Pull+instance Conflatable Pull++instance SocketType  Push where zmqSocketType = const push+instance Sender      Push+instance Conflatable Push++instance SocketType Stream where zmqSocketType = const stream+instance Sender     Stream+instance Receiver   Stream++-----------------------------------------------------------------------------++-- | Socket events.+data Event =+    In     -- ^ @ZMQ_POLLIN@ (incoming messages)+  | Out    -- ^ @ZMQ_POLLOUT@ (outgoing messages, i.e. at least 1 byte can be written)+  | Err    -- ^ @ZMQ_POLLERR@+  deriving (Eq, Ord, Read, Show)++-- | A 'Poll' value contains the object to poll (a 0MQ socket or a file+-- descriptor), the set of 'Event's which are of interest and--optionally--+-- a callback-function which is invoked iff the set of interested events+-- overlaps with the actual events.+data Poll s m where+    Sock :: s t -> [Event] -> Maybe ([Event] -> m ()) -> Poll s m+    File :: Fd -> [Event] -> Maybe ([Event] -> m ()) -> Poll s m++-- | Return the runtime version of the underlying 0MQ library as a+-- (major, minor, patch) triple.+version :: IO (Int, Int, Int)+version =+    with 0 $ \major_ptr ->+    with 0 $ \minor_ptr ->+    with 0 $ \patch_ptr ->+        c_zmq_version major_ptr minor_ptr patch_ptr >>+        tupleUp <$> peek major_ptr <*> peek minor_ptr <*> peek patch_ptr+  where+    tupleUp a b c = (fromIntegral a, fromIntegral b, fromIntegral c)++init :: Size -> IO Context+init n = do+    c <- context+    setIoThreads n c+    return c+{-# DEPRECATED init "Use context" #-}++-- | Initialize a 0MQ context.+-- Equivalent to <http://api.zeromq.org/4-0:zmq-ctx-new zmq_ctx_new>.+context :: IO Context+context = Context <$> throwIfNull "init" c_zmq_ctx_new++-- | Terminate a 0MQ context.+-- Equivalent to <http://api.zeromq.org/4-0:zmq-ctx-term zmq_ctx_term>.+term :: Context -> IO ()+term c = throwIfMinus1Retry_ "term" . c_zmq_ctx_term . _ctx $ c++-- | Shutdown a 0MQ context.+-- Equivalent to <http://api.zeromq.org/4-0:zmq-ctx-shutdown zmq_ctx_shutdown>.+shutdown :: Context -> IO ()+shutdown = throwIfMinus1_ "shutdown" . c_zmq_ctx_shutdown . _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 :: (Context -> IO a) -> IO a+withContext act =+  bracket (throwIfNull "withContext (new)" $ c_zmq_ctx_new)+          (throwIfMinus1Retry_ "withContext (term)" . c_zmq_ctx_term)+          (act . Context)++-- | Run an action with a 0MQ socket. The socket will be closed after running+-- the supplied action even if an error occurs. The socket supplied to your+-- action will /not/ be valid after the action terminates.+withSocket :: SocketType a => Context -> a -> (Socket a -> IO b) -> IO b+withSocket c t = bracket (socket c t) close++-- | 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 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 = closeSock . _socketRepr++-- | Subscribe Socket to given subscription.+subscribe :: Subscriber a => Socket a -> SB.ByteString -> IO ()+subscribe s = setByteStringOpt s B.subscribe++-- | Unsubscribe Socket from given subscription.+unsubscribe :: Subscriber a => Socket a -> SB.ByteString -> IO ()+unsubscribe s = setByteStringOpt s B.unsubscribe++-- Read Only++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_EVENTS>.+events :: Socket a -> IO [Event]+events s = toEvents <$> getIntOpt s B.events 0++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_FD>.+fileDescriptor :: Socket a -> IO Fd+fileDescriptor s = Fd . fromIntegral <$> getInt32Option B.filedesc s++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_RCVMORE>.+moreToReceive :: Socket a -> IO Bool+moreToReceive s = (== 1) <$> getInt32Option B.receiveMore s++-- Read++-- | <http://api.zeromq.org/4-0:zmq-ctx-get zmq_ctx_get ZMQ_IO_THREADS>.+ioThreads :: Context -> IO Word+ioThreads = ctxIntOption "ioThreads" _ioThreads++-- | <http://api.zeromq.org/4-0:zmq-ctx-get zmq_ctx_get ZMQ_MAX_SOCKETS>.+maxSockets :: Context -> IO Word+maxSockets = ctxIntOption "maxSockets" _maxSockets++-- | Restricts the outgoing and incoming socket buffers to a single message.+conflate :: Conflatable a => Socket a -> IO Bool+conflate s = (== 1) <$> getInt32Option B.conflate s++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_IMMEDIATE>.+immediate :: Socket a -> IO Bool+immediate s = (== 1) <$> getInt32Option B.immediate s++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_IDENTITY>.+identity :: Socket a -> IO SB.ByteString+identity s = getByteStringOpt s B.identity++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_AFFINITY>.+affinity :: Socket a -> IO Word64+affinity s = getIntOpt s B.affinity 0++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_MAXMSGSIZE>.+maxMessageSize :: Socket a -> IO Int64+maxMessageSize s = getIntOpt s B.maxMessageSize 0++ipv4Only :: Socket a -> IO Bool+ipv4Only s = (== 1) <$> getInt32Option B.ipv4Only s+{-# DEPRECATED ipv4Only "Use ipv6" #-}++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_IPV6>.+ipv6 :: Socket a -> IO Bool+ipv6 s = (== 1) <$> getInt32Option B.ipv6 s++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_BACKLOG>.+backlog :: Socket a -> IO Int+backlog = getInt32Option B.backlog++delayAttachOnConnect :: Socket a -> IO Bool+delayAttachOnConnect s = (== 1) <$> getInt32Option B.delayAttachOnConnect s+{-# DEPRECATED delayAttachOnConnect "Use immediate" #-}++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_LINGER>.+linger :: Socket a -> IO Int+linger = getInt32Option B.linger++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_LAST_ENDPOINT>.+lastEndpoint :: Socket a -> IO String+lastEndpoint s = getStrOpt s B.lastEndpoint++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_RATE>.+rate :: Socket a -> IO Int+rate = getInt32Option B.rate++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_RCVBUF>.+receiveBuffer :: Socket a -> IO Int+receiveBuffer = getInt32Option B.receiveBuf++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_RECONNECT_IVL>.+reconnectInterval :: Socket a -> IO Int+reconnectInterval = getInt32Option B.reconnectIVL++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_RECONNECT_IVL_MAX>.+reconnectIntervalMax :: Socket a -> IO Int+reconnectIntervalMax = getInt32Option B.reconnectIVLMax++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_RECOVERY_IVL>.+recoveryInterval :: Socket a -> IO Int+recoveryInterval = getInt32Option B.recoveryIVL++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_SNDBUF>.+sendBuffer :: Socket a -> IO Int+sendBuffer = getInt32Option B.sendBuf++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_MULTICAST_HOPS>.+mcastHops :: Socket a -> IO Int+mcastHops = getInt32Option B.mcastHops++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_RCVHWM>.+receiveHighWM :: Socket a -> IO Int+receiveHighWM = getInt32Option B.receiveHighWM++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_RCVTIMEO>.+receiveTimeout :: Socket a -> IO Int+receiveTimeout = getInt32Option B.receiveTimeout++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_SNDTIMEO>.+sendTimeout :: Socket a -> IO Int+sendTimeout = getInt32Option B.sendTimeout++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_SNDHWM>.+sendHighWM :: Socket a -> IO Int+sendHighWM = getInt32Option B.sendHighWM++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_TCP_KEEPALIVE>.+tcpKeepAlive :: Socket a -> IO Switch+tcpKeepAlive = fmap (toSwitch "Invalid ZMQ_TCP_KEEPALIVE")+             . getInt32Option B.tcpKeepAlive++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_TCP_KEEPALIVE_CNT>.+tcpKeepAliveCount :: Socket a -> IO Int+tcpKeepAliveCount = getInt32Option B.tcpKeepAliveCount++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_TCP_KEEPALIVE_IDLE>.+tcpKeepAliveIdle :: Socket a -> IO Int+tcpKeepAliveIdle = getInt32Option B.tcpKeepAliveIdle++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_TCP_KEEPALIVE_INTVL>.+tcpKeepAliveInterval :: Socket a -> IO Int+tcpKeepAliveInterval = getInt32Option B.tcpKeepAliveInterval++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_MECHANISM>.+mechanism :: Socket a -> IO SecurityMechanism+mechanism = fmap (fromMechanism "Invalid ZMQ_MECHANISM")+          . getInt32Option B.mechanism++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_PLAIN_SERVER>.+plainServer :: Socket a -> IO Bool+plainServer = fmap (== 1) . getInt32Option B.plainServer++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_PLAIN_USERNAME>.+plainUserName :: Socket a -> IO SB.ByteString+plainUserName s = getByteStringOpt s B.plainUserName++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_PLAIN_PASSWORD>.+plainPassword :: Socket a -> IO SB.ByteString+plainPassword s = getByteStringOpt s B.plainPassword++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_ZAP_DOMAIN>.+zapDomain :: Socket a -> IO SB.ByteString+zapDomain s = getByteStringOpt s B.zapDomain++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_CURVE_PUBLICKEY>.+curvePublicKey :: KeyFormat f -> Socket a -> IO SB.ByteString+curvePublicKey f s = getKey f s B.curvePublicKey++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_CURVE_SERVERKEY>.+curveServerKey :: KeyFormat f -> Socket a -> IO SB.ByteString+curveServerKey f s = getKey f s B.curveServerKey++-- | <http://api.zeromq.org/4-0:zmq-getsockopt zmq_getsockopt ZMQ_CURVE_SECRETKEY>.+curveSecretKey :: KeyFormat f -> Socket a -> IO SB.ByteString+curveSecretKey f s = getKey f s B.curveSecretKey++-- Write++-- | <http://api.zeromq.org/4-0:zmq-ctx-set zmq_ctx_get ZMQ_IO_THREADS>.+setIoThreads :: Word -> Context -> IO ()+setIoThreads n = setCtxIntOption "ioThreads" _ioThreads n++-- | <http://api.zeromq.org/4-0:zmq-ctx-set zmq_ctx_get ZMQ_MAX_SOCKETS>.+setMaxSockets :: Word -> Context -> IO ()+setMaxSockets n = setCtxIntOption "maxSockets" _maxSockets n++-- | Restrict the outgoing and incoming socket buffers to a single message.+setConflate :: Conflatable a => Bool -> Socket a -> IO ()+setConflate x s = setIntOpt s B.conflate (bool2cint x)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_IMMEDIATE>.+setImmediate :: Bool -> Socket a -> IO ()+setImmediate x s = setIntOpt s B.immediate (bool2cint x)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_IDENTITY>.+setIdentity :: Restricted (N1, N254) SB.ByteString -> Socket a -> IO ()+setIdentity x s = setByteStringOpt s B.identity (rvalue x)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_AFFINITY>.+setAffinity :: Word64 -> Socket a -> IO ()+setAffinity x s = setIntOpt s B.affinity x++setDelayAttachOnConnect :: Bool -> Socket a -> IO ()+setDelayAttachOnConnect x s = setIntOpt s B.delayAttachOnConnect (bool2cint x)+{-# DEPRECATED setDelayAttachOnConnect "Use setImmediate" #-}++-- | <http://api.zeromq.org/4-0:zmq-setsockopt 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)++setIpv4Only :: Bool -> Socket a -> IO ()+setIpv4Only x s = setIntOpt s B.ipv4Only (bool2cint x)+{-# DEPRECATED setIpv4Only "Use setIpv6" #-}++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_IPV6>.+setIpv6 :: Bool -> Socket a -> IO ()+setIpv6 x s = setIntOpt s B.ipv6 (bool2cint x)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_PLAIN_SERVER>.+setPlainServer :: Bool -> Socket a -> IO ()+setPlainServer x s = setIntOpt s B.plainServer (bool2cint x)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_CURVE_SERVER>.+setCurveServer :: Bool -> Socket a -> IO ()+setCurveServer x s = setIntOpt s B.curveServer (bool2cint x)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_PLAIN_USERNAME>.+setPlainUserName :: Restricted (N1, N254) SB.ByteString -> Socket a -> IO ()+setPlainUserName x s = setByteStringOpt s B.plainUserName (rvalue x)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_PLAIN_USERNAME>.+setPlainPassword :: Restricted (N1, N254) SB.ByteString -> Socket a -> IO ()+setPlainPassword x s = setByteStringOpt s B.plainPassword (rvalue x)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_LINGER>.+setLinger :: Integral i => Restricted (Nneg1, Int32) i -> Socket a -> IO ()+setLinger = setInt32OptFromRestricted B.linger++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_RCVTIMEO>.+setReceiveTimeout :: Integral i => Restricted (Nneg1, Int32) i -> Socket a -> IO ()+setReceiveTimeout = setInt32OptFromRestricted B.receiveTimeout++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_ROUTER_MANDATORY>.+setRouterMandatory :: Bool -> Socket Router -> IO ()+setRouterMandatory x s = setIntOpt s B.routerMandatory (bool2cint x)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_SNDTIMEO>.+setSendTimeout :: Integral i => Restricted (Nneg1, Int32) i -> Socket a -> IO ()+setSendTimeout = setInt32OptFromRestricted B.sendTimeout++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_RATE>.+setRate :: Integral i => Restricted (N1, Int32) i -> Socket a -> IO ()+setRate = setInt32OptFromRestricted B.rate++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_MULTICAST_HOPS>.+setMcastHops :: Integral i => Restricted (N1, Int32) i -> Socket a -> IO ()+setMcastHops = setInt32OptFromRestricted B.mcastHops++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_BACKLOG>.+setBacklog :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO ()+setBacklog = setInt32OptFromRestricted B.backlog++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_CURVE_PUBLICKEY>.+setCurvePublicKey :: KeyFormat f -> Restricted f SB.ByteString -> Socket a -> IO ()+setCurvePublicKey _ k s = setByteStringOpt s B.curvePublicKey (rvalue k)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_CURVE_SECRETKEY>.+setCurveSecretKey :: KeyFormat f -> Restricted f SB.ByteString -> Socket a -> IO ()+setCurveSecretKey _ k s = setByteStringOpt s B.curveSecretKey (rvalue k)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_CURVE_SERVERKEY>.+setCurveServerKey :: KeyFormat f -> Restricted f SB.ByteString -> Socket a -> IO ()+setCurveServerKey _ k s = setByteStringOpt s B.curveServerKey (rvalue k)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_PROBE_ROUTER>.+setProbeRouter :: SendProbe a => Bool -> Socket a -> IO ()+setProbeRouter x s = setIntOpt s B.probeRouter (bool2cint x)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_RCVBUF>.+setReceiveBuffer :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO ()+setReceiveBuffer = setInt32OptFromRestricted B.receiveBuf++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_RECONNECT_IVL>.+setReconnectInterval :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO ()+setReconnectInterval = setInt32OptFromRestricted B.reconnectIVL++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_RECONNECT_IVL_MAX>.+setReconnectIntervalMax :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO ()+setReconnectIntervalMax = setInt32OptFromRestricted B.reconnectIVLMax++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_REQ_CORRELATE>.+setReqCorrelate :: Bool -> Socket Req -> IO ()+setReqCorrelate x s = setIntOpt s B.reqCorrelate (bool2cint x)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_REQ_RELAXED>.+setReqRelaxed :: Bool -> Socket Req -> IO ()+setReqRelaxed x s = setIntOpt s B.reqRelaxed (bool2cint x)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_SNDBUF>.+setSendBuffer :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO ()+setSendBuffer = setInt32OptFromRestricted B.sendBuf++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_RECOVERY_IVL>.+setRecoveryInterval :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO ()+setRecoveryInterval = setInt32OptFromRestricted B.recoveryIVL++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_RCVHWM>.+setReceiveHighWM :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO ()+setReceiveHighWM = setInt32OptFromRestricted B.receiveHighWM++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_SNDHWM>.+setSendHighWM :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO ()+setSendHighWM = setInt32OptFromRestricted B.sendHighWM++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_TCP_ACCEPT_FILTER>.+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 = setByteStringOpt sock tcpAcceptFilter dat++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_TCP_KEEPALIVE>.+setTcpKeepAlive :: Switch -> Socket a -> IO ()+setTcpKeepAlive x s = setIntOpt s B.tcpKeepAlive (fromSwitch x :: CInt)++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_TCP_KEEPALIVE_CNT>.+setTcpKeepAliveCount :: Integral i => Restricted (Nneg1, Int32) i -> Socket a -> IO ()+setTcpKeepAliveCount = setInt32OptFromRestricted B.tcpKeepAliveCount++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_TCP_KEEPALIVE_IDLE>.+setTcpKeepAliveIdle :: Integral i => Restricted (Nneg1, Int32) i -> Socket a -> IO ()+setTcpKeepAliveIdle = setInt32OptFromRestricted B.tcpKeepAliveIdle++-- | <http://api.zeromq.org/4-0:zmq-setsockopt zmq_setsockopt ZMQ_TCP_KEEPALIVE_INTVL>.+setTcpKeepAliveInterval :: Integral i => Restricted (Nneg1, Int32) i -> Socket a -> IO ()+setTcpKeepAliveInterval = setInt32OptFromRestricted B.tcpKeepAliveInterval++-- | <http://api.zeromq.org/4-0:zmq-setsockopt 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. <http://api.zeromq.org/4-0:zmq-bind zmq_bind>).+bind :: Socket a -> String -> IO ()+bind sock str = onSocket "bind" sock $+    throwIfMinus1Retry_ "bind" . withCString str . c_zmq_bind++-- | Unbind the socket from the given address+-- (cf. <http://api.zeromq.org/4-0:zmq-unbind zmq_unbind>).+unbind :: Socket a -> String -> IO ()+unbind sock str = onSocket "unbind" sock $+    throwIfMinus1Retry_ "unbind" . withCString str . c_zmq_unbind++-- | Connect the socket to the given address+-- (cf. <http://api.zeromq.org/4-0:zmq-connect zmq_connect>).+connect :: Socket a -> String -> IO ()+connect sock str = onSocket "connect" sock $+    throwIfMinus1Retry_ "connect" . withCString str . c_zmq_connect++-- | Send the given 'SB.ByteString' over the socket+-- (cf. <http://api.zeromq.org/4-0:zmq-sendmsg zmq_sendmsg>).+--+-- /Note/: This function always calls @zmq_sendmsg@ in a non-blocking way,+-- i.e. there is no need to provide the @ZMQ_DONTWAIT@ flag as this is used+-- by default. Still 'send' is blocking the thread as long as the message+-- can not be queued on the socket using GHC's 'threadWaitWrite'.+send :: Sender a => Socket a -> [Flag] -> SB.ByteString -> IO ()+send sock fls val = bracket (messageOf val) messageClose $ \m ->+  onSocket "send" sock $ \s ->+    retry "send" (waitWrite sock) $+          c_zmq_sendmsg s (msgPtr m) (combineFlags (DontWait : fls))++-- | Send the given 'LB.ByteString' over the socket+-- (cf. <http://api.zeromq.org/4-0:zmq-sendmsg zmq_sendmsg>).+--+-- This is operationally identical to @send socket (Strict.concat+-- (Lazy.toChunks lbs)) flags@ but may be more efficient.+--+-- /Note/: This function always calls @zmq_sendmsg@ in a non-blocking way,+-- i.e. there is no need to provide the @ZMQ_DONTWAIT@ flag as this is used+-- by default. Still 'send'' is blocking the thread as long as the message+-- can not be queued on the socket using GHC's 'threadWaitWrite'.+send' :: Sender a => Socket a -> [Flag] -> LB.ByteString -> IO ()+send' sock fls val = bracket (messageOfLazy val) messageClose $ \m ->+  onSocket "send'" sock $ \s ->+    retry "send'" (waitWrite sock) $+          c_zmq_sendmsg s (msgPtr m) (combineFlags (DontWait : fls))++-- | Send a multi-part message.+-- This function applies the 'SendMore' 'Flag' between all message parts.+-- 0MQ guarantees atomic delivery of a multi-part message+-- (cf. <http://api.zeromq.org/4-0:zmq-sendmsg zmq_sendmsg>).+sendMulti :: Sender a => Socket a -> NonEmpty SB.ByteString -> IO ()+sendMulti sock msgs = do+    mapM_ (send sock [SendMore]) (S.init msgs)+    send sock [] (S.last msgs)++-- | Receive a 'ByteString' from socket+-- (cf. <http://api.zeromq.org/4-0:zmq-recvmsg zmq_recvmsg>).+--+-- /Note/: This function always calls @zmq_recvmsg@ in a non-blocking way,+-- i.e. there is no need to provide the @ZMQ_DONTWAIT@ flag as this is used+-- by default. Still 'receive' is blocking the thread as long as no data+-- is available using GHC's 'threadWaitRead'.+receive :: Receiver a => Socket a -> IO (SB.ByteString)+receive sock = bracket messageInit messageClose $ \m ->+  onSocket "receive" sock $ \s -> do+    retry "receive" (waitRead sock) $+          c_zmq_recvmsg s (msgPtr m) (flagVal dontWait)+    data_ptr <- c_zmq_msg_data (msgPtr m)+    size     <- c_zmq_msg_size (msgPtr m)+    SB.packCStringLen (data_ptr, fromIntegral size)++-- | Receive a multi-part message.+-- This function collects all message parts send via 'sendMulti'.+receiveMulti :: Receiver a => Socket a -> IO [SB.ByteString]+receiveMulti sock = recvall []+  where+    recvall acc = do+        msg <- receive sock+        moreToReceive sock >>= next (msg:acc)++    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+-- (cf. <http://api.zeromq.org/4-0:zmq-socket-monitor zmq_socket_monitor>).+--+-- 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 . _socketRepr $ 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)+        evt <- peek ptr+        str <- receive soc+        return . Just $ eventMessage str evt++-- | Polls for events on the given 'Poll' descriptors. Returns a list of+-- events per descriptor which have occured.+-- (cf. <http://api.zeromq.org/4-0:zmq-poll zmq_poll>)+poll :: (SocketLike s, MonadIO m) => Timeout -> [Poll s m] -> m [[Event]]+poll _    [] = return []+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 :: (SocketLike s, MonadIO m) => Poll s m -> ZMQPoll+    toZMQPoll (Sock s e _) =+        ZMQPoll (_socket . _socketRepr . toSocket $ s) 0 (combine (map fromEvent e)) 0++    toZMQPoll (File (Fd s) e _) =+        ZMQPoll nullPtr (fromIntegral s) (combine (map fromEvent e)) 0++    fromZMQPoll :: (SocketLike s, MonadIO m) => (Poll s 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)+        forM f (unless (P.null (e `intersect` e')) . ($ e)) >> 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++wait' :: (Fd -> IO ()) -> ZMQPollEvent -> Socket a -> IO ()+wait' w f s = do+    fd <- getIntOpt s B.filedesc 0+    w (Fd fd)+    evs <- getInt32Option B.events s+    unless (testev evs) $+        wait' w f s+  where+    testev e = e .&. fromIntegral (pollVal f) /= 0++-- | Wait until data is available for reading from the given Socket.+-- After this function returns, a call to 'receive' will essentially be+-- non-blocking.+waitRead :: Socket a -> IO ()+waitRead = wait' threadWaitRead pollIn++-- | Wait until data can be written to the given Socket.+-- After this function returns, a call to 'send' will essentially be+-- non-blocking.+waitWrite :: Socket a -> IO ()+waitWrite = wait' threadWaitWrite pollOut++-- | Starts built-in 0MQ proxy+-- (cf. <http://api.zeromq.org/4-0:zmq-proxy zmq_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 ->+        throwIfMinus1Retry_ "c_zmq_proxy" $ c_zmq_proxy f b c+  where+    c = maybe nullPtr (_socket . _socketRepr) capture++-- | Generate a new curve key pair.+-- (cf. <http://api.zeromq.org/4-0:zmq-curve-keypair zmq_curve_keypair>)+curveKeyPair :: MonadIO m => m (Restricted Div5 SB.ByteString, Restricted Div5 SB.ByteString)+curveKeyPair = liftIO $+    allocaBytes 41 $ \cstr1 ->+    allocaBytes 41 $ \cstr2 -> do+        throwIfMinus1_ "c_zmq_curve_keypair" $ c_zmq_curve_keypair cstr1 cstr2+        public  <- toRestricted <$> SB.packCString cstr1+        private <- toRestricted <$> SB.packCString cstr2+        maybe (fail errmsg) return ((,) <$> public <*> private)+      where+        errmsg = "curveKeyPair: invalid key-lengths produced"+
+ src/System/ZMQ4/Base.hsc view
@@ -0,0 +1,360 @@+{-# LANGUAGE CPP                      #-}+{-# LANGUAGE ForeignFunctionInterface #-}++module System.ZMQ4.Base where++import Foreign+import Foreign.C.Types+import Foreign.C.String+import Control.Applicative++#include <zmq.h>++#if ZMQ_VERSION_MAJOR != 4+    #error *** INVALID 0MQ VERSION (must be 4.x) ***+#endif++-----------------------------------------------------------------------------+-- Message++newtype ZMQMsg = ZMQMsg+  { content :: Ptr ()+  } deriving (Eq, Ord)++instance Storable ZMQMsg where+    alignment _        = #{alignment zmq_msg_t}+    sizeOf    _        = #{size zmq_msg_t}+    peek p             = ZMQMsg <$> #{peek zmq_msg_t, _} p+    poke p (ZMQMsg c)  = #{poke zmq_msg_t, _} p c++-----------------------------------------------------------------------------+-- Poll++data ZMQPoll = ZMQPoll+    { pSocket  :: {-# UNPACK #-} !ZMQSocket+    , pFd      :: {-# UNPACK #-} !CInt+    , pEvents  :: {-# UNPACK #-} !CShort+    , pRevents :: {-# UNPACK #-} !CShort+    }++instance Storable ZMQPoll where+    alignment _ = #{alignment zmq_pollitem_t}+    sizeOf    _ = #{size zmq_pollitem_t}+    peek p = do+        s  <- #{peek zmq_pollitem_t, socket} p+        f  <- #{peek zmq_pollitem_t, fd} p+        e  <- #{peek zmq_pollitem_t, events} p+        re <- #{peek zmq_pollitem_t, revents} p+        return $ ZMQPoll s f e re+    poke p (ZMQPoll s f e re) = do+        #{poke zmq_pollitem_t, socket} p s+        #{poke zmq_pollitem_t, fd} p f+        #{poke zmq_pollitem_t, events} p e+        #{poke zmq_pollitem_t, revents} p re++type ZMQMsgPtr  = Ptr ZMQMsg+type ZMQCtx     = Ptr ()+type ZMQSocket  = Ptr ()+type ZMQPollPtr = Ptr ZMQPoll++#let alignment t = "%lu", (unsigned long)offsetof(struct {char x__; t (y__); }, y__)++-----------------------------------------------------------------------------+-- Socket Types++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+  , stream   = ZMQ_STREAM+}++-----------------------------------------------------------------------------+-- Socket Options++newtype ZMQOption = ZMQOption+  { optVal :: CInt+  } deriving (Eq, Ord)++#{enum ZMQOption, ZMQOption+  , affinity             = ZMQ_AFFINITY+  , backlog              = ZMQ_BACKLOG+  , conflate             = ZMQ_CONFLATE+  , curve                = ZMQ_CURVE+  , curvePublicKey       = ZMQ_CURVE_PUBLICKEY+  , curveSecretKey       = ZMQ_CURVE_SECRETKEY+  , curveServer          = ZMQ_CURVE_SERVER+  , curveServerKey       = ZMQ_CURVE_SERVERKEY+  , delayAttachOnConnect = ZMQ_DELAY_ATTACH_ON_CONNECT+  , events               = ZMQ_EVENTS+  , filedesc             = ZMQ_FD+  , identity             = ZMQ_IDENTITY+  , immediate            = ZMQ_IMMEDIATE+  , ipv4Only             = ZMQ_IPV4ONLY+  , ipv6                 = ZMQ_IPV6+  , lastEndpoint         = ZMQ_LAST_ENDPOINT+  , linger               = ZMQ_LINGER+  , maxMessageSize       = ZMQ_MAXMSGSIZE+  , mcastHops            = ZMQ_MULTICAST_HOPS+  , mechanism            = ZMQ_MECHANISM+  , null                 = ZMQ_NULL+  , plain                = ZMQ_PLAIN+  , plainPassword        = ZMQ_PLAIN_PASSWORD+  , plainServer          = ZMQ_PLAIN_SERVER+  , plainUserName        = ZMQ_PLAIN_USERNAME+  , probeRouter          = ZMQ_PROBE_ROUTER+  , 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+  , reqCorrelate         = ZMQ_REQ_CORRELATE+  , reqRelaxed           = ZMQ_REQ_RELAXED+  , 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+  , zapDomain            = ZMQ_ZAP_DOMAIN+}++-----------------------------------------------------------------------------+-- Context Options++newtype ZMQCtxOption = ZMQCtxOption+  { ctxOptVal :: CInt+  } deriving (Eq, Ord)++#{enum ZMQCtxOption, ZMQCtxOption+  , _ioThreads = ZMQ_IO_THREADS+  , _maxSockets = ZMQ_MAX_SOCKETS+}++-----------------------------------------------------------------------------+-- Event Type++newtype ZMQEventType = ZMQEventType+  { eventTypeVal :: Word16+  } deriving (Eq, Ord, Show)++#{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+  , monitorStopped = ZMQ_EVENT_MONITOR_STOPPED+}++-----------------------------------------------------------------------------+-- Event++data ZMQEvent = ZMQEvent+  { zeEvent :: {-# UNPACK #-} !ZMQEventType+  , zeValue :: {-# UNPACK #-} !Int32+  }++instance Storable ZMQEvent where+    alignment _ = #{alignment zmq_event_t}+    sizeOf    _ = #{size zmq_event_t}+    peek e = ZMQEvent+        <$> (ZMQEventType <$> #{peek zmq_event_t, event} e)+        <*> #{peek zmq_event_t, value} e+    poke e (ZMQEvent (ZMQEventType a) b) = do+        #{poke zmq_event_t, event} e a+        #{poke zmq_event_t, value} e b++-----------------------------------------------------------------------------+-- Security Mechanism++newtype ZMQSecMechanism = ZMQSecMechanism+  { secMechanism :: Int+  } deriving (Eq, Ord, Show)++#{enum ZMQSecMechanism, ZMQSecMechanism+  , secNull  = ZMQ_NULL+  , secPlain = ZMQ_PLAIN+  , secCurve = ZMQ_CURVE+}++-----------------------------------------------------------------------------+-- Message Options++newtype ZMQMsgOption = ZMQMsgOption+  { msgOptVal :: CInt+  } deriving (Eq, Ord)++#{enum ZMQMsgOption, ZMQMsgOption+  , more = ZMQ_MORE+}++-----------------------------------------------------------------------------+-- Flags++newtype ZMQFlag = ZMQFlag+  { flagVal :: CInt+  } deriving (Eq, Ord)++#{enum ZMQFlag, ZMQFlag+  , dontWait = ZMQ_DONTWAIT+  , sndMore  = ZMQ_SNDMORE+}++-----------------------------------------------------------------------------+-- Poll Events++newtype ZMQPollEvent = ZMQPollEvent+  { pollVal :: CShort+  } deriving (Eq, Ord)++#{enum ZMQPollEvent, ZMQPollEvent,+    pollIn    = ZMQ_POLLIN,+    pollOut   = ZMQ_POLLOUT,+    pollerr   = ZMQ_POLLERR+}++-----------------------------------------------------------------------------+-- function declarations++-- general initialization++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_ctx_new"+    c_zmq_ctx_new :: IO ZMQCtx++foreign import ccall unsafe "zmq.h zmq_ctx_shutdown"+    c_zmq_ctx_shutdown :: ZMQCtx -> IO CInt++foreign import ccall unsafe "zmq.h zmq_ctx_term"+    c_zmq_ctx_term :: 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"+    c_zmq_msg_init :: ZMQMsgPtr -> IO CInt++foreign import ccall unsafe "zmq.h zmq_msg_init_size"+    c_zmq_msg_init_size :: ZMQMsgPtr -> CSize -> IO CInt++foreign import ccall unsafe "zmq.h zmq_msg_close"+    c_zmq_msg_close :: ZMQMsgPtr -> IO CInt++foreign import ccall unsafe "zmq.h zmq_msg_data"+    c_zmq_msg_data :: ZMQMsgPtr -> IO (Ptr a)++foreign import ccall unsafe "zmq.h zmq_msg_size"+    c_zmq_msg_size :: ZMQMsgPtr -> IO CSize++foreign import ccall unsafe "zmq.h zmq_msg_get"+    c_zmq_msg_get :: ZMQMsgPtr -> CInt -> IO CInt++foreign import ccall unsafe "zmq.h zmq_msg_set"+    c_zmq_msg_set :: ZMQMsgPtr -> CInt -> CInt -> IO CInt++-- socket++foreign import ccall unsafe "zmq.h zmq_socket"+    c_zmq_socket :: ZMQCtx -> CInt -> IO ZMQSocket++foreign import ccall unsafe "zmq.h zmq_close"+    c_zmq_close :: ZMQSocket -> IO CInt++foreign import ccall unsafe "zmq.h zmq_setsockopt"+    c_zmq_setsockopt :: ZMQSocket+                     -> CInt   -- option+                     -> Ptr () -- option value+                     -> CSize  -- option value size+                     -> IO CInt++foreign import ccall unsafe "zmq.h zmq_getsockopt"+    c_zmq_getsockopt :: ZMQSocket+                     -> CInt       -- option+                     -> Ptr ()     -- option value+                     -> Ptr CSize  -- option value size ptr+                     -> IO CInt++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++foreign import ccall unsafe "zmq.h zmq_sendmsg"+    c_zmq_sendmsg :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt++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++-- errors++foreign import ccall unsafe "zmq.h zmq_errno"+    c_zmq_errno :: IO CInt++foreign import ccall unsafe "zmq.h zmq_strerror"+    c_zmq_strerror :: CInt -> IO CString++-- proxy++foreign import ccall safe "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++-- Z85 encode/decode++foreign import ccall unsafe "zmq.h zmq_z85_encode"+    c_zmq_z85_encode :: CString -> Ptr Word8 -> CSize -> IO CString++foreign import ccall unsafe "zmq.h zmq_z85_decode"+    c_zmq_z85_decode :: Ptr Word8 -> CString -> IO (Ptr Word8)++-- curve crypto++foreign import ccall unsafe "zmq.h zmq_curve_keypair"+    c_zmq_curve_keypair :: CString -> CString -> IO CInt+
+ src/System/ZMQ4/Error.hs view
@@ -0,0 +1,99 @@+{-# LANGUAGE DeriveDataTypeable #-}++-- We use our own functions for throwing exceptions in order to get+-- the actual error message via 'zmq_strerror'. 0MQ defines additional+-- error numbers besides those defined by the operating system, so+-- 'zmq_strerror' should be used in preference to 'strerror' which is+-- used by the standard throw* functions in 'Foreign.C.Error'.+module System.ZMQ4.Error where++import Control.Applicative+import Control.Monad+import Control.Exception+import Text.Printf+import Data.Typeable (Typeable)++import Foreign hiding (throwIf, throwIf_, void)+import Foreign.C.Error+import Foreign.C.String+import Foreign.C.Types (CInt)++import System.ZMQ4.Base++-- | ZMQError encapsulates information about errors, which occur+-- when using the native 0MQ API, such as error number and message.+data ZMQError = ZMQError+    { errno   :: Int     -- ^ Error number value.+    , source  :: String  -- ^ Source where this error originates from.+    , message :: String  -- ^ Actual error message.+    } deriving (Eq, Ord, Typeable)++instance Show ZMQError where+    show e = printf "ZMQError { errno = %d, source = \"%s\", message = \"%s\" }"+                (errno e) (source e) (message e)++instance Exception ZMQError++throwError :: String -> IO a+throwError src = do+    (Errno e) <- zmqErrno+    msg       <- zmqErrnoMessage e+    throwIO $ ZMQError (fromIntegral e) src msg++throwIf :: (a -> Bool) -> String -> IO a -> IO a+throwIf p src act = do+    r <- act+    if p r then throwError src else return r++throwIf_ :: (a -> Bool) -> String -> IO a -> IO ()+throwIf_ p src act = void $ throwIf p src act++throwIfRetry :: (a -> Bool) -> String -> IO a -> IO a+throwIfRetry p src act = do+    r <- act+    if p r then zmqErrno >>= k else return r+  where+    k e | e == eINTR = throwIfRetry p src act+        | otherwise  = throwError src++throwIfRetry_ :: (a -> Bool) -> String -> IO a -> IO ()+throwIfRetry_ p src act = void $ throwIfRetry p src act++throwIfMinus1 :: (Eq a, Num a) => String -> IO a -> IO a+throwIfMinus1 = throwIf (== -1)++throwIfMinus1_ :: (Eq a, Num a) => String -> IO a -> IO ()+throwIfMinus1_ = throwIf_ (== -1)++throwIfNull :: String -> IO (Ptr a) -> IO (Ptr a)+throwIfNull = throwIf (== nullPtr)++throwIfMinus1Retry :: (Eq a, Num a) => String -> IO a -> IO a+throwIfMinus1Retry = throwIfRetry (== -1)++throwIfMinus1Retry_ :: (Eq a, Num a) => String -> IO a -> IO ()+throwIfMinus1Retry_ = throwIfRetry_ (== -1)++throwIfRetryMayBlock :: (a -> Bool) -> String -> IO a -> IO b -> IO a+throwIfRetryMayBlock p src f on_block = do+    r <- f+    if p r then zmqErrno >>= k else return r+  where+    k e | e == eINTR                      = throwIfRetryMayBlock p src f on_block+        | e == eWOULDBLOCK || e == eAGAIN = on_block >> throwIfRetryMayBlock p src f on_block+        | otherwise                       = throwError src++throwIfRetryMayBlock_ :: (a -> Bool) -> String -> IO a -> IO b -> IO ()+throwIfRetryMayBlock_ p src f on_block = void $ throwIfRetryMayBlock p src f on_block++throwIfMinus1RetryMayBlock :: (Eq a, Num a) => String -> IO a -> IO b -> IO a+throwIfMinus1RetryMayBlock = throwIfRetryMayBlock (== -1)++throwIfMinus1RetryMayBlock_ :: (Eq a, Num a) => String -> IO a -> IO b -> IO ()+throwIfMinus1RetryMayBlock_ = throwIfRetryMayBlock_ (== -1)++zmqErrnoMessage :: CInt -> IO String+zmqErrnoMessage e = c_zmq_strerror e >>= peekCString++zmqErrno :: IO Errno+zmqErrno = Errno <$> c_zmq_errno
+ src/System/ZMQ4/Internal.hs view
@@ -0,0 +1,371 @@+{-# LANGUAGE GADTs              #-}+{-# LANGUAGE StandaloneDeriving #-}++module System.ZMQ4.Internal+    ( Context           (..)+    , Socket            (..)+    , SocketRepr        (..)+    , SocketType        (..)+    , SocketLike        (..)+    , Message           (..)+    , Flag              (..)+    , Timeout+    , Size+    , Switch            (..)+    , EventType         (..)+    , EventMsg          (..)+    , SecurityMechanism (..)+    , KeyFormat         (..)++    , messageOf+    , messageOfLazy+    , messageClose+    , messageInit+    , messageInitSize+    , setIntOpt+    , setStrOpt+    , getIntOpt+    , getStrOpt+    , getInt32Option+    , setInt32OptFromRestricted+    , ctxIntOption+    , setCtxIntOption+    , getByteStringOpt+    , setByteStringOpt++    , z85Encode+    , z85Decode++    , toZMQFlag+    , combine+    , combineFlags+    , mkSocketRepr+    , closeSock+    , onSocket++    , bool2cint+    , toSwitch+    , fromSwitch+    , events2cint+    , eventMessage++    , toMechanism+    , fromMechanism++    , getKey+    ) where++import Control.Applicative+import Control.Monad (foldM_, when, void)+import Control.Monad.IO.Class+import Control.Exception+import Data.IORef (IORef, mkWeakIORef, readIORef, atomicModifyIORef)++import Foreign hiding (throwIfNull, void)+import Foreign.C.String+import Foreign.C.Types (CInt, CSize)++import Data.IORef (newIORef)+import Data.Restricted++import System.Posix.Types (Fd(..))+import System.ZMQ4.Base+import System.ZMQ4.Error++import qualified Data.ByteString        as SB+import qualified Data.ByteString.Lazy   as LB+import qualified Data.ByteString.Unsafe as UB++type Timeout = Int64+type Size    = Word++-- | Flags to apply on send operations (cf. man zmq_send)+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+  | MonitorStoppedEvent+  | AllEvents+  deriving (Eq, Ord, Show)++-- | Event Message to receive when monitoring socket events.+data EventMsg =+    Connected      !SB.ByteString !Fd+  | ConnectDelayed !SB.ByteString+  | ConnectRetried !SB.ByteString !Int+  | Listening      !SB.ByteString !Fd+  | BindFailed     !SB.ByteString !Int+  | Accepted       !SB.ByteString !Fd+  | AcceptFailed   !SB.ByteString !Int+  | Closed         !SB.ByteString !Fd+  | CloseFailed    !SB.ByteString !Int+  | Disconnected   !SB.ByteString !Fd+  | MonitorStopped !SB.ByteString !Int+  deriving (Eq, Show)++data SecurityMechanism+  = Null+  | Plain+  | Curve+  deriving (Eq, Show)++data KeyFormat a where+  BinaryFormat :: KeyFormat Div4+  TextFormat   :: KeyFormat Div5++deriving instance Eq   (KeyFormat a)+deriving instance Show (KeyFormat a)++-- | A 0MQ context representation.+newtype Context = Context { _ctx :: ZMQCtx }++-- | A 0MQ Socket.+newtype Socket a = Socket+  { _socketRepr :: SocketRepr }++data SocketRepr = SocketRepr+  { _socket   :: ZMQSocket+  , _sockLive :: IORef Bool+  }++-- | Socket types.+class SocketType a where+    zmqSocketType :: a -> ZMQSocketType++class SocketLike s where+    toSocket :: s t -> Socket t++instance SocketLike Socket where+    toSocket = id++-- A 0MQ Message representation.+newtype Message = Message { msgPtr :: ZMQMsgPtr }++-- internal helpers:++onSocket :: String -> Socket a -> (ZMQSocket -> IO b) -> IO b+onSocket _func (Socket (SocketRepr sock _state)) act = act sock+{-# INLINE onSocket #-}++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 (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)+    data_ptr <- c_zmq_msg_data (msgPtr msg)+    copyBytes data_ptr cstr len+    return msg++messageOfLazy :: LB.ByteString -> IO Message+messageOfLazy lbs = do+    msg <- messageInitSize (fromIntegral len)+    data_ptr <- c_zmq_msg_data (msgPtr msg)+    let fn offset bs = UB.unsafeUseAsCStringLen bs $ \(cstr, str_len) -> do+        copyBytes (data_ptr `plusPtr` offset) cstr str_len+        return (offset + str_len)+    foldM_ fn 0 (LB.toChunks lbs)+    return msg+ where+    len = LB.length lbs++messageClose :: Message -> IO ()+messageClose (Message ptr) = do+    throwIfMinus1_ "messageClose" $ c_zmq_msg_close ptr+    free ptr++messageInit :: IO Message+messageInit = do+    ptr <- new (ZMQMsg nullPtr)+    throwIfMinus1_ "messageInit" $ c_zmq_msg_init ptr+    return (Message ptr)++messageInitSize :: Size -> IO Message+messageInitSize s = do+    ptr <- new (ZMQMsg nullPtr)+    throwIfMinus1_ "messageInitSize" $+        c_zmq_msg_init_size ptr (fromIntegral s)+    return (Message ptr)++setIntOpt :: (Storable b, Integral b) => Socket a -> ZMQOption -> b -> IO ()+setIntOpt sock (ZMQOption o) i = onSocket "setIntOpt" sock $ \s ->+    throwIfMinus1Retry_ "setIntOpt" $ with i $ \ptr ->+        c_zmq_setsockopt s (fromIntegral o)+                           (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 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+    bracket (new i) free $ \iptr ->+        bracket (new (fromIntegral . sizeOf $ i :: CSize)) free $ \jptr -> do+            throwIfMinus1Retry_ "getIntOpt" $+                c_zmq_getsockopt s (fromIntegral o) (castPtr iptr) jptr+            peek iptr++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_ "getCStrOpt" $+            c_zmq_getsockopt s (fromIntegral o) (castPtr bPtr) sPtr+        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)++setInt32OptFromRestricted :: Integral i => ZMQOption -> Restricted r 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)++z85Encode :: (MonadIO m) => Restricted Div4 SB.ByteString -> m SB.ByteString+z85Encode b = liftIO $ UB.unsafeUseAsCStringLen (rvalue b) $ \(c, s) ->+    allocaBytes ((s * 5) `div` 4 + 1) $ \w -> do+        void . throwIfNull "z85Encode" $+            c_zmq_z85_encode w (castPtr c) (fromIntegral s)+        SB.packCString w++z85Decode :: (MonadIO m) => Restricted Div5 SB.ByteString -> m SB.ByteString+z85Decode b = liftIO $ SB.useAsCStringLen (rvalue b) $ \(c, s) -> do+    let size = (s * 4) `div` 5+    allocaBytes size $ \w -> do+        void . throwIfNull "z85Decode" $+            c_zmq_z85_decode (castPtr w) (castPtr c)+        SB.packCStringLen (w, size)++getKey :: KeyFormat f -> Socket a -> ZMQOption -> IO SB.ByteString+getKey kf sock (ZMQOption o) = onSocket "getKey" sock $ \s -> do+    let len = case kf of+            BinaryFormat -> 32+            TextFormat   -> 41+    with len $ \lenptr -> allocaBytes len $ \w -> do+        throwIfMinus1Retry_ "getKey" $+            c_zmq_getsockopt s (fromIntegral o) (castPtr w) (castPtr lenptr)+        SB.packCString w++toZMQFlag :: Flag -> ZMQFlag+toZMQFlag DontWait = dontWait+toZMQFlag SendMore = sndMore++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+bool2cint False = 0++toSwitch :: (Show a, Integral a) => String -> a -> Switch+toSwitch _ (-1) = Default+toSwitch _   0  = Off+toSwitch _   1  = On+toSwitch m   n  = error $ m ++ ": " ++ show n++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+toZMQEventType MonitorStoppedEvent = monitorStopped++toMechanism :: SecurityMechanism -> ZMQSecMechanism+toMechanism Null  = secNull+toMechanism Plain = secPlain+toMechanism Curve = secCurve++fromMechanism :: String -> Int -> SecurityMechanism+fromMechanism s m+    | m == secMechanism secNull  = Null+    | m == secMechanism secPlain = Plain+    | m == secMechanism secCurve = Curve+    | otherwise                  = error $ s ++ ": " ++ show m++events2cint :: [EventType] -> CInt+events2cint = fromIntegral . foldr ((.|.) . eventTypeVal . toZMQEventType) 0++eventMessage :: SB.ByteString -> ZMQEvent -> EventMsg+eventMessage str (ZMQEvent e v)+    | e == connected      = Connected      str (Fd . fromIntegral $ v)+    | e == connectDelayed = ConnectDelayed str+    | e == connectRetried = ConnectRetried str (fromIntegral $ v)+    | e == listening      = Listening      str (Fd . fromIntegral $ v)+    | e == bindFailed     = BindFailed     str (fromIntegral $ v)+    | e == accepted       = Accepted       str (Fd . fromIntegral $ v)+    | e == acceptFailed   = AcceptFailed   str (fromIntegral $ v)+    | e == closed         = Closed         str (Fd . fromIntegral $ v)+    | e == closeFailed    = CloseFailed    str (fromIntegral $ v)+    | e == disconnected   = Disconnected   str (fromIntegral $ v)+    | e == monitorStopped = MonitorStopped str (fromIntegral $ v)+    | otherwise           = error $ "unknown event type: " ++ show e
+ src/System/ZMQ4/Monadic.hs view
@@ -0,0 +1,606 @@+{-# LANGUAGE RankNTypes                         #-}+{-# OPTIONS_GHC -fno-warn-warnings-deprecations #-}++-- |+-- Module      : System.ZMQ4.Monadic+-- Copyright   : (c) 2013 Toralf Wittner+-- License     : MIT+-- Maintainer  : Toralf Wittner <tw@dtex.org>+-- Stability   : experimental+-- Portability : non-portable+--+-- This modules exposes a monadic interface of 'System.ZMQ4'. 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.ZMQ4.Monadic+  ( -- * Type Definitions+    ZMQ+  , Socket+  , Z.Flag              (SendMore)+  , Z.Switch            (..)+  , Z.Timeout+  , Z.Event             (..)+  , Z.EventType         (..)+  , Z.EventMsg          (..)+  , Z.Poll              (..)+  , Z.KeyFormat         (..)+  , Z.SecurityMechanism (..)++  -- ** Socket type-classes+  , Z.SocketType+  , Z.Sender+  , Z.Receiver+  , Z.Subscriber+  , Z.SocketLike+  , Z.Conflatable+  , Z.SendProbe++  -- ** Socket Types+  , Z.Pair   (..)+  , Z.Pub    (..)+  , Z.Sub    (..)+  , Z.XPub   (..)+  , Z.XSub   (..)+  , Z.Req    (..)+  , Z.Rep    (..)+  , Z.Dealer (..)+  , Z.Router (..)+  , Z.Pull   (..)+  , Z.Push   (..)+  , Z.Stream (..)++  -- * 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+  , conflate+  , curvePublicKey+  , curveSecretKey+  , curveServerKey+  , delayAttachOnConnect+  , events+  , fileDescriptor+  , identity+  , immediate+  , ipv4Only+  , ipv6+  , lastEndpoint+  , linger+  , maxMessageSize+  , mcastHops+  , mechanism+  , moreToReceive+  , plainServer+  , plainPassword+  , plainUserName+  , rate+  , receiveBuffer+  , receiveHighWM+  , receiveTimeout+  , reconnectInterval+  , reconnectIntervalMax+  , recoveryInterval+  , sendBuffer+  , sendHighWM+  , sendTimeout+  , tcpKeepAlive+  , tcpKeepAliveCount+  , tcpKeepAliveIdle+  , tcpKeepAliveInterval+  , zapDomain++  -- * Socket Options (Write)+  , setAffinity+  , setBacklog+  , setConflate+  , setCurveServer+  , setCurvePublicKey+  , setCurveSecretKey+  , setCurveServerKey+  , setDelayAttachOnConnect+  , setIdentity+  , setImmediate+  , setIpv4Only+  , setIpv6+  , setLinger+  , setMaxMessageSize+  , setMcastHops+  , setPlainServer+  , setPlainPassword+  , setPlainUserName+  , setProbeRouter+  , setRate+  , setReceiveBuffer+  , setReceiveHighWM+  , setReceiveTimeout+  , setReconnectInterval+  , setReconnectIntervalMax+  , setRecoveryInterval+  , setReqCorrelate+  , setReqRelaxed+  , 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+  , I.z85Encode+  , I.z85Decode+  , Z.curveKeyPair+  ) where++import Control.Applicative+import Control.Concurrent.Async (Async)+import Control.Monad+import Control.Monad.Trans.Reader+import Control.Monad.IO.Class+import Control.Monad.Catch+import Control.Monad.CatchIO+import Data.Int+import Data.IORef+import Data.List.NonEmpty (NonEmpty)+import Data.Restricted+import Data.Word+import Data.ByteString (ByteString)+import System.Posix.Types (Fd)++import qualified Control.Concurrent.Async as A+import qualified Control.Exception        as E+import qualified Control.Monad.Catch      as C+import qualified Control.Monad.CatchIO    as M+import qualified Data.ByteString.Lazy     as Lazy+import qualified System.ZMQ4              as Z+import qualified System.ZMQ4.Internal     as I++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.ZMQ4.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 I.SocketLike (Socket z) where+    toSocket = _unsocket++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 MonadCatch (ZMQ z) where+    throwM          = ZMQ . C.throwM+    catch (ZMQ m) f = ZMQ $ m `C.catch` (_unzmq . f)++    mask a = ZMQ . ReaderT $ \env ->+        C.mask $ \restore ->+            let f (ZMQ (ReaderT b)) = ZMQ $ ReaderT (restore . b)+            in runReaderT (_unzmq (a $ f)) env++    uninterruptibleMask a = ZMQ . ReaderT $ \env ->+        C.uninterruptibleMask $ \restore ->+            let f (ZMQ (ReaderT b)) = ZMQ $ ReaderT (restore . b)+            in runReaderT (_unzmq (a $ f)) env++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.ZMQ4.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 term (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 'Control.Concurrent.Async.async'.+-- /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.ZMQ4.Context' will only be disposed after+-- 'async' finishes as well.+async :: ZMQ z a -> ZMQ z (Async a)+async z = ZMQ $ do+    e <- ask+    liftIO $ atomicModifyIORef (_refcount e) $ \n -> (succ n, ())+    liftIO . A.async $ (runReaderT (_unzmq z) e) `E.finally` term e++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 -> NonEmpty 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++conflate :: Z.Conflatable t => Socket z t -> ZMQ z Bool+conflate = liftIO . Z.conflate . _unsocket++curvePublicKey :: Z.KeyFormat f -> Socket z t -> ZMQ z ByteString+curvePublicKey f = liftIO . Z.curvePublicKey f . _unsocket++curveSecretKey :: Z.KeyFormat f -> Socket z t -> ZMQ z ByteString+curveSecretKey f = liftIO . Z.curveSecretKey f . _unsocket++curveServerKey :: Z.KeyFormat f -> Socket z t -> ZMQ z ByteString+curveServerKey f = liftIO . Z.curveServerKey f . _unsocket++delayAttachOnConnect :: Socket z t -> ZMQ z Bool+delayAttachOnConnect = liftIO . Z.delayAttachOnConnect . _unsocket+{-# DEPRECATED delayAttachOnConnect "Use immediate" #-}++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++immediate :: Socket z t -> ZMQ z Bool+immediate = liftIO . Z.immediate . _unsocket++ipv4Only :: Socket z t -> ZMQ z Bool+ipv4Only = liftIO . Z.ipv4Only . _unsocket+{-# DEPRECATED ipv4Only "Use ipv6" #-}++ipv6 :: Socket z t -> ZMQ z Bool+ipv6 = liftIO . Z.ipv6 . _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++mechanism :: Socket z t -> ZMQ z Z.SecurityMechanism+mechanism = liftIO . Z.mechanism . _unsocket++moreToReceive :: Socket z t -> ZMQ z Bool+moreToReceive = liftIO . Z.moreToReceive . _unsocket++plainServer :: Socket z t -> ZMQ z Bool+plainServer = liftIO . Z.plainServer . _unsocket++plainPassword :: Socket z t -> ZMQ z ByteString+plainPassword = liftIO . Z.plainPassword . _unsocket++plainUserName :: Socket z t -> ZMQ z ByteString+plainUserName = liftIO . Z.plainUserName . _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++zapDomain :: Socket z t -> ZMQ z ByteString+zapDomain = liftIO . Z.zapDomain . _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++setConflate :: Z.Conflatable t => Bool -> Socket z t -> ZMQ z ()+setConflate i = liftIO . Z.setConflate i . _unsocket++setCurvePublicKey :: Z.KeyFormat f -> Restricted f ByteString -> Socket z t -> ZMQ z ()+setCurvePublicKey f k = liftIO . Z.setCurvePublicKey f k . _unsocket++setCurveSecretKey :: Z.KeyFormat f -> Restricted f ByteString -> Socket z t -> ZMQ z ()+setCurveSecretKey f k = liftIO . Z.setCurveSecretKey f k . _unsocket++setCurveServer :: Bool -> Socket z t -> ZMQ z ()+setCurveServer b = liftIO . Z.setCurveServer b . _unsocket++setCurveServerKey :: Z.KeyFormat f -> Restricted f ByteString -> Socket z t -> ZMQ z ()+setCurveServerKey f k = liftIO . Z.setCurveServerKey f k . _unsocket++setDelayAttachOnConnect :: Bool -> Socket z t -> ZMQ z ()+setDelayAttachOnConnect d = liftIO . Z.setDelayAttachOnConnect d . _unsocket+{-# DEPRECATED setDelayAttachOnConnect "Use setImmediate" #-}++setIdentity :: Restricted (N1, N254) ByteString -> Socket z t -> ZMQ z ()+setIdentity i = liftIO . Z.setIdentity i . _unsocket++setImmediate :: Bool -> Socket z t -> ZMQ z ()+setImmediate i = liftIO . Z.setImmediate i . _unsocket++setIpv4Only :: Bool -> Socket z t -> ZMQ z ()+setIpv4Only i = liftIO . Z.setIpv4Only i . _unsocket+{-# DEPRECATED setIpv4Only "Use setIpv6" #-}++setIpv6 :: Bool -> Socket z t -> ZMQ z ()+setIpv6 i = liftIO . Z.setIpv6 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++setPlainServer :: Bool -> Socket z t -> ZMQ z ()+setPlainServer b = liftIO . Z.setPlainServer b . _unsocket++setPlainPassword :: Restricted (N1, N254) ByteString -> Socket z t -> ZMQ z ()+setPlainPassword s = liftIO . Z.setPlainPassword s . _unsocket++setPlainUserName :: Restricted (N1, N254) ByteString -> Socket z t -> ZMQ z ()+setPlainUserName s = liftIO . Z.setPlainUserName s . _unsocket++setProbeRouter :: Z.SendProbe t => Bool -> Socket z t -> ZMQ z ()+setProbeRouter b = liftIO . Z.setProbeRouter b . _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++setReqCorrelate :: Bool -> Socket z Z.Req -> ZMQ z ()+setReqCorrelate b = liftIO . Z.setReqCorrelate b . _unsocket++setReqRelaxed :: Bool -> Socket z Z.Req -> ZMQ z ()+setReqRelaxed b = liftIO . Z.setReqRelaxed b . _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++term :: ZMQEnv -> IO ()+term env = do+    n <- atomicModifyIORef (_refcount env) $ \n -> (pred n, n)+    when (n == 1) $ do+        readIORef (_sockets env) >>= mapM_ close'+        Z.term (_context env)+  where+    close' s = I.closeSock s `E.catch` (\e -> print (e :: E.SomeException))
+ tests/System/ZMQ4/Test/Properties.hs view
@@ -0,0 +1,194 @@+{-# LANGUAGE FlexibleInstances    #-}+{-# LANGUAGE OverloadedStrings    #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}++module System.ZMQ4.Test.Properties where++import Test.QuickCheck+import Test.QuickCheck.Monadic+import Test.Tools++import Control.Applicative+import Control.Concurrent.Async (wait)+import Data.Int+import Data.Word+import Data.Restricted+import Data.Maybe (fromJust)+import Data.ByteString (ByteString)+import System.ZMQ4.Monadic+import System.Posix.Types (Fd(..))+import qualified Data.ByteString as SB+import qualified Data.ByteString.Char8 as CB++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+    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 $ 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+    ieq :: (Integral i, Integral k) => i -> k -> Bool+    ieq i k  = (fromIntegral i :: Int) == (fromIntegral k :: Int)++prop_subscribe :: (Subscriber a, SocketType a) => a -> ByteString -> Property+prop_subscribe t subs = monadicIO $ run $+    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 $ runZMQ $ do+        sender   <- socket a+        receiver <- socket b+        bind receiver "inproc://endpoint"+        x <- async $ receive receiver+        connect sender "inproc://endpoint"+        send sender [] msg+        liftIO $ wait x+    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 $ 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 . filter (/= '\0') <$> arbitrary++data GetOpt =+    Events          Int+  | Filedesc        Fd+  | ReceiveMore     Bool+  deriving Show++data SetOpt =+    Affinity        Word64+  | Backlog         (Restricted (N0, Int32) Int)+  | Identity        (Restricted (N1, N254) ByteString)+  | Ipv4Only        Bool+  | Linger          (Restricted (Nneg1, Int32) Int)+  | MaxMessageSize  (Restricted (Nneg1, Int64) Int64)+  | McastHops       (Restricted (N1, Int32) Int)+  | Rate            (Restricted (N1, Int32) Int)+  | ReceiveBuf      (Restricted (N0, Int32) Int)+  | ReceiveHighWM   (Restricted (N0, Int32) Int)+  | ReceiveTimeout  (Restricted (Nneg1, Int32) Int)+  | ReconnectIVL    (Restricted (N0, Int32) Int)+  | ReconnectIVLMax (Restricted (N0, Int32) Int)+  | RecoveryIVL     (Restricted (N0, Int32) Int)+  | SendBuf         (Restricted (N0, Int32) Int)+  | SendHighWM      (Restricted (N0, Int32) Int)+  | SendTimeout     (Restricted (Nneg1, Int32) Int)+  deriving Show++instance Arbitrary GetOpt where+    arbitrary = oneof [+        Events                       <$> arbitrary+      , Filedesc . Fd . fromIntegral <$> (arbitrary :: Gen Int32)+      , ReceiveMore                  <$> arbitrary+      ]++instance Arbitrary SetOpt where+    arbitrary = oneof [+        Affinity                   <$> (arbitrary :: Gen Word64)+      , Ipv4Only                   <$> (arbitrary :: Gen Bool)+      , Backlog         . toR0     <$> (arbitrary :: Gen Int32) `suchThat` (>=  0)+      , Linger          . toRneg1  <$> (arbitrary :: Gen Int32) `suchThat` (>= -1)+      , Rate            . toR1     <$> (arbitrary :: Gen Int32) `suchThat` (>   0)+      , ReceiveBuf      . toR0     <$> (arbitrary :: Gen Int32) `suchThat` (>=  0)+      , ReconnectIVL    . toR0     <$> (arbitrary :: Gen Int32) `suchThat` (>=  0)+      , ReconnectIVLMax . toR0     <$> (arbitrary :: Gen Int32) `suchThat` (>=  0)+      , RecoveryIVL     . toR0     <$> (arbitrary :: Gen Int32) `suchThat` (>=  0)+      , SendBuf         . toR0     <$> (arbitrary :: Gen Int32) `suchThat` (>=  0)+      , McastHops       . toR1     <$> (arbitrary :: Gen Int32) `suchThat` (>   0)+      , ReceiveHighWM   . toR0     <$> (arbitrary :: Gen Int32) `suchThat` (>=  0)+      , ReceiveTimeout  . toRneg1  <$> (arbitrary :: Gen Int32) `suchThat` (>= -1)+      , SendHighWM      . toR0     <$> (arbitrary :: Gen Int32) `suchThat` (>=  0)+      , SendTimeout     . toRneg1  <$> (arbitrary :: Gen Int32) `suchThat` (>= -1)+      , MaxMessageSize  . toRneg1' <$> (arbitrary :: Gen Int64) `suchThat` (>= -1)+      , Identity . fromJust . toRestricted <$> arbitrary `suchThat` (\s -> SB.length s > 0 && SB.length s < 255)+      ]++toR1 :: Int32 -> Restricted (N1, Int32) Int+toR1 = fromJust . toRestricted . fromIntegral++toR0 :: Int32 -> Restricted (N0, Int32) Int+toR0 = fromJust . toRestricted . fromIntegral++toRneg1 :: Int32 -> Restricted (Nneg1, Int32) Int+toRneg1 = fromJust . toRestricted . fromIntegral++toRneg1' :: Int64 -> Restricted (Nneg1, Int64) Int64+toRneg1' = fromJust . toRestricted . fromIntegral
+ tests/Test/Tools.hs view
@@ -0,0 +1,35 @@+module Test.Tools (quickBatch', checkBatch') where++import Control.Monad+import System.Console.ANSI+import System.Exit+import Test.QuickCheck+import Test.QuickCheck.Checkers+import qualified Control.Exception as E++quickBatch' :: TestBatch -> IO ()+quickBatch' = checkBatch' (stdArgs { maxSuccess = 500 })++checkBatch' :: Args -> TestBatch -> IO ()+checkBatch' args (name, tsts) = do+    writeLn Cyan name+    forM_ tsts $ \(s, p) -> do+        write White ("    " ++ s ++ ": ")+        r <- quickCheckWithResult (args { chatty = False}) p+             `E.catch`+             ((\e -> write Red (show e) >> exitFailure) :: E.SomeException -> IO a)+        case r of+            Success _ _ m           -> write Green m+            GaveUp  _ _ m           -> write Magenta m >> exitFailure+            Failure _ _ _ _ _ _ _ m -> write Red m     >> exitFailure+            NoExpectedFailure _ _ m -> write Red m     >> exitFailure++write, writeLn :: Color -> String -> IO ()+write   c = withColour c . putStr+writeLn c = withColour c . putStrLn++withColour :: Color -> IO () -> IO ()+withColour c a = do+    setSGR [Reset, SetColor Foreground Vivid c]+    a+    setSGR [Reset]
+ tests/tests.hs view
@@ -0,0 +1,5 @@+import qualified System.ZMQ4.Test.Properties as Properties++main :: IO ()+main = Properties.tests+
+ zeromq4-haskell.cabal view
@@ -0,0 +1,85 @@+name:               zeromq4-haskell+version:            0.1+synopsis:           Bindings to ZeroMQ 4.x+category:           System, FFI+license:            MIT+license-file:       LICENSE+author:             Toralf Wittner+maintainer:         Toralf Wittner <tw@dtex.org>+copyright:          Copyright (c) 2010 - 2013 zeromq-haskell authors+homepage:           http://github.com/twittner/zeromq-haskell/+stability:          experimental+tested-With:        GHC == 7.6.3+cabal-version:      >= 1.8+build-type:         Simple+extra-source-files:+      README.md+    , AUTHORS+    , examples/*.hs+    , examples/Makefile+    , examples/perf/*.hs+    , examples/perf/Makefile+    , tests/*.hs+    , tests/System/ZMQ4/Test/*.hs++description:+    The 0MQ lightweight messaging kernel is a library which extends+    the standard socket interfaces with features traditionally provided+    by specialised messaging middleware products.+    .+    0MQ sockets provide an abstraction of asynchronous message queues,+    multiple messaging patterns, message filtering (subscriptions),+    seamless access to multiple transport protocols and more.+    .+    This library provides the Haskell language binding to 0MQ >= 4.x++source-repository head+    type:             git+    location:         https://github.com/twittner/zeromq-haskell++library+    hs-source-dirs:       src+    exposed-modules:+        System.ZMQ4+        System.ZMQ4.Monadic+        Data.Restricted++    other-modules:+        System.ZMQ4.Base+        System.ZMQ4.Internal+        System.ZMQ4.Error++    includes:             zmq.h+    ghc-options:          -Wall -O2 -fwarn-tabs -funbox-strict-fields+    build-depends:+          base >= 3 && < 5+        , async                     == 2.0.*+        , bytestring+        , containers+        , exceptions                >= 0.3+        , MonadCatchIO-transformers+        , semigroups                >= 0.8+        , transformers              >= 0.3++    if os(freebsd)+        extra-libraries:  zmq, pthread+    else+        extra-libraries:  zmq++test-suite zeromq-haskell-tests+    type:             exitcode-stdio-1.0+    hs-source-dirs:   tests+    main-is:          tests.hs+    other-modules:    Test.Tools+    ghc-options:      -Wall -threaded -rtsopts+    build-depends:+          zeromq4-haskell+        , ansi-terminal             >= 0.6+        , async                     == 2.0.*+        , base                      >= 3 && < 5+        , bytestring+        , checkers                  >= 0.3+        , containers+        , MonadCatchIO-transformers+        , QuickCheck                >= 2.6+        , transformers