diff --git a/AUTHORS b/AUTHORS new file mode 100644 --- /dev/null +++ b/AUTHORS @@ -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 + diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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 + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,6 @@ +module Main where + +import Distribution.Simple + +main = defaultMain + diff --git a/examples/Makefile b/examples/Makefile new file mode 100644 --- /dev/null +++ b/examples/Makefile @@ -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 + diff --git a/examples/display.hs b/examples/display.hs new file mode 100644 --- /dev/null +++ b/examples/display.hs @@ -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
[
, ...]" + 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 + diff --git a/examples/perf/Makefile b/examples/perf/Makefile new file mode 100644 --- /dev/null +++ b/examples/perf/Makefile @@ -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 + diff --git a/examples/perf/local_lat.hs b/examples/perf/local_lat.hs new file mode 100644 --- /dev/null +++ b/examples/perf/local_lat.hs @@ -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 " + diff --git a/examples/perf/local_thr.hs b/examples/perf/local_thr.hs new file mode 100644 --- /dev/null +++ b/examples/perf/local_thr.hs @@ -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 " + diff --git a/examples/perf/remote_lat.hs b/examples/perf/remote_lat.hs new file mode 100644 --- /dev/null +++ b/examples/perf/remote_lat.hs @@ -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 " + diff --git a/examples/perf/remote_thr.hs b/examples/perf/remote_thr.hs new file mode 100644 --- /dev/null +++ b/examples/perf/remote_thr.hs @@ -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 " + diff --git a/examples/prompt.hs b/examples/prompt.hs new file mode 100644 --- /dev/null +++ b/examples/prompt.hs @@ -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
" + 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) + diff --git a/src/Data/Restricted.hs b/src/Data/Restricted.hs new file mode 100644 --- /dev/null +++ b/src/Data/Restricted.hs @@ -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 diff --git a/src/System/ZMQ3.hs b/src/System/ZMQ3.hs new file mode 100644 --- /dev/null +++ b/src/System/ZMQ3.hs @@ -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 +-- +-- 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 + diff --git a/src/System/ZMQ3/Base.hsc b/src/System/ZMQ3/Base.hsc new file mode 100644 --- /dev/null +++ b/src/System/ZMQ3/Base.hsc @@ -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 + +#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 + diff --git a/src/System/ZMQ3/Error.hs b/src/System/ZMQ3/Error.hs new file mode 100644 --- /dev/null +++ b/src/System/ZMQ3/Error.hs @@ -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 + diff --git a/src/System/ZMQ3/Internal.hs b/src/System/ZMQ3/Internal.hs new file mode 100644 --- /dev/null +++ b/src/System/ZMQ3/Internal.hs @@ -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 diff --git a/tests/System/ZMQ3/Test/Properties.hs b/tests/System/ZMQ3/Test/Properties.hs new file mode 100644 --- /dev/null +++ b/tests/System/ZMQ3/Test/Properties.hs @@ -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 diff --git a/tests/tests.hs b/tests/tests.hs new file mode 100644 --- /dev/null +++ b/tests/tests.hs @@ -0,0 +1,6 @@ +import Test.Framework (defaultMain) +import qualified System.ZMQ3.Test.Properties as Properties + +main :: IO () +main = defaultMain Properties.tests + diff --git a/zeromq3-haskell.cabal b/zeromq3-haskell.cabal new file mode 100644 --- /dev/null +++ b/zeromq3-haskell.cabal @@ -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 + +