zeromq-haskell (empty) → 0.1
raw patch · 13 files changed
+977/−0 lines, 13 filesdep +basedep +bytestringdep +containerssetup-changed
Dependencies added: base, bytestring, containers, unix
Files
- LICENSE +165/−0
- README +60/−0
- Setup.hs +6/−0
- src/System/ZMQ.hs +321/−0
- src/System/ZMQ/Base.hsc +165/−0
- test/display.hs +23/−0
- test/perf/local_lat.hs +33/−0
- test/perf/local_thr.hs +51/−0
- test/perf/remote_lat.hs +38/−0
- test/perf/remote_thr.hs +29/−0
- test/poll.hs +31/−0
- test/prompt.hs +25/−0
- zeromq-haskell.cabal +30/−0
+ LICENSE view
@@ -0,0 +1,165 @@+ GNU LESSER GENERAL PUBLIC LICENSE+ Version 3, 29 June 2007++ Copyright (C) 2007 Free Software Foundation, Inc. <http://fsf.org/>+ Everyone is permitted to copy and distribute verbatim copies+ of this license document, but changing it is not allowed.+++ This version of the GNU Lesser General Public License incorporates+the terms and conditions of version 3 of the GNU General Public+License, supplemented by the additional permissions listed below.++ 0. Additional Definitions.++ As used herein, "this License" refers to version 3 of the GNU Lesser+General Public License, and the "GNU GPL" refers to version 3 of the GNU+General Public License.++ "The Library" refers to a covered work governed by this License,+other than an Application or a Combined Work as defined below.++ An "Application" is any work that makes use of an interface provided+by the Library, but which is not otherwise based on the Library.+Defining a subclass of a class defined by the Library is deemed a mode+of using an interface provided by the Library.++ A "Combined Work" is a work produced by combining or linking an+Application with the Library. The particular version of the Library+with which the Combined Work was made is also called the "Linked+Version".++ The "Minimal Corresponding Source" for a Combined Work means the+Corresponding Source for the Combined Work, excluding any source code+for portions of the Combined Work that, considered in isolation, are+based on the Application, and not on the Linked Version.++ The "Corresponding Application Code" for a Combined Work means the+object code and/or source code for the Application, including any data+and utility programs needed for reproducing the Combined Work from the+Application, but excluding the System Libraries of the Combined Work.++ 1. Exception to Section 3 of the GNU GPL.++ You may convey a covered work under sections 3 and 4 of this License+without being bound by section 3 of the GNU GPL.++ 2. Conveying Modified Versions.++ If you modify a copy of the Library, and, in your modifications, a+facility refers to a function or data to be supplied by an Application+that uses the facility (other than as an argument passed when the+facility is invoked), then you may convey a copy of the modified+version:++ a) under this License, provided that you make a good faith effort to+ ensure that, in the event an Application does not supply the+ function or data, the facility still operates, and performs+ whatever part of its purpose remains meaningful, or++ b) under the GNU GPL, with none of the additional permissions of+ this License applicable to that copy.++ 3. Object Code Incorporating Material from Library Header Files.++ The object code form of an Application may incorporate material from+a header file that is part of the Library. You may convey such object+code under terms of your choice, provided that, if the incorporated+material is not limited to numerical parameters, data structure+layouts and accessors, or small macros, inline functions and templates+(ten or fewer lines in length), you do both of the following:++ a) Give prominent notice with each copy of the object code that the+ Library is used in it and that the Library and its use are+ covered by this License.++ b) Accompany the object code with a copy of the GNU GPL and this license+ document.++ 4. Combined Works.++ You may convey a Combined Work under terms of your choice that,+taken together, effectively do not restrict modification of the+portions of the Library contained in the Combined Work and reverse+engineering for debugging such modifications, if you also do each of+the following:++ a) Give prominent notice with each copy of the Combined Work that+ the Library is used in it and that the Library and its use are+ covered by this License.++ b) Accompany the Combined Work with a copy of the GNU GPL and this license+ document.++ c) For a Combined Work that displays copyright notices during+ execution, include the copyright notice for the Library among+ these notices, as well as a reference directing the user to the+ copies of the GNU GPL and this license document.++ d) Do one of the following:++ 0) Convey the Minimal Corresponding Source under the terms of this+ License, and the Corresponding Application Code in a form+ suitable for, and under terms that permit, the user to+ recombine or relink the Application with a modified version of+ the Linked Version to produce a modified Combined Work, in the+ manner specified by section 6 of the GNU GPL for conveying+ Corresponding Source.++ 1) Use a suitable shared library mechanism for linking with the+ Library. A suitable mechanism is one that (a) uses at run time+ a copy of the Library already present on the user's computer+ system, and (b) will operate properly with a modified version+ of the Library that is interface-compatible with the Linked+ Version.++ e) Provide Installation Information, but only if you would otherwise+ be required to provide such information under section 6 of the+ GNU GPL, and only to the extent that such information is+ necessary to install and execute a modified version of the+ Combined Work produced by recombining or relinking the+ Application with a modified version of the Linked Version. (If+ you use option 4d0, the Installation Information must accompany+ the Minimal Corresponding Source and Corresponding Application+ Code. If you use option 4d1, you must provide the Installation+ Information in the manner specified by section 6 of the GNU GPL+ for conveying Corresponding Source.)++ 5. Combined Libraries.++ You may place library facilities that are a work based on the+Library side by side in a single library together with other library+facilities that are not Applications and are not covered by this+License, and convey such a combined library under terms of your+choice, if you do both of the following:++ a) Accompany the combined library with a copy of the same work based+ on the Library, uncombined with any other library facilities,+ conveyed under the terms of this License.++ b) Give prominent notice with the combined library that part of it+ is a work based on the Library, and explaining where to find the+ accompanying uncombined form of the same work.++ 6. Revised Versions of the GNU Lesser General Public License.++ The Free Software Foundation may publish revised and/or new versions+of the GNU Lesser General Public License from time to time. Such new+versions will be similar in spirit to the present version, but may+differ in detail to address new problems or concerns.++ Each version is given a distinguishing version number. If the+Library as you received it specifies that a certain numbered version+of the GNU Lesser General Public License "or any later version"+applies to it, you have the option of following the terms and+conditions either of that published version or of any later version+published by the Free Software Foundation. If the Library as you+received it does not specify a version number of the GNU Lesser+General Public License, you may choose any version of the GNU Lesser+General Public License ever published by the Free Software Foundation.++ If the Library as you received it specifies that a proxy can decide+whether future versions of the GNU Lesser General Public License shall+apply, that proxy's public statement of acceptance of any version is+permanent authorization for you to choose that version for the+Library.
+ README view
@@ -0,0 +1,60 @@+This library provides Haskell bindings to zeromq (http://zeromq.org).+++· Current status++Version 0.1 - This software currently has *alpha* status, i.e. it had+only seen very limited testing and changes to its API might happen.++This software was developed and tested on Linux 2.6.31 with GHC-6.12.1+using the current git HEAD of zeromq2 (http://github.com/sustrik/zeromq2).+++· Installation++As usual for Haskell packages this software is installed best via Cabal+(http://www.haskell.org/cabal). In addition to GHC it depends on 0MQ of course.+++· Usage++The API mostly follows 0MQ's. Public functions are:++- init+- term+- socket+- close+- setOption+- subscribe+- unsubscribe+- bind+- connect+- send+- flush+- receive+- poll++One difference to 0MQ's API is that sockets are parameterized types, i.e. there+is not one socket type but when creating a socket the desired socket type has+to be specified, e.g. 'P2P' and the resulting socket is of type 'Socket P2P'.+This additional type information is used to ensure that only options applicable+to the socket type can be set, hence ZMQ_SUBSCRIBE and ZMQ_UNSUBSCRIBE which +only apply to ZMQ_SUB sockets have their own functions ('subscribe' and+'unsubscribe') which can only be used with sockets of type 'Socket Sub'.++Other differences are mostly for convenience. Also one does not deal directly+with 0MQ messages, instead these are created internally as needed.+++· Examples++The test folder contains some simple tests mostly mimicking the ones that come+with 0MQ.+++· Bugs++If you find any bugs or other shortcomings I would greatly appreciate a bug+report, preferably via http://github.com/twittner/zeromq-haskell/issues or e-mail+to toralf.wittner@gmail.com+
+ Setup.hs view
@@ -0,0 +1,6 @@+module Main where++import Distribution.Simple++main = defaultMain+
+ src/System/ZMQ.hs view
@@ -0,0 +1,321 @@+{-# LANGUAGE ExistentialQuantification #-}+-- |+-- Module : System.ZMQ+-- Copyright : (c) 2010 Toralf Wittner+-- License : LGPL+-- Maintainer : toralf.wittner@gmail.com+-- Stability : experimental+-- Portability : non-portable+-- ++module System.ZMQ (++ Size,+ Context,+ Socket,+ Flag(..),+ SocketOption(..),+ Poll(..),+ PollEvent(..),++ P2P(..),+ Pub(..),+ Sub(..),+ Req(..),+ Rep(..),+ XReq(..),+ XRep(..),+ Up(..),+ Down(..),++ init,+ term,++ socket,+ close,+ setOption,+ System.ZMQ.subscribe,+ System.ZMQ.unsubscribe,+ bind,+ connect,+ send,+ flush,+ receive,++ poll++) where++import Prelude hiding (init)+import Control.Applicative+import Control.Exception+import Data.Int+import Data.Maybe+import System.ZMQ.Base+import qualified System.ZMQ.Base as B+import Foreign+import Foreign.C.Error+import Foreign.C.String+import Foreign.C.Types (CInt, CShort)+import qualified Data.ByteString as SB+import qualified Data.ByteString.Unsafe as UB+import System.Posix.Types (Fd(..))++-- ^ A 0MQ context representation.+newtype Context = Context { ctx :: ZMQCtx }++-- ^ A 0MQ Socket.+newtype Socket a = Socket { sock :: ZMQSocket }++-- A 0MQ Message representation.+newtype Message = Message { msgPtr :: ZMQMsgPtr }++type Timeout = Int64+type Size = Word++-- Socket types:++class SType a where+ zmqSocketType :: a -> ZMQSocketType++data P2P = P2P+instance SType P2P where+ zmqSocketType = const p2p++data Pub = Pub+instance SType Pub where+ zmqSocketType = const pub++data Sub = Sub+instance SType Sub where+ zmqSocketType = const sub++data Req = Req+instance SType Req where+ zmqSocketType = const request++data Rep = Rep+instance SType Rep where+ zmqSocketType = const response++data XReq = Xreq+instance SType XReq where+ zmqSocketType = const xrequest++data XRep = Xrep+instance SType XRep where+ zmqSocketType = const xresponse++data Up = Up+instance SType Up where+ zmqSocketType = const upstream++data Down = Down+instance SType Down where+ zmqSocketType = const downstream++-- Subscribable:++class SubsType a+instance SubsType Sub++-- ^ The option to set on 0MQ sockets (cf. man zmq_setsockopt)+data SocketOption =+ HighWM Int64 -- ^ ZMQ_HWM+ | LowWM Int64 -- ^ ZMQ_LWM+ | Swap Int64 -- ^ ZMQ_SWAP+ | Affinity Int64 -- ^ ZMQ_AFFINITY+ | Identity String -- ^ ZMQ_IDENTITY+ | Rate Word64 -- ^ ZMQ_RATE+ | RecoveryIVL Word64 -- ^ ZMQ_RECOVERY_IVL+ | McastLoop Word64 -- ^ ZMQ_MCAST_LOOP+ | SendBuf Word64 -- ^ ZMQ_SNDBUF+ | ReceiveBuf Word64 -- ^ ZMQ_RCVBUF+ deriving (Eq, Ord, Show)++-- ^ Flags to apply on send operations (cf. man zmq_send)+data Flag =+ NoBlock -- ^ ZMQ_NOBLOCK+ | NoFlush -- ^ ZMQ_NOFLUSH+ deriving (Eq, Ord, Show)++-- ^ The events to wait for in poll (cf. man zmq_poll)+data PollEvent =+ In -- ^ ZMQ_POLLIN+ | Out -- ^ ZMQ_POLLOUT+ | InOut -- ^ ZMQ_POLLIN | ZMQ_POLLOUT+ deriving (Eq, Ord, Show)++-- ^ Type representing a descriptor, poll is waiting for+-- (either a 0MQ socket or a file descriptor) plus the type+-- of event of wait for.+data Poll =+ forall a. S (Socket a) PollEvent+ | F Fd PollEvent++-- ^ Initialize a 0MQ context (cf. zmq_init).+init :: Size -> Size -> Bool -> IO Context+init appThreads ioThreads doPoll = do+ c <- throwErrnoIfNull "init" $+ c_zmq_init (fromIntegral appThreads)+ (fromIntegral ioThreads)+ (if doPoll then usePoll else 0)+ return (Context c)++-- ^ Terminate 0MQ context (cf. zmq_term).+term :: Context -> IO ()+term = throwErrnoIfMinus1_ "term" . c_zmq_term . ctx++-- ^ Create a new 0MQ socket within the given context.+socket :: SType a => Context -> a -> IO (Socket a)+socket (Context c) t =+ let zt = typeVal . zmqSocketType $ t+ in Socket <$> throwErrnoIfNull "socket" (c_zmq_socket c zt)++-- ^ Close a 0MQ socket.+close :: Socket a -> IO ()+close = throwErrnoIfMinus1_ "close" . c_zmq_close . sock++-- ^ Set the given option on the socket. Please note that there are+-- certain combatibility constraints w.r.t the socket type (cf. man+-- zmq_setsockopt).+--+-- Please note that subscribe/unsubscribe is handled with separate+-- functions.+setOption :: Socket a -> SocketOption -> IO ()+setOption s (HighWM o) = setIntOpt s highWM o+setOption s (LowWM o) = setIntOpt s lowWM o+setOption s (Swap o) = setIntOpt s swap o+setOption s (Affinity o) = setIntOpt s affinity o+setOption s (Identity o) = setStrOpt s identity o+setOption s (Rate o) = setIntOpt s rate o+setOption s (RecoveryIVL o) = setIntOpt s recoveryIVL o+setOption s (McastLoop o) = setIntOpt s mcastLoop o+setOption s (SendBuf o) = setIntOpt s sendBuf o+setOption s (ReceiveBuf o) = setIntOpt s receiveBuf o++-- ^ Subscribe Socket to given subscription.+subscribe :: SubsType a => Socket a -> String -> IO ()+subscribe s = setStrOpt s B.subscribe++-- ^ Unsubscribe Socket from given subscription.+unsubscribe :: SubsType a => Socket a -> String -> IO ()+unsubscribe s = setStrOpt s B.unsubscribe++-- ^ Bind the socket to the given address (zmq_bind)+bind :: Socket a -> String -> IO ()+bind (Socket s) str = throwErrnoIfMinus1_ "bind" $+ withCString str (c_zmq_bind s)++-- ^ Connect the socket to the given address (zmq_connect).+connect :: Socket a -> String -> IO ()+connect (Socket s) str = throwErrnoIfMinus1_ "connect" $+ withCString str (c_zmq_connect s)++-- ^ Send the given 'ByteString' over the socket (zmq_send).+send :: Socket a -> SB.ByteString -> [Flag] -> IO ()+send (Socket s) val fls = bracket (messageOf val) messageClose $ \m ->+ throwErrnoIfMinus1_ "send" $ c_zmq_send s (msgPtr m) (combine fls)++-- ^ Flush the given socket (useful for 'send's with 'NoFlush').+flush :: Socket a -> IO ()+flush = throwErrnoIfMinus1_ "flush" . c_zmq_flush . sock++-- ^ Receive a 'ByteString' from socket (zmq_recv).+receive :: Socket a -> [Flag] -> IO (SB.ByteString)+receive (Socket s) fls = bracket messageInit messageClose $ \m -> do+ throwErrnoIfMinus1_ "receive" $ c_zmq_recv s (msgPtr m) (combine fls)+ data_ptr <- c_zmq_msg_data (msgPtr m)+ size <- c_zmq_msg_size (msgPtr m)+ SB.packCStringLen (data_ptr, fromIntegral size)++-- ^ Polls for events on the given 'Poll' descriptors. Returns the+-- list of 'Poll' descriptors for which an event occured (cf. zmq_poll).+poll :: [Poll] -> Timeout -> IO [Poll]+poll fds to = do+ let len = length fds+ ps = map createZMQPoll fds+ withArray ps $ \ptr -> do+ throwErrnoIfMinus1_ "poll" $+ c_zmq_poll ptr (fromIntegral len) (fromIntegral to)+ ps' <- peekArray len ptr+ createPoll ps' []+ where+ createZMQPoll :: Poll -> ZMQPoll+ createZMQPoll (S (Socket s) e) =+ ZMQPoll s 0 (fromEvent e) 0+ createZMQPoll (F (Fd s) e) =+ ZMQPoll nullPtr (fromIntegral s) (fromEvent e) 0++ createPoll :: [ZMQPoll] -> [Poll] -> IO [Poll]+ createPoll [] fd = return fd+ createPoll (p:pp) fd = do+ let s = pSocket p;+ f = pFd p;+ r = toEvent $ pRevents p+ if isJust r+ then createPoll pp (newPoll s f r:fd)+ else createPoll pp fd++ newPoll :: ZMQSocket -> CInt -> Maybe PollEvent -> Poll+ newPoll s 0 r = S (Socket s) (fromJust r)+ newPoll _ f r = F (Fd f) (fromJust r)++ fromEvent :: PollEvent -> CShort+ fromEvent In = fromIntegral . pollVal $ pollIn+ fromEvent Out = fromIntegral . pollVal $ pollOut+ fromEvent InOut = fromIntegral . pollVal $ pollInOut++ toEvent :: CShort -> Maybe PollEvent+ toEvent e | e == (fromIntegral . pollVal $ pollIn) = Just In+ | e == (fromIntegral . pollVal $ pollOut) = Just Out+ | e == (fromIntegral . pollVal $ pollInOut) = Just InOut+ | otherwise = Nothing+++-- internal helpers:++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++messageClose :: Message -> IO ()+messageClose (Message ptr) = do+ throwErrnoIfMinus1_ "messageClose" $ c_zmq_msg_close ptr+ free ptr++messageInit :: IO Message+messageInit = do+ ptr <- new (ZMQMsg nullPtr)+ throwErrnoIfMinus1_ "messageInit" $ c_zmq_msg_init ptr+ return (Message ptr)++messageInitSize :: Size -> IO Message+messageInitSize s = do+ ptr <- new (ZMQMsg nullPtr)+ throwErrnoIfMinus1_ "messageInitSize" $+ c_zmq_msg_init_size ptr (fromIntegral s)+ return (Message ptr)++setIntOpt :: (Storable b, Integral b) => Socket a -> ZMQOption -> b -> IO ()+setIntOpt (Socket s) (ZMQOption o) i = throwErrnoIfMinus1_ "setIntOpt" $+ bracket (newStablePtr i) freeStablePtr $ \ptr ->+ c_zmq_setsockopt s (fromIntegral o)+ (castStablePtrToPtr ptr)+ (fromIntegral . sizeOf $ i)++setStrOpt :: Socket a -> ZMQOption -> String -> IO ()+setStrOpt (Socket s) (ZMQOption o) str = throwErrnoIfMinus1_ "setStrOpt" $+ withCStringLen str $ \(cstr, len) ->+ c_zmq_setsockopt s (fromIntegral o) (castPtr cstr) (fromIntegral len)++toZMQFlag :: Flag -> ZMQFlag+toZMQFlag NoBlock = noBlock+toZMQFlag NoFlush = noFlush++combine :: [Flag] -> CInt+combine = fromIntegral . foldr ((.|.) . flagVal . toZMQFlag) 0+
+ src/System/ZMQ/Base.hsc view
@@ -0,0 +1,165 @@+{-# LANGUAGE CPP, ForeignFunctionInterface #-}+-- |+-- Module : System.ZMQ.Base+-- Copyright : (c) 2010 Toralf Wittner+-- License : LGPL+-- Maintainer : toralf.wittner@gmail.com+-- Stability : experimental+-- Portability : non-portable+-- ++module System.ZMQ.Base where++import Control.Applicative+import Foreign+import Foreign.C.Types+import Foreign.C.String++#include <zmq.h>++data ZMQMsg = ZMQMsg { content :: !(Ptr ()) }++instance Storable ZMQMsg where+ alignment _ = #{alignment zmq_msg_t}+ sizeOf _ = #{size zmq_msg_t}+ peek p = ZMQMsg <$> #{peek zmq_msg_t, content} p+ poke p (ZMQMsg c) = #{poke zmq_msg_t, content} p c++data ZMQPoll = ZMQPoll+ { pSocket :: ZMQSocket+ , pFd :: CInt+ , pEvents :: CShort+ , pRevents :: CShort+ }++instance Storable ZMQPoll where+ alignment _ = #{alignment zmq_pollitem_t}+ sizeOf _ = #{size zmq_pollitem_t}+ peek p = do+ s <- #{peek zmq_pollitem_t, socket} p+ f <- #{peek zmq_pollitem_t, fd} p+ e <- #{peek zmq_pollitem_t, events} p+ re <- #{peek zmq_pollitem_t, revents} p+ return $ ZMQPoll s f e re+ poke p (ZMQPoll s f e re) = do+ #{poke zmq_pollitem_t, socket} p s+ #{poke zmq_pollitem_t, fd} p f+ #{poke zmq_pollitem_t, events} p e+ #{poke zmq_pollitem_t, revents} p re++usePoll :: CInt+usePoll = #const ZMQ_POLL++type ZMQMsgPtr = Ptr ZMQMsg+type ZMQCtx = Ptr ()+type ZMQSocket = Ptr ()+type ZMQPollPtr = Ptr ZMQPoll++#let alignment t = "%lu", (unsigned long)offsetof(struct {char x__; t (y__); }, y__)++newtype ZMQSocketType = ZMQSocketType { typeVal :: CInt } deriving (Eq, Ord)++#{enum ZMQSocketType, ZMQSocketType,+ p2p = ZMQ_P2P,+ pub = ZMQ_PUB,+ sub = ZMQ_SUB,+ request = ZMQ_REQ,+ response = ZMQ_REP,+ xrequest = ZMQ_XREQ,+ xresponse = ZMQ_XREP,+ upstream = ZMQ_UPSTREAM,+ downstream = ZMQ_DOWNSTREAM+}++newtype ZMQOption = ZMQOption { optVal :: CInt } deriving (Eq, Ord)++#{enum ZMQOption, ZMQOption,+ highWM = ZMQ_HWM,+ lowWM = ZMQ_LWM,+ swap = ZMQ_SWAP,+ affinity = ZMQ_AFFINITY,+ identity = ZMQ_IDENTITY,+ subscribe = ZMQ_SUBSCRIBE,+ unsubscribe = ZMQ_UNSUBSCRIBE,+ rate = ZMQ_RATE,+ recoveryIVL = ZMQ_RECOVERY_IVL,+ mcastLoop = ZMQ_MCAST_LOOP,+ sendBuf = ZMQ_SNDBUF,+ receiveBuf = ZMQ_RCVBUF+}++newtype ZMQFlag = ZMQFlag { flagVal :: CInt } deriving (Eq, Ord)++#{enum ZMQFlag, ZMQFlag,+ noBlock = ZMQ_NOBLOCK,+ noFlush = ZMQ_NOFLUSH+}++newtype ZMQPollEvent = ZMQPollEvent { pollVal :: CShort } deriving (Eq, Ord)++#{enum ZMQPollEvent, ZMQPollEvent,+ pollIn = ZMQ_POLLIN,+ pollOut = ZMQ_POLLOUT,+ pollInOut = ZMQ_POLLIN | ZMQ_POLLOUT+}++-- general initialization++foreign import ccall unsafe "zmq.h zmq_init"+ c_zmq_init :: CInt -> CInt -> CInt -> IO ZMQCtx++foreign import ccall unsafe "zmq.h zmq_term"+ c_zmq_term :: ZMQCtx -> IO CInt++-- zmq_msg_t related++foreign import ccall unsafe "zmq.h zmq_msg_init"+ c_zmq_msg_init :: ZMQMsgPtr -> IO CInt++foreign import ccall unsafe "zmq.h zmq_msg_init_size"+ c_zmq_msg_init_size :: ZMQMsgPtr -> CSize -> IO CInt++foreign import ccall unsafe "zmq.h zmq_msg_close"+ c_zmq_msg_close :: ZMQMsgPtr -> IO CInt++foreign import ccall unsafe "zmq.h zmq_msg_data"+ c_zmq_msg_data :: ZMQMsgPtr -> IO (Ptr a)++foreign import ccall unsafe "zmq.h zmq_msg_size"+ c_zmq_msg_size :: ZMQMsgPtr -> IO CSize++-- socket++foreign import ccall unsafe "zmq.h zmq_socket"+ c_zmq_socket :: ZMQCtx -> CInt -> IO ZMQSocket++foreign import ccall unsafe "zmq.h zmq_close"+ c_zmq_close :: ZMQSocket -> IO CInt++foreign import ccall unsafe "zmq.h zmq_setsockopt"+ c_zmq_setsockopt :: ZMQSocket+ -> CInt -- option+ -> Ptr () -- option value+ -> CSize -- option value size+ -> IO CInt++foreign import ccall unsafe "zmq.h zmq_bind"+ c_zmq_bind :: ZMQSocket -> CString -> IO CInt++foreign import ccall unsafe "zmq.h zmq_connect"+ c_zmq_connect :: ZMQSocket -> CString -> IO CInt++foreign import ccall unsafe "zmq.h zmq_send"+ c_zmq_send :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt++foreign import ccall unsafe "zmq.h zmq_flush"+ c_zmq_flush :: ZMQSocket -> IO CInt++foreign import ccall unsafe "zmq.h zmq_recv"+ c_zmq_recv :: ZMQSocket -> ZMQMsgPtr -> CInt -> IO CInt++-- poll++foreign import ccall unsafe "zmq.h zmq_poll"+ c_zmq_poll :: ZMQPollPtr -> CInt -> CLong -> IO CInt+
+ test/display.hs view
@@ -0,0 +1,23 @@+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB++main :: IO ()+main = do+ args <- getArgs+ when (length args /= 1) $ do+ hPutStrLn stderr "usage: display <address>"+ exitFailure+ let addr = head args+ c <- ZMQ.init 1 1 False+ s <- ZMQ.socket c ZMQ.Sub+ ZMQ.subscribe s ""+ ZMQ.connect s addr+ forever $ do+ line <- ZMQ.receive s []+ SB.putStrLn line+ hFlush stdout+
+ test/perf/local_lat.hs view
@@ -0,0 +1,33 @@+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB++main :: IO ()+main = do+ args <- getArgs+ when (length args /= 3) $ do+ hPutStrLn stderr usage+ exitFailure+ let bindTo = args !! 0+ size = read $ args !! 1+ rounds = read $ args !! 2+ c <- ZMQ.init 1 1 False+ s <- ZMQ.socket c ZMQ.Rep+ ZMQ.bind s bindTo+ loop s rounds size+ ZMQ.close s+ ZMQ.term c+ where+ loop s r sz = unless (r <= 0) $ do+ msg <- ZMQ.receive s []+ when (SB.length msg /= sz) $+ error "message of incorrect size received"+ ZMQ.send s msg []+ loop s (r - 1) sz++usage :: String+usage = "usage: local_lat <bind-to> <message-size> <roundtrip-count>"+
+ test/perf/local_thr.hs view
@@ -0,0 +1,51 @@+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import Data.Time.Clock+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB+import Text.Printf++main :: IO ()+main = do+ args <- getArgs+ when (length args /= 3) $ do+ hPutStrLn stderr usage+ exitFailure+ let bindTo = args !! 0+ size = read $ args !! 1 :: Int+ count = read $ args !! 2 :: Int+ c <- ZMQ.init 1 1 False+ s <- ZMQ.socket c ZMQ.Sub+ ZMQ.subscribe s ""+ ZMQ.bind s bindTo+ receive s size+ start <- getCurrentTime+ loop s (count - 1) size+ end <- getCurrentTime+ printStat start end size count+ ZMQ.close s+ ZMQ.term c+ where+ receive s sz = do+ msg <- ZMQ.receive s []+ when (SB.length msg /= sz) $+ error "message of incorrect size received"++ loop s c sz = unless (c <= 0) $ do+ receive s sz+ loop s (c - 1) sz++ printStat start end size count = do+ let elapsed = fromRational . toRational $ diffUTCTime end start :: Double+ through = fromIntegral count / elapsed+ mbits = (through * fromIntegral size * 8) / 1000000+ printf "message size: %d [B]\n" size+ printf "message count: %d\n" count+ printf "mean throughput: %.3f [msg/s]\n" through+ printf "mean throughput: %.3f [Mb/s]\n" mbits++usage :: String+usage = "usage: local_thr <bind-to> <message-size> <message-count>"+
+ test/perf/remote_lat.hs view
@@ -0,0 +1,38 @@+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import Data.Time.Clock+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB++main :: IO ()+main = do+ args <- getArgs+ when (length args /= 3) $ do+ hPutStrLn stderr usage+ exitFailure+ let connTo = args !! 0+ size = read $ args !! 1+ rounds = read $ args !! 2+ message = SB.replicate size 0x65+ c <- ZMQ.init 1 1 False+ s <- ZMQ.socket c ZMQ.Req+ ZMQ.connect s connTo+ start <- getCurrentTime+ loop s rounds message+ end <- getCurrentTime+ print (diffUTCTime end start)+ ZMQ.close s+ ZMQ.term c+ where+ loop s r msg = unless (r <= 0) $ do+ ZMQ.send s msg []+ msg' <- ZMQ.receive s []+ when (SB.length msg' /= SB.length msg) $+ error "message of incorrect size received"+ loop s (r - 1) msg++usage :: String+usage = "usage: remote_lat <connect-to> <message-size> <roundtrip-count>"+
+ test/perf/remote_thr.hs view
@@ -0,0 +1,29 @@+import Control.Monad+import Control.Concurrent+import System.IO+import System.Exit+import System.Environment+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB++main :: IO ()+main = do+ args <- getArgs+ when (length args /= 3) $ do+ hPutStrLn stderr usage+ exitFailure+ let connTo = args !! 0+ size = read $ args !! 1+ count = read $ args !! 2+ message = SB.replicate size 0x65+ c <- ZMQ.init 1 1 False+ s <- ZMQ.socket c ZMQ.Pub+ ZMQ.connect s connTo+ replicateM_ count $ ZMQ.send s message []+ threadDelay 10000000+ ZMQ.close s+ ZMQ.term c++usage :: String+usage = "usage: remote_thr <connect-to> <message-size> <message-count>"+
+ test/poll.hs view
@@ -0,0 +1,31 @@+import Control.Applicative+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString as SB++main :: IO ()+main = do+ args <- getArgs+ when (length args /= 1) $ do+ hPutStrLn stderr usage+ exitFailure+ c <- ZMQ.init 1 1 True+ s <- ZMQ.socket c ZMQ.Rep+ let bindTo = head args+ toPoll = [ZMQ.S s ZMQ.In]+ ZMQ.bind s bindTo+ forever $+ ZMQ.poll toPoll 1000000 >>= receive+ where+ receive [] = return ()+ receive ((ZMQ.S s e):ss) = do+ msg <- ZMQ.receive s []+ ZMQ.send s msg []+ receive ss++usage :: String+usage = "usage: poll <bind-to>"+
+ test/prompt.hs view
@@ -0,0 +1,25 @@+{-# LANGUAGE OverloadedStrings #-}+import Control.Applicative+import Control.Monad+import System.IO+import System.Exit+import System.Environment+import qualified System.ZMQ as ZMQ+import qualified Data.ByteString.UTF8 as SB+import qualified Data.ByteString.Char8 as SB++main :: IO ()+main = do+ args <- getArgs+ when (length args /= 2) $ do+ hPutStrLn stderr "usage: prompt <address> <username>"+ exitFailure+ let addr = args !! 0+ name = SB.append (SB.fromString $ args !! 1) ": "+ c <- ZMQ.init 1 1 False+ s <- ZMQ.socket c ZMQ.Pub+ ZMQ.connect s addr+ forever $ do+ line <- SB.fromString <$> getLine+ ZMQ.send s (SB.append name line) []+
+ zeromq-haskell.cabal view
@@ -0,0 +1,30 @@+name: zeromq-haskell+version: 0.1+synopsis: bindings to zeromq +description: Bindings to zeromq (http://zeromq.org)+category: System, FFI+license: LGPL+license-file: LICENSE+author: Toralf Wittner+maintainer: toralf.wittner@gmail.com+copyright: Copyright (c) 2010 Toralf Wittner+homepage: http://github.com/twittner/zeromq-haskell/+stability: experimental+tested-With: GHC == 6.12.1+cabal-version: >= 1.6.0+build-type: Simple+extra-source-files: README, test/*.hs, test/perf/*.hs++library+ exposed-modules: System.ZMQ+ other-modules: System.ZMQ.Base+ ghc-options: -Wall -O2+ extensions: CPP,+ ForeignFunctionInterface,+ ExistentialQuantification + build-depends: base >= 3 && < 5,+ unix,+ containers,+ bytestring+ hs-source-dirs: src+