diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2015, Matthew Peddie + +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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/examples/client.hs b/examples/client.hs new file mode 100644 --- /dev/null +++ b/examples/client.hs @@ -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 diff --git a/examples/proxy.hs b/examples/proxy.hs new file mode 100644 --- /dev/null +++ b/examples/proxy.hs @@ -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 diff --git a/examples/server.hs b/examples/server.hs new file mode 100644 --- /dev/null +++ b/examples/server.hs @@ -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") diff --git a/pipes-zeromq4.cabal b/pipes-zeromq4.cabal new file mode 100644 --- /dev/null +++ b/pipes-zeromq4.cabal @@ -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 +maintainer: Matthew Peddie +copyright: (c) 2015 Matthew Peddie +category: Network, Control, Pipes +build-type: Simple +cabal-version: >=1.10 +description: { + + integration for + using the +. +. +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 diff --git a/src/Pipes/ZMQ4.hs b/src/Pipes/ZMQ4.hs new file mode 100644 --- /dev/null +++ b/src/Pipes/ZMQ4.hs @@ -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)