diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,17 @@ +zeromq4-patterns +================ + +[![Travis](https://travis-ci.org/hverr/zeromq4-patterns.svg?branch=master)](https://travis-ci.org/hverr/zeromq4-patterns) +[![Hackage](https://img.shields.io/hackage/v/zeromq4-patterns.svg?maxAge=2592000)](https://hackage.haskell.org/package/zeromq4-patterns) +[![Stackage Nightly](http://stackage.org/package/zeromq4-patterns/badge/nightly)](http://stackage.org/nightly/package/zeromq4-patterns) +[![Stackage LTS](http://stackage.org/package/zeromq4-patterns/badge/lts)](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 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/example/Main.hs b/example/Main.hs new file mode 100644 --- /dev/null +++ b/example/Main.hs @@ -0,0 +1,4 @@ +module Main where + +main :: IO () +main = undefined diff --git a/src/System/ZMQ4/Patterns/Clone.hs b/src/System/ZMQ4/Patterns/Clone.hs new file mode 100644 --- /dev/null +++ b/src/System/ZMQ4/Patterns/Clone.hs @@ -0,0 +1,42 @@ +-- | This module implements the ZeroMQ reliable pub-sub (clone) pattern. +-- +-- See +-- 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) diff --git a/src/System/ZMQ4/Patterns/Clone/Internal.hs b/src/System/ZMQ4/Patterns/Clone/Internal.hs new file mode 100644 --- /dev/null +++ b/src/System/ZMQ4/Patterns/Clone/Internal.hs @@ -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) + diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -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 diff --git a/zeromq4-patterns.cabal b/zeromq4-patterns.cabal new file mode 100644 --- /dev/null +++ b/zeromq4-patterns.cabal @@ -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