packages feed

zeromq3-haskell (empty) → 0.1

raw patch · 20 files changed

+1823/−0 lines, 20 filesdep +QuickCheckdep +basedep +bytestringsetup-changed

Dependencies added: QuickCheck, base, bytestring, containers, test-framework, test-framework-quickcheck2, zeromq3-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+
+ 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,55 @@+This library provides Haskell bindings to 0MQ 3.x (http://zeromq.org).++Current status+--------------++This software currently has *beta* status, i.e. it had seen limited testing.++Version 0.1 - First release to provide bindings against 0MQ 3.1.0++Installation+------------++As usual for Haskell packages this software is installed best via Cabal+(http://www.haskell.org/cabal). In addition to GHC it depends on 0MQ 3.1.x+of course.++Notes+-----++zeromq3-haskell mostly follows 0MQ's API. One difference though is that sockets+are parameterized types, i.e. there is not one single socket type but when+creating a socket the desired socket type has to be specified, e.g. `Pair` and+the resulting socket is of type `Socket Pair`.+This additional type information is used to ensure that only options applicable+to the socket type can be set.++Other differences are mostly for convenience. Also one does not deal directly+with 0MQ messages, instead these are created internally as needed.++Finally note that `receive` is already non-blocking internally.+GHC's I/O manager is used to wait for data to be available, so from a client's+perspective `receive` appears to be blocking.++Differences to the 0MQ 2.x binding+----------------------------------++This library is based on the zeromq-haskell binding for 0MQ 2.x. Socket types+and options have been aligned with 0MQ 3.x and instead of using a big+`SocketOption` datatype, this library provides separate get and set functions for+each available option, e.g. `affinity`/`setAffinity`. For details, please refer+to the module's haddock documentation.++Examples+--------++The examples folder contains some simple tests mostly mimicking the ones that come+with 0MQ.++Bugs+----++If you find any bugs or other shortcomings I would greatly appreciate a bug+report, preferably via http://github.com/twittner/zeromq-haskell/issues or+e-mail to toralf.wittner@gmail.com+
+ 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 *.o *.hi+	@rm display prompt+
+ examples/display.hs view
@@ -0,0 +1,23 @@+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import Control.Exception+import qualified System.ZMQ3 as ZMQ+import qualified Data.ByteString as SB++main :: IO ()+main = do+    args <- getArgs+    when (length args < 1) $ do+        hPutStrLn stderr "usage: display <address> [<address>, ...]"+        exitFailure+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Sub $ \s -> do+            ZMQ.subscribe s ""+            mapM (ZMQ.connect s) args+            forever $ do+                line <- ZMQ.receive s+                SB.putStrLn line+                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 *.o *.hi+	@rm 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 qualified System.ZMQ3 as ZMQ+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+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Rep $ \s -> do+            ZMQ.bind s bindTo+            loop s rounds size+  where+    loop s r sz = unless (r <= 0) $ do+        msg <- ZMQ.receive s+        when (SB.length msg /= sz) $+            error "message of incorrect size received"+        ZMQ.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,49 @@+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import Data.Time.Clock+import qualified System.ZMQ3 as ZMQ+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 :: Int+        count  = read $ args !! 2 :: Int+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Sub $ \s -> do+            ZMQ.subscribe s ""+            ZMQ.bind s bindTo+            receive s size+            start <- getCurrentTime+            loop s (count - 1) size+            end <- getCurrentTime+            printStat start end size count+  where+    receive s sz = do+        msg <- ZMQ.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 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 qualified System.ZMQ3 as ZMQ+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+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Req $ \s -> do+            ZMQ.connect s connTo+            start <- getCurrentTime+            loop s rounds message+            end <- getCurrentTime+            print (diffUTCTime end start)+  where+    loop s r msg = unless (r <= 0) $ do+        ZMQ.send s [] msg+        msg' <- ZMQ.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 qualified System.ZMQ3 as ZMQ+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+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Pub $ \s -> do+            ZMQ.connect s connTo+            replicateM_ count $ ZMQ.send s [] message+            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 System.IO+import System.Exit+import System.Environment+import qualified System.ZMQ3 as ZMQ+import qualified Data.ByteString.UTF8 as SB+import qualified Data.ByteString.Char8 as SB++main :: IO ()+main = do+    args <- getArgs+    when (length args /= 2) $ do+        hPutStrLn stderr "usage: prompt <address> <username>"+        exitFailure+    let addr = args !! 0+        name = SB.append (SB.fromString $ args !! 1) ": "+    ZMQ.withContext 1 $ \c ->+        ZMQ.withSocket c ZMQ.Pub $ \s -> do+            ZMQ.bind s addr+            forever $ do+                line <- SB.fromString <$> getLine+                ZMQ.send s [] (SB.append name line)+
+ src/Data/Restricted.hs view
@@ -0,0 +1,168 @@+{-# LANGUAGE EmptyDataDecls         #-}+{-# LANGUAGE MultiParamTypeClasses  #-}+{-# LANGUAGE FlexibleInstances      #-}+{-# LANGUAGE UndecidableInstances   #-}+{-# LANGUAGE TypeSynonymInstances   #-}++-- |+-- Module      : Data.Restricted+-- Copyright   : (c) 2011-2012 Toralf Wittner+-- License     : MIT+-- Maintainer  : toralf.wittner@gmail.com+-- 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++) where++import Data.Int++-- | Type level restriction.+data Restricted l u v = Restricted !v deriving Show++-- | A uniform way to restrict values.+class Restriction l u v where++    -- | Create a restricted value. Returns 'Nothing' if+    -- the given value does not satisfy all restrictions.+    toRestricted :: v -> Maybe (Restricted l u 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 l u v++-- | Get the actual value.+rvalue :: Restricted l u 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++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"++-- 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)++-- 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
+ src/System/ZMQ3.hs view
@@ -0,0 +1,625 @@+-- |+-- Module      : System.ZMQ3+-- Copyright   : (c) 2010-2012 Toralf Wittner+-- License     : MIT+-- Maintainer  : toralf.wittner@gmail.com+-- Stability   : experimental+-- Portability : non-portable+--+-- 0MQ haskell binding. The API closely follows the C-API of 0MQ with+-- the main difference that sockets are typed.+-- The documentation of the individual socket types is copied from+-- 0MQ's man pages authored by Martin Sustrik. For details please+-- refer to <http://api.zeromq.org>+--+-- Differences to zeromq-haskell 2.x+--+-- /Socket Types/+--+-- * 'System.ZMQ.Up' and 'System.ZMQ.Down' no longer exist.+--+-- * Some 0MQ tutorials mention 'ZMQ_DEALER' and 'ZMQ_ROUTER'. These+-- are aliases of 'XRep' and 'XReq'.+--+-- * Renamed type-classes:+-- @'SType' -\> 'SocketType'@, @'SubsType' -\> 'Subscriber'@.+--+-- * New type-classes:+-- 'Sender', 'Receiver'+--+-- /Socket Options/+--+-- Instead of a single 'SocketOption' data-type, getter and setter+-- functions are provided, e.g. one would write: @'affinity' sock@ instead of+-- @getOption sock (Affinity 0)@+--+-- /Restrictions/+--+-- Many option setters 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'.+--+-- /Devices/+--+-- Devices are no longer present in 0MQ 3.x and consequently have been+-- removed form this binding as well.+--+-- /Poll/+--+-- Removed support for polling. This should not be necessary as 'send' and+-- 'receive' are internally non-blocking and use GHC's I/O manager to block+-- calling threads when send or receive would yield EAGAIN. This combined with+-- GHC's scalable threading model should relieve client code from the burden+-- to do it's own polling. For timeouts please consider 'System.Timeout.timeout'.+--+-- /Error Handling/+--+-- The type 'ZMQError' is introduced, together with inspection functions 'errno',+-- 'source' and 'message'. @zmq_strerror@ is used underneath to retrieve the+-- correct error message. ZMQError will be thrown when native 0MQ procedures return+-- an error status and it can be 'catch'ed as an 'Exception'.++module System.ZMQ3 (++    -- * Type Definitions+    Size+  , Context+  , Socket+  , Flag (SndMore)+  , Timeout+  , Event (..)++    -- ** Type Classes+  , SocketType+  , Sender+  , Receiver+  , Subscriber++    -- ** Socket Types+  , Pair(..)+  , Pub(..)+  , Sub(..)+  , XPub(..)+  , XSub(..)+  , Req(..)+  , Rep(..)+  , XReq(..)+  , XRep(..)+  , Pull(..)+  , Push(..)++    -- * General Operations+  , withContext+  , withSocket+  , bind+  , connect+  , send+  , send'+  , receive+  , version++  , System.ZMQ3.subscribe+  , System.ZMQ3.unsubscribe++    -- * Socket Options (Read)+  , System.ZMQ3.affinity+  , System.ZMQ3.backlog+  , System.ZMQ3.events+  , System.ZMQ3.fileDescriptor+  , System.ZMQ3.identity+  , System.ZMQ3.ipv4Only+  , System.ZMQ3.linger+  , System.ZMQ3.maxMessageSize+  , System.ZMQ3.mcastHops+  , System.ZMQ3.moreToReceive+  , System.ZMQ3.rate+  , System.ZMQ3.receiveBuffer+  , System.ZMQ3.receiveHighWM+  , System.ZMQ3.receiveTimeout+  , System.ZMQ3.reconnectInterval+  , System.ZMQ3.reconnectIntervalMax+  , System.ZMQ3.recoveryInterval+  , System.ZMQ3.sendBuffer+  , System.ZMQ3.sendHighWM+  , System.ZMQ3.sendTimeout++    -- * Socket Options (Write)+  , setAffinity+  , setBacklog+  , setIdentity+  , setLinger+  , setRate+  , setReceiveBuffer+  , setReconnectInterval+  , setReconnectIntervalMax+  , setRecoveryInterval+  , setSendBuffer+  , setIpv4Only+  , setMcastHops+  , setReceiveHighWM+  , setReceiveTimeout+  , setSendHighWM+  , setSendTimeout+  , setMaxMessageSize++    -- * Restrictions+  , Data.Restricted.restrict+  , Data.Restricted.toRestricted++    -- * Error Handling+  , ZMQError+  , errno+  , source+  , message++    -- * Low-level Functions+  , init+  , term+  , socket+  , close++) where++import Prelude hiding (init)+import Control.Applicative+import Control.Exception+import Control.Monad (unless, when)+import Data.Restricted+import Data.IORef (atomicModifyIORef)+import Foreign hiding (throwIf, throwIf_, throwIfNull)+import Foreign.C.String+import Foreign.C.Types (CInt)+import qualified Data.ByteString as SB+import qualified Data.ByteString.Lazy as LB+import System.Mem.Weak (addFinalizer)+import System.Posix.Types (Fd(..))+import System.ZMQ3.Base+import qualified System.ZMQ3.Base as B+import System.ZMQ3.Internal+import System.ZMQ3.Error++import GHC.Conc (threadWaitRead, threadWaitWrite)++-- | Socket to communicate with a single peer. Allows for only a+-- single connect or a single bind. There's no message routing+-- or message filtering involved. /Compatible peer sockets/: 'Pair'.+data Pair = Pair++-- | Socket to distribute data. 'receive' function is not+-- implemented for this socket type. Messages are distributed in+-- fanout fashion to all the peers. /Compatible peer sockets/: 'Sub'.+data Pub = Pub++-- | Socket to subscribe for data. 'send' function is not implemented+-- for this socket type. Initially, socket is subscribed for no+-- messages. Use 'subscribe' to specify which messages to subscribe for.+-- /Compatible peer sockets/: 'Pub'.+data Sub = Sub++-- | Same as 'Pub' except that you can receive subscriptions from the+-- peers in form of incoming messages. Subscription message is a byte 1+-- (for subscriptions) or byte 0 (for unsubscriptions) followed by the+-- subscription body.+-- /Compatible peer sockets/: 'Sub', 'XSub'.+data XPub = XPub++-- | Same as 'Sub' except that you subscribe by sending subscription+-- messages to the socket. Subscription message is a byte 1 (for subscriptions)+-- or byte 0 (for unsubscriptions) followed by the subscription body.+-- /Compatible peer sockets/: 'Pub', 'XPub'.+data XSub = XSub++-- | Socket to send requests and receive replies. Requests are+-- load-balanced among all the peers. This socket type allows only an+-- alternated sequence of send's and recv's.+-- /Compatible peer sockets/: 'Rep', 'Xrep'.+data Req = Req++-- | Socket to receive requests and send replies. This socket type+-- allows only an alternated sequence of receive's and send's. Each+-- send is routed to the peer that issued the last received request.+-- /Compatible peer sockets/: 'Req', 'XReq'.+data Rep = Rep++-- | Special socket type to be used in request/reply middleboxes+-- such as zmq_queue(7).  Requests forwarded using this socket type+-- should be tagged by a proper prefix identifying the original requester.+-- Replies received by this socket are tagged with a proper postfix+-- that can be use to route the reply back to the original requester.+-- /Compatible peer sockets/: 'Rep', 'XRep'.+data XReq = XReq++-- | Special socket type to be used in request/reply middleboxes+-- such as zmq_queue(7).  Requests received using this socket are already+-- properly tagged with prefix identifying the original requester. When+-- sending a reply via XRep socket the message should be tagged with a+-- prefix from a corresponding request.+-- /Compatible peer sockets/: 'Req', 'XReq'.+data XRep = XRep++-- | A socket of type Pull is used by a pipeline node to receive+-- messages from upstream pipeline nodes. Messages are fair-queued from+-- among all connected upstream nodes. The zmq_send() function is not+-- implemented for this socket type.+data Pull = Pull++-- | A socket of type Push is used by a pipeline node to send messages+-- to downstream pipeline nodes. Messages are load-balanced to all connected+-- downstream nodes. The zmq_recv() function is not implemented for this+-- socket type.+--+-- When a Push socket enters an exceptional state due to having reached+-- the high water mark for all downstream nodes, or if there are no+-- downstream nodes at all, then any zmq_send(3) operations on the socket+-- shall block until the exceptional state ends or at least one downstream+-- node becomes available for sending; messages are not discarded.+data Push = Push++-- | Socket types.+class SocketType a where+    zmqSocketType :: a -> ZMQSocketType++-- | Sockets which can 'subscribe'.+class Subscriber a++-- | Sockets which can 'send'.+class Sender a++-- | Sockets which can 'receive'.+class Receiver 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 SocketType Sub where zmqSocketType = const sub+instance Subscriber Sub+instance Receiver   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 SocketType Rep where zmqSocketType = const response+instance Sender     Rep+instance Receiver   Rep++instance SocketType XReq where zmqSocketType = const xrequest+instance Sender     XReq+instance Receiver   XReq++instance SocketType XRep where zmqSocketType = const xresponse+instance Sender     XRep+instance Receiver   XRep++instance SocketType Pull where zmqSocketType = const pull+instance Receiver   Pull++instance SocketType Push where zmqSocketType = const push+instance Sender     Push++-- | Socket events.+data Event =+    In     -- ^ ZMQ_POLLIN (incoming messages)+  | Out    -- ^ ZMQ_POLLOUT (outgoing messages, i.e. at least 1 byte can be written)+  | InOut  -- ^ ZMQ_POLLIN | ZMQ_POLLOUT+  | Native -- ^ ZMQ_POLLERR+  | None+  deriving (Eq, Ord, Show)++-- | 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)++-- | Initialize a 0MQ context (cf. zmq_init for details).  You should+-- normally prefer to use 'withContext' instead.+init :: Size -> IO Context+init ioThreads = do+    c <- throwIfNull "init" $ c_zmq_init (fromIntegral ioThreads)+    return (Context c)++-- | Terminate a 0MQ context (cf. zmq_term).  You should normally+-- prefer to use 'withContext' instead.+term :: Context -> IO ()+term = throwIfMinus1Retry_ "term" . c_zmq_term . ctx++-- | Run an action with a 0MQ context.  The 'Context' supplied to your+-- action will /not/ be valid after the action either returns or+-- throws an exception.+withContext :: Size -> (Context -> IO a) -> IO a+withContext ioThreads act =+  bracket (throwIfNull "withContext (init)" $ c_zmq_init (fromIntegral ioThreads))+          (throwIfMinus1Retry_ "withContext (term)" . c_zmq_term)+          (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 (Context c) t = do+  let zt = typeVal . zmqSocketType $ t+  s <- throwIfNull "socket" (c_zmq_socket c zt)+  sock@(Socket _ status) <- mkSocket s+  addFinalizer sock $ do+    alive <- atomicModifyIORef status (\b -> (False, b))+    when alive $ c_zmq_close s >> return () -- socket has not been closed yet+  return sock++-- | Close a 0MQ socket. 'withSocket' provides automatic socket closing and may+-- be safer to use.+close :: Socket a -> IO ()+close sock@(Socket _ status) = onSocket "close" sock $ \s -> do+  alive <- atomicModifyIORef status (\b -> (False, b))+  when alive $ throwIfMinus1_ "close" . c_zmq_close $ s++-- | Subscribe Socket to given subscription.+subscribe :: Subscriber a => Socket a -> String -> IO ()+subscribe s = setStrOpt s B.subscribe++-- | Unsubscribe Socket from given subscription.+unsubscribe :: Subscriber a => Socket a -> String -> IO ()+unsubscribe s = setStrOpt s B.unsubscribe++-- Read Only++-- | Cf. @zmq_getsockopt ZMQ_EVENTS@+events :: Socket a -> IO Event+events s = toEvent <$> getIntOpt s B.events 0++-- | Cf. @zmq_getsockopt ZMQ_FD@+fileDescriptor :: Socket a -> IO Fd+fileDescriptor s = Fd . fromIntegral <$> getInt32Option B.filedesc s++-- | Cf. @zmq_getsockopt ZMQ_RCVMORE@+moreToReceive :: Socket a -> IO Bool+moreToReceive s = (== 1) <$> getInt32Option B.receiveMore s++-- Read++-- | Cf. @zmq_getsockopt ZMQ_IDENTITY@+identity :: Socket a -> IO String+identity s = getStrOpt s B.identity++-- | Cf. @zmq_getsockopt ZMQ_AFFINITY@+affinity :: Socket a -> IO Word64+affinity s = getIntOpt s B.affinity 0++-- | Cf. @zmq_getsockopt ZMQ_MAXMSGSIZE@+maxMessageSize :: Socket a -> IO Int64+maxMessageSize s = getIntOpt s B.maxMessageSize 0++-- | Cf. @zmq_getsockopt ZMQ_IPV4ONLY@+ipv4Only :: Socket a -> IO Bool+ipv4Only s = (== 1) <$> getInt32Option B.ipv4Only s++-- | Cf. @zmq_getsockopt ZMQ_BACKLOG@+backlog :: Socket a -> IO Int+backlog = getInt32Option B.backlog++-- | Cf. @zmq_getsockopt ZMQ_LINGER@+linger :: Socket a -> IO Int+linger = getInt32Option B.linger++-- | Cf. @zmq_getsockopt ZMQ_RATE@+rate :: Socket a -> IO Int+rate = getInt32Option B.rate++-- | Cf. @zmq_getsockopt ZMQ_RCVBUF@+receiveBuffer :: Socket a -> IO Int+receiveBuffer = getInt32Option B.receiveBuf++-- | Cf. @zmq_getsockopt ZMQ_RECONNECT_IVL@+reconnectInterval :: Socket a -> IO Int+reconnectInterval = getInt32Option B.reconnectIVL++-- | Cf. @zmq_getsockopt ZMQ_RECONNECT_IVL_MAX@+reconnectIntervalMax :: Socket a -> IO Int+reconnectIntervalMax = getInt32Option B.reconnectIVLMax++-- | Cf. @zmq_getsockopt ZMQ_RECOVERY_IVL@+recoveryInterval :: Socket a -> IO Int+recoveryInterval = getInt32Option B.recoveryIVL++-- | Cf. @zmq_getsockopt ZMQ_SNDBUF@+sendBuffer :: Socket a -> IO Int+sendBuffer = getInt32Option B.sendBuf++-- | Cf. @zmq_getsockopt ZMQ_MULTICAST_HOPS@+mcastHops :: Socket a -> IO Int+mcastHops = getInt32Option B.mcastHops++-- | Cf. @zmq_getsockopt ZMQ_RCVHWM@+receiveHighWM :: Socket a -> IO Int+receiveHighWM = getInt32Option B.receiveHighWM++-- | Cf. @zmq_getsockopt ZMQ_RCVTIMEO@+receiveTimeout :: Socket a -> IO Int+receiveTimeout = getInt32Option B.receiveTimeout++-- | Cf. @zmq_getsockopt ZMQ_SNDTIMEO@+sendTimeout :: Socket a -> IO Int+sendTimeout = getInt32Option B.sendTimeout++-- | Cf. @zmq_getsockopt ZMQ_SNDHWM@+sendHighWM :: Socket a -> IO Int+sendHighWM = getInt32Option B.sendHighWM++-- Write++-- | Cf. @zmq_setsockopt ZMQ_IDENTITY@+setIdentity :: Restricted N1 N254 String -> Socket a -> IO ()+setIdentity x s = setStrOpt s B.identity (rvalue x)++-- | Cf. @zmq_setsockopt ZMQ_AFFINITY@+setAffinity :: Word64 -> Socket a -> IO ()+setAffinity x s = setIntOpt s B.affinity x++-- | Cf. @zmq_setsockopt ZMQ_MAXMSGSIZE@+setMaxMessageSize :: Integral i => Restricted Nneg1 Int64 i -> Socket a -> IO ()+setMaxMessageSize x s = setIntOpt s B.maxMessageSize ((fromIntegral . rvalue $ x) :: Int64)++-- | Cf. @zmq_setsockopt ZMQ_IPV4ONLY@+setIpv4Only :: Bool -> Socket a -> IO ()+setIpv4Only True s  = setIntOpt s B.ipv4Only (1 :: CInt)+setIpv4Only False s = setIntOpt s B.ipv4Only (0 :: CInt)++-- | Cf. @zmq_setsockopt ZMQ_LINGER@+setLinger :: Integral i => Restricted Nneg1 Int32 i -> Socket a -> IO ()+setLinger = setInt32OptFromRestricted B.linger++-- | Cf. @zmq_setsockopt ZMQ_RCVTIMEO@+setReceiveTimeout :: Integral i => Restricted Nneg1 Int32 i -> Socket a -> IO ()+setReceiveTimeout = setInt32OptFromRestricted B.receiveTimeout++-- | Cf. @zmq_setsockopt ZMQ_SNDTIMEO@+setSendTimeout :: Integral i => Restricted Nneg1 Int32 i -> Socket a -> IO ()+setSendTimeout = setInt32OptFromRestricted B.sendTimeout++-- | Cf. @zmq_setsockopt ZMQ_RATE@+setRate :: Integral i => Restricted N1 Int32 i -> Socket a -> IO ()+setRate = setInt32OptFromRestricted B.rate++-- | Cf. @zmq_setsockopt ZMQ_MULTICAST_HOPS@+setMcastHops :: Integral i => Restricted N1 Int32 i -> Socket a -> IO ()+setMcastHops = setInt32OptFromRestricted B.mcastHops++-- | Cf. @zmq_setsockopt ZMQ_BACKLOG@+setBacklog :: Integral i => Restricted N0 Int32 i -> Socket a -> IO ()+setBacklog = setInt32OptFromRestricted B.backlog++-- | Cf. @zmq_setsockopt ZMQ_RCVBUF@+setReceiveBuffer :: Integral i => Restricted N0 Int32 i -> Socket a -> IO ()+setReceiveBuffer = setInt32OptFromRestricted B.receiveBuf++-- | Cf. @zmq_setsockopt ZMQ_RECONNECT_IVL@+setReconnectInterval :: Integral i => Restricted N0 Int32 i -> Socket a -> IO ()+setReconnectInterval = setInt32OptFromRestricted B.reconnectIVL++-- | Cf. @zmq_setsockopt ZMQ_RECONNECT_IVL_MAX@+setReconnectIntervalMax :: Integral i => Restricted N0 Int32 i -> Socket a -> IO ()+setReconnectIntervalMax = setInt32OptFromRestricted B.reconnectIVLMax++-- | Cf. @zmq_setsockopt ZMQ_SNDBUF@+setSendBuffer :: Integral i => Restricted N0 Int32 i -> Socket a -> IO ()+setSendBuffer = setInt32OptFromRestricted B.sendBuf++-- | Cf. @zmq_setsockopt ZMQ_RECOVERY_IVL@+setRecoveryInterval :: Integral i => Restricted N0 Int32 i -> Socket a -> IO ()+setRecoveryInterval = setInt32OptFromRestricted B.recoveryIVL++-- | Cf. @zmq_setsockopt ZMQ_RCVHWM@+setReceiveHighWM :: Integral i => Restricted N0 Int32 i -> Socket a -> IO ()+setReceiveHighWM = setInt32OptFromRestricted B.receiveHighWM++-- | Cf. @zmq_setsockopt ZMQ_SNDHWM@+setSendHighWM :: Integral i => Restricted N0 Int32 i -> Socket a -> IO ()+setSendHighWM = setInt32OptFromRestricted B.sendHighWM++-- | Bind the socket to the given address (cf. zmq_bind)+bind :: Socket a -> String -> IO ()+bind sock str = onSocket "bind" sock $+    throwIfMinus1_ "bind" . withCString str . c_zmq_bind++-- | Connect the socket to the given address (cf. zmq_connect).+connect :: Socket a -> String -> IO ()+connect sock str = onSocket "connect" sock $+    throwIfMinus1_ "connect" . withCString str . c_zmq_connect++-- | Send the given 'SB.ByteString' over the socket (cf. 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) (combine (DontWait : fls))++-- | Send the given 'LB.ByteString' over the socket (cf. 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) (combine (DontWait : fls))++-- | Receive a 'ByteString' from socket (cf. 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)++-- Convert bit-masked word into Event.+toEvent :: Word32 -> Event+toEvent e | e == (fromIntegral . pollVal $ pollIn)    = In+          | e == (fromIntegral . pollVal $ pollOut)   = Out+          | e == (fromIntegral . pollVal $ pollInOut) = InOut+          | e == (fromIntegral . pollVal $ pollerr)   = Native+          | otherwise                                 = None++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++waitRead, waitWrite :: Socket a -> IO ()+waitRead = wait' threadWaitRead pollIn+waitWrite = wait' threadWaitWrite pollOut+
+ src/System/ZMQ3/Base.hsc view
@@ -0,0 +1,198 @@+{-# LANGUAGE CPP, ForeignFunctionInterface #-}+-- |+-- Module      : System.ZMQ3.Base+-- Copyright   : (c) 2010-2012 Toralf Wittner+-- License     : MIT+-- Maintainer  : toralf.wittner@gmail.com+-- Stability   : experimental+-- Portability : non-portable+--++module System.ZMQ3.Base where++import Foreign+import Foreign.C.Types+import Foreign.C.String+import Control.Applicative++#include <zmq.h>++#if ZMQ_VERSION_MAJOR != 3+    ERROR__Invalid_0MQ_Version+#endif++newtype ZMQMsg = ZMQMsg { content :: Ptr () }++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++data ZMQPoll = ZMQPoll+    { pSocket  :: ZMQSocket+    , pFd      :: CInt+    , pEvents  :: CShort+    , pRevents :: 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__)++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+  , xrequest   = ZMQ_XREQ+  , xresponse  = ZMQ_XREP+  , pull       = ZMQ_PULL+  , push       = ZMQ_PUSH+}++newtype ZMQOption = ZMQOption { optVal :: CInt } deriving (Eq, Ord)++#{enum ZMQOption, ZMQOption+  , affinity        = ZMQ_AFFINITY+  , backlog         = ZMQ_BACKLOG+  , events          = ZMQ_EVENTS+  , filedesc        = ZMQ_FD+  , identity        = ZMQ_IDENTITY+  , ipv4Only        = ZMQ_IPV4ONLY+  , linger          = ZMQ_LINGER+  , maxMessageSize  = ZMQ_MAXMSGSIZE+  , mcastHops       = ZMQ_MULTICAST_HOPS+  , rate            = ZMQ_RATE+  , receiveBuf      = ZMQ_RCVBUF+  , receiveHighWM   = ZMQ_RCVHWM+  , receiveMore     = ZMQ_RCVMORE+  , receiveTimeout  = ZMQ_RCVTIMEO+  , reconnectIVL    = ZMQ_RECONNECT_IVL+  , reconnectIVLMax = ZMQ_RECONNECT_IVL_MAX+  , recoveryIVL     = ZMQ_RECOVERY_IVL+  , sendBuf         = ZMQ_SNDBUF+  , sendHighWM      = ZMQ_SNDHWM+  , sendTimeout     = ZMQ_SNDTIMEO+  , subscribe       = ZMQ_SUBSCRIBE+  , unsubscribe     = ZMQ_UNSUBSCRIBE+}++newtype ZMQMsgOption = ZMQMsgOption { msgOptVal :: CInt } deriving (Eq, Ord)++#{enum ZMQMsgOption, ZMQMsgOption+  , more = ZMQ_MORE+}++newtype ZMQFlag = ZMQFlag { flagVal :: CInt } deriving (Eq, Ord)++#{enum ZMQFlag, ZMQFlag+  , dontWait = ZMQ_DONTWAIT+  , sndMore  = ZMQ_SNDMORE+}++newtype ZMQPollEvent = ZMQPollEvent { pollVal :: CShort } deriving (Eq, Ord)++#{enum ZMQPollEvent, ZMQPollEvent,+    pollIn    = ZMQ_POLLIN,+    pollOut   = ZMQ_POLLOUT,+    pollerr   = ZMQ_POLLERR,+    pollInOut = ZMQ_POLLIN | ZMQ_POLLOUT+}++-- 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_init"+    c_zmq_init :: CInt -> IO ZMQCtx++foreign import ccall unsafe "zmq.h zmq_term"+    c_zmq_term :: ZMQCtx -> 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++-- 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_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_getmsgopt"+    c_zmq_getmsgopt :: ZMQMsgPtr+                    -> CInt      -- option+                    -> Ptr ()    -- option value+                    -> Ptr CSize -- option value size ptr+                    -> IO CInt+-- error messages++foreign import ccall unsafe "zmq.h zmq_strerror"+    c_zmq_strerror :: CInt -> IO CString+
+ src/System/ZMQ3/Error.hs view
@@ -0,0 +1,94 @@+{-# 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.ZMQ3.Error where++import Control.Exception+import Text.Printf+import Data.Typeable (Typeable)++import Foreign hiding (throwIf, throwIf_)+import Foreign.C.Error+import Foreign.C.String+import Foreign.C.Types (CInt)++import System.ZMQ3.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) <- getErrno+    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 getErrno >>= 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 :: Num a => String -> IO a -> IO a+throwIfMinus1 = throwIf (== -1)++throwIfMinus1_ :: Num a => String -> IO a -> IO ()+throwIfMinus1_ = throwIf_ (== -1)++throwIfNull :: String -> IO (Ptr a) -> IO (Ptr a)+throwIfNull = throwIf (== nullPtr)++throwIfMinus1Retry :: Num a => String -> IO a -> IO a+throwIfMinus1Retry = throwIfRetry (== -1)++throwIfMinus1Retry_ :: 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 getErrno >>= 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 :: Num a => String -> IO a -> IO b -> IO a+throwIfMinus1RetryMayBlock = throwIfRetryMayBlock (== -1)++throwIfMinus1RetryMayBlock_ :: Num a => String -> IO a -> IO b -> IO ()+throwIfMinus1RetryMayBlock_ = throwIfRetryMayBlock_ (== -1)++zmqErrnoMessage :: CInt -> IO String+zmqErrnoMessage e = c_zmq_strerror e >>= peekCString+
+ src/System/ZMQ3/Internal.hs view
@@ -0,0 +1,162 @@+module System.ZMQ3.Internal+    ( Context(..)+    , Socket(..)+    , Message(..)+    , Flag(..)+    , Timeout+    , Size++    , messageOf+    , messageOfLazy+    , messageClose+    , messageInit+    , messageInitSize+    , setIntOpt+    , setStrOpt+    , getIntOpt+    , getStrOpt+    , getIntMsgOpt+    , getInt32Option+    , setInt32OptFromRestricted++    , toZMQFlag+    , combine+    , mkSocket+    , onSocket++    ) where++import Control.Applicative+import Control.Monad (foldM_)+import Control.Exception+import Data.IORef (IORef)++import Foreign+import Foreign.C.String+import Foreign.C.Types (CInt, CSize)++import qualified Data.ByteString as SB+import qualified Data.ByteString.Lazy as LB+import qualified Data.ByteString.Unsafe as UB+import Data.IORef (newIORef)+import Data.Restricted++import System.ZMQ3.Base+import System.ZMQ3.Error++type Timeout = Int64+type Size    = Word++-- | Flags to apply on send operations (cf. man zmq_send)+data Flag = DontWait -- ^ ZMQ_DONTWAIT+          | SndMore  -- ^ ZMQ_SNDMORE+  deriving (Eq, Ord, Show)++-- | A 0MQ context representation.+newtype Context = Context { ctx :: ZMQCtx }++-- | A 0MQ Socket.+data Socket a = Socket {+      _socket   :: ZMQSocket+    , _sockLive :: IORef Bool+    }++-- A 0MQ Message representation.+newtype Message = Message { msgPtr :: ZMQMsgPtr }++-- internal helpers:++onSocket :: String -> Socket a -> (ZMQSocket -> IO b) -> IO b+onSocket _func (Socket sock _state) act = act sock+{-# INLINE onSocket #-}++mkSocket :: ZMQSocket -> IO (Socket a)+mkSocket s = Socket s <$> newIORef True++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)++setStrOpt :: Socket a -> ZMQOption -> String -> IO ()+setStrOpt sock (ZMQOption o) str = onSocket "setStrOpt" sock $ \s ->+  throwIfMinus1Retry_ "setStrOpt" $ withCStringLen str $ \(cstr, len) ->+        c_zmq_setsockopt s (fromIntegral o)+                           (castPtr cstr)+                           (fromIntegral len)++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++getStrOpt :: Socket a -> ZMQOption -> IO String+getStrOpt sock (ZMQOption o) = onSocket "getStrOpt" sock $ \s ->+    bracket (mallocBytes 255) free $ \bPtr ->+    bracket (new (255 :: CSize)) free $ \sPtr -> do+        throwIfMinus1Retry_ "getStrOpt" $+            c_zmq_getsockopt s (fromIntegral o) (castPtr bPtr) sPtr+        peek sPtr >>= \len -> peekCStringLen (bPtr, fromIntegral len)++getIntMsgOpt :: (Storable a, Integral a) => Message -> ZMQMsgOption -> a -> IO a+getIntMsgOpt (Message m) (ZMQMsgOption o) i = do+    bracket (new i) free $ \iptr ->+        bracket (new (fromIntegral . sizeOf $ i :: CSize)) free $ \jptr -> do+            throwIfMinus1Retry_ "getIntMsgOpt" $+                c_zmq_getmsgopt m (fromIntegral o) (castPtr iptr) jptr+            peek iptr++getInt32Option :: ZMQOption -> Socket a -> IO Int+getInt32Option o s = fromIntegral <$> getIntOpt s o (0 :: CInt)++setInt32OptFromRestricted :: Integral i => ZMQOption -> Restricted l u i -> Socket b -> IO ()+setInt32OptFromRestricted o x s = setIntOpt s o ((fromIntegral . rvalue $ x) :: CInt)++toZMQFlag :: Flag -> ZMQFlag+toZMQFlag DontWait = dontWait+toZMQFlag SndMore = sndMore++combine :: [Flag] -> CInt+combine = fromIntegral . foldr ((.|.) . flagVal . toZMQFlag) 0
+ tests/System/ZMQ3/Test/Properties.hs view
@@ -0,0 +1,193 @@+{-# LANGUAGE FlexibleInstances #-}+module System.ZMQ3.Test.Properties where++import Control.Applicative+import Test.Framework (Test, testGroup)+import Test.Framework.Providers.QuickCheck2+import Test.QuickCheck+import Test.QuickCheck.Monadic++import Data.Int+import Data.Word+import Data.Restricted+import Data.Maybe (fromJust)+import Data.ByteString (ByteString)+import Control.Concurrent+import Control.Concurrent.MVar+import System.ZMQ3+import System.Posix.Types (Fd(..))+import qualified Data.ByteString as SB+import qualified Data.ByteString.Char8 as CB++tests :: [Test]+tests = [+    testGroup "0MQ Socket Properties" [+        testProperty "get socket option (Pair)" (prop_get_socket_option Pair)+      , testProperty "get socket option (Pub)"  (prop_get_socket_option Pub)+      , testProperty "get socket option (Sub)"  (prop_get_socket_option Sub)+      , testProperty "get socket option (XPub)" (prop_get_socket_option XPub)+      , testProperty "get socket option (XSub)" (prop_get_socket_option XSub)+      , testProperty "get socket option (Req)"  (prop_get_socket_option Req)+      , testProperty "get socket option (Rep)"  (prop_get_socket_option Rep)+      , testProperty "get socket option (XReq)" (prop_get_socket_option XReq)+      , testProperty "get socket option (XRep)" (prop_get_socket_option XRep)+      , testProperty "get socket option (Pull)" (prop_get_socket_option Pull)+      , testProperty "get socket option (Push)" (prop_get_socket_option Push)+      , testProperty "set/get socket option (Pair)" (prop_set_get_socket_option Pair)+      , testProperty "set/get socket option (Pub)"  (prop_set_get_socket_option Pub)+      , testProperty "set/get socket option (Sub)"  (prop_set_get_socket_option Sub)+      , testProperty "set/get socket option (XPub)" (prop_set_get_socket_option XPub)+      , testProperty "set/get socket option (XSub)" (prop_set_get_socket_option XSub)+      , testProperty "set/get socket option (Req)"  (prop_set_get_socket_option Req)+      , testProperty "set/get socket option (Rep)"  (prop_set_get_socket_option Rep)+      , testProperty "set/get socket option (XReq)" (prop_set_get_socket_option XReq)+      , testProperty "set/get socket option (XRep)" (prop_set_get_socket_option XRep)+      , testProperty "set/get socket option (Pull)" (prop_set_get_socket_option Pull)+      , testProperty "set/get socket option (Push)" (prop_set_get_socket_option Push)+      , testProperty "(un-)subscribe"               (prop_subscribe Sub)+      ]+  , testGroup "0MQ Messages" [+        testProperty "msg send == msg received (Req/Rep)"   (prop_send_receive Req Rep)+      , testProperty "msg send == msg received (Push/Pull)" (prop_send_receive Push Pull)+      , testProperty "msg send == msg received (Pair/Pair)" (prop_send_receive Pair Pair)+      , testProperty "publish/subscribe (Pub/Sub)"          (prop_pub_sub Pub Sub)+      ]+  ]++prop_get_socket_option :: SocketType t => t -> GetOpt -> Property+prop_get_socket_option t opt = monadicIO $ run $ do+    withContext 1 $ \c ->+        withSocket c t $ \s ->+            case opt of+                Events _      -> events s         >> return ()+                Filedesc _    -> fileDescriptor s >> return ()+                ReceiveMore _ -> moreToReceive s  >> return ()++prop_set_get_socket_option :: SocketType t => t -> SetOpt -> Property+prop_set_get_socket_option t opt = monadicIO $ do+    r <- run $+        withContext 1 $ \c ->+            withSocket c t $ \s ->+                case opt of+                    Identity val        -> (== (rvalue val)) <$> (setIdentity val s >> identity s)+                    Ipv4Only val        -> (== val)          <$> (setIpv4Only val s >> ipv4Only s)+                    Affinity val        -> (eq val)          <$> (setAffinity val s >> affinity s)+                    Backlog val         -> (eq (rvalue val)) <$> (setBacklog val s >> backlog s)+                    Linger val          -> (eq (rvalue val)) <$> (setLinger val s >> linger s)+                    Rate val            -> (eq (rvalue val)) <$> (setRate val s >> rate s)+                    ReceiveBuf val      -> (eq (rvalue val)) <$> (setReceiveBuffer val s >> receiveBuffer s)+                    ReconnectIVL val    -> (eq (rvalue val)) <$> (setReconnectInterval val s >> reconnectInterval s)+                    ReconnectIVLMax val -> (eq (rvalue val)) <$> (setReconnectIntervalMax val s >> reconnectIntervalMax s)+                    RecoveryIVL val     -> (eq (rvalue val)) <$> (setRecoveryInterval val s >> recoveryInterval s)+                    SendBuf val         -> (eq (rvalue val)) <$> (setSendBuffer val s >> sendBuffer s)+                    MaxMessageSize val  -> (eq (rvalue val)) <$> (setMaxMessageSize val s >> maxMessageSize s)+                    McastHops val       -> (eq (rvalue val)) <$> (setMcastHops val s >> mcastHops s)+                    ReceiveHighWM val   -> (eq (rvalue val)) <$> (setReceiveHighWM val s >> receiveHighWM s)+                    ReceiveTimeout val  -> (eq (rvalue val)) <$> (setReceiveTimeout val s >> receiveTimeout s)+                    SendHighWM val      -> (eq (rvalue val)) <$> (setSendHighWM val s >> sendHighWM s)+                    SendTimeout val     -> (eq (rvalue val)) <$> (setSendTimeout val s >> sendTimeout s)+    assert r+  where+    eq :: (Integral i, Integral k) => i -> k -> Bool+    eq i k  = fromIntegral i == fromIntegral k++prop_subscribe :: (Subscriber a, SocketType a) => a -> String -> Property+prop_subscribe t subs = monadicIO $ run $+    withContext 1 $ \c ->+        withSocket c t $ \s -> do+            subscribe s subs+            unsubscribe s subs++prop_send_receive :: (SocketType a, SocketType b, Receiver b, Sender a) => a -> b -> ByteString -> Property+prop_send_receive a b msg = monadicIO $ do+    msg' <- run $ withContext 0 $ \c ->+                    withSocket c a $ \sender ->+                    withSocket c b $ \receiver -> do+                        sync <- newEmptyMVar :: IO (MVar ByteString)+                        bind receiver "inproc://endpoint"+                        forkIO $ receive receiver >>= putMVar sync+                        connect sender "inproc://endpoint"+                        send sender [] msg+                        takeMVar sync+    assert (msg == msg')++prop_pub_sub :: (SocketType a, Subscriber b, SocketType b, Sender a, Receiver b) => a -> b -> ByteString -> Property+prop_pub_sub a b msg = monadicIO $ do+    msg' <- run $ withContext 0 $ \c ->+                    withSocket c a $ \pub ->+                    withSocket c b $ \sub -> do+                        subscribe sub ""+                        bind sub "inproc://endpoint"+                        connect pub "inproc://endpoint"+                        send pub [] msg+                        receive sub+    assert (msg == msg')++instance Arbitrary ByteString where+    arbitrary = CB.pack <$> arbitrary++data GetOpt =+    Events          Int+  | Filedesc        Fd+  | ReceiveMore     Bool+  deriving Show++data SetOpt =+    Affinity        Word64+  | Backlog         (Restricted N0 Int32 Int)+  | Identity        (Restricted N1 N254 String)+  | 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 . show <$> 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/tests.hs view
@@ -0,0 +1,6 @@+import Test.Framework (defaultMain)+import qualified System.ZMQ3.Test.Properties as Properties++main :: IO ()+main = defaultMain Properties.tests+
+ zeromq3-haskell.cabal view
@@ -0,0 +1,69 @@+name:               zeromq3-haskell+version:            0.1+synopsis:           Bindings to ZeroMQ 3.x+category:           System, FFI+license:            MIT+license-file:       LICENSE+author:             Toralf Wittner+maintainer:         toralf.wittner@gmail.com+copyright:          Copyright (c) 2010 - 2012 zeromq-haskell authors+homepage:           http://github.com/twittner/zeromq-haskell/+stability:          experimental+tested-With:        GHC == 7.0.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/ZMQ3/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 >= 3.1.0++library+    hs-source-dirs:       src+    exposed-modules:      System.ZMQ3, Data.Restricted+    other-modules:        System.ZMQ3.Base+                        , System.ZMQ3.Internal+                        , System.ZMQ3.Error+    includes:             zmq.h+    ghc-options:          -Wall -O2+    extensions:           CPP+                        , ForeignFunctionInterface+    build-depends:        base >= 3 && < 5+                        , containers+                        , bytestring++    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+    build-depends:    zeromq3-haskell+                    , base >= 3 && < 5+                    , containers+                    , bytestring+                    , test-framework >= 0.4+                    , test-framework-quickcheck2 >= 0.2+                    , QuickCheck >= 2.4+    ghc-options:      -Wall -threaded -rtsopts++source-repository head+    type:             git+    location:         https://github.com/twittner/zeromq-haskell++