zeromq-haskell 0.3.1 → 0.4

raw patch · 7 files changed

+186/−11 lines, 7 files

Files

• AUTHORS +5/−0
• LICENSE +1/−1
• README +5/−5
• src/System/ZMQ.hs +52/−1
• src/System/ZMQ/Base.hsc +16/−0
• test/queue.hs +103/−0
• zeromq-haskell.cabal +4/−4

@@ -0,0 +1,5 @@+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+

@@ -1,4 +1,4 @@-Copyright (c) 2010 Toralf Wittner, David Himmelstrup+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

@@ -3,11 +3,11 @@  · Current status -Version 0.3.x - This software currently has *alpha* status, i.e. it had-only seen very limited testing and changes to its API might happen.+Version 0.4 - This software currently has *beta* status, i.e. it had+seen limited testing. Changes to its API may still happen. -This software was developed and tested on Linux 2.6.32 with GHC-6.12.3-using the current git HEAD of zeromq2 (http://github.com/sustrik/zeromq2).+This software was developed and tested on Linux 2.6.35 with GHC-6.12.3+using zeromq-2.0.8.   · Installation@@ -36,7 +36,7 @@  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'.+to be specified, e.g. 'Pair' and the resulting socket is of type 'Socket Pair'. This additional type information is used to ensure that only options applicable to the socket type can be set, hence ZMQ_SUBSCRIBE and ZMQ_UNSUBSCRIBE which  only apply to ZMQ_SUB sockets have their own functions ('subscribe' and

@@ -21,6 +21,7 @@     SocketOption(..),     Poll(..),     PollEvent(..),+    Device(..),      SType,     SubsType,@@ -31,6 +32,8 @@     Rep(..),     XReq(..),     XRep(..),+    Pull(..),+    Push(..),     Up(..),     Down(..), @@ -48,8 +51,10 @@     send',     receive, -    poll+    poll, +    device+ ) where  import Prelude hiding (init)@@ -144,6 +149,29 @@ instance SType XRep where     zmqSocketType = const xresponse +-- | A socket of type ZMQ_PULL is used by a pipeline node to receive+-- messages from upstream pipeline nodes. Messages are fair-queued from+-- among all connected upstream nodes. The zmq_send() function is not+-- implemented for this socket type.+data Pull = Pull+instance SType Pull where+    zmqSocketType = const pull++-- | A socket of type ZMQ_PUSH is used by a pipeline node to send messages+-- to downstream pipeline nodes. Messages are load-balanced to all connected+-- downstream nodes. The zmq_recv() function is not implemented for this+-- socket type.+--+-- When a ZMQ_PUSH socket enters an exceptional state due to having reached+-- the high water mark for all downstream nodes, or if there are no+-- downstream nodes at all, then any zmq_send(3) operations on the socket+-- shall block until the exceptional state ends or at least one downstream+-- node becomes available for sending; messages are not discarded.+data Push = Push+instance SType Push where+    zmqSocketType = const push++{-# DEPRECATED Up "Use Pull instead." #-} -- | Socket to receive messages from up the stream. Messages are -- fair-queued from among all the connected peers. Send function is not -- implemented for this socket type. /Compatible peer sockets/: 'Down'.@@ -151,6 +179,7 @@ instance SType Up where     zmqSocketType = const upstream +{-# DEPRECATED Down "Use Push instead." #-} -- | Socket to send messages down stream. Messages are load-balanced -- among all the connected peers. Send function is not implemented for -- this socket type. /Compatible peer sockets/: 'Up'.@@ -247,6 +276,7 @@ -- If it cannot be performed immediatley an error will be thrown (errno -- is set to EAGAIN). data Flag = NoBlock -- ^ ZMQ_NOBLOCK+          | SndMore -- ^ ZMQ_SNDMORE   deriving (Eq, Ord, Show)  -- | The events to wait for in poll (cf. man zmq_poll)@@ -263,6 +293,13 @@     forall a. S (Socket a) PollEvent   | F Fd PollEvent +-- | Type representing ZeroMQ devices, as used with zmq_device+data Device =+    Streamer  -- ^ ZMQ_STREAMER+  | Forwarder -- ^ ZMQ_FORWARDER+  | Queue     -- ^ ZMQ_QUEUE+  deriving (Eq, Ord, Show)+ -- | Initialize a 0MQ context (cf. zmq_init for details). init :: Size -> IO Context init ioThreads = do@@ -381,7 +418,20 @@               | e == (fromIntegral . pollVal $ pollInOut) = Just InOut               | otherwise                                 = Nothing +-- | Launch a ZeroMQ device (zmq_device).+--+-- Please note that this call never returns.+device :: Device -> Socket a -> Socket b -> IO ()+device device' (Socket insocket) (Socket outsocket) =+    throwErrnoIfMinus1_ "device" $+        c_zmq_device (fromDevice device') insocket outsocket+ where+    fromDevice :: Device -> CInt+    fromDevice Streamer  = fromIntegral . deviceType $ deviceStreamer+    fromDevice Forwarder = fromIntegral . deviceType $ deviceForwarder+    fromDevice Queue     = fromIntegral . deviceType $ deviceQueue + -- internal helpers:  messageOf :: SB.ByteString -> IO Message@@ -435,6 +485,7 @@  toZMQFlag :: Flag -> ZMQFlag toZMQFlag NoBlock = noBlock+toZMQFlag SndMore = sndMore  combine :: [Flag] -> CInt combine = fromIntegral . foldr ((.|.) . flagVal . toZMQFlag) 0

@@ -64,6 +64,8 @@     response   = ZMQ_REP,     xrequest   = ZMQ_XREQ,     xresponse  = ZMQ_XREP,+    pull       = ZMQ_PULL,+    push       = ZMQ_PUSH,     upstream   = ZMQ_UPSTREAM,     downstream = ZMQ_DOWNSTREAM }@@ -88,6 +90,7 @@  #{enum ZMQFlag, ZMQFlag,     noBlock = ZMQ_NOBLOCK+  , sndMore = ZMQ_SNDMORE }  newtype ZMQPollEvent = ZMQPollEvent { pollVal :: CShort } deriving (Eq, Ord)@@ -98,6 +101,14 @@     pollInOut = ZMQ_POLLIN | ZMQ_POLLOUT } +newtype ZMQDevice = ZMQDevice { deviceType :: CInt } deriving (Eq, Ord)++#{enum ZMQDevice, ZMQDevice,+    deviceStreamer  = ZMQ_STREAMER,+    deviceForwarder = ZMQ_FORWARDER,+    deviceQueue     = ZMQ_QUEUE+}+ -- general initialization  foreign import ccall unsafe "zmq.h zmq_init"@@ -154,4 +165,9 @@  foreign import ccall safe "zmq.h zmq_poll"     c_zmq_poll :: ZMQPollPtr -> CInt -> CLong -> IO CInt++-- device++foreign import ccall safe "zmq.h zmq_device"+    c_zmq_device :: CInt -> ZMQSocket -> ZMQSocket -> IO CInt 

@@ -0,0 +1,103 @@+-- Demo application for a ZeroMQ 'queue' device+--+-- Compile using:+--+-- ghc --make -threaded queue.hs++import Control.Concurrent (forkOS, threadDelay)+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)++import Control.Monad (forever, forM_, replicateM, replicateM_)++import qualified Data.ByteString.Char8 as SB++import qualified System.ZMQ as ZMQ++main :: IO ()+main = do+    context <- ZMQ.init 1++    lock <- newEmptyMVar++    _ <- forkOS $ launchQueue context lock+    _ <- takeMVar lock++    forM_ [0..numWorkers] $ \i ->+        forkOS $ launchWorker context i++    locks <- replicateM numClients newEmptyMVar+    forM_ (zip [0..numClients] locks) $ \(i, lock') ->+        forkOS $ launchClient context i lock'++    -- Wait untill all clients signal completion+    forM_ locks takeMVar++    -- We can't clean up the context since our queue device is still running,+    -- and can't be killed...+    -- ZMQ.term context++ where+    numWorkers :: Int+    numWorkers = 5++    numClients :: Int+    numClients = 2 * numWorkers++    workersAddress :: String+    workersAddress = "inproc://workers"++    clientsAddress :: String+    clientsAddress = "tcp://127.0.0.1:5555"++    message :: SB.ByteString+    message = SB.replicate 10 '\0'++    delay :: Int+    delay = 1000000++    launchQueue :: ZMQ.Context -> MVar () -> IO ()+    launchQueue context lock = do+        workers <- ZMQ.socket context ZMQ.Xreq+        ZMQ.bind workers workersAddress++        clients <- ZMQ.socket context ZMQ.Xrep+        ZMQ.bind clients clientsAddress++        putMVar lock ()++        ZMQ.device ZMQ.Queue clients workers++        -- This isn't reached+        ZMQ.close workers+        ZMQ.close clients++    launchWorker :: ZMQ.Context -> Int -> IO ()+    launchWorker context i = do+        socket <- ZMQ.socket context ZMQ.Rep+        ZMQ.connect socket workersAddress++        forever $ do+            request <- ZMQ.receive socket []+            putStrLn $+                "Message received in worker " ++ show i ++ ": " +++                SB.unpack request+            threadDelay delay -- Do some 'work'+            ZMQ.send socket message []+            putStrLn $ "Reply sent in worker " ++ show i++        -- This isn't reached+        ZMQ.close socket++    launchClient :: ZMQ.Context -> Int -> MVar () -> IO ()+    launchClient context i lock = do+        socket <- ZMQ.socket context ZMQ.Req+        ZMQ.connect socket clientsAddress++        putStrLn $ "Sending message in client " ++ show i+        ZMQ.send socket (SB.pack $ show i) []+        _ <- ZMQ.receive socket []+        putStrLn $ "Reply received in client " ++ show i++        ZMQ.close socket++        putMVar lock ()

@@ -1,5 +1,5 @@ name:               zeromq-haskell-version:            0.3.1+version:            0.4 synopsis:           bindings to zeromq  description:        Bindings to zeromq (http://zeromq.org) category:           System, FFI@@ -7,13 +7,13 @@ license-file:       LICENSE author:             Toralf Wittner maintainer:         toralf.wittner@gmail.com-copyright:          Copyright (c) 2010 Toralf Wittner, David Himmelstrup+copyright:          Copyright (c) 2010 zeromq-haskell authors homepage:           http://github.com/twittner/zeromq-haskell/ stability:          experimental-tested-With:        GHC == 6.12.1+tested-With:        GHC == 6.12.3 cabal-version:      >= 1.6.0 build-type:         Simple-extra-source-files: README, test/*.hs, test/perf/*.hs+extra-source-files: README, AUTHORS, test/*.hs, test/perf/*.hs  library   exposed-modules:  System.ZMQ