zeromq4-patterns 0.1.0.0 → 0.2.0.0
raw patch · 8 files changed
+265/−97 lines, 8 files
Files
- README.md +7/−2
- src/System/ZMQ4/Patterns/Clone.hs +3/−3
- src/System/ZMQ4/Patterns/Clone/Internal.hs +17/−17
- src/System/ZMQ4/Patterns/RequestReply.hs +89/−0
- test/Spec.hs +6/−73
- test/Test/System/ZMQ4/Patterns/Clone/Internal.hs +76/−0
- test/Test/System/ZMQ4/Patterns/RequestReply.hs +63/−0
- zeromq4-patterns.cabal +4/−2
README.md view
@@ -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
src/System/ZMQ4/Patterns/Clone.hs view
@@ -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)
src/System/ZMQ4/Patterns/Clone/Internal.hs view
@@ -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
+ src/System/ZMQ4/Patterns/RequestReply.hs view
@@ -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)
test/Spec.hs view
@@ -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
+ test/Test/System/ZMQ4/Patterns/Clone/Internal.hs view
@@ -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
+ test/Test/System/ZMQ4/Patterns/RequestReply.hs view
@@ -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"
zeromq4-patterns.cabal view
@@ -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