pipes-zeromq4 0.1.0.0 → 0.2.0.0
raw patch · 5 files changed
+156/−35 lines, 5 filesdep ~basenew-component:exe:serverPipeline
Dependency ranges changed: base
Files
- examples/client.hs +2/−0
- examples/server.hs +1/−0
- examples/serverPipeline.hs +26/−0
- pipes-zeromq4.cabal +24/−3
- src/Pipes/ZMQ4.hs +103/−32
examples/client.hs view
@@ -19,6 +19,8 @@ liftIO $ threadDelay 1000000) ["1", "2", "3" :: B.ByteString] +-- | Please run me concurrently with either of the 'server' or+-- 'serverPipeline' programs. main :: IO () main = Z.withContext $ \ctx -> runSafeT . runEffect $
examples/server.hs view
@@ -20,6 +20,7 @@ nxt <- respond (b :| moar) echo nxt +-- | Please run me concurrently with the 'client' program. main :: IO () main = Z.withContext $ \ctx -> runSafeT . runEffect $
+ examples/serverPipeline.hs view
@@ -0,0 +1,26 @@+{-# OPTIONS_GHC -Wall #-}+{-# LANGUAGE OverloadedStrings #-}++module Main where++import Control.Monad+import Pipes.ZMQ4+import qualified System.ZMQ4 as Z++import qualified Data.ByteString as B+import Data.List.NonEmpty (NonEmpty(..))++echo :: MonadIO m => Pipe [B.ByteString] (NonEmpty B.ByteString) m ()+echo = forever $ do+ nxt <- await+ case nxt of+ [] -> liftIO . putStrLn $ "Received empty list."+ bs@(b:moar) -> do+ liftIO . putStrLn $ "Received data " ++ show (B.concat bs) ++ "."+ yield (b :| moar)++-- | Please run me concurrently with the 'client' program.+main :: IO ()+main = Z.withContext $ \ctx ->+ runSafeT . runEffect $+ setupPipeline ctx Z.Rep (`Z.bind` "ipc:///tmp/server") echo
pipes-zeromq4.cabal view
@@ -1,7 +1,7 @@ name: pipes-zeromq4-version: 0.1.0.0+version: 0.2.0.0 synopsis: Pipes integration for ZeroMQ messaging-homepage: http://github.com/peddie/pipes-zeromq4+homepage: https://github.com/peddie/pipes-zeromq4 license: BSD3 license-file: LICENSE author: Matthew Peddie <mpeddie@gmail.com>@@ -23,7 +23,7 @@ library exposed-modules: Pipes.ZMQ4- build-depends: base >=4.7 && <4.8+ build-depends: base >=4.6 && <4.9 , zeromq4-haskell >=0.6 && <0.7 , bytestring >=0.10 && <0.11 , semigroups >=0.16 && <0.17@@ -32,6 +32,9 @@ hs-source-dirs: src default-language: Haskell2010 +source-repository head+ type: git+ location: git://github.com/peddie/pipes-zeromq4 flag examples description: Build usage examples@@ -79,6 +82,24 @@ Buildable: False hs-source-dirs: examples main-is: server.hs+ default-language: Haskell2010+ build-depends: base >= 4.5 && < 5+ , pipes-zeromq4+ , pipes >=4.1 && <4.2+ , pipes-safe >=2.2 && <2.3+ , zeromq4-haskell >=0.6 && <0.7+ , bytestring >= 0.10 && < 0.11+ , semigroups >=0.16 && <0.17+ ghc-options: -O2 -rtsopts -threaded+ ghc-prof-options: -O2 -prof -fprof-auto -fprof-cafs -rtsopts -threaded++executable serverPipeline+ if flag(examples)+ Buildable: True+ else+ Buildable: False+ hs-source-dirs: examples+ main-is: serverPipeline.hs default-language: Haskell2010 build-depends: base >= 4.5 && < 5 , pipes-zeromq4
src/Pipes/ZMQ4.hs view
@@ -28,6 +28,10 @@ with exceptions and termination. If you need to avoid this layer of safety, please don't hesitate to contact the author for support. +Currently, everything runs in the same thread as the rest of the+'Effect', so+__if something blocks forever on a ZMQ receive or send, you may be in trouble.__+ -} module Pipes.ZMQ4 (@@ -41,9 +45,14 @@ -- __Note that according to the ZeroMQ manual pages__, the correct -- order of operations is to perform all socket configuration before -- running 'Z.connect' or 'Z.bind'.++ -- ** One-directional sources and sinks (for use with "Pipes"'s 'Pipe', 'Producer' and 'Consumer') setupProducer , setupConsumer- , setupBi+ , setupPipe+ , setupPipePair+ , setupPipeline+ -- ** Bidirectional sources and sinks (for use with "Pipes.Core"'s 'Proxy', 'Server' and 'Client'). , setupClient , setupServer -- ** Low-level safe setup@@ -52,8 +61,8 @@ -- friends instead of these functions unless you know what you're -- doing and need something else. , setup- , receiveLoop- , sendLoop+ , receiveMessage+ , sendMessage -- * Helpers , toNonEmpty -- * Re-exported modules@@ -78,7 +87,7 @@ -- defined as -- -- @--- setupProducer ctx ty opts = setup ctx ty opts receiveLoop+-- setupProducer ctx ty opts = setup ctx ty opts $ forever . receiveMessage -- @ setup :: (MonadSafe m, Base m ~ IO, Z.SocketType sockty) =>@@ -92,25 +101,25 @@ _ <- liftIO $ opts sock act sock --- | This is a low-level function for simply passing all messages+-- | This is a low-level function for simply passing a message -- received on a socket downstream. -- -- @--- receiveLoop sock = forever $ liftIO (Z.receiveMulti sock) >>= yield+-- receiveMessage sock = forever $ liftIO (Z.receiveMulti sock) >>= yield -- @-receiveLoop :: (Z.Receiver sck, MonadIO m) =>- Z.Socket sck -> Producer [B.ByteString] m ()-receiveLoop sock = forever $ liftIO (Z.receiveMulti sock) >>= yield+receiveMessage :: (Z.Receiver sck, MonadIO m) =>+ Z.Socket sck -> Producer' [B.ByteString] m ()+receiveMessage sock = liftIO (Z.receiveMulti sock) >>= yield --- | This is a low-level function for simply sending all messages+-- | This is a low-level function for simply sending a message -- available from upstream out on the socket. -- -- @--- sendLoop sock = forever $ await >>= liftIO . Z.sendMulti sock+-- sendMessage sock = await >>= liftIO . Z.sendMulti sock -- @-sendLoop :: (Z.Sender sck, MonadIO m) =>+sendMessage :: (Z.Sender sck, MonadIO m) => Z.Socket sck -> Consumer' (NonEmpty B.ByteString) m ()-sendLoop sock = forever $ await >>= liftIO . Z.sendMulti sock+sendMessage sock = await >>= liftIO . Z.sendMulti sock -- | Create a 'Producer' of message data from the given ZeroMQ -- parameters. All messages received on the socket will be sent@@ -121,7 +130,7 @@ -> sockty -- ^ ZMQ socket type -> (Z.Socket sockty -> IO ()) -- ^ Setup function -> Producer [B.ByteString] m () -- ^ Message source-setupProducer ctx ty opts = setup ctx ty opts receiveLoop+setupProducer ctx ty opts = setup ctx ty opts $ forever . receiveMessage -- | Create a 'Consumer' of message data from the given ZeroMQ -- parameters. All data successfully 'await'ed from upstream will be@@ -136,33 +145,90 @@ -> sockty -- ^ ZMQ socket type -> (Z.Socket sockty -> IO ()) -- ^ Setup function -> Consumer' (NonEmpty B.ByteString) m () -- ^ Message sink-setupConsumer ctx ty opts = setup ctx ty opts sendLoop+setupConsumer ctx ty opts = setup ctx ty opts $ forever . sendMessage --- | Create both a 'Producer' and a 'Consumer' of message data, both--- corresponding to the same socket, from the given ZeroMQ parameters.--- This is like 'setupProducer' and 'setupConsumer' combined; the--- socket type must be both a 'Z.Sender' and a 'Z.Receiver' (for--- example, a 'Z.Dealer'). Messages received over the socket are--- 'yield'ed by the 'Producer'; messages 'await'ed by the 'Consumer'--- are sent over the socket.+-- | Create a 'Pipe' out of a socket from the given ZeroMQ parameters.+-- All data successfully 'await'ed from upstream will be sent out the+-- socket, and all message data received on the socket will be+-- 'yield'ed downstream. ----- See also the descriptions of 'setupProducer' and 'setupConsumer'.-setupBi :: (MonadSafe m, Base m ~ IO, MonadSafe m1, Base m1 ~ IO,- Z.SocketType sockty, Z.Sender sockty, Z.Receiver sockty) =>- Z.Context -- ^ ZMQ context- -> sockty -- ^ ZMQ socket type- -> (Z.Socket sockty -> IO ()) -- ^ Setup function- -> m1 (Producer [B.ByteString] m (),- Consumer (NonEmpty B.ByteString) m ()) -- ^ Message (source, sink) pair-setupBi ctx ty opts = setup ctx ty opts $ \sock -> return- (receiveLoop sock, sendLoop sock)+-- You can use this 'Pipe' to replace a processing stage in your+-- pipeline with a ZMQ transaction. Note that the ZMQ actions here+-- are blocking; for example, if you make a 'Z.Req' socket, the+-- pipeline will block when the message from upstream is transmitted+-- until the response from the remote ZMQ socket is received.+-- Consider using the @pipes-concurrency@ package if you are sure+-- you're using the right socket type but you don't want everything to+-- block here.+setupPipe :: (MonadSafe m, Base m ~ IO,+ Z.SocketType sockty,+ Z.Sender sockty, Z.Receiver sockty) =>+ Z.Context -- ^ ZMQ context+ -> sockty -- ^ ZMQ socket type+ -> (Z.Socket sockty -> IO ()) -- ^ Setup function for socket+ -> Pipe (NonEmpty B.ByteString) [B.ByteString] m () -- ^ Pipeline stage via ZMQ+setupPipe ctx ty opts = setup ctx ty opts $ \sock ->+ forever $ sendMessage sock >> receiveMessage sock +-- | 'setupPipePair' is the same as 'setupPipe' except that it allows+-- you to use two different sockets, one for sending and the other for+-- receiving.+setupPipePair :: (MonadSafe m, Base m ~ IO,+ Z.SocketType txsock, Z.SocketType rxsock,+ Z.Sender txsock, Z.Receiver rxsock) =>+ Z.Context -- ^ ZMQ context+ -> txsock -- ^ ZMQ transmit socket type+ -> rxsock -- ^ ZMQ receive socket type+ -> (Z.Socket txsock -> IO ()) -- ^ Setup function for transmit socket+ -> (Z.Socket rxsock -> IO ()) -- ^ Setup function for receive socket+ -> Pipe (NonEmpty B.ByteString) [B.ByteString] m () -- ^ Pipeline stage via ZMQ+setupPipePair ctx txty rxty txopts rxopts =+ setup ctx txty txopts $ \tx ->+ setup ctx rxty rxopts $ \rx ->+ forever $ sendMessage tx >> receiveMessage rx++-- | Create an 'Effect' out of a socket from the given ZeroMQ+-- parameters and a pipe to use to connect the resulting 'Producer'+-- and 'Consumer'.+--+-- There are two ways to use a single socket to do both ZMQ sends and+-- receives in the same pipeline. 'setupPipe' places the ZMQ socket+-- in the middle of the pipeline. 'setupPipeline' places the same ZMQ+-- socket at both ends: the provided 'Pipe' is used to connect the+-- receiving Producer and sending Consumer generated from the ZMQ+-- socket. In other words, for @setupPipeline ctx socktype opts+-- myPipe@,+--+-- @+-- socket RX >-> myPipe >-> socket TX+-- @+--+-- See 'setupPipe' for some caveats. See @examples/serverPipeline.hs@+-- for a usage example.+setupPipeline :: (MonadSafe m, Base m ~ IO,+ Z.SocketType sockty,+ Z.Sender sockty, Z.Receiver sockty) =>+ Z.Context -- ^ ZMQ context+ -> sockty -- ^ ZMQ socket type+ -> (Z.Socket sockty -> IO ()) -- ^ Setup function for socket+ -> Pipe [B.ByteString] (NonEmpty B.ByteString) m () -- ^ Pipeline to run in the middle+ -> Effect m ()+setupPipeline ctx ty opts inpipe = setup ctx ty opts $ \sock ->+ forever $+ (liftIO (Z.receiveMulti sock) >>= yield) >->+ inpipe >->+ (await >>= liftIO . Z.sendMulti sock)+ -- | -- This is a low-level function for passing all messages received on -- the socket upstream and sending the responses out on the same -- socket. -- -- @+-- clientLoop sock = forever $+-- liftIO (Z.receiveMulti sock) >>=+-- request >>=+-- liftIO . Z.sendMulti sock -- @ clientLoop :: (Z.Receiver sockty, Z.Sender sockty, MonadIO m) => Z.Socket sockty@@ -189,6 +255,11 @@ -- socket. -- -- @+-- serverLoop sock msg =+-- liftIO (Z.sendMulti sock msg) >>+-- liftIO (Z.receiveMulti sock) >>=+-- respond >>=+-- serverLoop sock -- @ serverLoop :: (Z.Receiver sockty, Z.Sender sockty, MonadIO m) => Z.Socket sockty