diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -10,8 +10,13 @@ [zeromq-guide]: http://zguide.zeromq.org/ -## Reliable Pub-Sub (Clone pattern) +## Implementations -Haskell implementation of the [ZeroMQ Reliable Pub-Sub (Clone) pattern][zeromq-clone]. +- [`System.ZMQ4.Patterns.Clone`][clone-module]: [ZeroMQ Reliable Pub-Sub (Clone) pattern][zeromq-clone] +- [`System.ZMQ4.Patterns.RequestReply`][requestreply-module]: [ZeroMQ Request-Reply pattern][zeromq-requestreply] + [clone-module]: https://hackage.haskell.org/package/zeromq4-patterns/docs/System-ZMQ4-Patterns-Clone.html [zeromq-clone]: http://zguide.zeromq.org/page:all#Reliable-Pub-Sub-Clone-Pattern + + [requestreply-module]: https://hackage.haskell.org/package/zeromq4-patterns/docs/System-ZMQ4-Patterns-RequestReply.html + [zeromq-requestreply]: http://zguide.zeromq.org/page:all#Ask-and-Ye-Shall-Receive diff --git a/src/System/ZMQ4/Patterns/Clone.hs b/src/System/ZMQ4/Patterns/Clone.hs --- a/src/System/ZMQ4/Patterns/Clone.hs +++ b/src/System/ZMQ4/Patterns/Clone.hs @@ -35,8 +35,8 @@ -- module System.ZMQ4.Patterns.Clone ( -- * Server and client - server -, client + publisher +, subscriber ) where -import System.ZMQ4.Patterns.Clone.Internal (server, client) +import System.ZMQ4.Patterns.Clone.Internal (publisher, subscriber) diff --git a/src/System/ZMQ4/Patterns/Clone/Internal.hs b/src/System/ZMQ4/Patterns/Clone/Internal.hs --- a/src/System/ZMQ4/Patterns/Clone/Internal.hs +++ b/src/System/ZMQ4/Patterns/Clone/Internal.hs @@ -10,10 +10,10 @@ , iSTATE_REQUEST -- * Server -, server +, publisher -- * Client -, client +, subscriber ) where import Control.Concurrent.Async (Async, waitBoth, uninterruptibleCancel) @@ -66,21 +66,21 @@ -- object cached, and use at as a snapshot. Sequencing -- is automatically handled using a 'Word64' sequence -- counter. -server :: Binary a - => String -- ^ Bind address of the PUB socket - -> String -- ^ Bind address of the ROUTER socket - -> MVar a -- ^ Channel of incoming messages - -> IO () -server !pubAddr !routerAddr !chan = runZMQ $ do +publisher :: Binary a + => String -- ^ Bind address of the PUB socket + -> String -- ^ Bind address of the ROUTER socket + -> MVar a -- ^ Channel of incoming messages + -> IO () +publisher !pubAddr !routerAddr !chan = runZMQ $ do ready <- liftIO newEmptyMVar routerC <- liftIO . atomically $ newTVar Nothing - withAsync (publisher ready routerC) $ \t1 -> + withAsync (publisher' ready routerC) $ \t1 -> withAsync (router ready routerC) $ \t2 -> void . liftIO $ waitBoth t1 t2 where - publisher :: forall z void. MVar () -> TVar (Maybe ByteString) -> ZMQ z void - publisher ready routerC = do + publisher' :: forall z void. MVar () -> TVar (Maybe ByteString) -> ZMQ z void + publisher' ready routerC = do -- bind socket pub <- socket Pub bind pub pubAddr @@ -136,12 +136,12 @@ -- -- Note that this pattern is not 100% reliable. Messages might be dropped -- between the initial state request and the first update. -client :: Binary a - => String -- ^ Address of the server's PUB socket - -> String -- ^ Address of the server's ROUTER socket - -> MVar a -- ^ CHannel where incoming messsages will be written to - -> IO () -client !pubAddr !routerAddr !chan = runZMQ $ do +subscriber :: Binary a + => String -- ^ Address of the server's PUB socket + -> String -- ^ Address of the server's ROUTER socket + -> MVar a -- ^ CHannel where incoming messsages will be written to + -> IO () +subscriber !pubAddr !routerAddr !chan = runZMQ $ do -- connect to both sockets sub <- socket Sub connect sub pubAddr diff --git a/src/System/ZMQ4/Patterns/RequestReply.hs b/src/System/ZMQ4/Patterns/RequestReply.hs new file mode 100644 --- /dev/null +++ b/src/System/ZMQ4/Patterns/RequestReply.hs @@ -0,0 +1,89 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE FunctionalDependencies #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ScopedTypeVariables #-} +module System.ZMQ4.Patterns.RequestReply ( + -- * Type class + RequestReply(..) + + -- * Server and client +, responder +, request +) where + +import Control.Monad (forever) +import Control.Monad.IO.Class (liftIO) + +import Data.Binary +import Data.Proxy (Proxy) +import qualified Data.ByteString.Lazy as BL + +import System.ZMQ4.Monadic + +-- | A request-reply type class. +-- +-- @a@ is the request type, @b@ is the response type. +-- +-- Example: +-- +-- >>> {-# LANGUAGE DataKinds #-} +-- >>> {-# LANGUAGE TypeApplications #-} +-- >>> +-- >>> import Control.Concurrent.Async +-- >>> import Data.Binary +-- >>> +-- >>> data A = A deriving (Binary, Show) +-- >>> data B = B deriving (Binary, Show) +-- >>> +-- >>> instance RequestReply A B where +-- >>> reply _ = B +-- >>> +-- >>> main :: IO () +-- >>> main = withAsync (responder @A Proxy "tcp://*:5000") $ \_ -> +-- >>> requester "tcp://127.0.0.1:5000" A >>= print +-- +class (Binary a, Binary b) => RequestReply a b | a -> b where + reply :: a -> IO b + +-- | Start responding using the given type class. +-- +-- See 'RequestReply' for an example. +-- +-- Silently ignores a request when decoding fails +responder :: forall a b . + RequestReply a b + => Proxy a -- ^ Proxy type of the request type + -> String -- ^ Address to bind to + -> IO () +responder _ addr = runZMQ $ do + rep <- socket Rep + bind rep addr + + forever $ (decodeOrFail . BL.fromStrict <$> receive rep) >>= \case + Left _ -> return () + Right (_, _, a :: a) -> do + (b :: b) <- liftIO $ reply a + let !msg = encode b + send' rep [] msg + + +-- | Request a reply. +-- +-- See 'RequestReply' for an example. +-- +-- Throws an error when the response cannot be decoded. +request :: forall a b. + RequestReply a b + => String -- ^ Address of the REP socket + -> a -- ^ The request + -> IO b -- ^ The reply +request addr x = runZMQ $ do + req <- socket Req + connect req addr + + let !msg = encode x + send' req [] msg + + bs <- receive req + return $! decode (BL.fromStrict bs) diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -1,79 +1,12 @@ -{-# LANGUAGE BangPatterns #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} module Main where -import Control.Concurrent -import Control.Concurrent.Async - -import Data.Binary (encode, decode) -import Data.Word (Word64) - -import System.ZMQ4.Patterns.Clone.Internal +import Test.Framework (defaultMain) -import Test.Framework (Test, defaultMain, testGroup) -import Test.Framework.Providers.QuickCheck2 (testProperty) -import Test.QuickCheck +import qualified Test.System.ZMQ4.Patterns.Clone.Internal +import qualified Test.System.ZMQ4.Patterns.RequestReply main :: IO () -main = defaultMain [tests] - -tests :: Test -tests = testGroup "System.ZMQ4.Patterns.Clone.Internal" [ - testProperty "binary message" prop_binary_message - , testProperty "server client" prop_server_client +main = defaultMain [ + Test.System.ZMQ4.Patterns.Clone.Internal.tests + , Test.System.ZMQ4.Patterns.RequestReply.tests ] - -newtype TestMessage = TestMessage (Message Int) deriving (Eq, Show) - -instance Arbitrary TestMessage where - arbitrary = TestMessage <$> (Message <$> arbitrary <*> arbitrary) - -prop_binary_message :: TestMessage -> Bool -prop_binary_message (TestMessage m) = m == decode (encode m) - - -data ServerClientTest = ServerClientTest { - serverClientTestMessages :: ![Word64] - } deriving (Show) - -instance Arbitrary ServerClientTest where - arbitrary = do - xs <- sized $ \n' -> - let n = fromIntegral (max 1 n') in - return [1..n] - return $! ServerClientTest xs - -prop_server_client :: ServerClientTest -> Property -prop_server_client test = within (10*1000*1000) $ ioProperty $ do - let pubAddr = "ipc:///tmp/zeromq4-clone-pattern-test-pub.socket" - routerAddr = "ipc:///tmp/zeromq4-clone-pattern-test-router.socket" - - pushC <- newEmptyMVar - recvC <- newEmptyMVar - - withAsync (server pubAddr routerAddr pushC) $ \_ -> - withAsync (client pubAddr routerAddr recvC) $ \_ -> - withAsync (pushAll pushC) $ \_ -> - receiveAll recvC - where - messages = serverClientTestMessages test - - pushAll :: MVar Word64 -> IO () - pushAll c = mapM_ (\x -> putMVar c x >> threadDelay (1000)) - messages - - receiveAll :: MVar Word64 -> IO Property - receiveAll c = do - initSeq <- takeMVar c - let next :: Word64 -> IO Property - next !s | s == (fromIntegral $ length messages) = return $ property True - | otherwise = do ms' <- race (threadDelay (100*1000)) - (takeMVar c) - case ms' of - Left () -> return $ property True - Right s' -> - if s' <= s then - return $ counterexample "out of order" False - else - next s' - next initSeq diff --git a/test/Test/System/ZMQ4/Patterns/Clone/Internal.hs b/test/Test/System/ZMQ4/Patterns/Clone/Internal.hs new file mode 100644 --- /dev/null +++ b/test/Test/System/ZMQ4/Patterns/Clone/Internal.hs @@ -0,0 +1,76 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +module Test.System.ZMQ4.Patterns.Clone.Internal (tests) where + +import Control.Concurrent +import Control.Concurrent.Async + +import Data.Binary (encode, decode) +import Data.Word (Word64) + +import System.ZMQ4.Patterns.Clone.Internal + +import Test.Framework (Test, testGroup) +import Test.Framework.Providers.QuickCheck2 (testProperty) +import Test.QuickCheck + +tests :: Test +tests = testGroup "System.ZMQ4.Patterns.Clone.Internal" [ + testProperty "binary message" prop_binary_message + , testProperty "server client" prop_server_client + ] + +newtype TestMessage = TestMessage (Message Int) deriving (Eq, Show) + +instance Arbitrary TestMessage where + arbitrary = TestMessage <$> (Message <$> arbitrary <*> arbitrary) + +prop_binary_message :: TestMessage -> Bool +prop_binary_message (TestMessage m) = m == decode (encode m) + + +newtype ServerClientTest = ServerClientTest { + serverClientTestMessages :: [Word64] + } deriving (Show) + +instance Arbitrary ServerClientTest where + arbitrary = do + xs <- sized $ \n' -> + let n = fromIntegral (max 1 n') in + return [1..n] + return $! ServerClientTest xs + +prop_server_client :: ServerClientTest -> Property +prop_server_client test = within (10*1000*1000) $ ioProperty $ do + let pubAddr = "ipc:///tmp/zeromq4-clone-pattern-test-pub.socket" + routerAddr = "ipc:///tmp/zeromq4-clone-pattern-test-router.socket" + + pushC <- newEmptyMVar + recvC <- newEmptyMVar + + withAsync (publisher pubAddr routerAddr pushC) $ \_ -> + withAsync (subscriber pubAddr routerAddr recvC) $ \_ -> + withAsync (pushAll pushC) $ \_ -> + receiveAll recvC + where + messages = serverClientTestMessages test + + pushAll :: MVar Word64 -> IO () + pushAll c = mapM_ (\x -> putMVar c x >> threadDelay (100)) + messages + + receiveAll :: MVar Word64 -> IO Property + receiveAll c = do + initSeq <- takeMVar c + let next :: Word64 -> IO Property + next !s | s == (fromIntegral $ length messages) = return $ property True + | otherwise = do ms' <- race (threadDelay (100*1000)) + (takeMVar c) + case ms' of + Left () -> return $ property True + Right s' -> + if s' <= s then + return $ counterexample "out of order" False + else + next s' + next initSeq diff --git a/test/Test/System/ZMQ4/Patterns/RequestReply.hs b/test/Test/System/ZMQ4/Patterns/RequestReply.hs new file mode 100644 --- /dev/null +++ b/test/Test/System/ZMQ4/Patterns/RequestReply.hs @@ -0,0 +1,63 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE FunctionalDependencies #-} +{-# LANGUAGE TypeApplications #-} +module Test.System.ZMQ4.Patterns.RequestReply (tests) where + +import Control.Concurrent.Async + +import Control.Monad (replicateM) + +import Data.Binary +import Data.Proxy + +import GHC.Generics (Generic) + +import System.ZMQ4.Patterns.RequestReply + +import Test.Framework (Test, testGroup) +import Test.Framework.Providers.QuickCheck2 (testProperty) +import Test.QuickCheck + +tests :: Test +tests = testGroup "System.ZMQ4.PAtterns.RequestReply" [ + testProperty "server client" prop_server_client + ] + +data Request = ReqA | ReqB deriving (Eq, Generic, Show) + +instance Binary Request + +instance Arbitrary Request where + arbitrary = elements [ReqA, ReqB] + +data Response = RepA | RepB deriving (Eq, Generic, Show) + +instance Binary Response + +instance RequestReply Request Response where + reply ReqA = return RepA + reply ReqB = return RepB + +newtype TestSetup = TestSetup [Request] deriving (Show) + +instance Arbitrary TestSetup where + arbitrary = TestSetup <$> replicateM 10 arbitrary + +prop_server_client :: TestSetup -> Property +prop_server_client (TestSetup reqs) = within (10*1000*1000) $ ioProperty $ + withAsync (responder @Request Proxy addr) $ \_ -> + checkAll reqs + + where + checkAll :: [Request] -> IO Property + checkAll [] = return (property True) + checkAll (x:xs) = do + y <- request addr x + let flag = case x of ReqA -> (y == RepA) + ReqB -> (y == RepB) + if flag then + checkAll xs + else + return (property False) + + addr = "ipc:///tmp/zeromq4-patterns-test-req-rep.socket" diff --git a/zeromq4-patterns.cabal b/zeromq4-patterns.cabal --- a/zeromq4-patterns.cabal +++ b/zeromq4-patterns.cabal @@ -1,5 +1,5 @@ name: zeromq4-patterns -version: 0.1.0.0 +version: 0.2.0.0 synopsis: Haskell implementation of several ZeroMQ patterns. description: Haskell implementation of several ZeroMQ patterns that you can find in the @@ -24,6 +24,7 @@ hs-source-dirs: src exposed-modules: System.ZMQ4.Patterns.Clone , System.ZMQ4.Patterns.Clone.Internal + , System.ZMQ4.Patterns.RequestReply build-depends: base >=4.7 && <5 , async , binary @@ -47,7 +48,8 @@ type: exitcode-stdio-1.0 main-is: Spec.hs hs-source-dirs: test - ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall + other-modules: Test.System.ZMQ4.Patterns.Clone.Internal + , Test.System.ZMQ4.Patterns.RequestReply build-depends: base >=4.7 && <5 , zeromq4-patterns , QuickCheck