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 +Alexander Vershilov added support for zmq_proxy 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,4 @@ +zeromq4-haskell +--------------- + +This library provides Haskell bindings to [0MQ 4.0](http://zeromq.org). 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 -f *.o *.hi + -rm -f 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,21 @@ +{-# LANGUAGE OverloadedStrings #-} +import Control.Monad +import System.Exit +import System.IO +import System.Environment +import System.ZMQ4.Monadic +import qualified Data.ByteString.Char8 as CS + +main :: IO () +main = do + args <- getArgs + when (length args < 1) $ do + hPutStrLn stderr "usage: display
[
, ...]" + exitFailure + runZMQ $ do + sub <- socket Sub + subscribe sub "" + mapM_ (connect sub) args + forever $ do + receive sub >>= liftIO . CS.putStrLn + liftIO $ 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 -f *.o *.hi + -rm -f 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 System.ZMQ4.Monadic +import qualified Data.ByteString as SB + +main :: IO () +main = do + args <- getArgs + when (length args /= 3) $ do + hPutStrLn stderr usage + exitFailure + let bindTo = args !! 0 + size = read $ args !! 1 + rounds = read $ args !! 2 + runZMQ $ do + s <- socket Rep + bind s bindTo + loop s rounds size + where + loop s r sz = unless (r <= 0) $ do + msg <- receive s + when (SB.length msg /= sz) $ + error "message of incorrect size received" + send s [] msg + loop s (r - 1) sz + +usage :: String +usage = "usage: local_lat " + 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,52 @@ +{-# LANGUAGE OverloadedStrings #-} +import Control.Concurrent +import Control.Monad +import System.IO +import System.Exit +import System.Environment +import Data.Time.Clock +import System.ZMQ4.Monadic +import qualified Data.ByteString as SB +import Text.Printf + +main :: IO () +main = do + args <- getArgs + when (length args /= 3) $ do + hPutStrLn stderr usage + exitFailure + let bindTo = args !! 0 + size = read $ args !! 1 + count = read $ args !! 2 + runZMQ $ do + s <- socket Sub + subscribe s "" + bind s bindTo + receive' s size + start <- liftIO $ getCurrentTime + loop s count size + end <- liftIO $ getCurrentTime + liftIO $ printStat start end size count + where + receive' s sz = do + msg <- receive s + when (SB.length msg /= sz) $ + error "message of incorrect size received" + + loop s c sz = unless (c < 0) $ do + receive' s sz + loop s (c - 1) sz + + printStat :: UTCTime -> UTCTime -> Int -> Int -> IO () + printStat start end size count = do + let elapsed = fromRational . toRational $ diffUTCTime end start :: Double + through = fromIntegral count / elapsed + mbits = (through * fromIntegral size * 8) / 1000000 + printf "message size: %d [B]\n" size + printf "message count: %d\n" count + printf "mean throughput: %.3f [msg/s]\n" through + printf "mean throughput: %.3f [Mb/s]\n" mbits + +usage :: String +usage = "usage: local_thr " + 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 System.ZMQ4.Monadic +import qualified Data.ByteString as SB + +main :: IO () +main = do + args <- getArgs + when (length args /= 3) $ do + hPutStrLn stderr usage + exitFailure + let connTo = args !! 0 + size = read $ args !! 1 + rounds = read $ args !! 2 + message = SB.replicate size 0x65 + runZMQ $ do + s <- socket Req + connect s connTo + start <- liftIO $ getCurrentTime + loop s rounds message + end <- liftIO $ getCurrentTime + liftIO $ print (diffUTCTime end start) + where + loop s r msg = unless (r <= 0) $ do + send s [] msg + msg' <- receive s + when (SB.length msg' /= SB.length msg) $ + error "message of incorrect size received" + loop s (r - 1) msg + +usage :: String +usage = "usage: remote_lat " + 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 System.ZMQ4.Monadic +import qualified Data.ByteString as SB + +main :: IO () +main = do + args <- getArgs + when (length args /= 3) $ do + hPutStrLn stderr usage + exitFailure + let connTo = args !! 0 + size = read $ args !! 1 + count = read $ args !! 2 + message = SB.replicate size 0x65 + runZMQ $ do + s <- socket Pub + connect s connTo + replicateM_ count $ send s [] message + liftIO $ threadDelay 10000000 + +usage :: String +usage = "usage: remote_thr " + 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 Data.Monoid +import Data.String +import System.IO +import System.Exit +import System.Environment +import System.ZMQ4.Monadic +import qualified Data.ByteString.UTF8 as SB + +main :: IO () +main = do + args <- getArgs + when (length args /= 2) $ do + hPutStrLn stderr "usage: prompt
" + exitFailure + let addr = head args + name = fromString (args !! 1) <> ": " + runZMQ $ do + pub <- socket Pub + bind pub addr + forever $ do + line <- liftIO $ SB.fromString <$> getLine + send pub [] (name <> line) 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,211 @@ +{-# LANGUAGE EmptyDataDecls #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE UndecidableInstances #-} +{-# LANGUAGE TypeSynonymInstances #-} +{-# LANGUAGE OverloadedStrings #-} + +-- | +-- Module : Data.Restricted +-- Copyright : (c) 2011-2013 Toralf Wittner +-- License : MIT +-- Maintainer : Toralf Wittner +-- Stability : experimental +-- Portability : non-portable +-- +-- Type-level restricted data. +-- This module allows for type declarations which embed certain restrictions, +-- such as value bounds. E.g. @Restricted N0 N1 Int@ denotes an 'Int' which can +-- only have values [0 .. 1]. When creating such a value, the constructor functions +-- 'restrict' or 'toRestricted' ensure that the restrictions are obeyed. Code +-- that consumes restricted types does not need to check the constraints. +-- +-- /N.B./ This module is more or less tailored to be used within 'System.ZMQ3'. +-- Therefore the provided type level restrictions are limited. +module Data.Restricted ( + + Restricted + , Restriction (..) + , rvalue + + , Nneg1 + , N1 + , N0 + , N254 + , Inf + , Div4 + , Div5 + +) where + +import Data.Int +import Data.ByteString (ByteString) +import qualified Data.ByteString as B + +-- | Type level restriction. +newtype Restricted r v = Restricted v deriving Show + +-- | A uniform way to restrict values. +class Restriction r v where + + -- | Create a restricted value. Returns 'Nothing' if + -- the given value does not satisfy all restrictions. + toRestricted :: v -> Maybe (Restricted r v) + + -- | Create a restricted value. If the given value + -- does not satisfy the restrictions, a modified + -- variant is used instead, e.g. if an integer is + -- larger than the upper bound, the upper bound + -- value is used. + restrict :: v -> Restricted r v + +-- | Get the actual value. +rvalue :: Restricted r v -> v +rvalue (Restricted v) = v + +-- | type level -1 +data Nneg1 + +-- | type-level 0 +data N0 + +-- | type-level 1 +data N1 + +-- | type-level 254 +data N254 + +-- | type-level infinity +data Inf + +-- | divisable by 4 +data Div4 + +-- | divisable by 5 +data Div5 + +instance Show Nneg1 where show _ = "Nneg1" +instance Show N0 where show _ = "N0" +instance Show N1 where show _ = "N1" +instance Show N254 where show _ = "N254" +instance Show Inf where show _ = "Inf" +instance Show Div4 where show _ = "Div4" +instance Show Div5 where show _ = "Div5" + +-- Natural numbers + +instance (Integral a) => Restriction (N0, Inf) a where + toRestricted = toIntRLB 0 + restrict = intRLB 0 + +instance (Integral a) => Restriction (N0, Int32) a where + toRestricted = toIntR 0 (maxBound :: Int32) + restrict = intR 0 (maxBound :: Int32) + +instance (Integral a) => Restriction (N0, Int64) a where + toRestricted = toIntR 0 (maxBound :: Int64) + restrict = intR 0 (maxBound :: Int64) + +-- Positive natural numbers + +instance (Integral a) => Restriction (N1, Inf) a where + toRestricted = toIntRLB 1 + restrict = intRLB 1 + +instance (Integral a) => Restriction (N1, Int32) a where + toRestricted = toIntR 1 (maxBound :: Int32) + restrict = intR 1 (maxBound :: Int32) + +instance (Integral a) => Restriction (N1, Int64) a where + toRestricted = toIntR 1 (maxBound :: Int64) + restrict = intR 1 (maxBound :: Int64) + +-- From -1 ranges + +instance (Integral a) => Restriction (Nneg1, Inf) a where + toRestricted = toIntRLB (-1) + restrict = intRLB (-1) + +instance (Integral a) => Restriction (Nneg1, Int32) a where + toRestricted = toIntR (-1) (maxBound :: Int32) + restrict = intR (-1) (maxBound :: Int32) + +instance (Integral a) => Restriction (Nneg1, Int64) a where + toRestricted = toIntR (-1) (maxBound :: Int64) + restrict = intR (-1) (maxBound :: Int64) + +-- Other ranges + +instance Restriction (N1, N254) String where + toRestricted s | check (1, 254) (length s) = Just $ Restricted s + | otherwise = Nothing + + restrict s | length s < 1 = Restricted " " + | otherwise = Restricted (take 254 s) + +instance Restriction (N1, N254) ByteString where + toRestricted s | check (1, 254) (B.length s) = Just $ Restricted s + | otherwise = Nothing + + restrict s | B.length s < 1 = Restricted (B.singleton 0x20) + | otherwise = Restricted (B.take 254 s) + +-- Other constraints + +instance Restriction Div4 ByteString where + toRestricted s | B.length s `mod` 4 == 0 = Just $ Restricted s + | otherwise = Nothing + + restrict = fitByRem 4 + +instance Restriction Div5 ByteString where + toRestricted s | B.length s `mod` 5 == 0 = Just $ Restricted s + | otherwise = Nothing + + restrict = fitByRem 5 + +-- Helpers + +toIntR :: (Integral i, Integral j) => i -> j -> i -> Maybe (Restricted (a, b) i) +toIntR lb ub i | check (lb, fromIntegral ub) i = Just $ Restricted i + | otherwise = Nothing + +intR :: (Integral i, Integral j) => i -> j -> i -> Restricted (a, b) i +intR lb ub = Restricted . lbfit lb . ubfit (fromIntegral ub) + +toIntRLB :: Integral i => i -> i -> Maybe (Restricted (a, b) i) +toIntRLB lb i | lbcheck lb i = Just $ Restricted i + | otherwise = Nothing + +intRLB :: Integral i => i -> i -> Restricted (a, b) i +intRLB lb = Restricted . lbfit lb + +-- Bounds checks + +lbcheck :: Ord a => a -> a -> Bool +lbcheck lb a = a >= lb + +ubcheck :: Ord a => a -> a -> Bool +ubcheck ub a = a <= ub + +check :: Ord a => (a, a) -> a -> Bool +check (lb, ub) a = lbcheck lb a && ubcheck ub a + +-- Fit + +lbfit :: Integral a => a -> a -> a +lbfit lb a | a >= lb = a + | otherwise = lb + +ubfit :: Integral a => a -> a -> a +ubfit ub a | a <= ub = a + | otherwise = ub + +fitByRem :: Int -> ByteString -> Restricted r ByteString +fitByRem r s = + let len = B.length s + x = len `mod` r + in if x == 0 + then Restricted s + else Restricted (B.take (len - x) s) + diff --git a/src/System/ZMQ4.hs b/src/System/ZMQ4.hs new file mode 100644 --- /dev/null +++ b/src/System/ZMQ4.hs @@ -0,0 +1,972 @@ +{-# LANGUAGE GADTs #-} + +-- | +-- Module : System.ZMQ4 +-- Copyright : (c) 2010-2013 Toralf Wittner +-- License : MIT +-- Maintainer : Toralf Wittner +-- Stability : experimental +-- Portability : non-portable +-- +-- 0MQ haskell binding. The API closely follows the C-API of 0MQ with +-- the main difference being that sockets are typed. +-- +-- /Notes/ +-- +-- Many option settings use a 'Restriction' to further constrain the +-- range of possible values of their integral types. For example +-- the maximum message size can be given as -1, which means no limit +-- or by greater values, which denote the message size in bytes. The +-- type of 'setMaxMessageSize' is therefore: +-- +-- @setMaxMessageSize :: Integral i +-- => Restricted (Nneg1, Int64) i +-- -> Socket a +-- -> IO ()@ +-- +-- which means any integral value in the range of @-1@ to +-- (@maxBound :: Int64@) can be given. To create a restricted +-- value from plain value, use 'toRestricted' or 'restrict'. + +module System.ZMQ4 + ( -- * Type Definitions + -- ** Socket Types + Pair (..) + , Pub (..) + , Sub (..) + , XPub (..) + , XSub (..) + , Req (..) + , Rep (..) + , Dealer (..) + , Router (..) + , XReq + , XRep + , Pull (..) + , Push (..) + , Stream (..) + + -- ** Socket type-classes + , SocketType + , Sender + , Receiver + , Subscriber + , SocketLike + , Conflatable + , SendProbe + + -- ** Various type definitions + , Size + , Context + , Socket + , Flag (SendMore) + , Switch (..) + , Timeout + , Event (..) + , EventType (..) + , EventMsg (..) + , Poll (..) + , KeyFormat (..) + , SecurityMechanism (..) + + -- * General Operations + , withContext + , withSocket + , bind + , unbind + , connect + , send + , send' + , sendMulti + , receive + , receiveMulti + , version + , monitor + , poll + + , System.ZMQ4.subscribe + , System.ZMQ4.unsubscribe + + -- * Context Options (Read) + , ioThreads + , maxSockets + + -- * Context Options (Write) + , setIoThreads + , setMaxSockets + + -- * Socket Options (Read) + , System.ZMQ4.affinity + , System.ZMQ4.backlog + , System.ZMQ4.conflate + , System.ZMQ4.curvePublicKey + , System.ZMQ4.curveSecretKey + , System.ZMQ4.curveServerKey + , System.ZMQ4.delayAttachOnConnect + , System.ZMQ4.events + , System.ZMQ4.fileDescriptor + , System.ZMQ4.identity + , System.ZMQ4.immediate + , System.ZMQ4.ipv4Only + , System.ZMQ4.ipv6 + , System.ZMQ4.lastEndpoint + , System.ZMQ4.linger + , System.ZMQ4.maxMessageSize + , System.ZMQ4.mcastHops + , System.ZMQ4.mechanism + , System.ZMQ4.moreToReceive + , System.ZMQ4.plainServer + , System.ZMQ4.plainPassword + , System.ZMQ4.plainUserName + , System.ZMQ4.rate + , System.ZMQ4.receiveBuffer + , System.ZMQ4.receiveHighWM + , System.ZMQ4.receiveTimeout + , System.ZMQ4.reconnectInterval + , System.ZMQ4.reconnectIntervalMax + , System.ZMQ4.recoveryInterval + , System.ZMQ4.sendBuffer + , System.ZMQ4.sendHighWM + , System.ZMQ4.sendTimeout + , System.ZMQ4.tcpKeepAlive + , System.ZMQ4.tcpKeepAliveCount + , System.ZMQ4.tcpKeepAliveIdle + , System.ZMQ4.tcpKeepAliveInterval + , System.ZMQ4.zapDomain + + -- * Socket Options (Write) + , setAffinity + , setBacklog + , setConflate + , setCurveServer + , setCurvePublicKey + , setCurveSecretKey + , setCurveServerKey + , setDelayAttachOnConnect + , setIdentity + , setImmediate + , setIpv4Only + , setIpv6 + , setLinger + , setMaxMessageSize + , setMcastHops + , setPlainServer + , setPlainPassword + , setPlainUserName + , setProbeRouter + , setRate + , setReceiveBuffer + , setReceiveHighWM + , setReceiveTimeout + , setReconnectInterval + , setReconnectIntervalMax + , setRecoveryInterval + , setReqCorrelate + , setReqRelaxed + , setRouterMandatory + , setSendBuffer + , setSendHighWM + , setSendTimeout + , setTcpAcceptFilter + , setTcpKeepAlive + , setTcpKeepAliveCount + , setTcpKeepAliveIdle + , setTcpKeepAliveInterval + , setXPubVerbose + + -- * Restrictions + , Data.Restricted.restrict + , Data.Restricted.toRestricted + + -- * Error Handling + , ZMQError + , errno + , source + , message + + -- * Low-level Functions + , init + , term + , shutdown + , context + , socket + , close + , waitRead + , waitWrite + , z85Encode + , z85Decode + + -- * Utils + , proxy + , curveKeyPair + ) where + +import Prelude hiding (init) +import Control.Applicative +import Control.Exception +import Control.Monad (unless) +import Control.Monad.IO.Class +import Data.List (intersect, foldl') +import Data.List.NonEmpty (NonEmpty) +import Data.Restricted +import Data.Traversable (forM) +import Foreign hiding (throwIf, throwIf_, throwIfNull, void) +import Foreign.C.String +import Foreign.C.Types (CInt, CShort) +import System.Posix.Types (Fd(..)) +import System.ZMQ4.Base +import System.ZMQ4.Internal +import System.ZMQ4.Error + +import qualified Data.ByteString as SB +import qualified Data.ByteString.Lazy as LB +import qualified Data.List.NonEmpty as S +import qualified Prelude as P +import qualified System.ZMQ4.Base as B + +import GHC.Conc (threadWaitRead, threadWaitWrite) + +----------------------------------------------------------------------------- +-- Socket Types + +-- | +data Pair = Pair + +-- | +data Pub = Pub + +-- | +data Sub = Sub + +-- | +data XPub = XPub + +-- | +data XSub = XSub + +-- | +data Req = Req + +-- | +data Rep = Rep + +-- | +data Dealer = Dealer + +-- | +data Router = Router + +-- | +data Pull = Pull + +-- | +data Push = Push + +-- | +data Stream = Stream + +type XReq = Dealer +{-# DEPRECATED XReq "Use Dealer" #-} + +type XRep = Router +{-# DEPRECATED XRep "Use Router" #-} + +----------------------------------------------------------------------------- +-- Socket Type Classifications + +-- | Sockets which can 'subscribe'. +class Subscriber a + +-- | Sockets which can 'send'. +class Sender a + +-- | Sockets which can 'receive'. +class Receiver a + +-- | Sockets which can be 'conflate'd. +class Conflatable a + +-- | Sockets which can send probes (cf. 'setProbeRouter'). +class SendProbe a + +instance SocketType Pair where zmqSocketType = const pair +instance Sender Pair +instance Receiver Pair + +instance SocketType Pub where zmqSocketType = const pub +instance Sender Pub +instance Conflatable Pub + +instance SocketType Sub where zmqSocketType = const sub +instance Subscriber Sub +instance Receiver Sub +instance Conflatable Sub + +instance SocketType XPub where zmqSocketType = const xpub +instance Sender XPub +instance Receiver XPub + +instance SocketType XSub where zmqSocketType = const xsub +instance Sender XSub +instance Receiver XSub + +instance SocketType Req where zmqSocketType = const request +instance Sender Req +instance Receiver Req +instance SendProbe Req + +instance SocketType Rep where zmqSocketType = const response +instance Sender Rep +instance Receiver Rep + +instance SocketType Dealer where zmqSocketType = const dealer +instance Sender Dealer +instance Receiver Dealer +instance Conflatable Dealer +instance SendProbe Dealer + +instance SocketType Router where zmqSocketType = const router +instance Sender Router +instance Receiver Router +instance SendProbe Router + +instance SocketType Pull where zmqSocketType = const pull +instance Receiver Pull +instance Conflatable Pull + +instance SocketType Push where zmqSocketType = const push +instance Sender Push +instance Conflatable Push + +instance SocketType Stream where zmqSocketType = const stream +instance Sender Stream +instance Receiver Stream + +----------------------------------------------------------------------------- + +-- | Socket events. +data Event = + In -- ^ @ZMQ_POLLIN@ (incoming messages) + | Out -- ^ @ZMQ_POLLOUT@ (outgoing messages, i.e. at least 1 byte can be written) + | Err -- ^ @ZMQ_POLLERR@ + deriving (Eq, Ord, Read, Show) + +-- | A 'Poll' value contains the object to poll (a 0MQ socket or a file +-- descriptor), the set of 'Event's which are of interest and--optionally-- +-- a callback-function which is invoked iff the set of interested events +-- overlaps with the actual events. +data Poll s m where + Sock :: s t -> [Event] -> Maybe ([Event] -> m ()) -> Poll s m + File :: Fd -> [Event] -> Maybe ([Event] -> m ()) -> Poll s m + +-- | Return the runtime version of the underlying 0MQ library as a +-- (major, minor, patch) triple. +version :: IO (Int, Int, Int) +version = + with 0 $ \major_ptr -> + with 0 $ \minor_ptr -> + with 0 $ \patch_ptr -> + c_zmq_version major_ptr minor_ptr patch_ptr >> + tupleUp <$> peek major_ptr <*> peek minor_ptr <*> peek patch_ptr + where + tupleUp a b c = (fromIntegral a, fromIntegral b, fromIntegral c) + +init :: Size -> IO Context +init n = do + c <- context + setIoThreads n c + return c +{-# DEPRECATED init "Use context" #-} + +-- | Initialize a 0MQ context. +-- Equivalent to . +context :: IO Context +context = Context <$> throwIfNull "init" c_zmq_ctx_new + +-- | Terminate a 0MQ context. +-- Equivalent to . +term :: Context -> IO () +term c = throwIfMinus1Retry_ "term" . c_zmq_ctx_term . _ctx $ c + +-- | Shutdown a 0MQ context. +-- Equivalent to . +shutdown :: Context -> IO () +shutdown = throwIfMinus1_ "shutdown" . c_zmq_ctx_shutdown . _ctx + +-- | Run an action with a 0MQ context. The 'Context' supplied to your +-- action will /not/ be valid after the action either returns or +-- throws an exception. +withContext :: (Context -> IO a) -> IO a +withContext act = + bracket (throwIfNull "withContext (new)" $ c_zmq_ctx_new) + (throwIfMinus1Retry_ "withContext (term)" . c_zmq_ctx_term) + (act . Context) + +-- | Run an action with a 0MQ socket. The socket will be closed after running +-- the supplied action even if an error occurs. The socket supplied to your +-- action will /not/ be valid after the action terminates. +withSocket :: SocketType a => Context -> a -> (Socket a -> IO b) -> IO b +withSocket c t = bracket (socket c t) close + +-- | Create a new 0MQ socket within the given context. 'withSocket' provides +-- automatic socket closing and may be safer to use. +socket :: SocketType a => Context -> a -> IO (Socket a) +socket c t = Socket <$> mkSocketRepr t c + +-- | Close a 0MQ socket. 'withSocket' provides automatic socket closing and may +-- be safer to use. +close :: Socket a -> IO () +close = closeSock . _socketRepr + +-- | Subscribe Socket to given subscription. +subscribe :: Subscriber a => Socket a -> SB.ByteString -> IO () +subscribe s = setByteStringOpt s B.subscribe + +-- | Unsubscribe Socket from given subscription. +unsubscribe :: Subscriber a => Socket a -> SB.ByteString -> IO () +unsubscribe s = setByteStringOpt s B.unsubscribe + +-- Read Only + +-- | . +events :: Socket a -> IO [Event] +events s = toEvents <$> getIntOpt s B.events 0 + +-- | . +fileDescriptor :: Socket a -> IO Fd +fileDescriptor s = Fd . fromIntegral <$> getInt32Option B.filedesc s + +-- | . +moreToReceive :: Socket a -> IO Bool +moreToReceive s = (== 1) <$> getInt32Option B.receiveMore s + +-- Read + +-- | . +ioThreads :: Context -> IO Word +ioThreads = ctxIntOption "ioThreads" _ioThreads + +-- | . +maxSockets :: Context -> IO Word +maxSockets = ctxIntOption "maxSockets" _maxSockets + +-- | Restricts the outgoing and incoming socket buffers to a single message. +conflate :: Conflatable a => Socket a -> IO Bool +conflate s = (== 1) <$> getInt32Option B.conflate s + +-- | . +immediate :: Socket a -> IO Bool +immediate s = (== 1) <$> getInt32Option B.immediate s + +-- | . +identity :: Socket a -> IO SB.ByteString +identity s = getByteStringOpt s B.identity + +-- | . +affinity :: Socket a -> IO Word64 +affinity s = getIntOpt s B.affinity 0 + +-- | . +maxMessageSize :: Socket a -> IO Int64 +maxMessageSize s = getIntOpt s B.maxMessageSize 0 + +ipv4Only :: Socket a -> IO Bool +ipv4Only s = (== 1) <$> getInt32Option B.ipv4Only s +{-# DEPRECATED ipv4Only "Use ipv6" #-} + +-- | . +ipv6 :: Socket a -> IO Bool +ipv6 s = (== 1) <$> getInt32Option B.ipv6 s + +-- | . +backlog :: Socket a -> IO Int +backlog = getInt32Option B.backlog + +delayAttachOnConnect :: Socket a -> IO Bool +delayAttachOnConnect s = (== 1) <$> getInt32Option B.delayAttachOnConnect s +{-# DEPRECATED delayAttachOnConnect "Use immediate" #-} + +-- | . +linger :: Socket a -> IO Int +linger = getInt32Option B.linger + +-- | . +lastEndpoint :: Socket a -> IO String +lastEndpoint s = getStrOpt s B.lastEndpoint + +-- | . +rate :: Socket a -> IO Int +rate = getInt32Option B.rate + +-- | . +receiveBuffer :: Socket a -> IO Int +receiveBuffer = getInt32Option B.receiveBuf + +-- | . +reconnectInterval :: Socket a -> IO Int +reconnectInterval = getInt32Option B.reconnectIVL + +-- | . +reconnectIntervalMax :: Socket a -> IO Int +reconnectIntervalMax = getInt32Option B.reconnectIVLMax + +-- | . +recoveryInterval :: Socket a -> IO Int +recoveryInterval = getInt32Option B.recoveryIVL + +-- | . +sendBuffer :: Socket a -> IO Int +sendBuffer = getInt32Option B.sendBuf + +-- | . +mcastHops :: Socket a -> IO Int +mcastHops = getInt32Option B.mcastHops + +-- | . +receiveHighWM :: Socket a -> IO Int +receiveHighWM = getInt32Option B.receiveHighWM + +-- | . +receiveTimeout :: Socket a -> IO Int +receiveTimeout = getInt32Option B.receiveTimeout + +-- | . +sendTimeout :: Socket a -> IO Int +sendTimeout = getInt32Option B.sendTimeout + +-- | . +sendHighWM :: Socket a -> IO Int +sendHighWM = getInt32Option B.sendHighWM + +-- | . +tcpKeepAlive :: Socket a -> IO Switch +tcpKeepAlive = fmap (toSwitch "Invalid ZMQ_TCP_KEEPALIVE") + . getInt32Option B.tcpKeepAlive + +-- | . +tcpKeepAliveCount :: Socket a -> IO Int +tcpKeepAliveCount = getInt32Option B.tcpKeepAliveCount + +-- | . +tcpKeepAliveIdle :: Socket a -> IO Int +tcpKeepAliveIdle = getInt32Option B.tcpKeepAliveIdle + +-- | . +tcpKeepAliveInterval :: Socket a -> IO Int +tcpKeepAliveInterval = getInt32Option B.tcpKeepAliveInterval + +-- | . +mechanism :: Socket a -> IO SecurityMechanism +mechanism = fmap (fromMechanism "Invalid ZMQ_MECHANISM") + . getInt32Option B.mechanism + +-- | . +plainServer :: Socket a -> IO Bool +plainServer = fmap (== 1) . getInt32Option B.plainServer + +-- | . +plainUserName :: Socket a -> IO SB.ByteString +plainUserName s = getByteStringOpt s B.plainUserName + +-- | . +plainPassword :: Socket a -> IO SB.ByteString +plainPassword s = getByteStringOpt s B.plainPassword + +-- | . +zapDomain :: Socket a -> IO SB.ByteString +zapDomain s = getByteStringOpt s B.zapDomain + +-- | . +curvePublicKey :: KeyFormat f -> Socket a -> IO SB.ByteString +curvePublicKey f s = getKey f s B.curvePublicKey + +-- | . +curveServerKey :: KeyFormat f -> Socket a -> IO SB.ByteString +curveServerKey f s = getKey f s B.curveServerKey + +-- | . +curveSecretKey :: KeyFormat f -> Socket a -> IO SB.ByteString +curveSecretKey f s = getKey f s B.curveSecretKey + +-- Write + +-- | . +setIoThreads :: Word -> Context -> IO () +setIoThreads n = setCtxIntOption "ioThreads" _ioThreads n + +-- | . +setMaxSockets :: Word -> Context -> IO () +setMaxSockets n = setCtxIntOption "maxSockets" _maxSockets n + +-- | Restrict the outgoing and incoming socket buffers to a single message. +setConflate :: Conflatable a => Bool -> Socket a -> IO () +setConflate x s = setIntOpt s B.conflate (bool2cint x) + +-- | . +setImmediate :: Bool -> Socket a -> IO () +setImmediate x s = setIntOpt s B.immediate (bool2cint x) + +-- | . +setIdentity :: Restricted (N1, N254) SB.ByteString -> Socket a -> IO () +setIdentity x s = setByteStringOpt s B.identity (rvalue x) + +-- | . +setAffinity :: Word64 -> Socket a -> IO () +setAffinity x s = setIntOpt s B.affinity x + +setDelayAttachOnConnect :: Bool -> Socket a -> IO () +setDelayAttachOnConnect x s = setIntOpt s B.delayAttachOnConnect (bool2cint x) +{-# DEPRECATED setDelayAttachOnConnect "Use setImmediate" #-} + +-- | . +setMaxMessageSize :: Integral i => Restricted (Nneg1, Int64) i -> Socket a -> IO () +setMaxMessageSize x s = setIntOpt s B.maxMessageSize ((fromIntegral . rvalue $ x) :: Int64) + +setIpv4Only :: Bool -> Socket a -> IO () +setIpv4Only x s = setIntOpt s B.ipv4Only (bool2cint x) +{-# DEPRECATED setIpv4Only "Use setIpv6" #-} + +-- | . +setIpv6 :: Bool -> Socket a -> IO () +setIpv6 x s = setIntOpt s B.ipv6 (bool2cint x) + +-- | . +setPlainServer :: Bool -> Socket a -> IO () +setPlainServer x s = setIntOpt s B.plainServer (bool2cint x) + +-- | . +setCurveServer :: Bool -> Socket a -> IO () +setCurveServer x s = setIntOpt s B.curveServer (bool2cint x) + +-- | . +setPlainUserName :: Restricted (N1, N254) SB.ByteString -> Socket a -> IO () +setPlainUserName x s = setByteStringOpt s B.plainUserName (rvalue x) + +-- | . +setPlainPassword :: Restricted (N1, N254) SB.ByteString -> Socket a -> IO () +setPlainPassword x s = setByteStringOpt s B.plainPassword (rvalue x) + +-- | . +setLinger :: Integral i => Restricted (Nneg1, Int32) i -> Socket a -> IO () +setLinger = setInt32OptFromRestricted B.linger + +-- | . +setReceiveTimeout :: Integral i => Restricted (Nneg1, Int32) i -> Socket a -> IO () +setReceiveTimeout = setInt32OptFromRestricted B.receiveTimeout + +-- | . +setRouterMandatory :: Bool -> Socket Router -> IO () +setRouterMandatory x s = setIntOpt s B.routerMandatory (bool2cint x) + +-- | . +setSendTimeout :: Integral i => Restricted (Nneg1, Int32) i -> Socket a -> IO () +setSendTimeout = setInt32OptFromRestricted B.sendTimeout + +-- | . +setRate :: Integral i => Restricted (N1, Int32) i -> Socket a -> IO () +setRate = setInt32OptFromRestricted B.rate + +-- | . +setMcastHops :: Integral i => Restricted (N1, Int32) i -> Socket a -> IO () +setMcastHops = setInt32OptFromRestricted B.mcastHops + +-- | . +setBacklog :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO () +setBacklog = setInt32OptFromRestricted B.backlog + +-- | . +setCurvePublicKey :: KeyFormat f -> Restricted f SB.ByteString -> Socket a -> IO () +setCurvePublicKey _ k s = setByteStringOpt s B.curvePublicKey (rvalue k) + +-- | . +setCurveSecretKey :: KeyFormat f -> Restricted f SB.ByteString -> Socket a -> IO () +setCurveSecretKey _ k s = setByteStringOpt s B.curveSecretKey (rvalue k) + +-- | . +setCurveServerKey :: KeyFormat f -> Restricted f SB.ByteString -> Socket a -> IO () +setCurveServerKey _ k s = setByteStringOpt s B.curveServerKey (rvalue k) + +-- | . +setProbeRouter :: SendProbe a => Bool -> Socket a -> IO () +setProbeRouter x s = setIntOpt s B.probeRouter (bool2cint x) + +-- | . +setReceiveBuffer :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO () +setReceiveBuffer = setInt32OptFromRestricted B.receiveBuf + +-- | . +setReconnectInterval :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO () +setReconnectInterval = setInt32OptFromRestricted B.reconnectIVL + +-- | . +setReconnectIntervalMax :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO () +setReconnectIntervalMax = setInt32OptFromRestricted B.reconnectIVLMax + +-- | . +setReqCorrelate :: Bool -> Socket Req -> IO () +setReqCorrelate x s = setIntOpt s B.reqCorrelate (bool2cint x) + +-- | . +setReqRelaxed :: Bool -> Socket Req -> IO () +setReqRelaxed x s = setIntOpt s B.reqRelaxed (bool2cint x) + +-- | . +setSendBuffer :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO () +setSendBuffer = setInt32OptFromRestricted B.sendBuf + +-- | . +setRecoveryInterval :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO () +setRecoveryInterval = setInt32OptFromRestricted B.recoveryIVL + +-- | . +setReceiveHighWM :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO () +setReceiveHighWM = setInt32OptFromRestricted B.receiveHighWM + +-- | . +setSendHighWM :: Integral i => Restricted (N0, Int32) i -> Socket a -> IO () +setSendHighWM = setInt32OptFromRestricted B.sendHighWM + +-- | . +setTcpAcceptFilter :: Maybe SB.ByteString -> Socket a -> IO () +setTcpAcceptFilter Nothing sock = onSocket "setTcpAcceptFilter" sock $ \s -> + throwIfMinus1Retry_ "setStrOpt" $ + c_zmq_setsockopt s (optVal tcpAcceptFilter) nullPtr 0 +setTcpAcceptFilter (Just dat) sock = setByteStringOpt sock tcpAcceptFilter dat + +-- | . +setTcpKeepAlive :: Switch -> Socket a -> IO () +setTcpKeepAlive x s = setIntOpt s B.tcpKeepAlive (fromSwitch x :: CInt) + +-- | . +setTcpKeepAliveCount :: Integral i => Restricted (Nneg1, Int32) i -> Socket a -> IO () +setTcpKeepAliveCount = setInt32OptFromRestricted B.tcpKeepAliveCount + +-- | . +setTcpKeepAliveIdle :: Integral i => Restricted (Nneg1, Int32) i -> Socket a -> IO () +setTcpKeepAliveIdle = setInt32OptFromRestricted B.tcpKeepAliveIdle + +-- | . +setTcpKeepAliveInterval :: Integral i => Restricted (Nneg1, Int32) i -> Socket a -> IO () +setTcpKeepAliveInterval = setInt32OptFromRestricted B.tcpKeepAliveInterval + +-- | . +setXPubVerbose :: Bool -> Socket XPub -> IO () +setXPubVerbose x s = setIntOpt s B.xpubVerbose (bool2cint x) + +-- | Bind the socket to the given address +-- (cf. ). +bind :: Socket a -> String -> IO () +bind sock str = onSocket "bind" sock $ + throwIfMinus1Retry_ "bind" . withCString str . c_zmq_bind + +-- | Unbind the socket from the given address +-- (cf. ). +unbind :: Socket a -> String -> IO () +unbind sock str = onSocket "unbind" sock $ + throwIfMinus1Retry_ "unbind" . withCString str . c_zmq_unbind + +-- | Connect the socket to the given address +-- (cf. ). +connect :: Socket a -> String -> IO () +connect sock str = onSocket "connect" sock $ + throwIfMinus1Retry_ "connect" . withCString str . c_zmq_connect + +-- | Send the given 'SB.ByteString' over the socket +-- (cf. ). +-- +-- /Note/: This function always calls @zmq_sendmsg@ in a non-blocking way, +-- i.e. there is no need to provide the @ZMQ_DONTWAIT@ flag as this is used +-- by default. Still 'send' is blocking the thread as long as the message +-- can not be queued on the socket using GHC's 'threadWaitWrite'. +send :: Sender a => Socket a -> [Flag] -> SB.ByteString -> IO () +send sock fls val = bracket (messageOf val) messageClose $ \m -> + onSocket "send" sock $ \s -> + retry "send" (waitWrite sock) $ + c_zmq_sendmsg s (msgPtr m) (combineFlags (DontWait : fls)) + +-- | Send the given 'LB.ByteString' over the socket +-- (cf. ). +-- +-- This is operationally identical to @send socket (Strict.concat +-- (Lazy.toChunks lbs)) flags@ but may be more efficient. +-- +-- /Note/: This function always calls @zmq_sendmsg@ in a non-blocking way, +-- i.e. there is no need to provide the @ZMQ_DONTWAIT@ flag as this is used +-- by default. Still 'send'' is blocking the thread as long as the message +-- can not be queued on the socket using GHC's 'threadWaitWrite'. +send' :: Sender a => Socket a -> [Flag] -> LB.ByteString -> IO () +send' sock fls val = bracket (messageOfLazy val) messageClose $ \m -> + onSocket "send'" sock $ \s -> + retry "send'" (waitWrite sock) $ + c_zmq_sendmsg s (msgPtr m) (combineFlags (DontWait : fls)) + +-- | Send a multi-part message. +-- This function applies the 'SendMore' 'Flag' between all message parts. +-- 0MQ guarantees atomic delivery of a multi-part message +-- (cf. ). +sendMulti :: Sender a => Socket a -> NonEmpty SB.ByteString -> IO () +sendMulti sock msgs = do + mapM_ (send sock [SendMore]) (S.init msgs) + send sock [] (S.last msgs) + +-- | Receive a 'ByteString' from socket +-- (cf. ). +-- +-- /Note/: This function always calls @zmq_recvmsg@ in a non-blocking way, +-- i.e. there is no need to provide the @ZMQ_DONTWAIT@ flag as this is used +-- by default. Still 'receive' is blocking the thread as long as no data +-- is available using GHC's 'threadWaitRead'. +receive :: Receiver a => Socket a -> IO (SB.ByteString) +receive sock = bracket messageInit messageClose $ \m -> + onSocket "receive" sock $ \s -> do + retry "receive" (waitRead sock) $ + c_zmq_recvmsg s (msgPtr m) (flagVal dontWait) + data_ptr <- c_zmq_msg_data (msgPtr m) + size <- c_zmq_msg_size (msgPtr m) + SB.packCStringLen (data_ptr, fromIntegral size) + +-- | Receive a multi-part message. +-- This function collects all message parts send via 'sendMulti'. +receiveMulti :: Receiver a => Socket a -> IO [SB.ByteString] +receiveMulti sock = recvall [] + where + recvall acc = do + msg <- receive sock + moreToReceive sock >>= next (msg:acc) + + next acc True = recvall acc + next acc False = return (reverse acc) + +-- | Setup socket monitoring, i.e. a 'Pair' socket which +-- sends monitoring events about the given 'Socket' to the +-- given address. +socketMonitor :: [EventType] -> String -> Socket a -> IO () +socketMonitor es addr soc = onSocket "socketMonitor" soc $ \s -> + withCString addr $ \a -> + throwIfMinus1_ "zmq_socket_monitor" $ + c_zmq_socket_monitor s a (events2cint es) + +-- | Monitor socket events +-- (cf. ). +-- +-- This function returns a function which can be invoked to retrieve +-- the next socket event, potentially blocking until the next one becomes +-- available. When applied to 'False', monitoring will terminate, i.e. +-- internal monitoring resources will be disposed. Consequently after +-- 'monitor' has been invoked, the returned function must be applied +-- /once/ to 'False'. +monitor :: [EventType] -> Context -> Socket a -> IO (Bool -> IO (Maybe EventMsg)) +monitor es ctx sock = do + let addr = "inproc://" ++ show (_socket . _socketRepr $ sock) + s <- socket ctx Pair + socketMonitor es addr sock + connect s addr + next s <$> messageInit + where + next soc m False = messageClose m `finally` close soc >> return Nothing + next soc m True = onSocket "recv" soc $ \s -> do + retry "recv" (waitRead soc) $ c_zmq_recvmsg s (msgPtr m) (flagVal dontWait) + ptr <- c_zmq_msg_data (msgPtr m) + evt <- peek ptr + str <- receive soc + return . Just $ eventMessage str evt + +-- | Polls for events on the given 'Poll' descriptors. Returns a list of +-- events per descriptor which have occured. +-- (cf. ) +poll :: (SocketLike s, MonadIO m) => Timeout -> [Poll s m] -> m [[Event]] +poll _ [] = return [] +poll to desc = do + let len = length desc + let ps = map toZMQPoll desc + ps' <- liftIO $ withArray ps $ \ptr -> do + throwIfMinus1Retry_ "poll" $ + c_zmq_poll ptr (fromIntegral len) (fromIntegral to) + peekArray len ptr + mapM fromZMQPoll (zip desc ps') + where + toZMQPoll :: (SocketLike s, MonadIO m) => Poll s m -> ZMQPoll + toZMQPoll (Sock s e _) = + ZMQPoll (_socket . _socketRepr . toSocket $ s) 0 (combine (map fromEvent e)) 0 + + toZMQPoll (File (Fd s) e _) = + ZMQPoll nullPtr (fromIntegral s) (combine (map fromEvent e)) 0 + + fromZMQPoll :: (SocketLike s, MonadIO m) => (Poll s m, ZMQPoll) -> m [Event] + fromZMQPoll (p, zp) = do + let e = toEvents . fromIntegral . pRevents $ zp + let (e', f) = case p of + (Sock _ x g) -> (x, g) + (File _ x g) -> (x, g) + forM f (unless (P.null (e `intersect` e')) . ($ e)) >> return e + + fromEvent :: Event -> CShort + fromEvent In = fromIntegral . pollVal $ pollIn + fromEvent Out = fromIntegral . pollVal $ pollOut + fromEvent Err = fromIntegral . pollVal $ pollerr + +-- Convert bit-masked word into Event list. +toEvents :: Word32 -> [Event] +toEvents e = foldl' (\es f -> f e es) [] tests + where + tests = + [ \i xs -> if i .&. (fromIntegral . pollVal $ pollIn) /= 0 then In:xs else xs + , \i xs -> if i .&. (fromIntegral . pollVal $ pollOut) /= 0 then Out:xs else xs + , \i xs -> if i .&. (fromIntegral . pollVal $ pollerr) /= 0 then Err:xs else xs + ] + +retry :: String -> IO () -> IO CInt -> IO () +retry msg wait act = throwIfMinus1RetryMayBlock_ msg act wait + +wait' :: (Fd -> IO ()) -> ZMQPollEvent -> Socket a -> IO () +wait' w f s = do + fd <- getIntOpt s B.filedesc 0 + w (Fd fd) + evs <- getInt32Option B.events s + unless (testev evs) $ + wait' w f s + where + testev e = e .&. fromIntegral (pollVal f) /= 0 + +-- | Wait until data is available for reading from the given Socket. +-- After this function returns, a call to 'receive' will essentially be +-- non-blocking. +waitRead :: Socket a -> IO () +waitRead = wait' threadWaitRead pollIn + +-- | Wait until data can be written to the given Socket. +-- After this function returns, a call to 'send' will essentially be +-- non-blocking. +waitWrite :: Socket a -> IO () +waitWrite = wait' threadWaitWrite pollOut + +-- | Starts built-in 0MQ proxy +-- (cf. ) +-- +-- Proxy connects front to back socket +-- +-- Before calling proxy all sockets should be binded +-- +-- If the capture socket is not Nothing, the proxy shall send all +-- messages, received on both frontend and backend, to the capture socket. +proxy :: Socket a -> Socket b -> Maybe (Socket c) -> IO () +proxy front back capture = + onSocket "proxy-front" front $ \f -> + onSocket "proxy-back" back $ \b -> + throwIfMinus1Retry_ "c_zmq_proxy" $ c_zmq_proxy f b c + where + c = maybe nullPtr (_socket . _socketRepr) capture + +-- | Generate a new curve key pair. +-- (cf. ) +curveKeyPair :: MonadIO m => m (Restricted Div5 SB.ByteString, Restricted Div5 SB.ByteString) +curveKeyPair = liftIO $ + allocaBytes 41 $ \cstr1 -> + allocaBytes 41 $ \cstr2 -> do + throwIfMinus1_ "c_zmq_curve_keypair" $ c_zmq_curve_keypair cstr1 cstr2 + public <- toRestricted <$> SB.packCString cstr1 + private <- toRestricted <$> SB.packCString cstr2 + maybe (fail errmsg) return ((,) <$> public <*> private) + where + errmsg = "curveKeyPair: invalid key-lengths produced" + diff --git a/src/System/ZMQ4/Base.hsc b/src/System/ZMQ4/Base.hsc new file mode 100644 --- /dev/null +++ b/src/System/ZMQ4/Base.hsc @@ -0,0 +1,360 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE ForeignFunctionInterface #-} + +module System.ZMQ4.Base where + +import Foreign +import Foreign.C.Types +import Foreign.C.String +import Control.Applicative + +#include + +#if ZMQ_VERSION_MAJOR != 4 + #error *** INVALID 0MQ VERSION (must be 4.x) *** +#endif + +----------------------------------------------------------------------------- +-- Message + +newtype ZMQMsg = ZMQMsg + { content :: Ptr () + } deriving (Eq, Ord) + +instance Storable ZMQMsg where + alignment _ = #{alignment zmq_msg_t} + sizeOf _ = #{size zmq_msg_t} + peek p = ZMQMsg <$> #{peek zmq_msg_t, _} p + poke p (ZMQMsg c) = #{poke zmq_msg_t, _} p c + +----------------------------------------------------------------------------- +-- Poll + +data ZMQPoll = ZMQPoll + { pSocket :: {-# UNPACK #-} !ZMQSocket + , pFd :: {-# UNPACK #-} !CInt + , pEvents :: {-# UNPACK #-} !CShort + , pRevents :: {-# UNPACK #-} !CShort + } + +instance Storable ZMQPoll where + alignment _ = #{alignment zmq_pollitem_t} + sizeOf _ = #{size zmq_pollitem_t} + peek p = do + s <- #{peek zmq_pollitem_t, socket} p + f <- #{peek zmq_pollitem_t, fd} p + e <- #{peek zmq_pollitem_t, events} p + re <- #{peek zmq_pollitem_t, revents} p + return $ ZMQPoll s f e re + poke p (ZMQPoll s f e re) = do + #{poke zmq_pollitem_t, socket} p s + #{poke zmq_pollitem_t, fd} p f + #{poke zmq_pollitem_t, events} p e + #{poke zmq_pollitem_t, revents} p re + +type ZMQMsgPtr = Ptr ZMQMsg +type ZMQCtx = Ptr () +type ZMQSocket = Ptr () +type ZMQPollPtr = Ptr ZMQPoll + +#let alignment t = "%lu", (unsigned long)offsetof(struct {char x__; t (y__); }, y__) + +----------------------------------------------------------------------------- +-- Socket Types + +newtype ZMQSocketType = ZMQSocketType + { typeVal :: CInt + } deriving (Eq, Ord) + +#{enum ZMQSocketType, ZMQSocketType + , pair = ZMQ_PAIR + , pub = ZMQ_PUB + , sub = ZMQ_SUB + , xpub = ZMQ_XPUB + , xsub = ZMQ_XSUB + , request = ZMQ_REQ + , response = ZMQ_REP + , dealer = ZMQ_DEALER + , router = ZMQ_ROUTER + , pull = ZMQ_PULL + , push = ZMQ_PUSH + , stream = ZMQ_STREAM +} + +----------------------------------------------------------------------------- +-- Socket Options + +newtype ZMQOption = ZMQOption + { optVal :: CInt + } deriving (Eq, Ord) + +#{enum ZMQOption, ZMQOption + , affinity = ZMQ_AFFINITY + , backlog = ZMQ_BACKLOG + , conflate = ZMQ_CONFLATE + , curve = ZMQ_CURVE + , curvePublicKey = ZMQ_CURVE_PUBLICKEY + , curveSecretKey = ZMQ_CURVE_SECRETKEY + , curveServer = ZMQ_CURVE_SERVER + , curveServerKey = ZMQ_CURVE_SERVERKEY + , delayAttachOnConnect = ZMQ_DELAY_ATTACH_ON_CONNECT + , events = ZMQ_EVENTS + , filedesc = ZMQ_FD + , identity = ZMQ_IDENTITY + , immediate = ZMQ_IMMEDIATE + , ipv4Only = ZMQ_IPV4ONLY + , ipv6 = ZMQ_IPV6 + , lastEndpoint = ZMQ_LAST_ENDPOINT + , linger = ZMQ_LINGER + , maxMessageSize = ZMQ_MAXMSGSIZE + , mcastHops = ZMQ_MULTICAST_HOPS + , mechanism = ZMQ_MECHANISM + , null = ZMQ_NULL + , plain = ZMQ_PLAIN + , plainPassword = ZMQ_PLAIN_PASSWORD + , plainServer = ZMQ_PLAIN_SERVER + , plainUserName = ZMQ_PLAIN_USERNAME + , probeRouter = ZMQ_PROBE_ROUTER + , rate = ZMQ_RATE + , receiveBuf = ZMQ_RCVBUF + , receiveHighWM = ZMQ_RCVHWM + , receiveMore = ZMQ_RCVMORE + , receiveTimeout = ZMQ_RCVTIMEO + , reconnectIVL = ZMQ_RECONNECT_IVL + , reconnectIVLMax = ZMQ_RECONNECT_IVL_MAX + , recoveryIVL = ZMQ_RECOVERY_IVL + , reqCorrelate = ZMQ_REQ_CORRELATE + , reqRelaxed = ZMQ_REQ_RELAXED + , routerMandatory = ZMQ_ROUTER_MANDATORY + , sendBuf = ZMQ_SNDBUF + , sendHighWM = ZMQ_SNDHWM + , sendTimeout = ZMQ_SNDTIMEO + , subscribe = ZMQ_SUBSCRIBE + , tcpAcceptFilter = ZMQ_TCP_ACCEPT_FILTER + , tcpKeepAlive = ZMQ_TCP_KEEPALIVE + , tcpKeepAliveCount = ZMQ_TCP_KEEPALIVE_CNT + , tcpKeepAliveIdle = ZMQ_TCP_KEEPALIVE_IDLE + , tcpKeepAliveInterval = ZMQ_TCP_KEEPALIVE_INTVL + , unsubscribe = ZMQ_UNSUBSCRIBE + , xpubVerbose = ZMQ_XPUB_VERBOSE + , zapDomain = ZMQ_ZAP_DOMAIN +} + +----------------------------------------------------------------------------- +-- Context Options + +newtype ZMQCtxOption = ZMQCtxOption + { ctxOptVal :: CInt + } deriving (Eq, Ord) + +#{enum ZMQCtxOption, ZMQCtxOption + , _ioThreads = ZMQ_IO_THREADS + , _maxSockets = ZMQ_MAX_SOCKETS +} + +----------------------------------------------------------------------------- +-- Event Type + +newtype ZMQEventType = ZMQEventType + { eventTypeVal :: Word16 + } deriving (Eq, Ord, Show) + +#{enum ZMQEventType, ZMQEventType + , connected = ZMQ_EVENT_CONNECTED + , connectDelayed = ZMQ_EVENT_CONNECT_DELAYED + , connectRetried = ZMQ_EVENT_CONNECT_RETRIED + , listening = ZMQ_EVENT_LISTENING + , bindFailed = ZMQ_EVENT_BIND_FAILED + , accepted = ZMQ_EVENT_ACCEPTED + , acceptFailed = ZMQ_EVENT_ACCEPT_FAILED + , closed = ZMQ_EVENT_CLOSED + , closeFailed = ZMQ_EVENT_CLOSE_FAILED + , disconnected = ZMQ_EVENT_DISCONNECTED + , allEvents = ZMQ_EVENT_ALL + , monitorStopped = ZMQ_EVENT_MONITOR_STOPPED +} + +----------------------------------------------------------------------------- +-- Event + +data ZMQEvent = ZMQEvent + { zeEvent :: {-# UNPACK #-} !ZMQEventType + , zeValue :: {-# UNPACK #-} !Int32 + } + +instance Storable ZMQEvent where + alignment _ = #{alignment zmq_event_t} + sizeOf _ = #{size zmq_event_t} + peek e = ZMQEvent + <$> (ZMQEventType <$> #{peek zmq_event_t, event} e) + <*> #{peek zmq_event_t, value} e + poke e (ZMQEvent (ZMQEventType a) b) = do + #{poke zmq_event_t, event} e a + #{poke zmq_event_t, value} e b + +----------------------------------------------------------------------------- +-- Security Mechanism + +newtype ZMQSecMechanism = ZMQSecMechanism + { secMechanism :: Int + } deriving (Eq, Ord, Show) + +#{enum ZMQSecMechanism, ZMQSecMechanism + , secNull = ZMQ_NULL + , secPlain = ZMQ_PLAIN + , secCurve = ZMQ_CURVE +} + +----------------------------------------------------------------------------- +-- Message Options + +newtype ZMQMsgOption = ZMQMsgOption + { msgOptVal :: CInt + } deriving (Eq, Ord) + +#{enum ZMQMsgOption, ZMQMsgOption + , more = ZMQ_MORE +} + +----------------------------------------------------------------------------- +-- Flags + +newtype ZMQFlag = ZMQFlag + { flagVal :: CInt + } deriving (Eq, Ord) + +#{enum ZMQFlag, ZMQFlag + , dontWait = ZMQ_DONTWAIT + , sndMore = ZMQ_SNDMORE +} + +----------------------------------------------------------------------------- +-- Poll Events + +newtype ZMQPollEvent = ZMQPollEvent + { pollVal :: CShort + } deriving (Eq, Ord) + +#{enum ZMQPollEvent, ZMQPollEvent, + pollIn = ZMQ_POLLIN, + pollOut = ZMQ_POLLOUT, + pollerr = ZMQ_POLLERR +} + +----------------------------------------------------------------------------- +-- function declarations + +-- general initialization + +foreign import ccall unsafe "zmq.h zmq_version" + c_zmq_version :: Ptr CInt -> Ptr CInt -> Ptr CInt -> IO () + +foreign import ccall unsafe "zmq.h zmq_ctx_new" + c_zmq_ctx_new :: IO ZMQCtx + +foreign import ccall unsafe "zmq.h zmq_ctx_shutdown" + c_zmq_ctx_shutdown :: ZMQCtx -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_ctx_term" + c_zmq_ctx_term :: ZMQCtx -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_ctx_get" + c_zmq_ctx_get :: ZMQCtx -> CInt -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_ctx_set" + c_zmq_ctx_set :: ZMQCtx -> CInt -> CInt -> IO CInt + +-- zmq_msg_t related + +foreign import ccall unsafe "zmq.h zmq_msg_init" + c_zmq_msg_init :: ZMQMsgPtr -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_msg_init_size" + c_zmq_msg_init_size :: ZMQMsgPtr -> CSize -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_msg_close" + c_zmq_msg_close :: ZMQMsgPtr -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_msg_data" + c_zmq_msg_data :: ZMQMsgPtr -> IO (Ptr a) + +foreign import ccall unsafe "zmq.h zmq_msg_size" + c_zmq_msg_size :: ZMQMsgPtr -> IO CSize + +foreign import ccall unsafe "zmq.h zmq_msg_get" + c_zmq_msg_get :: ZMQMsgPtr -> CInt -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_msg_set" + c_zmq_msg_set :: ZMQMsgPtr -> CInt -> CInt -> IO CInt + +-- socket + +foreign import ccall unsafe "zmq.h zmq_socket" + c_zmq_socket :: ZMQCtx -> CInt -> IO ZMQSocket + +foreign import ccall unsafe "zmq.h zmq_close" + c_zmq_close :: ZMQSocket -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_setsockopt" + c_zmq_setsockopt :: ZMQSocket + -> CInt -- option + -> Ptr () -- option value + -> CSize -- option value size + -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_getsockopt" + c_zmq_getsockopt :: ZMQSocket + -> CInt -- option + -> Ptr () -- option value + -> Ptr CSize -- option value size ptr + -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_bind" + c_zmq_bind :: ZMQSocket -> CString -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_unbind" + c_zmq_unbind :: ZMQSocket -> CString -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_connect" + c_zmq_connect :: ZMQSocket -> CString -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_sendmsg" + c_zmq_sendmsg :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_recvmsg" + c_zmq_recvmsg :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt + +foreign import ccall unsafe "zmq.h zmq_socket_monitor" + c_zmq_socket_monitor :: ZMQSocket -> CString -> CInt -> IO CInt + +-- errors + +foreign import ccall unsafe "zmq.h zmq_errno" + c_zmq_errno :: IO CInt + +foreign import ccall unsafe "zmq.h zmq_strerror" + c_zmq_strerror :: CInt -> IO CString + +-- proxy + +foreign import ccall safe "zmq.h zmq_proxy" + c_zmq_proxy :: ZMQSocket -> ZMQSocket -> ZMQSocket -> IO CInt + +-- poll + +foreign import ccall safe "zmq.h zmq_poll" + c_zmq_poll :: ZMQPollPtr -> CInt -> CLong -> IO CInt + +-- Z85 encode/decode + +foreign import ccall unsafe "zmq.h zmq_z85_encode" + c_zmq_z85_encode :: CString -> Ptr Word8 -> CSize -> IO CString + +foreign import ccall unsafe "zmq.h zmq_z85_decode" + c_zmq_z85_decode :: Ptr Word8 -> CString -> IO (Ptr Word8) + +-- curve crypto + +foreign import ccall unsafe "zmq.h zmq_curve_keypair" + c_zmq_curve_keypair :: CString -> CString -> IO CInt + diff --git a/src/System/ZMQ4/Error.hs b/src/System/ZMQ4/Error.hs new file mode 100644 --- /dev/null +++ b/src/System/ZMQ4/Error.hs @@ -0,0 +1,99 @@ +{-# LANGUAGE DeriveDataTypeable #-} + +-- We use our own functions for throwing exceptions in order to get +-- the actual error message via 'zmq_strerror'. 0MQ defines additional +-- error numbers besides those defined by the operating system, so +-- 'zmq_strerror' should be used in preference to 'strerror' which is +-- used by the standard throw* functions in 'Foreign.C.Error'. +module System.ZMQ4.Error where + +import Control.Applicative +import Control.Monad +import Control.Exception +import Text.Printf +import Data.Typeable (Typeable) + +import Foreign hiding (throwIf, throwIf_, void) +import Foreign.C.Error +import Foreign.C.String +import Foreign.C.Types (CInt) + +import System.ZMQ4.Base + +-- | ZMQError encapsulates information about errors, which occur +-- when using the native 0MQ API, such as error number and message. +data ZMQError = ZMQError + { errno :: Int -- ^ Error number value. + , source :: String -- ^ Source where this error originates from. + , message :: String -- ^ Actual error message. + } deriving (Eq, Ord, Typeable) + +instance Show ZMQError where + show e = printf "ZMQError { errno = %d, source = \"%s\", message = \"%s\" }" + (errno e) (source e) (message e) + +instance Exception ZMQError + +throwError :: String -> IO a +throwError src = do + (Errno e) <- zmqErrno + msg <- zmqErrnoMessage e + throwIO $ ZMQError (fromIntegral e) src msg + +throwIf :: (a -> Bool) -> String -> IO a -> IO a +throwIf p src act = do + r <- act + if p r then throwError src else return r + +throwIf_ :: (a -> Bool) -> String -> IO a -> IO () +throwIf_ p src act = void $ throwIf p src act + +throwIfRetry :: (a -> Bool) -> String -> IO a -> IO a +throwIfRetry p src act = do + r <- act + if p r then zmqErrno >>= k else return r + where + k e | e == eINTR = throwIfRetry p src act + | otherwise = throwError src + +throwIfRetry_ :: (a -> Bool) -> String -> IO a -> IO () +throwIfRetry_ p src act = void $ throwIfRetry p src act + +throwIfMinus1 :: (Eq a, Num a) => String -> IO a -> IO a +throwIfMinus1 = throwIf (== -1) + +throwIfMinus1_ :: (Eq a, Num a) => String -> IO a -> IO () +throwIfMinus1_ = throwIf_ (== -1) + +throwIfNull :: String -> IO (Ptr a) -> IO (Ptr a) +throwIfNull = throwIf (== nullPtr) + +throwIfMinus1Retry :: (Eq a, Num a) => String -> IO a -> IO a +throwIfMinus1Retry = throwIfRetry (== -1) + +throwIfMinus1Retry_ :: (Eq a, Num a) => String -> IO a -> IO () +throwIfMinus1Retry_ = throwIfRetry_ (== -1) + +throwIfRetryMayBlock :: (a -> Bool) -> String -> IO a -> IO b -> IO a +throwIfRetryMayBlock p src f on_block = do + r <- f + if p r then zmqErrno >>= k else return r + where + k e | e == eINTR = throwIfRetryMayBlock p src f on_block + | e == eWOULDBLOCK || e == eAGAIN = on_block >> throwIfRetryMayBlock p src f on_block + | otherwise = throwError src + +throwIfRetryMayBlock_ :: (a -> Bool) -> String -> IO a -> IO b -> IO () +throwIfRetryMayBlock_ p src f on_block = void $ throwIfRetryMayBlock p src f on_block + +throwIfMinus1RetryMayBlock :: (Eq a, Num a) => String -> IO a -> IO b -> IO a +throwIfMinus1RetryMayBlock = throwIfRetryMayBlock (== -1) + +throwIfMinus1RetryMayBlock_ :: (Eq a, Num a) => String -> IO a -> IO b -> IO () +throwIfMinus1RetryMayBlock_ = throwIfRetryMayBlock_ (== -1) + +zmqErrnoMessage :: CInt -> IO String +zmqErrnoMessage e = c_zmq_strerror e >>= peekCString + +zmqErrno :: IO Errno +zmqErrno = Errno <$> c_zmq_errno diff --git a/src/System/ZMQ4/Internal.hs b/src/System/ZMQ4/Internal.hs new file mode 100644 --- /dev/null +++ b/src/System/ZMQ4/Internal.hs @@ -0,0 +1,371 @@ +{-# LANGUAGE GADTs #-} +{-# LANGUAGE StandaloneDeriving #-} + +module System.ZMQ4.Internal + ( Context (..) + , Socket (..) + , SocketRepr (..) + , SocketType (..) + , SocketLike (..) + , Message (..) + , Flag (..) + , Timeout + , Size + , Switch (..) + , EventType (..) + , EventMsg (..) + , SecurityMechanism (..) + , KeyFormat (..) + + , messageOf + , messageOfLazy + , messageClose + , messageInit + , messageInitSize + , setIntOpt + , setStrOpt + , getIntOpt + , getStrOpt + , getInt32Option + , setInt32OptFromRestricted + , ctxIntOption + , setCtxIntOption + , getByteStringOpt + , setByteStringOpt + + , z85Encode + , z85Decode + + , toZMQFlag + , combine + , combineFlags + , mkSocketRepr + , closeSock + , onSocket + + , bool2cint + , toSwitch + , fromSwitch + , events2cint + , eventMessage + + , toMechanism + , fromMechanism + + , getKey + ) where + +import Control.Applicative +import Control.Monad (foldM_, when, void) +import Control.Monad.IO.Class +import Control.Exception +import Data.IORef (IORef, mkWeakIORef, readIORef, atomicModifyIORef) + +import Foreign hiding (throwIfNull, void) +import Foreign.C.String +import Foreign.C.Types (CInt, CSize) + +import Data.IORef (newIORef) +import Data.Restricted + +import System.Posix.Types (Fd(..)) +import System.ZMQ4.Base +import System.ZMQ4.Error + +import qualified Data.ByteString as SB +import qualified Data.ByteString.Lazy as LB +import qualified Data.ByteString.Unsafe as UB + +type Timeout = Int64 +type Size = Word + +-- | Flags to apply on send operations (cf. man zmq_send) +data Flag = + DontWait -- ^ ZMQ_DONTWAIT + | SendMore -- ^ ZMQ_SNDMORE + deriving (Eq, Ord, Show) + +-- | Configuration switch +data Switch = + Default -- ^ Use default setting + | On -- ^ Activate setting + | Off -- ^ De-activate setting + deriving (Eq, Ord, Show) + +-- | Event types to monitor. +data EventType = + ConnectedEvent + | ConnectDelayedEvent + | ConnectRetriedEvent + | ListeningEvent + | BindFailedEvent + | AcceptedEvent + | AcceptFailedEvent + | ClosedEvent + | CloseFailedEvent + | DisconnectedEvent + | MonitorStoppedEvent + | AllEvents + deriving (Eq, Ord, Show) + +-- | Event Message to receive when monitoring socket events. +data EventMsg = + Connected !SB.ByteString !Fd + | ConnectDelayed !SB.ByteString + | ConnectRetried !SB.ByteString !Int + | Listening !SB.ByteString !Fd + | BindFailed !SB.ByteString !Int + | Accepted !SB.ByteString !Fd + | AcceptFailed !SB.ByteString !Int + | Closed !SB.ByteString !Fd + | CloseFailed !SB.ByteString !Int + | Disconnected !SB.ByteString !Fd + | MonitorStopped !SB.ByteString !Int + deriving (Eq, Show) + +data SecurityMechanism + = Null + | Plain + | Curve + deriving (Eq, Show) + +data KeyFormat a where + BinaryFormat :: KeyFormat Div4 + TextFormat :: KeyFormat Div5 + +deriving instance Eq (KeyFormat a) +deriving instance Show (KeyFormat a) + +-- | A 0MQ context representation. +newtype Context = Context { _ctx :: ZMQCtx } + +-- | A 0MQ Socket. +newtype Socket a = Socket + { _socketRepr :: SocketRepr } + +data SocketRepr = SocketRepr + { _socket :: ZMQSocket + , _sockLive :: IORef Bool + } + +-- | Socket types. +class SocketType a where + zmqSocketType :: a -> ZMQSocketType + +class SocketLike s where + toSocket :: s t -> Socket t + +instance SocketLike Socket where + toSocket = id + +-- A 0MQ Message representation. +newtype Message = Message { msgPtr :: ZMQMsgPtr } + +-- internal helpers: + +onSocket :: String -> Socket a -> (ZMQSocket -> IO b) -> IO b +onSocket _func (Socket (SocketRepr sock _state)) act = act sock +{-# INLINE onSocket #-} + +mkSocketRepr :: SocketType t => t -> Context -> IO SocketRepr +mkSocketRepr t c = do + let ty = typeVal (zmqSocketType t) + s <- throwIfNull "mkSocketRepr" (c_zmq_socket (_ctx c) ty) + ref <- newIORef True + addFinalizer ref $ do + alive <- readIORef ref + when alive $ c_zmq_close s >> return () + return (SocketRepr s ref) + where + addFinalizer r f = mkWeakIORef r f >> return () + +closeSock :: SocketRepr -> IO () +closeSock (SocketRepr s status) = do + alive <- atomicModifyIORef status (\b -> (False, b)) + when alive $ throwIfMinus1_ "close" . c_zmq_close $ s + +messageOf :: SB.ByteString -> IO Message +messageOf b = UB.unsafeUseAsCStringLen b $ \(cstr, len) -> do + msg <- messageInitSize (fromIntegral len) + data_ptr <- c_zmq_msg_data (msgPtr msg) + copyBytes data_ptr cstr len + return msg + +messageOfLazy :: LB.ByteString -> IO Message +messageOfLazy lbs = do + msg <- messageInitSize (fromIntegral len) + data_ptr <- c_zmq_msg_data (msgPtr msg) + let fn offset bs = UB.unsafeUseAsCStringLen bs $ \(cstr, str_len) -> do + copyBytes (data_ptr `plusPtr` offset) cstr str_len + return (offset + str_len) + foldM_ fn 0 (LB.toChunks lbs) + return msg + where + len = LB.length lbs + +messageClose :: Message -> IO () +messageClose (Message ptr) = do + throwIfMinus1_ "messageClose" $ c_zmq_msg_close ptr + free ptr + +messageInit :: IO Message +messageInit = do + ptr <- new (ZMQMsg nullPtr) + throwIfMinus1_ "messageInit" $ c_zmq_msg_init ptr + return (Message ptr) + +messageInitSize :: Size -> IO Message +messageInitSize s = do + ptr <- new (ZMQMsg nullPtr) + throwIfMinus1_ "messageInitSize" $ + c_zmq_msg_init_size ptr (fromIntegral s) + return (Message ptr) + +setIntOpt :: (Storable b, Integral b) => Socket a -> ZMQOption -> b -> IO () +setIntOpt sock (ZMQOption o) i = onSocket "setIntOpt" sock $ \s -> + throwIfMinus1Retry_ "setIntOpt" $ with i $ \ptr -> + c_zmq_setsockopt s (fromIntegral o) + (castPtr ptr) + (fromIntegral . sizeOf $ i) + +setCStrOpt :: ZMQSocket -> ZMQOption -> CStringLen -> IO CInt +setCStrOpt s (ZMQOption o) (cstr, len) = + c_zmq_setsockopt s (fromIntegral o) (castPtr cstr) (fromIntegral len) + +setByteStringOpt :: Socket a -> ZMQOption -> SB.ByteString -> IO () +setByteStringOpt sock opt str = onSocket "setByteStringOpt" sock $ \s -> + throwIfMinus1Retry_ "setByteStringOpt" . UB.unsafeUseAsCStringLen str $ setCStrOpt s opt + +setStrOpt :: Socket a -> ZMQOption -> String -> IO () +setStrOpt sock opt str = onSocket "setStrOpt" sock $ \s -> + throwIfMinus1Retry_ "setStrOpt" . withCStringLen str $ setCStrOpt s opt + +getIntOpt :: (Storable b, Integral b) => Socket a -> ZMQOption -> b -> IO b +getIntOpt sock (ZMQOption o) i = onSocket "getIntOpt" sock $ \s -> do + bracket (new i) free $ \iptr -> + bracket (new (fromIntegral . sizeOf $ i :: CSize)) free $ \jptr -> do + throwIfMinus1Retry_ "getIntOpt" $ + c_zmq_getsockopt s (fromIntegral o) (castPtr iptr) jptr + peek iptr + +getCStrOpt :: (CStringLen -> IO s) -> Socket a -> ZMQOption -> IO s +getCStrOpt peekA sock (ZMQOption o) = onSocket "getCStrOpt" sock $ \s -> + bracket (mallocBytes 255) free $ \bPtr -> + bracket (new (255 :: CSize)) free $ \sPtr -> do + throwIfMinus1Retry_ "getCStrOpt" $ + c_zmq_getsockopt s (fromIntegral o) (castPtr bPtr) sPtr + peek sPtr >>= \len -> peekA (bPtr, fromIntegral len) + +getStrOpt :: Socket a -> ZMQOption -> IO String +getStrOpt = getCStrOpt peekCStringLen + +getByteStringOpt :: Socket a -> ZMQOption -> IO SB.ByteString +getByteStringOpt = getCStrOpt SB.packCStringLen + +getInt32Option :: ZMQOption -> Socket a -> IO Int +getInt32Option o s = fromIntegral <$> getIntOpt s o (0 :: CInt) + +setInt32OptFromRestricted :: Integral i => ZMQOption -> Restricted r i -> Socket b -> IO () +setInt32OptFromRestricted o x s = setIntOpt s o ((fromIntegral . rvalue $ x) :: CInt) + +ctxIntOption :: Integral i => String -> ZMQCtxOption -> Context -> IO i +ctxIntOption name opt ctx = fromIntegral <$> + (throwIfMinus1 name $ c_zmq_ctx_get (_ctx ctx) (ctxOptVal opt)) + +setCtxIntOption :: Integral i => String -> ZMQCtxOption -> i -> Context -> IO () +setCtxIntOption name opt val ctx = throwIfMinus1_ name $ + c_zmq_ctx_set (_ctx ctx) (ctxOptVal opt) (fromIntegral val) + +z85Encode :: (MonadIO m) => Restricted Div4 SB.ByteString -> m SB.ByteString +z85Encode b = liftIO $ UB.unsafeUseAsCStringLen (rvalue b) $ \(c, s) -> + allocaBytes ((s * 5) `div` 4 + 1) $ \w -> do + void . throwIfNull "z85Encode" $ + c_zmq_z85_encode w (castPtr c) (fromIntegral s) + SB.packCString w + +z85Decode :: (MonadIO m) => Restricted Div5 SB.ByteString -> m SB.ByteString +z85Decode b = liftIO $ SB.useAsCStringLen (rvalue b) $ \(c, s) -> do + let size = (s * 4) `div` 5 + allocaBytes size $ \w -> do + void . throwIfNull "z85Decode" $ + c_zmq_z85_decode (castPtr w) (castPtr c) + SB.packCStringLen (w, size) + +getKey :: KeyFormat f -> Socket a -> ZMQOption -> IO SB.ByteString +getKey kf sock (ZMQOption o) = onSocket "getKey" sock $ \s -> do + let len = case kf of + BinaryFormat -> 32 + TextFormat -> 41 + with len $ \lenptr -> allocaBytes len $ \w -> do + throwIfMinus1Retry_ "getKey" $ + c_zmq_getsockopt s (fromIntegral o) (castPtr w) (castPtr lenptr) + SB.packCString w + +toZMQFlag :: Flag -> ZMQFlag +toZMQFlag DontWait = dontWait +toZMQFlag SendMore = sndMore + +combineFlags :: [Flag] -> CInt +combineFlags = fromIntegral . combine . map (flagVal . toZMQFlag) + +combine :: (Integral i, Bits i) => [i] -> i +combine = foldr (.|.) 0 + +bool2cint :: Bool -> CInt +bool2cint True = 1 +bool2cint False = 0 + +toSwitch :: (Show a, Integral a) => String -> a -> Switch +toSwitch _ (-1) = Default +toSwitch _ 0 = Off +toSwitch _ 1 = On +toSwitch m n = error $ m ++ ": " ++ show n + +fromSwitch :: Integral a => Switch -> a +fromSwitch Default = -1 +fromSwitch Off = 0 +fromSwitch On = 1 + +toZMQEventType :: EventType -> ZMQEventType +toZMQEventType AllEvents = allEvents +toZMQEventType ConnectedEvent = connected +toZMQEventType ConnectDelayedEvent = connectDelayed +toZMQEventType ConnectRetriedEvent = connectRetried +toZMQEventType ListeningEvent = listening +toZMQEventType BindFailedEvent = bindFailed +toZMQEventType AcceptedEvent = accepted +toZMQEventType AcceptFailedEvent = acceptFailed +toZMQEventType ClosedEvent = closed +toZMQEventType CloseFailedEvent = closeFailed +toZMQEventType DisconnectedEvent = disconnected +toZMQEventType MonitorStoppedEvent = monitorStopped + +toMechanism :: SecurityMechanism -> ZMQSecMechanism +toMechanism Null = secNull +toMechanism Plain = secPlain +toMechanism Curve = secCurve + +fromMechanism :: String -> Int -> SecurityMechanism +fromMechanism s m + | m == secMechanism secNull = Null + | m == secMechanism secPlain = Plain + | m == secMechanism secCurve = Curve + | otherwise = error $ s ++ ": " ++ show m + +events2cint :: [EventType] -> CInt +events2cint = fromIntegral . foldr ((.|.) . eventTypeVal . toZMQEventType) 0 + +eventMessage :: SB.ByteString -> ZMQEvent -> EventMsg +eventMessage str (ZMQEvent e v) + | e == connected = Connected str (Fd . fromIntegral $ v) + | e == connectDelayed = ConnectDelayed str + | e == connectRetried = ConnectRetried str (fromIntegral $ v) + | e == listening = Listening str (Fd . fromIntegral $ v) + | e == bindFailed = BindFailed str (fromIntegral $ v) + | e == accepted = Accepted str (Fd . fromIntegral $ v) + | e == acceptFailed = AcceptFailed str (fromIntegral $ v) + | e == closed = Closed str (Fd . fromIntegral $ v) + | e == closeFailed = CloseFailed str (fromIntegral $ v) + | e == disconnected = Disconnected str (fromIntegral $ v) + | e == monitorStopped = MonitorStopped str (fromIntegral $ v) + | otherwise = error $ "unknown event type: " ++ show e diff --git a/src/System/ZMQ4/Monadic.hs b/src/System/ZMQ4/Monadic.hs new file mode 100644 --- /dev/null +++ b/src/System/ZMQ4/Monadic.hs @@ -0,0 +1,606 @@ +{-# LANGUAGE RankNTypes #-} +{-# OPTIONS_GHC -fno-warn-warnings-deprecations #-} + +-- | +-- Module : System.ZMQ4.Monadic +-- Copyright : (c) 2013 Toralf Wittner +-- License : MIT +-- Maintainer : Toralf Wittner +-- Stability : experimental +-- Portability : non-portable +-- +-- This modules exposes a monadic interface of 'System.ZMQ4'. Actions run +-- inside a 'ZMQ' monad and 'Socket's are guaranteed not to leak outside +-- their corresponding 'runZMQ' scope. Running 'ZMQ' computations +-- asynchronously is directly supported through 'async'. +module System.ZMQ4.Monadic + ( -- * Type Definitions + ZMQ + , Socket + , Z.Flag (SendMore) + , Z.Switch (..) + , Z.Timeout + , Z.Event (..) + , Z.EventType (..) + , Z.EventMsg (..) + , Z.Poll (..) + , Z.KeyFormat (..) + , Z.SecurityMechanism (..) + + -- ** Socket type-classes + , Z.SocketType + , Z.Sender + , Z.Receiver + , Z.Subscriber + , Z.SocketLike + , Z.Conflatable + , Z.SendProbe + + -- ** Socket Types + , Z.Pair (..) + , Z.Pub (..) + , Z.Sub (..) + , Z.XPub (..) + , Z.XSub (..) + , Z.Req (..) + , Z.Rep (..) + , Z.Dealer (..) + , Z.Router (..) + , Z.Pull (..) + , Z.Push (..) + , Z.Stream (..) + + -- * General Operations + , version + , runZMQ + , async + , socket + + -- * ZMQ Options (Read) + , ioThreads + , maxSockets + + -- * ZMQ Options (Write) + , setIoThreads + , setMaxSockets + + -- * Socket operations + , close + , bind + , unbind + , connect + , send + , send' + , sendMulti + , receive + , receiveMulti + , subscribe + , unsubscribe + , proxy + , monitor + , Z.poll + + -- * Socket Options (Read) + , affinity + , backlog + , conflate + , curvePublicKey + , curveSecretKey + , curveServerKey + , delayAttachOnConnect + , events + , fileDescriptor + , identity + , immediate + , ipv4Only + , ipv6 + , lastEndpoint + , linger + , maxMessageSize + , mcastHops + , mechanism + , moreToReceive + , plainServer + , plainPassword + , plainUserName + , rate + , receiveBuffer + , receiveHighWM + , receiveTimeout + , reconnectInterval + , reconnectIntervalMax + , recoveryInterval + , sendBuffer + , sendHighWM + , sendTimeout + , tcpKeepAlive + , tcpKeepAliveCount + , tcpKeepAliveIdle + , tcpKeepAliveInterval + , zapDomain + + -- * Socket Options (Write) + , setAffinity + , setBacklog + , setConflate + , setCurveServer + , setCurvePublicKey + , setCurveSecretKey + , setCurveServerKey + , setDelayAttachOnConnect + , setIdentity + , setImmediate + , setIpv4Only + , setIpv6 + , setLinger + , setMaxMessageSize + , setMcastHops + , setPlainServer + , setPlainPassword + , setPlainUserName + , setProbeRouter + , setRate + , setReceiveBuffer + , setReceiveHighWM + , setReceiveTimeout + , setReconnectInterval + , setReconnectIntervalMax + , setRecoveryInterval + , setReqCorrelate + , setReqRelaxed + , setRouterMandatory + , setSendBuffer + , setSendHighWM + , setSendTimeout + , setTcpAcceptFilter + , setTcpKeepAlive + , setTcpKeepAliveCount + , setTcpKeepAliveIdle + , setTcpKeepAliveInterval + , setXPubVerbose + + -- * Error Handling + , Z.ZMQError + , Z.errno + , Z.source + , Z.message + + -- * Re-exports + , Control.Monad.IO.Class.liftIO + , Data.Restricted.restrict + , Data.Restricted.toRestricted + + -- * Low-level Functions + , waitRead + , waitWrite + , I.z85Encode + , I.z85Decode + , Z.curveKeyPair + ) where + +import Control.Applicative +import Control.Concurrent.Async (Async) +import Control.Monad +import Control.Monad.Trans.Reader +import Control.Monad.IO.Class +import Control.Monad.Catch +import Control.Monad.CatchIO +import Data.Int +import Data.IORef +import Data.List.NonEmpty (NonEmpty) +import Data.Restricted +import Data.Word +import Data.ByteString (ByteString) +import System.Posix.Types (Fd) + +import qualified Control.Concurrent.Async as A +import qualified Control.Exception as E +import qualified Control.Monad.Catch as C +import qualified Control.Monad.CatchIO as M +import qualified Data.ByteString.Lazy as Lazy +import qualified System.ZMQ4 as Z +import qualified System.ZMQ4.Internal as I + +data ZMQEnv = ZMQEnv + { _refcount :: !(IORef Word) + , _context :: !Z.Context + , _sockets :: !(IORef [I.SocketRepr]) + } + +-- | The ZMQ monad is modeled after 'Control.Monad.ST' and encapsulates +-- a 'System.ZMQ4.Context'. It uses the uninstantiated type variable 'z' to +-- distinguish different invoctions of 'runZMQ' and to prevent +-- unintented use of 'Socket's outside their scope. Cf. the paper +-- of John Launchbury and Simon Peyton Jones /Lazy Functional State Threads/. +newtype ZMQ z a = ZMQ { _unzmq :: ReaderT ZMQEnv IO a } + +-- | The ZMQ socket, parameterised by 'SocketType' and belonging to +-- a particular 'ZMQ' thread. +newtype Socket z t = Socket { _unsocket :: Z.Socket t } + +instance I.SocketLike (Socket z) where + toSocket = _unsocket + +instance Monad (ZMQ z) where + return = ZMQ . return + (ZMQ m) >>= f = ZMQ $ m >>= _unzmq . f + +instance MonadIO (ZMQ z) where + liftIO m = ZMQ $! liftIO m + +instance MonadCatch (ZMQ z) where + throwM = ZMQ . C.throwM + catch (ZMQ m) f = ZMQ $ m `C.catch` (_unzmq . f) + + mask a = ZMQ . ReaderT $ \env -> + C.mask $ \restore -> + let f (ZMQ (ReaderT b)) = ZMQ $ ReaderT (restore . b) + in runReaderT (_unzmq (a $ f)) env + + uninterruptibleMask a = ZMQ . ReaderT $ \env -> + C.uninterruptibleMask $ \restore -> + let f (ZMQ (ReaderT b)) = ZMQ $ ReaderT (restore . b) + in runReaderT (_unzmq (a $ f)) env + +instance MonadCatchIO (ZMQ z) where + catch (ZMQ m) f = ZMQ $ m `M.catch` (_unzmq . f) + block (ZMQ m) = ZMQ $ block m + unblock (ZMQ m) = ZMQ $ unblock m + +instance Functor (ZMQ z) where + fmap = liftM + +instance Applicative (ZMQ z) where + pure = return + (<*>) = ap + +-- | Return the value computed by the given 'ZMQ' monad. Rank-2 +-- polymorphism is used to prevent leaking of 'z'. +-- An invocation of 'runZMQ' will internally create a 'System.ZMQ4.Context' +-- and all actions are executed relative to this context. On finish the +-- context will be disposed, but see 'async'. +runZMQ :: MonadIO m => (forall z. ZMQ z a) -> m a +runZMQ z = liftIO $ E.bracket make term (runReaderT (_unzmq z)) + where + make = ZMQEnv <$> newIORef 1 <*> Z.context <*> newIORef [] + +-- | Run the given 'ZMQ' computation asynchronously, i.e. this function +-- runs the computation in a new thread using 'Control.Concurrent.Async.async'. +-- /N.B./ reference counting is used to prolong the lifetime of the +-- 'System.ZMQ.Context' encapsulated in 'ZMQ' as necessary, e.g.: +-- +-- @ +-- runZMQ $ do +-- s <- socket Pair +-- async $ do +-- liftIO (threadDelay 10000000) +-- identity s >>= liftIO . print +-- @ +-- +-- Here, 'runZMQ' will finish before the code section in 'async', but due to +-- reference counting, the 'System.ZMQ4.Context' will only be disposed after +-- 'async' finishes as well. +async :: ZMQ z a -> ZMQ z (Async a) +async z = ZMQ $ do + e <- ask + liftIO $ atomicModifyIORef (_refcount e) $ \n -> (succ n, ()) + liftIO . A.async $ (runReaderT (_unzmq z) e) `E.finally` term e + +ioThreads :: ZMQ z Word +ioThreads = onContext Z.ioThreads + +setIoThreads :: Word -> ZMQ z () +setIoThreads = onContext . Z.setIoThreads + +maxSockets :: ZMQ z Word +maxSockets = onContext Z.maxSockets + +setMaxSockets :: Word -> ZMQ z () +setMaxSockets = onContext . Z.setMaxSockets + +socket :: Z.SocketType t => t -> ZMQ z (Socket z t) +socket t = ZMQ $ do + c <- asks _context + s <- asks _sockets + x <- liftIO $ I.mkSocketRepr t c + liftIO $ atomicModifyIORef s $ \ss -> (x:ss, ()) + return (Socket (I.Socket x)) + +version :: ZMQ z (Int, Int, Int) +version = liftIO $! Z.version + +-- * Socket operations + +close :: Socket z t -> ZMQ z () +close = liftIO . Z.close . _unsocket + +bind :: Socket z t -> String -> ZMQ z () +bind s = liftIO . Z.bind (_unsocket s) + +unbind :: Socket z t -> String -> ZMQ z () +unbind s = liftIO . Z.unbind (_unsocket s) + +connect :: Socket z t -> String -> ZMQ z () +connect s = liftIO . Z.connect (_unsocket s) + +send :: Z.Sender t => Socket z t -> [Z.Flag] -> ByteString -> ZMQ z () +send s f = liftIO . Z.send (_unsocket s) f + +send' :: Z.Sender t => Socket z t -> [Z.Flag] -> Lazy.ByteString -> ZMQ z () +send' s f = liftIO . Z.send' (_unsocket s) f + +sendMulti :: Z.Sender t => Socket z t -> NonEmpty ByteString -> ZMQ z () +sendMulti s = liftIO . Z.sendMulti (_unsocket s) + +receive :: Z.Receiver t => Socket z t -> ZMQ z ByteString +receive = liftIO . Z.receive . _unsocket + +receiveMulti :: Z.Receiver t => Socket z t -> ZMQ z [ByteString] +receiveMulti = liftIO . Z.receiveMulti . _unsocket + +subscribe :: Z.Subscriber t => Socket z t -> ByteString -> ZMQ z () +subscribe s = liftIO . Z.subscribe (_unsocket s) + +unsubscribe :: Z.Subscriber t => Socket z t -> ByteString -> ZMQ z () +unsubscribe s = liftIO . Z.unsubscribe (_unsocket s) + +proxy :: Socket z a -> Socket z b -> Maybe (Socket z c) -> ZMQ z () +proxy a b c = liftIO $ Z.proxy (_unsocket a) (_unsocket b) (_unsocket <$> c) + +monitor :: [Z.EventType] -> Socket z t -> ZMQ z (Bool -> IO (Maybe Z.EventMsg)) +monitor es s = onContext $ \ctx -> Z.monitor es ctx (_unsocket s) + +-- * Socket Options (Read) + +affinity :: Socket z t -> ZMQ z Word64 +affinity = liftIO . Z.affinity . _unsocket + +backlog :: Socket z t -> ZMQ z Int +backlog = liftIO . Z.backlog . _unsocket + +conflate :: Z.Conflatable t => Socket z t -> ZMQ z Bool +conflate = liftIO . Z.conflate . _unsocket + +curvePublicKey :: Z.KeyFormat f -> Socket z t -> ZMQ z ByteString +curvePublicKey f = liftIO . Z.curvePublicKey f . _unsocket + +curveSecretKey :: Z.KeyFormat f -> Socket z t -> ZMQ z ByteString +curveSecretKey f = liftIO . Z.curveSecretKey f . _unsocket + +curveServerKey :: Z.KeyFormat f -> Socket z t -> ZMQ z ByteString +curveServerKey f = liftIO . Z.curveServerKey f . _unsocket + +delayAttachOnConnect :: Socket z t -> ZMQ z Bool +delayAttachOnConnect = liftIO . Z.delayAttachOnConnect . _unsocket +{-# DEPRECATED delayAttachOnConnect "Use immediate" #-} + +events :: Socket z t -> ZMQ z [Z.Event] +events = liftIO . Z.events . _unsocket + +fileDescriptor :: Socket z t -> ZMQ z Fd +fileDescriptor = liftIO . Z.fileDescriptor . _unsocket + +identity :: Socket z t -> ZMQ z ByteString +identity = liftIO . Z.identity . _unsocket + +immediate :: Socket z t -> ZMQ z Bool +immediate = liftIO . Z.immediate . _unsocket + +ipv4Only :: Socket z t -> ZMQ z Bool +ipv4Only = liftIO . Z.ipv4Only . _unsocket +{-# DEPRECATED ipv4Only "Use ipv6" #-} + +ipv6 :: Socket z t -> ZMQ z Bool +ipv6 = liftIO . Z.ipv6 . _unsocket + +lastEndpoint :: Socket z t -> ZMQ z String +lastEndpoint = liftIO . Z.lastEndpoint . _unsocket + +linger :: Socket z t -> ZMQ z Int +linger = liftIO . Z.linger . _unsocket + +maxMessageSize :: Socket z t -> ZMQ z Int64 +maxMessageSize = liftIO . Z.maxMessageSize . _unsocket + +mcastHops :: Socket z t -> ZMQ z Int +mcastHops = liftIO . Z.mcastHops . _unsocket + +mechanism :: Socket z t -> ZMQ z Z.SecurityMechanism +mechanism = liftIO . Z.mechanism . _unsocket + +moreToReceive :: Socket z t -> ZMQ z Bool +moreToReceive = liftIO . Z.moreToReceive . _unsocket + +plainServer :: Socket z t -> ZMQ z Bool +plainServer = liftIO . Z.plainServer . _unsocket + +plainPassword :: Socket z t -> ZMQ z ByteString +plainPassword = liftIO . Z.plainPassword . _unsocket + +plainUserName :: Socket z t -> ZMQ z ByteString +plainUserName = liftIO . Z.plainUserName . _unsocket + +rate :: Socket z t -> ZMQ z Int +rate = liftIO . Z.rate . _unsocket + +receiveBuffer :: Socket z t -> ZMQ z Int +receiveBuffer = liftIO . Z.receiveBuffer . _unsocket + +receiveHighWM :: Socket z t -> ZMQ z Int +receiveHighWM = liftIO . Z.receiveHighWM . _unsocket + +receiveTimeout :: Socket z t -> ZMQ z Int +receiveTimeout = liftIO . Z.receiveTimeout . _unsocket + +reconnectInterval :: Socket z t -> ZMQ z Int +reconnectInterval = liftIO . Z.reconnectInterval . _unsocket + +reconnectIntervalMax :: Socket z t -> ZMQ z Int +reconnectIntervalMax = liftIO . Z.reconnectIntervalMax . _unsocket + +recoveryInterval :: Socket z t -> ZMQ z Int +recoveryInterval = liftIO . Z.recoveryInterval . _unsocket + +sendBuffer :: Socket z t -> ZMQ z Int +sendBuffer = liftIO . Z.sendBuffer . _unsocket + +sendHighWM :: Socket z t -> ZMQ z Int +sendHighWM = liftIO . Z.sendHighWM . _unsocket + +sendTimeout :: Socket z t -> ZMQ z Int +sendTimeout = liftIO . Z.sendTimeout . _unsocket + +tcpKeepAlive :: Socket z t -> ZMQ z Z.Switch +tcpKeepAlive = liftIO . Z.tcpKeepAlive . _unsocket + +tcpKeepAliveCount :: Socket z t -> ZMQ z Int +tcpKeepAliveCount = liftIO . Z.tcpKeepAliveCount . _unsocket + +tcpKeepAliveIdle :: Socket z t -> ZMQ z Int +tcpKeepAliveIdle = liftIO . Z.tcpKeepAliveIdle . _unsocket + +tcpKeepAliveInterval :: Socket z t -> ZMQ z Int +tcpKeepAliveInterval = liftIO . Z.tcpKeepAliveInterval . _unsocket + +zapDomain :: Socket z t -> ZMQ z ByteString +zapDomain = liftIO . Z.zapDomain . _unsocket + +-- * Socket Options (Write) + +setAffinity :: Word64 -> Socket z t -> ZMQ z () +setAffinity a = liftIO . Z.setAffinity a . _unsocket + +setBacklog :: Integral i => Restricted (N0, Int32) i -> Socket z t -> ZMQ z () +setBacklog b = liftIO . Z.setBacklog b . _unsocket + +setConflate :: Z.Conflatable t => Bool -> Socket z t -> ZMQ z () +setConflate i = liftIO . Z.setConflate i . _unsocket + +setCurvePublicKey :: Z.KeyFormat f -> Restricted f ByteString -> Socket z t -> ZMQ z () +setCurvePublicKey f k = liftIO . Z.setCurvePublicKey f k . _unsocket + +setCurveSecretKey :: Z.KeyFormat f -> Restricted f ByteString -> Socket z t -> ZMQ z () +setCurveSecretKey f k = liftIO . Z.setCurveSecretKey f k . _unsocket + +setCurveServer :: Bool -> Socket z t -> ZMQ z () +setCurveServer b = liftIO . Z.setCurveServer b . _unsocket + +setCurveServerKey :: Z.KeyFormat f -> Restricted f ByteString -> Socket z t -> ZMQ z () +setCurveServerKey f k = liftIO . Z.setCurveServerKey f k . _unsocket + +setDelayAttachOnConnect :: Bool -> Socket z t -> ZMQ z () +setDelayAttachOnConnect d = liftIO . Z.setDelayAttachOnConnect d . _unsocket +{-# DEPRECATED setDelayAttachOnConnect "Use setImmediate" #-} + +setIdentity :: Restricted (N1, N254) ByteString -> Socket z t -> ZMQ z () +setIdentity i = liftIO . Z.setIdentity i . _unsocket + +setImmediate :: Bool -> Socket z t -> ZMQ z () +setImmediate i = liftIO . Z.setImmediate i . _unsocket + +setIpv4Only :: Bool -> Socket z t -> ZMQ z () +setIpv4Only i = liftIO . Z.setIpv4Only i . _unsocket +{-# DEPRECATED setIpv4Only "Use setIpv6" #-} + +setIpv6 :: Bool -> Socket z t -> ZMQ z () +setIpv6 i = liftIO . Z.setIpv6 i . _unsocket + +setLinger :: Integral i => Restricted (Nneg1, Int32) i -> Socket z t -> ZMQ z () +setLinger l = liftIO . Z.setLinger l . _unsocket + +setMaxMessageSize :: Integral i => Restricted (Nneg1, Int64) i -> Socket z t -> ZMQ z () +setMaxMessageSize s = liftIO . Z.setMaxMessageSize s . _unsocket + +setMcastHops :: Integral i => Restricted (N1, Int32) i -> Socket z t -> ZMQ z () +setMcastHops k = liftIO . Z.setMcastHops k . _unsocket + +setPlainServer :: Bool -> Socket z t -> ZMQ z () +setPlainServer b = liftIO . Z.setPlainServer b . _unsocket + +setPlainPassword :: Restricted (N1, N254) ByteString -> Socket z t -> ZMQ z () +setPlainPassword s = liftIO . Z.setPlainPassword s . _unsocket + +setPlainUserName :: Restricted (N1, N254) ByteString -> Socket z t -> ZMQ z () +setPlainUserName s = liftIO . Z.setPlainUserName s . _unsocket + +setProbeRouter :: Z.SendProbe t => Bool -> Socket z t -> ZMQ z () +setProbeRouter b = liftIO . Z.setProbeRouter b . _unsocket + +setRate :: Integral i => Restricted (N1, Int32) i -> Socket z t -> ZMQ z () +setRate r = liftIO . Z.setRate r . _unsocket + +setReceiveBuffer :: Integral i => Restricted (N0, Int32) i -> Socket z t -> ZMQ z () +setReceiveBuffer k = liftIO . Z.setReceiveBuffer k . _unsocket + +setReceiveHighWM :: Integral i => Restricted (N0, Int32) i -> Socket z t -> ZMQ z () +setReceiveHighWM k = liftIO . Z.setReceiveHighWM k . _unsocket + +setReceiveTimeout :: Integral i => Restricted (Nneg1, Int32) i -> Socket z t -> ZMQ z () +setReceiveTimeout t = liftIO . Z.setReceiveTimeout t . _unsocket + +setReconnectInterval :: Integral i => Restricted (N0, Int32) i -> Socket z t -> ZMQ z () +setReconnectInterval i = liftIO . Z.setReconnectInterval i . _unsocket + +setReconnectIntervalMax :: Integral i => Restricted (N0, Int32) i -> Socket z t -> ZMQ z () +setReconnectIntervalMax i = liftIO . Z.setReconnectIntervalMax i . _unsocket + +setRecoveryInterval :: Integral i => Restricted (N0, Int32) i -> Socket z t -> ZMQ z () +setRecoveryInterval i = liftIO . Z.setRecoveryInterval i . _unsocket + +setReqCorrelate :: Bool -> Socket z Z.Req -> ZMQ z () +setReqCorrelate b = liftIO . Z.setReqCorrelate b . _unsocket + +setReqRelaxed :: Bool -> Socket z Z.Req -> ZMQ z () +setReqRelaxed b = liftIO . Z.setReqRelaxed b . _unsocket + +setRouterMandatory :: Bool -> Socket z Z.Router -> ZMQ z () +setRouterMandatory b = liftIO . Z.setRouterMandatory b . _unsocket + +setSendBuffer :: Integral i => Restricted (N0, Int32) i -> Socket z t -> ZMQ z () +setSendBuffer i = liftIO . Z.setSendBuffer i . _unsocket + +setSendHighWM :: Integral i => Restricted (N0, Int32) i -> Socket z t -> ZMQ z () +setSendHighWM i = liftIO . Z.setSendHighWM i . _unsocket + +setSendTimeout :: Integral i => Restricted (Nneg1, Int32) i -> Socket z t -> ZMQ z () +setSendTimeout i = liftIO . Z.setSendTimeout i . _unsocket + +setTcpAcceptFilter :: Maybe ByteString -> Socket z t -> ZMQ z () +setTcpAcceptFilter s = liftIO . Z.setTcpAcceptFilter s . _unsocket + +setTcpKeepAlive :: Z.Switch -> Socket z t -> ZMQ z () +setTcpKeepAlive s = liftIO . Z.setTcpKeepAlive s . _unsocket + +setTcpKeepAliveCount :: Integral i => Restricted (Nneg1, Int32) i -> Socket z t -> ZMQ z () +setTcpKeepAliveCount c = liftIO . Z.setTcpKeepAliveCount c . _unsocket + +setTcpKeepAliveIdle :: Integral i => Restricted (Nneg1, Int32) i -> Socket z t -> ZMQ z () +setTcpKeepAliveIdle i = liftIO . Z.setTcpKeepAliveIdle i . _unsocket + +setTcpKeepAliveInterval :: Integral i => Restricted (Nneg1, Int32) i -> Socket z t -> ZMQ z () +setTcpKeepAliveInterval i = liftIO . Z.setTcpKeepAliveInterval i . _unsocket + +setXPubVerbose :: Bool -> Socket z Z.XPub -> ZMQ z () +setXPubVerbose b = liftIO . Z.setXPubVerbose b . _unsocket + +-- * Low Level Functions + +waitRead :: Socket z t -> ZMQ z () +waitRead = liftIO . Z.waitRead . _unsocket + +waitWrite :: Socket z t -> ZMQ z () +waitWrite = liftIO . Z.waitWrite . _unsocket + +-- * Internal + +onContext :: (Z.Context -> IO a) -> ZMQ z a +onContext f = ZMQ $! asks _context >>= liftIO . f + +term :: ZMQEnv -> IO () +term env = do + n <- atomicModifyIORef (_refcount env) $ \n -> (pred n, n) + when (n == 1) $ do + readIORef (_sockets env) >>= mapM_ close' + Z.term (_context env) + where + close' s = I.closeSock s `E.catch` (\e -> print (e :: E.SomeException)) diff --git a/tests/System/ZMQ4/Test/Properties.hs b/tests/System/ZMQ4/Test/Properties.hs new file mode 100644 --- /dev/null +++ b/tests/System/ZMQ4/Test/Properties.hs @@ -0,0 +1,194 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE OverloadedStrings #-} +{-# OPTIONS_GHC -fno-warn-orphans #-} + +module System.ZMQ4.Test.Properties where + +import Test.QuickCheck +import Test.QuickCheck.Monadic +import Test.Tools + +import Control.Applicative +import Control.Concurrent.Async (wait) +import Data.Int +import Data.Word +import Data.Restricted +import Data.Maybe (fromJust) +import Data.ByteString (ByteString) +import System.ZMQ4.Monadic +import System.Posix.Types (Fd(..)) +import qualified Data.ByteString as SB +import qualified Data.ByteString.Char8 as CB + +tests :: IO () +tests = do + quickBatch' ("0MQ Socket Properties" + , [ ("get socket option (Pair)", property $ prop_get_socket_option Pair) + , ("get socket option (Pub)", property $ prop_get_socket_option Pub) + , ("get socket option (Sub)", property $ prop_get_socket_option Sub) + , ("get socket option (XPub)", property $ prop_get_socket_option XPub) + , ("get socket option (XSub)", property $ prop_get_socket_option XSub) + , ("get socket option (Req)", property $ prop_get_socket_option Req) + , ("get socket option (Rep)", property $ prop_get_socket_option Rep) + , ("get socket option (Dealer)", property $ prop_get_socket_option Dealer) + , ("get socket option (Router)", property $ prop_get_socket_option Router) + , ("get socket option (Pull)", property $ prop_get_socket_option Pull) + , ("get socket option (Push)", property $ prop_get_socket_option Push) + , ("set;get socket option (Pair)", property $ prop_set_get_socket_option Pair) + , ("set;get socket option (Pub)", property $ prop_set_get_socket_option Pub) + , ("set;get socket option (Sub)", property $ prop_set_get_socket_option Sub) + , ("set;get socket option (XPub)", property $ prop_set_get_socket_option XPub) + , ("set;get socket option (XSub)", property $ prop_set_get_socket_option XSub) + , ("set;get socket option (Req)", property $ prop_set_get_socket_option Req) + , ("set;get socket option (Rep)", property $ prop_set_get_socket_option Rep) + , ("set;get socket option (Dealer)", property $ prop_set_get_socket_option Dealer) + , ("set;get socket option (Router)", property $ prop_set_get_socket_option Router) + , ("set;get socket option (Pull)", property $ prop_set_get_socket_option Pull) + , ("set;get socket option (Push)", property $ prop_set_get_socket_option Push) + , ("(un-)subscribe", property $ prop_subscribe Sub) + ]) + + quickBatch' ("0MQ Messages" + , [ ("msg send == msg received (Req/Rep)", property $ prop_send_receive Req Rep) + , ("msg send == msg received (Push/Pull)", property $ prop_send_receive Push Pull) + , ("msg send == msg received (Pair/Pair)", property $ prop_send_receive Pair Pair) + + -- , ("publish/subscribe", property $ prop_pub_sub Pub Sub) + -- (disabled due to LIBZMQ-270 [https://zeromq.jira.com/browse/LIBZMQ-270]) + ]) + +prop_get_socket_option :: SocketType t => t -> GetOpt -> Property +prop_get_socket_option t opt = monadicIO $ run $ do + runZMQ $ do + s <- socket t + case opt of + Events _ -> events s >> return () + Filedesc _ -> fileDescriptor s >> return () + ReceiveMore _ -> moreToReceive s >> return () + +prop_set_get_socket_option :: SocketType t => t -> SetOpt -> Property +prop_set_get_socket_option t opt = monadicIO $ do + r <- run $ runZMQ $ do + s <- socket t + case opt of + Identity val -> (== (rvalue val)) <$> (setIdentity val s >> identity s) + Ipv4Only val -> (== val) <$> (setIpv4Only val s >> ipv4Only s) + Affinity val -> (ieq val) <$> (setAffinity val s >> affinity s) + Backlog val -> (ieq (rvalue val)) <$> (setBacklog val s >> backlog s) + Linger val -> (ieq (rvalue val)) <$> (setLinger val s >> linger s) + Rate val -> (ieq (rvalue val)) <$> (setRate val s >> rate s) + ReceiveBuf val -> (ieq (rvalue val)) <$> (setReceiveBuffer val s >> receiveBuffer s) + ReconnectIVL val -> (ieq (rvalue val)) <$> (setReconnectInterval val s >> reconnectInterval s) + ReconnectIVLMax val -> (ieq (rvalue val)) <$> (setReconnectIntervalMax val s >> reconnectIntervalMax s) + RecoveryIVL val -> (ieq (rvalue val)) <$> (setRecoveryInterval val s >> recoveryInterval s) + SendBuf val -> (ieq (rvalue val)) <$> (setSendBuffer val s >> sendBuffer s) + MaxMessageSize val -> (ieq (rvalue val)) <$> (setMaxMessageSize val s >> maxMessageSize s) + McastHops val -> (ieq (rvalue val)) <$> (setMcastHops val s >> mcastHops s) + ReceiveHighWM val -> (ieq (rvalue val)) <$> (setReceiveHighWM val s >> receiveHighWM s) + ReceiveTimeout val -> (ieq (rvalue val)) <$> (setReceiveTimeout val s >> receiveTimeout s) + SendHighWM val -> (ieq (rvalue val)) <$> (setSendHighWM val s >> sendHighWM s) + SendTimeout val -> (ieq (rvalue val)) <$> (setSendTimeout val s >> sendTimeout s) + assert r + where + ieq :: (Integral i, Integral k) => i -> k -> Bool + ieq i k = (fromIntegral i :: Int) == (fromIntegral k :: Int) + +prop_subscribe :: (Subscriber a, SocketType a) => a -> ByteString -> Property +prop_subscribe t subs = monadicIO $ run $ + runZMQ $ do + s <- socket t + subscribe s subs + unsubscribe s subs + +prop_send_receive :: (SocketType a, SocketType b, Receiver b, Sender a) => a -> b -> ByteString -> Property +prop_send_receive a b msg = monadicIO $ do + msg' <- run $ runZMQ $ do + sender <- socket a + receiver <- socket b + bind receiver "inproc://endpoint" + x <- async $ receive receiver + connect sender "inproc://endpoint" + send sender [] msg + liftIO $ wait x + assert (msg == msg') + +prop_pub_sub :: (SocketType a, Subscriber b, SocketType b, Sender a, Receiver b) => a -> b -> ByteString -> Property +prop_pub_sub a b msg = monadicIO $ do + msg' <- run $ runZMQ $ do + pub <- socket a + sub <- socket b + subscribe sub "" + bind sub "inproc://endpoint" + connect pub "inproc://endpoint" + send pub [] msg + receive sub + assert (msg == msg') + +instance Arbitrary ByteString where + arbitrary = CB.pack . filter (/= '\0') <$> arbitrary + +data GetOpt = + Events Int + | Filedesc Fd + | ReceiveMore Bool + deriving Show + +data SetOpt = + Affinity Word64 + | Backlog (Restricted (N0, Int32) Int) + | Identity (Restricted (N1, N254) ByteString) + | Ipv4Only Bool + | Linger (Restricted (Nneg1, Int32) Int) + | MaxMessageSize (Restricted (Nneg1, Int64) Int64) + | McastHops (Restricted (N1, Int32) Int) + | Rate (Restricted (N1, Int32) Int) + | ReceiveBuf (Restricted (N0, Int32) Int) + | ReceiveHighWM (Restricted (N0, Int32) Int) + | ReceiveTimeout (Restricted (Nneg1, Int32) Int) + | ReconnectIVL (Restricted (N0, Int32) Int) + | ReconnectIVLMax (Restricted (N0, Int32) Int) + | RecoveryIVL (Restricted (N0, Int32) Int) + | SendBuf (Restricted (N0, Int32) Int) + | SendHighWM (Restricted (N0, Int32) Int) + | SendTimeout (Restricted (Nneg1, Int32) Int) + deriving Show + +instance Arbitrary GetOpt where + arbitrary = oneof [ + Events <$> arbitrary + , Filedesc . Fd . fromIntegral <$> (arbitrary :: Gen Int32) + , ReceiveMore <$> arbitrary + ] + +instance Arbitrary SetOpt where + arbitrary = oneof [ + Affinity <$> (arbitrary :: Gen Word64) + , Ipv4Only <$> (arbitrary :: Gen Bool) + , Backlog . toR0 <$> (arbitrary :: Gen Int32) `suchThat` (>= 0) + , Linger . toRneg1 <$> (arbitrary :: Gen Int32) `suchThat` (>= -1) + , Rate . toR1 <$> (arbitrary :: Gen Int32) `suchThat` (> 0) + , ReceiveBuf . toR0 <$> (arbitrary :: Gen Int32) `suchThat` (>= 0) + , ReconnectIVL . toR0 <$> (arbitrary :: Gen Int32) `suchThat` (>= 0) + , ReconnectIVLMax . toR0 <$> (arbitrary :: Gen Int32) `suchThat` (>= 0) + , RecoveryIVL . toR0 <$> (arbitrary :: Gen Int32) `suchThat` (>= 0) + , SendBuf . toR0 <$> (arbitrary :: Gen Int32) `suchThat` (>= 0) + , McastHops . toR1 <$> (arbitrary :: Gen Int32) `suchThat` (> 0) + , ReceiveHighWM . toR0 <$> (arbitrary :: Gen Int32) `suchThat` (>= 0) + , ReceiveTimeout . toRneg1 <$> (arbitrary :: Gen Int32) `suchThat` (>= -1) + , SendHighWM . toR0 <$> (arbitrary :: Gen Int32) `suchThat` (>= 0) + , SendTimeout . toRneg1 <$> (arbitrary :: Gen Int32) `suchThat` (>= -1) + , MaxMessageSize . toRneg1' <$> (arbitrary :: Gen Int64) `suchThat` (>= -1) + , Identity . fromJust . toRestricted <$> arbitrary `suchThat` (\s -> SB.length s > 0 && SB.length s < 255) + ] + +toR1 :: Int32 -> Restricted (N1, Int32) Int +toR1 = fromJust . toRestricted . fromIntegral + +toR0 :: Int32 -> Restricted (N0, Int32) Int +toR0 = fromJust . toRestricted . fromIntegral + +toRneg1 :: Int32 -> Restricted (Nneg1, Int32) Int +toRneg1 = fromJust . toRestricted . fromIntegral + +toRneg1' :: Int64 -> Restricted (Nneg1, Int64) Int64 +toRneg1' = fromJust . toRestricted . fromIntegral diff --git a/tests/Test/Tools.hs b/tests/Test/Tools.hs new file mode 100644 --- /dev/null +++ b/tests/Test/Tools.hs @@ -0,0 +1,35 @@ +module Test.Tools (quickBatch', checkBatch') where + +import Control.Monad +import System.Console.ANSI +import System.Exit +import Test.QuickCheck +import Test.QuickCheck.Checkers +import qualified Control.Exception as E + +quickBatch' :: TestBatch -> IO () +quickBatch' = checkBatch' (stdArgs { maxSuccess = 500 }) + +checkBatch' :: Args -> TestBatch -> IO () +checkBatch' args (name, tsts) = do + writeLn Cyan name + forM_ tsts $ \(s, p) -> do + write White (" " ++ s ++ ": ") + r <- quickCheckWithResult (args { chatty = False}) p + `E.catch` + ((\e -> write Red (show e) >> exitFailure) :: E.SomeException -> IO a) + case r of + Success _ _ m -> write Green m + GaveUp _ _ m -> write Magenta m >> exitFailure + Failure _ _ _ _ _ _ _ m -> write Red m >> exitFailure + NoExpectedFailure _ _ m -> write Red m >> exitFailure + +write, writeLn :: Color -> String -> IO () +write c = withColour c . putStr +writeLn c = withColour c . putStrLn + +withColour :: Color -> IO () -> IO () +withColour c a = do + setSGR [Reset, SetColor Foreground Vivid c] + a + setSGR [Reset] diff --git a/tests/tests.hs b/tests/tests.hs new file mode 100644 --- /dev/null +++ b/tests/tests.hs @@ -0,0 +1,5 @@ +import qualified System.ZMQ4.Test.Properties as Properties + +main :: IO () +main = Properties.tests + diff --git a/zeromq4-haskell.cabal b/zeromq4-haskell.cabal new file mode 100644 --- /dev/null +++ b/zeromq4-haskell.cabal @@ -0,0 +1,85 @@ +name: zeromq4-haskell +version: 0.1 +synopsis: Bindings to ZeroMQ 4.x +category: System, FFI +license: MIT +license-file: LICENSE +author: Toralf Wittner +maintainer: Toralf Wittner +copyright: Copyright (c) 2010 - 2013 zeromq-haskell authors +homepage: http://github.com/twittner/zeromq-haskell/ +stability: experimental +tested-With: GHC == 7.6.3 +cabal-version: >= 1.8 +build-type: Simple +extra-source-files: + README.md + , AUTHORS + , examples/*.hs + , examples/Makefile + , examples/perf/*.hs + , examples/perf/Makefile + , tests/*.hs + , tests/System/ZMQ4/Test/*.hs + +description: + The 0MQ lightweight messaging kernel is a library which extends + the standard socket interfaces with features traditionally provided + by specialised messaging middleware products. + . + 0MQ sockets provide an abstraction of asynchronous message queues, + multiple messaging patterns, message filtering (subscriptions), + seamless access to multiple transport protocols and more. + . + This library provides the Haskell language binding to 0MQ >= 4.x + +source-repository head + type: git + location: https://github.com/twittner/zeromq-haskell + +library + hs-source-dirs: src + exposed-modules: + System.ZMQ4 + System.ZMQ4.Monadic + Data.Restricted + + other-modules: + System.ZMQ4.Base + System.ZMQ4.Internal + System.ZMQ4.Error + + includes: zmq.h + ghc-options: -Wall -O2 -fwarn-tabs -funbox-strict-fields + build-depends: + base >= 3 && < 5 + , async == 2.0.* + , bytestring + , containers + , exceptions >= 0.3 + , MonadCatchIO-transformers + , semigroups >= 0.8 + , transformers >= 0.3 + + if os(freebsd) + extra-libraries: zmq, pthread + else + extra-libraries: zmq + +test-suite zeromq-haskell-tests + type: exitcode-stdio-1.0 + hs-source-dirs: tests + main-is: tests.hs + other-modules: Test.Tools + ghc-options: -Wall -threaded -rtsopts + build-depends: + zeromq4-haskell + , ansi-terminal >= 0.6 + , async == 2.0.* + , base >= 3 && < 5 + , bytestring + , checkers >= 0.3 + , containers + , MonadCatchIO-transformers + , QuickCheck >= 2.6 + , transformers