zeromq4-patterns (empty) → 0.1.0.0
raw patch · 8 files changed
+409/−0 lines, 8 filesdep +QuickCheckdep +asyncdep +basesetup-changed
Dependencies added: QuickCheck, async, base, binary, bytestring, exceptions, stm, test-framework, test-framework-quickcheck2, transformers, zeromq4-haskell, zeromq4-patterns
Files
- LICENSE +30/−0
- README.md +17/−0
- Setup.hs +2/−0
- example/Main.hs +4/−0
- src/System/ZMQ4/Patterns/Clone.hs +42/−0
- src/System/ZMQ4/Patterns/Clone/Internal.hs +175/−0
- test/Spec.hs +79/−0
- zeromq4-patterns.cabal +60/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Henri Verroken (c) 2018++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 Henri Verroken 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.
+ README.md view
@@ -0,0 +1,17 @@+zeromq4-patterns+================++[](https://travis-ci.org/hverr/zeromq4-patterns)+[](https://hackage.haskell.org/package/zeromq4-patterns)+[](http://stackage.org/nightly/package/zeromq4-patterns)+[](http://stackage.org/lts/package/zeromq4-patterns)++Haskell implementation of several ZeroMQ patterns that you can find in the [official ZeroMQ guide][zeromq-guide]++ [zeromq-guide]: http://zguide.zeromq.org/++## Reliable Pub-Sub (Clone pattern)++Haskell implementation of the [ZeroMQ Reliable Pub-Sub (Clone) pattern][zeromq-clone].++ [zeromq-clone]: http://zguide.zeromq.org/page:all#Reliable-Pub-Sub-Clone-Pattern
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ example/Main.hs view
@@ -0,0 +1,4 @@+module Main where++main :: IO ()+main = undefined
+ src/System/ZMQ4/Patterns/Clone.hs view
@@ -0,0 +1,42 @@+-- | This module implements the ZeroMQ reliable pub-sub (clone) pattern.+--+-- See+-- <http://zguide.zeromq.org/page:all#Reliable-Pub-Sub-Clone-Pattern+-- the ZeroMQ guide> for more information.+--+-- Example usage+--+-- @+-- import Control.Concurrent (threadDelay)+-- import Control.Concurrent.MVar+-- import Control.Concurrent.Async+--+-- import Control.Monad (forever)+--+-- import System.ZMQ4.Patterns.Clone (server, client)+--+-- -- | Produce an infinite stream of numbers+-- producer :: IO ()+-- producer = do+-- chan <- newEmptyMVar+-- withAsync (server "tcp:\/\/:5009" "tcp:\/\/:5010" chan) $ \\_ ->+-- let send i = putMVar chan i >> threadDelay (100*1000)+-- mapM_ send [(1 :: Integer)..]+--+-- -- | Consume the stream of numbers+-- consumer :: IO ()+-- consumer = do+-- chan <- newEmptyMVar+-- withAsync (client "tcp:\/\/127.0.0.1:5009" "tcp:\/\/127.0.0.1:5010" chan) $ \\_ ->+-- forever $ do+-- n <- takeMVar chan+-- print n+-- @+--+module System.ZMQ4.Patterns.Clone (+ -- * Server and client+ server+, client+) where++import System.ZMQ4.Patterns.Clone.Internal (server, client)
+ src/System/ZMQ4/Patterns/Clone/Internal.hs view
@@ -0,0 +1,175 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedLists #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE RecordWildCards #-}+module System.ZMQ4.Patterns.Clone.Internal (+ -- * Shared+ Message(..)+, iSTATE_REQUEST++ -- * Server+, server++ -- * Client+, client+) where++import Control.Concurrent.Async (Async, waitBoth, uninterruptibleCancel)+import Control.Concurrent.MVar+import Control.Concurrent.STM++import Control.Monad (forever, void, when)+import Control.Monad.Catch+import Control.Monad.IO.Class (liftIO)++import Data.Binary+import Data.ByteString (ByteString)+import Data.Word (Word64)+import qualified Data.ByteString.Lazy as BL++import System.ZMQ4.Monadic++-- | Actual message sent between server and client.+data Message a = Message {+ messageVersion :: {-# UNPACK #-} !Word64+ , messageValue :: !a+ } deriving (Eq, Show)++instance Binary a => Binary (Message a) where+ put Message{..} = do+ put messageVersion+ put messageValue++ get = do+ messageVersion <- get+ messageValue <- get+ return $! Message{..}++-- | Message sent by a client to request a state snapshot+iSTATE_REQUEST :: ByteString+iSTATE_REQUEST = "STATE_REQUEST?"++-- | A server publishing messages to a client.+--+-- This function will start serving messages on the+-- current thread.+--+-- It will send all objects that are pushed on the+-- channel to a PUB socket.+--+-- It will listen for snapshot requests on a ROUTER+-- socket.+--+-- The server will automatically keep the last sent+-- 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+ ready <- liftIO newEmptyMVar+ routerC <- liftIO . atomically $ newTVar Nothing++ 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+ -- bind socket+ pub <- socket Pub+ bind pub pubAddr++ -- wait for router+ () <- liftIO $ takeMVar ready++ -- handle all messages+ let next :: Word64 -> ZMQ z void+ next !seq' = do a <- liftIO $ takeMVar chan+ let !env = Message seq' a+ !msg = BL.toStrict (encode env)+ !jmsg = Just msg+ liftIO . atomically $ writeTVar routerC jmsg+ send pub [] msg+ next (seq'+1)++ next 0++ router :: forall z void. MVar () -> TVar (Maybe ByteString) -> ZMQ z void+ router ready routerC = do+ -- bind socket+ rout <- socket Router+ bind rout routerAddr++ -- send ready+ liftIO $ putMVar ready ()++ -- handle all messages+ let getSnapshot :: IO ByteString+ getSnapshot = liftIO . atomically $ readTVar routerC >>= \case+ Nothing -> retry+ Just x -> return x++ forever $ do+ receiveMulti rout >>= \case+ [identity', req] -> when (req == iSTATE_REQUEST) $ do+ snapshot <- liftIO getSnapshot+ sendMulti rout [identity', snapshot]+ _ -> return ()++-- | A client receiving messages from a server.+--+-- This function will start reading messages on the+-- current thread.+--+-- It will connect to the server's PUB and ROUTER+-- ports. It will backlog messages from the PUB+-- socket, while requesting the initial state snapshot+-- from the ROUTER socket. Afterwards, it will forever+-- read from the PUB socket, and process all+-- in-sequence updates.+--+-- 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+ -- connect to both sockets+ sub <- socket Sub+ connect sub pubAddr+ subscribe sub ""++ dealer <- socket Dealer+ connect dealer routerAddr++ -- get inital snapshot+ send dealer [] iSTATE_REQUEST+ msg <- receive dealer+ let !env = decode (BL.fromStrict msg)+ !a = messageValue env+ liftIO $ putMVar chan a++ -- get remaining updates+ forever $ do+ msg' <- receive sub+ let !env' = decode (BL.fromStrict msg')+ !a' = messageValue env'++ when (messageVersion env' > messageVersion env) $+ liftIO $ putMVar chan a'+++-- | Helper function+withAsync :: ZMQ z a -> (Async a -> ZMQ z b) -> ZMQ z b+withAsync action inner = mask $ \restore -> do+ a <- async (restore action)+ restore (inner a) `finally` liftIO (uninterruptibleCancel a)+
+ test/Spec.hs view
@@ -0,0 +1,79 @@+{-# 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 (Test, defaultMain, testGroup)+import Test.Framework.Providers.QuickCheck2 (testProperty)+import Test.QuickCheck++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+ ]++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
+ zeromq4-patterns.cabal view
@@ -0,0 +1,60 @@+name: zeromq4-patterns+version: 0.1.0.0+synopsis: Haskell implementation of several ZeroMQ patterns.+description:+ Haskell implementation of several ZeroMQ patterns that you can find in the+ official ZeroMQ guide.+homepage: https://github.com/hverr/zeromq4-patterns#readme+bug-reports: https://github.com/hverr/zeromq4-patterns/issues+author: Henri Verroken+maintainer: henriverroken@gmail.com+copyright: 2018 Henri Verroken+license: BSD3+license-file: LICENSE+build-type: Simple+cabal-version: >= 1.10+category: System+extra-source-files: README.md++source-repository head+ type: git+ location: https://github.com/hverr/zeromq4-patterns++library+ hs-source-dirs: src+ exposed-modules: System.ZMQ4.Patterns.Clone+ , System.ZMQ4.Patterns.Clone.Internal+ build-depends: base >=4.7 && <5+ , async+ , binary+ , bytestring+ , exceptions+ , stm+ , transformers+ , zeromq4-haskell+ default-language: Haskell2010+ ghc-options: -Wall++executable zeromq4-patterns-exe+ main-is: Main.hs+ hs-source-dirs: example+ build-depends: base >=4.7 && <5+ , zeromq4-patterns+ default-language: Haskell2010+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall++test-suite zeromq4-patterns-test+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ hs-source-dirs: test+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall+ build-depends: base >=4.7 && <5+ , zeromq4-patterns+ , QuickCheck+ , async+ , binary+ , bytestring+ , test-framework+ , test-framework-quickcheck2+ default-language: Haskell2010+ ghc-options: -Wall