patterns 0.0.1 → 0.0.2
raw patch · 7 files changed
+45/−47 lines, 7 filesdep −splitPVP ok
version bump matches the API change (PVP)
Dependencies removed: split
API changes (from Hackage documentation)
Files
- patterns.cabal +3/−2
- src/Network/Mom/Patterns.hs +4/−0
- src/Network/Mom/Patterns/Basic.hs +24/−23
- src/Network/Mom/Patterns/Device.hs +7/−8
- src/Network/Mom/Patterns/Enumerator.hs +0/−2
- src/Service.hs +6/−9
- src/Types.hs +1/−3
patterns.cabal view
@@ -1,5 +1,5 @@ Name: patterns-Version: 0.0.1+Version: 0.0.2 Cabal-Version: >= 1.8 Copyright: Copyright (c) Tobias Schoofs, 2011 - 2012 License: LGPL@@ -32,7 +32,6 @@ bytestring >= 0.9.1.9, utf8-string >= 0.3.6, containers >= 0.3.0.0,- split >= 0.1.4.1, zeromq-haskell >= 0.8.3, enumerator >= 0.4.11, mtl >= 2.0.1.0,@@ -45,4 +44,6 @@ Network.Mom.Patterns.Device, Network.Mom.Patterns.Enumerator other-modules: Factory, Types, Service++ ghc-options: -Wall
src/Network/Mom/Patterns.hs view
@@ -75,6 +75,10 @@ together to request or provide services, publish or subscribe data or to allocate work to processing nodes.++ Note that, since the patterns package is based on ZMQ,+ applications based on patterns must be compiled with the + /-threaded/ flag. -}
src/Network/Mom/Patterns/Basic.hs view
@@ -234,14 +234,12 @@ ------------------------------------------------------------------------ -- prepare service for single client ------------------------------------------------------------------------- | n <= 1 = (+ | n <= 1 = Z.withSocket ctx Z.Rep $ \client -> do link t ac client Z.withSocket ctx Z.Sub $ \cmd -> do- trycon cmd sockname retries- Z.subscribe cmd ""- ready- poll False [Z.S cmd Z.In, Z.S client Z.In] (go client) param)+ conCmd cmd sockname ready+ poll False [Z.S cmd Z.In, Z.S client Z.In] (go client) param `catch` (\e -> onerr Fatal e name param >>= \_ -> return ()) ------------------------------------------------------------------------ -- prepare service for multiple clients @@ -252,7 +250,7 @@ zs <- replicateM n newEmptyMVar withQueue ctx ("Queue " ++ name) (ac, t) (Address add [], Bind) onQErr $ \_ -> do- _ <- mapM (\(a,z) -> start add a z) (zip as zs)+ _ <- mapM (uncurry $ start add) (zip as zs) mapM_ takeMVar as -- wait for workers to start ready -- report state to service mapM_ takeMVar zs) -- wait for workers to terminate@@ -650,14 +648,12 @@ Z.Context -> String -> String -> String -> IO () -> IO () publish period ac oconv onerr - fetch ctx name sockname param ready = (+ fetch ctx name sockname param ready = Z.withSocket ctx Z.Pub $ \sock -> do Z.bind sock (acAdd ac) Z.withSocket ctx Z.Sub $ \cmd -> do- trycon cmd sockname retries- Z.subscribe cmd ""- ready- periodicSend False period cmd (go sock) param)+ conCmd cmd sockname ready+ periodicSend False period cmd (go sock) param `catch` (\e -> onerr Fatal e name param) ------------------------------------------------------------------------ -- do the job periodically@@ -745,15 +741,13 @@ String -> String -> Parameter -> IO () -> IO () subscribe sub ac iconv onerr dump - ctx name sockname param ready = (+ ctx name sockname param ready = Z.withSocket ctx Z.Sub $ \sock -> do trycon sock (acAdd ac) retries mapM_ (Z.subscribe sock) sub Z.withSocket ctx Z.Sub $ \cmd -> do- trycon cmd sockname retries- Z.subscribe cmd ""- ready- poll False [Z.S cmd Z.In, Z.S sock Z.In] (go sock) param)+ conCmd cmd sockname ready+ poll False [Z.S cmd Z.In, Z.S sock Z.In] (go sock) param `catch` (\e -> onerr Fatal e name param) where go sock p = do eiR <- E.run (rcvEnum sock iconv $$ dump ctx p)@@ -839,13 +833,13 @@ -- | Unsubscribe a topic ------------------------------------------------------------------------ unsubscribe :: Sub i -> Topic -> IO ()- unsubscribe s t = Z.unsubscribe (subSock s) t+ unsubscribe s = Z.unsubscribe (subSock s) ------------------------------------------------------------------------ -- | Subscribe another topic ------------------------------------------------------------------------ resubscribe :: Sub i -> Topic -> IO ()- resubscribe s t = Z.subscribe (subSock s) t+ resubscribe s = Z.subscribe (subSock s) ------------------------------------------------------------------------ -- The working horse behind the scene@@ -909,14 +903,12 @@ Dump i -> Z.Context -> String -> String -> String -> IO () -> IO ()- pull ac iconv onerr dump ctx name sockname param ready = (+ pull ac iconv onerr dump ctx name sockname param ready = Z.withSocket ctx Z.Pull $ \sock -> do trycon sock (acAdd ac) retries Z.withSocket ctx Z.Sub $ \cmd -> do- trycon cmd sockname retries- Z.subscribe cmd ""- ready- poll False [Z.S cmd Z.In, Z.S sock Z.In] (go sock) param)+ conCmd cmd sockname ready+ poll False [Z.S cmd Z.In, Z.S sock Z.In] (go sock) param `catch` (\e -> onerr Fatal e name param) ------------------------------------------------------------------------ -- do the job @@ -1094,3 +1086,12 @@ ------------------------------------------------------------------------ receive :: Peer i -> E.Iteratee i IO a -> IO (Either SomeException a) receive p it = E.run (rcvEnum (peeSock p) (peeIn p) $$ it)++ ------------------------------------------------------------------------+ -- connect to command socket+ ------------------------------------------------------------------------+ conCmd :: Z.Socket Z.Sub -> String -> IO () -> IO ()+ conCmd cmd sockname ready = do+ trycon cmd sockname retries+ Z.subscribe cmd ""+ ready
src/Network/Mom/Patterns/Device.hs view
@@ -27,7 +27,6 @@ import Types import Service - import qualified Data.ByteString.Char8 as B import qualified Data.Enumerator as E import Data.Enumerator (($$)) import qualified Data.Enumerator.List as EL@@ -139,11 +138,11 @@ (AccessPoint, LinkType) -> OnError_ -> (Service -> IO a) -> IO a- withQueue ctx name (dealer, l1) (router, l2) onerr act = + withQueue ctx name (dealer, l1) (router, l2) onerr = withDevice ctx name noparam (-1) [pollEntry "clients" XDealer dealer l1 [], pollEntry "servers" XRouter router l2 []]- return return onerr (\_ -> return ()) (\_ -> putThrough) act+ return return onerr (\_ -> return ()) (\_ -> putThrough) ------------------------------------------------------------------------ -- | Starts a Forwarder;@@ -188,11 +187,11 @@ (AccessPoint, LinkType) -> OnError_ -> (Service -> IO a) -> IO a- withForwarder ctx name topics (sub, l1) (pub, l2) onerr act = + withForwarder ctx name topics (sub, l1) (pub, l2) onerr = withDevice ctx name noparam (-1) [pollEntry "subscriber" XSub sub l1 topics, pollEntry "publisher" XPub pub l2 []]- return return onerr (\_ -> return ()) (\_ -> putThrough) act+ return return onerr (\_ -> return ()) (\_ -> putThrough) ------------------------------------------------------------------------ -- | Starts a pipeline;@@ -242,11 +241,11 @@ (AccessPoint, LinkType) -> OnError_ -> (Service -> IO a) -> IO a- withPipeline ctx name (puller, l1) (pusher, l2) onerr act = + withPipeline ctx name (puller, l1) (pusher, l2) onerr = withDevice ctx name noparam (-1) [pollEntry "pull" XPull puller l1 [], pollEntry "push" XPipe pusher l2 []]- return return onerr (\_ -> return ()) (\_ -> putThrough) act+ return return onerr (\_ -> return ()) (\_ -> putThrough) ------------------------------------------------------------------------ -- | A transformer is an 'E.Iteratee'@@ -482,7 +481,7 @@ m <- newMVar xp finally (runDevice name m iconv oconv onerr ontmo trans sockname param imReady)- (do withMVar m (\xp' -> mapM_ closeS (xpPoll xp')) >> return ())+ (withMVar m (mapM_ closeS . xpPoll) >> return ()) closeS :: Z.Poll -> IO () closeS p = case p of
src/Network/Mom/Patterns/Enumerator.hs view
@@ -47,9 +47,7 @@ import Control.Applicative ((<$>)) import Control.Monad- import Control.Monad.Trans import Prelude hiding (catch)- import Control.Exception (AssertionFailed(..), catch, throwIO) import qualified System.ZMQ as Z
src/Service.hs view
@@ -13,6 +13,7 @@ import qualified Data.ByteString.Char8 as B import Data.Time.Clock+ import Data.List (isPrefixOf) import Data.Map (Map) import qualified Data.Map as Map @@ -133,15 +134,11 @@ "PAUSE" -> Right PAUSE "RESUME" -> Right RESUME x -> - if take 4 x == "APP " + if "APP" `isPrefixOf` x ||+ "OPT" `isPrefixOf` x ||+ "DEVICE" `isPrefixOf` x then Right $ read x- else - if take 4 x == "OPT "- then Right $ read x- else - if take 6 x == "DEVICE"- then Right $ read x- else Left $ "No Command: " ++ x+ else Left $ "No Command: " ++ x ------------------------------------------------------------------------ -- The work horse behind "with*" services@@ -378,4 +375,4 @@ mu2nominal m = (fromIntegral m / 1000000)::NominalDiffTime nominal2mu :: NominalDiffTime -> Int- nominal2mu n = ceiling (n * (fromIntegral (1000000::Int)))+ nominal2mu n = ceiling (n * fromIntegral (1000000::Int))
src/Types.hs view
@@ -30,8 +30,6 @@ import qualified Data.ByteString.Char8 as B import qualified Data.ByteString.UTF8 as U -- standard converters import Data.Char (toLower)- import Data.List (intercalate)- import Data.List.Split (splitOn) import qualified Data.Enumerator as E import Data.Enumerator (($$)) import qualified Data.Enumerator.List as EL (head)@@ -426,7 +424,7 @@ rcvEnum s iconv = go True where go more step = case step of - E.Continue k -> do+ E.Continue k -> if more then do x <- liftIO $ Z.receive s [] m <- liftIO $ Z.moreToReceive s