courier 0.1.0.8 → 0.1.0.9
raw patch · 7 files changed
+386/−26 lines, 7 filesdep −network-simple
Dependencies removed: network-simple
Files
- changes.md +13/−0
- courier.cabal +16/−7
- examples/HelloWorld.hs +31/−0
- src/Control/Concurrent/Mailbox.hs +12/−2
- src/Network/Endpoints.hs +44/−16
- src/Network/RPC.hs +267/−0
- tests/Tests.hs +3/−1
changes.md view
@@ -1,3 +1,16 @@+0.1.0.9++ * Added HelloWorld example, and updated code sample in documentation to reflect+ working with the current API+ + * Early implementation of synchronous RPCs on top of endpoints, with basic unit tests.+ Both single and group RPCs are implemented, as are calls with definite and indefinite+ waits for responses.++ * Removed dependency on network-simple++ * Improved documentation throughout.+ 0.1.0.8 * Enabled selective out of order mesage reception, by using a Mailbox, an extension of STM's
courier.cabal view
@@ -1,8 +1,5 @@--- Initial courier.cabal generated by cabal init. For further --- documentation, see http://haskell.org/cabal/users-guide/- name: courier-version: 0.1.0.8+version: 0.1.0.9 synopsis: A message-passing library for simplifying network applications description: Inspired by Erlang's simple message-passing facilities, courier provides roughly similar capabilities. Applications simply create one or more endpoints, @@ -10,13 +7,24 @@ send / receive messages to other endpoints just by referencing the name each endpoint bound to its transport. .+ A primary driver of this package's design is separation of concerns: many algorithms+ (especially for distributed applications) depend upon a message-passing foundation,+ but the algorithms are sufficiently complex that the details of how those messages+ are transmitted among nodes are best hidden away and solved separately from+ the distributed algorithm itself. With this in mind, this pacakge aims+ to provide a variety of transports as well as support for common communication+ idioms (e.g., in order message delivery, selective out of order message delivery,+ synchronous RPC, etc.). + . Applications may process messages in the order received at an endpoint, or use selective message reception to process the first message arriving at an endpoint that also matches a provided selection function. Through selective message reception,- applications may approximate the style of an Erlang application, and enjoy better composability- of message reception with multiple independent dispatch routines or message pumps.+ applications may approximate the style of an Erlang application, and enjoy better + composability of message reception with multiple independent dispatch routines or+ message pumps. extra-source-files: changes.md+ examples/HelloWorld.hs homepage: http://github.com/hargettp/courier license: MIT license-file: LICENSE@@ -42,6 +50,7 @@ Network.Transport.TCP Network.Transport.UDP Control.Concurrent.Mailbox+ Network.RPC ghc-options: -Wall other-modules:@@ -54,7 +63,6 @@ containers, hslogger, network,- network-simple, stm, text @@ -74,6 +82,7 @@ -- 3rd party modules async, cereal,+ containers, directory, hslogger, stm,
+ examples/HelloWorld.hs view
@@ -0,0 +1,31 @@+module HelloWorld (+ main+) where++ -- Just import this package to access the primary APIs+import Network.Endpoints++ -- A specific transport is necessary, however+import Network.Transport.TCP++-- Needed for serialization+import Data.Serialize++main :: IO ()+main = do+ let name1 = "endpoint1"+ name2 = "endpoint2"+ resolver = resolverFromList [(name1,"localhost:2000"),+ (name2,"localhost:2001")]+ transport <- newTCPTransport resolver+ endpoint1 <- newEndpoint [transport]+ endpoint2 <- newEndpoint [transport]+ Right () <- bindEndpoint endpoint1 name1+ Right () <- bindEndpoint endpoint2 name2+ sendMessage_ endpoint1 name2 $ encode "hello world!"+ msg <- receiveMessage endpoint2+ let Right txt = decode msg+ in print (txt :: String)+ Right () <- unbindEndpoint endpoint1 name1+ Right () <- unbindEndpoint endpoint2 name2+ shutdown transport
src/Control/Concurrent/Mailbox.hs view
@@ -14,14 +14,19 @@ -- a supplied test function, or block if no such match is possible with the messages currently -- in the queue. ----- As 'Mailbox' implements the same basic @read@/@write/@peek@ functions as a 'TQueue',--- it offers a superset of 'TQueue' functionality, extending it with the @find@/@select@/@handle@+-- As 'Mailbox' implements the same basic @read / write / peek@ group of functions as a 'TQueue',+-- it offers a superset of 'TQueue' functionality by extending it with the @find / select / handle@ -- groups of functions. Thus, applications can safely use 'Mailbox'es in place of 'TQueue's, -- but choose when to take the slight extra overhead of 'Mailbox' functionality. -- -- Because message selection in worst case requires fully traversing all messages in the queue, -- application designers are encouraged to understand this aspect when choosing to use 'Mailbox'es -- in their designs, or when using the additional features of 'Mailbox'es beyond that of 'TQueue's.+-- Dispatching messages with a 'Mailbox' is analogous to using a @case@ expression (O(n)) to dispatch+-- messages to a handler function, except that new cases can be added or removed at any time. In essence,+-- one can regard 'Mailbox'es as a useful means of creating an extensible message dispatch function.+-- If, however, if O(1) message dispatching time is necessary or desired, (using hashmaps, for example)+-- then 'Mailbox'es are not the correct choice. -- -- Despite this extra cost, 'Mailbox'es offer advantages to designers: --@@ -34,6 +39,11 @@ -- selective message reception, multiple concurrent message pumps are possible (with -- a small performance impact), each processing the messages they expect and with no -- need to be aware of other message pumps performing their own work on the same mailbox.+--+-- * Mixing synchronous and asynchronous programming styles: if restricted to in order message+-- delivery, an application must carefully construct all logic to avoid blocking its central message+-- loop. By supporting out of message delivery and multiple selective recipients, it becomes possible+-- to combine synchronous and asynchronous programming styles using the same 'Mailbox'. -- -- Basic framework for 'Mailbox' brazenly copied from "Control.Concurrent.STM.TQueue". -----------------------------------------------------------------------------
src/Network/Endpoints.hs view
@@ -29,7 +29,9 @@ newEndpoint, bindEndpoint,+ bindEndpoint_, unbindEndpoint,+ unbindEndpoint_, sendMessage, sendMessage_,@@ -76,27 +78,37 @@ -- -- A sample of how to use this library: -- +-- > module HelloWorld (+-- > main+-- > ) where+-- > -- > -- Just import this package to access the primary APIs -- > import Network.Endpoints -- > -- > -- A specific transport is necessary, however -- > import Network.Transport.TCP--- > --- > helloWorld :: IO ()--- > helloWorld = do--- > let name1 = "endpoint1"b--- > name2 = "endpoint2"--- > resolver = resolverFromList [(name1,"localhost:2000"),--- > (name2,"localhost:2001")]--- > transport <- newTCPTransport resolver--- > endpoint1 <- newEndpoint [transport]--- > endpoint2 <- newEndpoint [transport]--- > Right () <- bindEndpoint endpoint1 name1--- > Right () <- bindEndpoint endpoint2 name2--- > sendMessage endpoint1 name2 $ encode "hello world!"--- > msg <- receiveMessage endpoint2--- > print msg--- > shutdown transport+-- >+-- > -- Needed for serialization+-- > import Data.Serialize+-- >+-- > main :: IO ()+-- > main = do+-- > let name1 = "endpoint1"+-- > name2 = "endpoint2"+-- > resolver = resolverFromList [(name1,"localhost:2000"),+-- > (name2,"localhost:2001")]+-- > transport <- newTCPTransport resolver+-- > endpoint1 <- newEndpoint [transport]+-- > endpoint2 <- newEndpoint [transport]+-- > Right () <- bindEndpoint endpoint1 name1+-- > Right () <- bindEndpoint endpoint2 name2+-- > sendMessage_ endpoint1 name2 $ encode "hello world!"+-- > msg <- receiveMessage endpoint2+-- > let Right txt = decode msg+-- > in print (txt :: String)+-- > Right () <- unbindEndpoint endpoint1 name1+-- > Right () <- unbindEndpoint endpoint2 name2+-- > shutdown transport {-| Endpoints are a locus of communication, used for sending and receive messages.@@ -141,6 +153,14 @@ return $ Right () {-|+Invoke 'bindEndpoint', but ignore any returned result (success or failure).+-}+bindEndpoint_ :: Endpoint -> Name -> IO ()+bindEndpoint_ endpoint name = do+ _ <- bindEndpoint endpoint name+ return ()++{-| Unbind an 'Endpoint' from a 'Name', after which the 'Endpoint' will eventually not receive messages sent to that 'Name'. Note that there is no guarantee that after 'Unbind' succeeds that additional messages to that 'Name' will not be delivered: the only guarantee@@ -157,6 +177,14 @@ Just binding -> do unbind binding return $ Right ()++{-|+Invoke 'unbindEndpoint', but ignore any returned result (success or failure).+-}+unbindEndpoint_ :: Endpoint -> Name -> IO ()+unbindEndpoint_ endpoint name = do+ _ <- unbindEndpoint endpoint name+ return () {-| Send a 'Message' to specific 'Name' via the indicated 'Endpoint'. While a successful
+ src/Network/RPC.hs view
@@ -0,0 +1,267 @@+{-# LANGUAGE ExistentialQuantification #-}+-----------------------------------------------------------------------------+-- |+-- Module : Network.RPC+-- Copyright : (c) Phil Hargett 2014+-- License : MIT (see LICENSE file)+--+-- Maintainer : phil@haphazardhouse.net+-- Stability : experimental+-- Portability : non-portable (requires STM)+--+-- An implementation of synchronous remote procedure calls+-- (<http://en.wikipedia.org/wiki/Remote_procedure_call RPC>) on top of+-- 'Network.Endpoints.Endpoint's.+--+-- Applications exporting services for use by other applications via+-- RPC call 'handle' to start listening for incoming RPC requests+-- for a specific 'Method'. If multiple functions or 'Method's are exported,+-- then separate calls to 'handle' are necessary, one for each exported 'Method'.+-- Each call to 'handle' produces a 'HandleSite' which may be used to terminate+-- future handling of RPCs for that specific method by calling 'hangup' on the+-- returned 'HandleSite'.+--+-- Applications wishing to make RPCs to other applications or services do so+-- by first constructing a 'CallSite', and then 'call'ing specific methods+-- on the target handler through that 'CallSite'.+--+-- Both single and multiple target RPCs are available, as are variants that+-- either wait indefinitely or at most for a defined timeout.+--+-----------------------------------------------------------------------------++module Network.RPC (++ Method,++ newCallSite,+ CallSite,++ call,+ callWithTimeout,+ gcall,+ gcallWithTimeout,++ HandleSite,+ handle,+ hangup++) where++-- local imports++import Network.Endpoints++-- external imports++import Control.Concurrent+import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Monad++import qualified Data.Map as M+import qualified Data.Set as S+import Data.Serialize++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------++type Method = String++data Request a = (Serialize a) => Request {+ requestId :: String,+ requestCaller :: Name,+ requestMethod :: Method,+ requestArgs :: a+}++instance (Serialize a) => Serialize (Request a) where + put req = do+ put $ requestId req+ put $ requestCaller req+ put $ requestMethod req+ put $ requestArgs req+ get = do+ rid <- get+ caller <- get+ method <- get+ args <- get+ return $ Request rid caller method args++data Response b = (Serialize b) => Response {+ responseId :: String,+ responseFrom :: Name,+ responseValue :: b+}++instance (Serialize b) => Serialize (Response b) where+ put rsp = do+ put $ responseId rsp+ put $ responseFrom rsp+ put $ responseValue rsp++ get = do+ rid <- get+ from <- get+ val <- get+ return $ Response rid from val++{-|+A call site is a location for making RPCs: it includes an endpoint+and a name by which recipients can return the call+-}+data CallSite = CallSite Endpoint Name++{-|+Create a new 'CallSite' using the indicated 'Endpoint' for sending+RPCs and using the specified 'Name' for receiving responses.+-}+newCallSite :: Endpoint -> Name -> CallSite+newCallSite = CallSite++{-|+Call a method with the provided arguments on the recipient with the given name.+A request will be made through the 'CallSite''s 'Endpoint', and then+the caller will wait until a matching response is received.+-}+call :: (Serialize a, Serialize b) => CallSite -> Name -> Method -> a -> IO b+call (CallSite endpoint from) name method args = do+ let req = Request {requestId = "",requestCaller = from,requestMethod = method, requestArgs = args}+ sendMessage_ endpoint name $ encode req+ selectMessage endpoint $ \msg -> do+ case decode msg of+ Left _ -> Nothing+ Right (Response rid _ value) -> do+ if rid == (requestId req)+ then Just value+ else Nothing++{-|+Call a method with the provided arguments on the recipient with the given name.+A request will be made through the 'CallSite''s 'Endpoint', and then+the caller will wait until a matching response is received. If a response+is received within the provided timeout (measured in microseconds), then+return the value wrapped in 'Just'; otherwise, if the timeout expires+before the call returns, then return 'Nothing.+-}+callWithTimeout :: (Serialize a, Serialize b) => CallSite -> Name -> Method -> Int-> a -> IO (Maybe b)+callWithTimeout site name method delay args = do+ resultOrTimeout <- race callIt (threadDelay delay)+ case resultOrTimeout of+ Left value -> return $ Just value+ Right _ -> return Nothing+ where+ callIt = call site name method args++{-|+Group call or RPC: call a method with the provided arguments on all the recipients with the given names.+A request will be made through the 'CallSite''s 'Endpoint', and then+the caller will wait until all matching responses are received.+-}+gcall :: (Serialize a, Serialize b) => CallSite -> [Name] -> Method -> a -> IO (M.Map Name b)+gcall (CallSite endpoint from) names method args = do+ let req = Request {requestId = "",requestCaller = from,requestMethod = method, requestArgs = args}+ sendAll req+ recvAll req M.empty+ where+ sendAll req = do+ forM_ names $ \name -> sendMessage_ endpoint name $ encode req+ recv req = selectMessage endpoint $ \msg -> do+ case decode msg of+ Left _ -> Nothing+ Right (Response rid name value) -> do+ if (rid == (requestId req)) && (elem name names)+ then Just (name,value)+ else Nothing+ recvAll req results = do+ (replier,result) <- recv req+ let newResults = M.insert replier result results+ replied = S.fromList $ M.keys newResults+ expected = S.fromList names+ if S.null (S.difference expected replied)+ -- we have everything+ then return newResults+ -- still waiting on some results, so keep receiving+ else recvAll req newResults++{-|+Group call or RPC but with a timeout: call a method with the provided arguments on all the+recipients with the given names. A request will be made through the 'CallSite''s 'Endpoint',+and then the caller will wait until all matching responses are received or the timeout occurs.+The returned 'M.Map' has a key for every 'Name' that was a target of the call, and the value+of that key will be @Nothing@ if no response was received before the timeout, or @Just value@+if a response was received.+-}+gcallWithTimeout :: (Serialize a, Serialize b) => CallSite -> [Name] -> Method -> Int -> a -> IO (M.Map Name (Maybe b))+gcallWithTimeout (CallSite endpoint from) names method delay args = do+ let req = Request {requestId = "",requestCaller = from,requestMethod = method, requestArgs = args}+ sendAll req+ allResults <- atomically $ newTVar M.empty+ responses <- race (recvAll req allResults) (threadDelay delay)+ case responses of+ Left results -> return $ complete results+ Right _ -> do+ partialResults <- atomically $ readTVar allResults+ return $ complete partialResults+ where+ sendAll req = do+ forM_ names $ \name -> sendMessage_ endpoint name $ encode req+ recv req = selectMessage endpoint $ \msg -> do+ case decode msg of+ Left _ -> Nothing+ Right (Response rid name value) -> do+ if (rid == (requestId req)) && (elem name names)+ then Just (name,value)+ else Nothing+ recvAll :: (Serialize b) => Request a -> TVar (M.Map Name b) -> IO (M.Map Name b)+ recvAll req allResults = do+ (replier,result) <- recv req+ newResults <- atomically $ do+ modifyTVar allResults $ \results -> M.insert replier result results+ readTVar allResults+ let replied = S.fromList $ M.keys newResults+ expected = S.fromList names+ if S.null (S.difference expected replied)+ -- we have everything+ then return newResults+ -- still waiting on some results, so keep receiving+ else recvAll req allResults+ -- Make sure the final results have an entry for every name,+ -- but put Nothing for those handlers that did not return a result in time+ complete :: (Serialize b) => M.Map Name b -> M.Map Name (Maybe b)+ complete partial = foldl (\final name -> M.insert name (M.lookup name partial) final) M.empty names++{-|+A 'HandleSite' is a just reference to the actual handler of a specific method.+Mostly for invoking 'hangup' on the handler, once it is no longer needed.+-}+data HandleSite = HandleSite Name (Async ())++{-|+Handle all RPCs to invoke the indicated 'Method' on the specified 'Endpoint',+until 'hangup' is called on the returned 'HandleSite'.+-}+handle :: (Serialize a, Serialize b) => Endpoint -> Name -> Method -> (a -> IO b) -> IO HandleSite+handle endpoint name method fn = do+ task <- async $ handleCall+ return $ HandleSite name task+ where+ handleCall = do+ (caller,rid,args) <- selectMessage endpoint $ \msg -> do+ case decode msg of+ Left _ -> Nothing+ Right (Request rid caller rmethod args) -> do+ if rmethod == method+ then Just (caller,rid,args)+ else Nothing+ result <- fn args+ sendMessage_ endpoint caller $ encode $ Response rid name result+ handleCall++{-|+Stop handling incoming RPCs for the indicated 'HandleSite'.+-}+hangup :: HandleSite -> IO ()+hangup (HandleSite _ task) = do+ cancel task+ return ()
tests/Tests.hs view
@@ -19,12 +19,13 @@ import qualified TestMemory as M import qualified TestTCP as T import qualified TestUDP as U+import qualified TestRPC as R ----------------------------------------------------------------------------- ----------------------------------------------------------------------------- main :: IO ()-main = do +main = do initLogging defaultMain tests @@ -49,6 +50,7 @@ ++ M.tests ++ T.tests ++ U.tests+ ++ R.tests -- Endpoint tests