courier 0.1.0.10 → 0.1.0.11
raw patch · 5 files changed
+98/−18 lines, 5 files
Files
- changes.md +28/−0
- courier.cabal +2/−2
- src/Network/Endpoints.hs +26/−0
- src/Network/RPC.hs +30/−16
- src/Network/Transport.hs +12/−0
changes.md view
@@ -1,3 +1,31 @@+0.1.0.11++ * API breaking change: RPC functions no longer take / return instances of Serialize,+ but they take / return Message values instead. The reason for this change is that+ while courier does use Serialize internally to simplify a variety of operations,+ putting instances of Serialize in RPC functions was misleading: applications would+ still need to take care to ensure that different RPC requests / responses could+ be differentiated, otherwise courier might accidentally deserialize a message+ to the wrong type at an application's request (e.g., inside a hear callback). This+ can lead to decoding errors, which may actually causes exceptions to be thrown+ in pure methods. By leaving the interface to use Message values, its not only+ consistent with the rest of Endpoint behavior, but it clarifies the responsibility+ of the calling application to manage their Message payloads on their own.+ + * Improvements to RPC. Previously, it was not possible to differentiate correctly+ between requests and responses in the same mailbox: the only check was that+ deserialization succeeded, which wasn't a sufficient test, and it would be possible+ to deserialize a response as a request, etc. This lead to spurious errors.+ + * Added functions for detecting presence of messages in an an endpoint's mailbox+ (while still leaving them unconsumed): mostly added this for applications that+ need to react to the presence of a particular message, but not actual consume+ such messages.++ * Exposed the Request / Response types used by RPC. Applications may benefit+ from detecting them or otherwise manipulating them outside of normal RPC+ flow, they are exposed here for that purpose.+ 0.1.0.10 * Added hear primitive as one-shot counterpart to call. While handle
courier.cabal view
@@ -1,5 +1,5 @@ name: courier-version: 0.1.0.10+version: 0.1.0.11 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, @@ -11,7 +11,7 @@ (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+ the distributed algorithm itself. With this in mind, this package 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.).
src/Network/Endpoints.hs view
@@ -44,6 +44,8 @@ -- * Selective message reception selectMessage, selectMessageTimeout,+ detectMessage,+ detectMessageTimeout, dispatchMessage, dispatchMessageTimeout, @@ -272,6 +274,30 @@ selectMessageTimeout :: Endpoint -> Int -> (Message -> Maybe v) -> IO (Maybe v) selectMessageTimeout endpoint delay testFn = do resultOrTimeout <- race (selectMessage endpoint testFn) (threadDelay delay)+ case resultOrTimeout of+ Left result -> return $ Just result+ Right () -> return Nothing++{-|+Find a 'Message' in the 'Endpoint' 'Mailbox' matching the supplied+test function, or block until one is available. Note that any such message+is left in the mailbox, and thus repeated calls to this function could find the+message if it is not consumed immediately.+-}+detectMessage :: Endpoint -> (Message -> Maybe v) -> IO v+detectMessage endpoint testFn = do+ msg <- atomically $ findMailbox (endpointMailbox endpoint) testFn+ return msg++{-|+Find a 'Message' in the 'Endpoint' 'Mailbox' matching the supplied+test function, or block until either one is available or the timeout expires.+Note that any such message is left in the mailbox, and thus repeated calls+to this function could find the message if it is not consumed immediately.+-}+detectMessageTimeout :: Endpoint -> Int -> (Message -> Maybe v) -> IO (Maybe v)+detectMessageTimeout endpoint delay testFn = do+ resultOrTimeout <- race (detectMessage endpoint testFn) (threadDelay delay) case resultOrTimeout of Left result -> return $ Just result Right () -> return Nothing
src/Network/RPC.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE DeriveGeneric #-} ----------------------------------------------------------------------------- -- | -- Module : Network.RPC@@ -48,8 +49,11 @@ HandleSite, handle,- hangup+ hangup, + Request(..),+ Response(..)+ ) where -- local imports@@ -70,46 +74,56 @@ import Data.UUID.V4 import Data.Word +import GHC.Generics hiding (from)+ -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- type Method = String +data RPCMessageType = Req | Rsp deriving (Eq,Show,Enum,Generic)++instance Serialize RPCMessageType+ type RequestId = (Word32, Word32, Word32, Word32) -data Request a = (Serialize a) => Request {+data Request = Request { requestId :: RequestId, requestCaller :: Name, requestMethod :: Method,- requestArgs :: a+ requestArgs :: Message } -instance (Serialize a) => Serialize (Request a) where +instance Serialize Request where put req = do+ put Req put $ requestId req put $ requestCaller req put $ requestMethod req put $ requestArgs req get = do+ Req <- get rid <- get caller <- get method <- get args <- get return $ Request rid caller method args -data Response b = (Serialize b) => Response {+data Response = Response { responseId :: RequestId, responseFrom :: Name,- responseValue :: b+ responseValue :: Message } -instance (Serialize b) => Serialize (Response b) where+instance Serialize Response where put rsp = do+ put Rsp put $ responseId rsp put $ responseFrom rsp put $ responseValue rsp get = do+ Rsp <- get rid <- get from <- get val <- get@@ -130,10 +144,10 @@ {-| 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 -> Name -> Method -> Message -> IO Message call (CallSite endpoint from) name method args = do ruuid <- nextRandom let req = Request {requestId = toWords ruuid,requestCaller = from,requestMethod = method, requestArgs = args}@@ -154,7 +168,7 @@ 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 :: CallSite -> Name -> Method -> Int-> Message -> IO (Maybe Message) callWithTimeout site name method delay args = do resultOrTimeout <- race callIt (threadDelay delay) case resultOrTimeout of@@ -168,7 +182,7 @@ 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 -> [Name] -> Method -> Message -> IO (M.Map Name Message) gcall (CallSite endpoint from) names method args = do ruuid <- nextRandom let req = Request {requestId = toWords ruuid,requestCaller = from,requestMethod = method, requestArgs = args}@@ -203,7 +217,7 @@ 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 -> [Name] -> Method -> Int -> Message -> IO (M.Map Name (Maybe Message)) gcallWithTimeout (CallSite endpoint from) names method delay args = do ruuid <- nextRandom let req = Request {requestId = toWords ruuid,requestCaller = from,requestMethod = method, requestArgs = args}@@ -225,7 +239,7 @@ 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 :: Request -> TVar (M.Map Name Message) -> IO (M.Map Name Message) recvAll req allResults = do (replier,result) <- recv req newResults <- atomically $ do@@ -259,7 +273,7 @@ helps the original requestor of the RPC differentiate responses when the RPC was a group call. -}-hear :: (Serialize a,Serialize b) => Endpoint -> Name -> Method -> IO (a,Reply b)+hear :: Endpoint -> Name -> Method -> IO (Message,Reply Message) hear endpoint name method = do (caller,rid,args) <- selectMessage endpoint $ \msg -> do case decode msg of@@ -279,7 +293,7 @@ timeout (measured in microseconds), or return a 'Just' instance containing both the method arguments and a 'Reply' function useful for sending the reply. -}-hearTimeout :: (Serialize a,Serialize b) => Endpoint -> Name -> Method -> Int -> IO (Maybe (a,Reply b))+hearTimeout :: Endpoint -> Name -> Method -> Int -> IO (Maybe (Message,Reply Message)) hearTimeout endpoint name method timeout = do req <- selectMessageTimeout endpoint timeout $ \msg -> do case decode msg of@@ -305,7 +319,7 @@ 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 -> (Message -> IO Message) -> IO HandleSite handle endpoint name method fn = do task <- async $ handleCall return $ HandleSite name task
src/Network/Transport.hs view
@@ -14,6 +14,18 @@ -- that specific 'Transport' implementations should provide in order to deliver messages -- for 'Endpoint's. --+-- The definition of a transport is deliberately low-level in nature. Unless a specific+-- transport describes itself as supporting features like guaranteed delivery, applications+-- should NOT assume that message delivery is reliable.+--+-- For example, if a sender sends a message to a name that has not yet been bound, then +-- immediately waits on the response for that message, then the application may hang, +-- as the original message may have been dropped.+--+-- However, many application may find it easier to push features such as reliable+-- message delivery into a custom transport, leaving the application only having+-- to concern itself with the messages being delivered rather than how they arrive.+-- ----------------------------------------------------------------------------- module Network.Transport (