packages feed

pipes-zeromq4 (empty) → 0.1.0.0

raw patch · 7 files changed

+464/−0 lines, 7 filesdep +basedep +bytestringdep +pipessetup-changed

Dependencies added: base, bytestring, pipes, pipes-safe, pipes-zeromq4, semigroups, zeromq4-haskell

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2015, Matthew Peddie <mpeddie@gmail.com>++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Matthew Peddie nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ examples/client.hs view
@@ -0,0 +1,25 @@+{-# OPTIONS_GHC -Wall #-}+{-# LANGUAGE OverloadedStrings #-}++module Main where++import Pipes.ZMQ4+import Pipes.Core+import qualified System.ZMQ4 as Z+import Control.Concurrent (threadDelay)++import qualified Data.ByteString as B+import Data.List.NonEmpty (NonEmpty(..))++get :: MonadIO m => Client (NonEmpty B.ByteString) [B.ByteString] m ()+get = mapM_ (\x -> do+                liftIO . putStrLn $ "Requesting " ++ show x+                resp <- request $ x :| []+                liftIO . putStrLn $ "Received response " ++ show (B.concat resp)+                liftIO $ threadDelay 1000000)+      ["1", "2", "3" :: B.ByteString]++main :: IO ()+main = Z.withContext $ \ctx ->+  runSafeT . runEffect $+  setupServer ctx Z.Req (`Z.connect` "ipc:///tmp/server") +>> get
+ examples/proxy.hs view
@@ -0,0 +1,69 @@+{-# OPTIONS_GHC -Wall #-}+{-# LANGUAGE OverloadedStrings #-}++module Main where++import Control.Monad++import Pipes.ZMQ4+import qualified Pipes.Prelude as P+import qualified System.ZMQ4 as Z+import System.Environment (getArgs)+import Control.Concurrent (threadDelay)++proxy :: IO ()+proxy = Z.withContext $ \ctx ->+  runSafeT . runEffect $+  setupProducer ctx Z.Pull (`Z.bind` "ipc:///tmp/proxy_in") >->+  P.tee P.print >-> toNonEmpty >->+  setupConsumer ctx Z.Push (`Z.connect` "ipc:///tmp/proxy_out")++sender :: IO ()+sender = Z.withContext $ \ctx ->+  Z.withSocket ctx Z.Push $ \sock -> do+    Z.connect sock "ipc:///tmp/proxy_in"+    replicateM_ 3 $ Z.send sock [] "hello" >> threadDelay 1000000+    Z.send sock [] "done"++receiver :: IO ()+receiver = Z.withContext $ \ctx ->+  Z.withSocket ctx Z.Pull $ \sock -> do+    Z.bind sock "ipc:///tmp/proxy_out"+    go sock+    putStrLn "Received 'done'; exiting."+  where+    go sock = do+      msg <- Z.receive sock+      unless (msg == "done") $ do+        putStrLn $ "Received: " ++ show msg+        go sock++{-| You will need 3 terminals: A, B and C.  Once the program is+compiled, start the receiver in terminal A:+@+./proxy 22+@++Start the proxy in terminal B:+@+./proxy 22 22+@++Start the sender in terminal C:+@+./proxy+@++Once the sender starts, three ordinary messages (@"hello"@) should be+passed one second apart from the sender via the proxy to the receiver,+followed by a termination message (@"done"@) which causes the receiver+to exit.  The proxy remains running until interrupted (@Ctrl-C@).++-}+main :: IO ()+main = do+  as <- getArgs+  case as of+   [] -> sender+   [_] -> receiver+   _ -> proxy
+ examples/server.hs view
@@ -0,0 +1,26 @@+{-# OPTIONS_GHC -Wall #-}+{-# LANGUAGE OverloadedStrings #-}++module Main where++import Pipes.ZMQ4+import Pipes.Core+import qualified System.ZMQ4 as Z++import qualified Data.ByteString as B+import Data.List.NonEmpty (NonEmpty(..))++echo :: MonadIO m => [B.ByteString] -> Server [B.ByteString] (NonEmpty B.ByteString) m ()+echo [] = do+  liftIO . putStrLn $ "Received empty list."+  nxt <- respond ("" :| [])+  echo nxt+echo bs@(b:moar) = do+  liftIO . putStrLn $ "Received data " ++ show (B.concat bs) ++ "."+  nxt <- respond (b :| moar)+  echo nxt++main :: IO ()+main = Z.withContext $ \ctx ->+  runSafeT . runEffect $+  echo +>> setupClient ctx Z.Rep (`Z.bind` "ipc:///tmp/server")
+ pipes-zeromq4.cabal view
@@ -0,0 +1,91 @@+name:                pipes-zeromq4+version:             0.1.0.0+synopsis:            Pipes integration for ZeroMQ messaging+homepage:            http://github.com/peddie/pipes-zeromq4+license:             BSD3+license-file:        LICENSE+author:              Matthew Peddie <mpeddie@gmail.com>+maintainer:          Matthew Peddie <mpeddie@gmail.com>+copyright:           (c) 2015 Matthew Peddie+category:            Network, Control, Pipes+build-type:          Simple+cabal-version:       >=1.10+description: {++<https://hackage.haskell.org/package/pipes Pipes> integration for+<http://zeromq.org/ ZeroMQ> using the+<https://hackage.haskell.org/package/zeromq4-haskell zeromq4-haskell bindings>.+.+This package is very basic.  I am new to the @Pipes@ family of APIs+and would be overjoyed to receive feedback.++}++library+  exposed-modules:     Pipes.ZMQ4+  build-depends:       base >=4.7 && <4.8+                     , zeromq4-haskell >=0.6 && <0.7+                     , bytestring >=0.10 && <0.11+                     , semigroups >=0.16 && <0.17+                     , pipes >=4.1 && <4.2+                     , pipes-safe >=2.2 && <2.3+  hs-source-dirs:      src+  default-language:    Haskell2010+++flag examples+    description:    Build usage examples+    default:        False++executable proxy+  if flag(examples)+    Buildable: True+  else+    Buildable: False+  hs-source-dirs:      examples+  main-is:             proxy.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+  ghc-options:         -O2 -rtsopts -threaded+  ghc-prof-options:    -O2 -prof -fprof-auto -fprof-cafs -rtsopts -threaded++executable client+  if flag(examples)+    Buildable: True+  else+    Buildable: False+  hs-source-dirs:      examples+  main-is:             client.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 server+  if flag(examples)+    Buildable: True+  else+    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
+ src/Pipes/ZMQ4.hs view
@@ -0,0 +1,221 @@+{-# OPTIONS_GHC -Wall #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE RankNTypes #-}++{-|+Module      :  Pipes.ZMQ4+Copyright   :  (c) Matthew Peddie 2014+License     :  BSD3 (see the file zeromq4-pipes/LICENSE)++Maintainer  :  matt.peddie@planet.com+Stability   :  experimental+Portability :  GHC++This module provides functions to help you attach ZMQ sockets to+"Pipes" processing pipelines.++If you want to hook ZMQ sockets into a unidirectional pipeline+involving 'Producer's, 'Consumer's and 'Pipe's, see+@examples/proxy.hs@ for a short usage example for 'setupProducer' and+'setupConsumer'.++If you want to hook ZMQ sockets into a bidirectional or+non-'Pull'-based pipeline involving 'Client's, 'Server's and 'Proxy's,+see @examples/server.hs@ and @examples/client.hs@ for a short usage+example for 'setupServer' and 'setupClient'.++This module relies on the functions provided by "Pipes.Safe" to deal+with exceptions and termination.  If you need to avoid this layer of+safety, please don't hesitate to contact the author for support.++-}++module Pipes.ZMQ4 (+  -- * Create data sources and sinks++  -- | Each of these functions takes a setup function, which is is run+  -- before any messaging activities happen, so anything you need to+  -- do to configure the socket (e.g. 'Z.subscribe', set options,+  -- 'Z.connect' or 'Z.bind') should be done within it.+  --+  -- __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'.+  setupProducer+  , setupConsumer+  , setupBi+  , setupClient+  , setupServer+    -- ** Low-level safe setup++    -- | It's recommended to use 'setupProducer', 'setupConsumer' and+    -- friends instead of these functions unless you know what you're+    -- doing and need something else.+  , setup+  , receiveLoop+  , sendLoop+    -- * Helpers+  , toNonEmpty+    -- * Re-exported modules+  , module Pipes+  , module Pipes.Safe+  ) where++import Control.Monad (unless, forever)++import qualified System.ZMQ4 as Z++import qualified Data.ByteString      as B+import Data.List.NonEmpty (NonEmpty(..))++import Pipes+import Pipes.Safe+import Pipes.Core++-- | This is the low-level function for safely 'bracket'ing ZMQ socket+-- creation, so that any exceptions or 'Pipe' termination will not+-- result in abandoned sockets.  For example, 'setupProducer' is+-- defined as+--+-- @+-- setupProducer ctx ty opts = setup ctx ty opts receiveLoop+-- @+setup :: (MonadSafe m, Base m ~ IO,+          Z.SocketType sockty) =>+         Z.Context+      -> sockty+      -> (Z.Socket sockty -> IO ())+      -> (Z.Socket sockty -> m v)+      -> m v+setup ctx ty opts act =+  bracket (Z.socket ctx ty) (liftIO . Z.close) $ \sock -> do+    _ <- liftIO $ opts sock+    act sock++-- | This is a low-level function for simply passing all messages+-- received on a socket downstream.+--+-- @+-- receiveLoop 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++-- | This is a low-level function for simply sending all messages+-- available from upstream out on the socket.+--+-- @+-- sendLoop sock = forever $ await >>= liftIO . Z.sendMulti sock+-- @+sendLoop :: (Z.Sender sck, MonadIO m) =>+            Z.Socket sck -> Consumer' (NonEmpty B.ByteString) m ()+sendLoop sock = forever $ 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+-- downstream with 'yield'.+setupProducer :: (MonadSafe m, Base m ~ IO,+                  Z.SocketType sockty, Z.Receiver sockty) =>+                 Z.Context  -- ^ ZMQ context+              -> 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++-- | Create a 'Consumer' of message data from the given ZeroMQ+-- parameters.  All data successfully 'await'ed from upstream will be+-- sent out the socket.+--+-- The resulting 'Consumer' will only accept 'NonEmpty' lists of+-- 'B.ByteString' message parts.  See 'toNonEmpty' if this is a+-- sticking point for you.+setupConsumer :: (MonadSafe m, Base m ~ IO,+                  Z.SocketType sockty, Z.Sender sockty) =>+                 Z.Context  -- ^ ZMQ context+              -> 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++-- | 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.+--+-- 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)++-- |+-- 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 :: (Z.Receiver sockty, Z.Sender sockty, MonadIO m) =>+              Z.Socket sockty+           -> Client' [B.ByteString] (NonEmpty B.ByteString) m ()+clientLoop sock = forever $+                  liftIO (Z.receiveMulti sock) >>=+                  request >>=+                  liftIO . Z.sendMulti sock++-- | Create a 'Client'' from the given ZeroMQ parameters.  The 'Client''+-- passes all messages it receives on the ZMQ socket upstream with+-- 'request' and sends all corresponding replies back out the socket.+setupClient :: (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+           -> Client [B.ByteString] (NonEmpty B.ByteString) m ()+setupClient ctx ty opts = setup ctx ty opts clientLoop++-- |+-- This is a low-level function for passing all messages received on+-- the socket upstream and sending the responses out on the same+-- socket.+--+-- @+-- @+serverLoop :: (Z.Receiver sockty, Z.Sender sockty, MonadIO m) =>+              Z.Socket sockty+           -> NonEmpty B.ByteString+           -> Server' (NonEmpty B.ByteString) [B.ByteString] m ()+serverLoop sock msg = liftIO (Z.sendMulti sock msg) >>+                      liftIO (Z.receiveMulti sock) >>=+                      respond >>=+                      serverLoop sock++-- | Create a 'Server'' from the given ZeroMQ parameters.  The 'Server''+-- sends all received requests out on the ZMQ socket and passes all+-- messages it receives over the ZMQ socket back downstream with+-- 'respond'.+setupServer :: (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+            -> NonEmpty B.ByteString            -- ^ Server request input+            -> Server (NonEmpty B.ByteString) [B.ByteString] m ()+setupServer ctx ty opts n = setup ctx ty opts (`serverLoop` n)++-- | This is simply an adapter between lists of message parts and the+-- 'NonEmpty' lists demanded by ZMQ 'Consumer's.  If an empty list+-- arrives in the input, it will be ignored.+toNonEmpty :: Monad m => Pipe [B.ByteString] (NonEmpty B.ByteString) m ()+toNonEmpty = forever $ do+  msg <- await+  unless (null msg) $ yield (head msg :| tail msg)