courier 0.1.1.4 → 0.1.1.5

raw patch · 9 files changed

+338/−9 lines, 9 filesdep ~basePVP: minor bump suggested

API additions: PVP suggests at least a minor version bump

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• changes.md +7/−0
• courier.cabal +4/−2
• src/Network/RPC.hs +5/−5
• src/Network/RPC/Typed.hs +165/−0
• src/Network/Transport.hs +26/−1
• tests/TestRPCTyped.hs +106/−0
• tests/TestUtils.hs +8/−0
• tests/Tests.hs +2/−0
• tests/TransportTestSuite.hs +15/−1

@@ -1,3 +1,10 @@+0.1.1.5++  * Added withClient / withServer helpers+  * Removed support for 7.6 / 7.8, as they are before the FAMP proposal was         +    implemented (https://wiki.haskell.org/Functor-Applicative-Monad_Proposal)+  * Added typed equivalents for a selection of functions in Network.RPC+ 0.1.1.4    * Corrected an issue with memory transport where a message to an unbound destination would be dropped

@@ -1,5 +1,5 @@ name:                courier-version:             0.1.1.4+version:             0.1.1.5 synopsis:            A message-passing library for simplifying network applications tested-with:         GHC == 7.6.3, GHC == 7.8.4, GHC == 7.10.2, GHC == 7.10.3, GHC == 8.0.1 description:         Inspired by Erlang's simple message-passing facilities, courier provides roughly similar@@ -30,6 +30,7 @@                      tests/TestMailbox.hs                      tests/TestMemory.hs                      tests/TestRPC.hs+                     tests/TestRPCTyped.hs                      tests/TestTCP.hs                      tests/TestUDP.hs                      tests/TransportTestSuite.hs@@ -62,13 +63,14 @@                   Network.Transport.Sockets.UDP                   Control.Concurrent.Mailbox                   Network.RPC+                  Network.RPC.Typed    ghc-options:     -Wall    -- other-modules: -  build-depends:       base >=4.6 && <5,+  build-depends:       base >=4.8 && <5,                        async,                        bytestring,                        cereal,

@@ -51,7 +51,7 @@     hearAllTimeout,     Reply, -    HandleSite,+    HandleSite(..),     handle,     handleAll,     hangup,@@ -101,7 +101,7 @@ -} newtype RequestId = RequestId (Word32, Word32, Word32, Word32) deriving (Generic,Eq,Show) -instance Serialize (RequestId)+instance Serialize RequestId  {-| Create a new identifier for 'Request's@@ -285,7 +285,7 @@                 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 :: 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  {-|@@ -374,12 +374,12 @@ -} hearAll :: Endpoint -> Name -> IO (Method,Message,Reply Message) hearAll endpoint name = do-    (caller,rid,method,args) <- selectMessage endpoint anySelector+    (caller,rid,method,args) <- selectMessage endpoint selectorForAll     return (method,args,reply caller rid)     where         reply caller rid result =           sendMessage endpoint caller $ encode $ Response rid name result-        anySelector msg =+        selectorForAll msg =           case decode msg of             Left _ -> Nothing             Right (Request rid caller method args) -> Just (caller,rid,method,args)

@@ -0,0 +1,165 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  Network.RPC.Typed+-- Copyright   :  (c) Phil Hargett 2015+-- License     :  MIT (see LICENSE file)+--+-- Maintainer  :  phil@haphazardhouse.net+-- Stability   :  experimental+-- Portability :  non-portable (uses STM)+--+--+-----------------------------------------------------------------------------++module Network.RPC.Typed (++  call,+  callWithTimeout,+  gcallWithTimeout,+  hear,+  hearTimeout,+  handle,+  typedMethodSelector,++  module Network.RPC++) where++-- local imports++import Network.Endpoints+import Network.RPC hiding (call,callWithTimeout,gcallWithTimeout,hear,hearTimeout,handle)+import qualified Network.RPC as R++-- external imports++import Control.Concurrent+import Control.Concurrent.Async+import qualified Data.Map as M+import Data.Serialize++-- import Debug.Trace++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------++{-|+Call a method with the provided arguments on the recipient with the given name.++The caller will wait until a matching response is received.+-}+call :: (Serialize a,Serialize b) => CallSite -> Name -> Method -> a -> IO b+call cs name method args = do+  result <- R.call cs name method $ encode args+  let Right value = decode result+  return value++{-|+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 r) => CallSite -> Name -> Method -> Int-> a -> IO  (Maybe r)+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 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 r) => CallSite -> [Name] -> Method -> Int -> a -> IO (M.Map Name (Maybe r))+gcallWithTimeout cs names method delay args = do+  responses <- R.gcallWithTimeout cs names method delay $ encode args+  return $ decodeResponses responses+  where+    decodeResponses = M.map decodeResponse+      where+        decodeResponse maybeMsg = case maybeMsg of+          Nothing -> Nothing+          Just msg -> case decode msg of+            Left _ -> Nothing+            Right response -> Just response++{-|+Wait for a single incoming request to invoke the indicated 'Method' on the specified+'Endpoint'. Return both the method arguments and a 'Reply' function useful for sending+the reply.  A good pattern for using 'hear' will pattern match the result to a tuple of+the form @(args,reply)@, then use the args as needed to compute a result, and then+finally send the result back to the client by simply passing the result to reply: @reply result@.++The invoker of 'hear' must supply the 'Name' they have bound to the 'Endpoint', as this+helps the original requestor of the RPC differentiate responses when the RPC was a group+call.+-}+hear :: (Serialize a, Serialize r) => Endpoint -> Name -> Method -> IO (a,Reply r)+hear endpoint name method = do+  (caller,rid,args) <- selectMessage endpoint $ typedMethodSelector method+  return (args, reply caller rid)+  where+    reply caller rid result =+      sendMessage endpoint caller $ encode $ Response rid name $ encode result++{-|+Same as 'hear', except return 'Nothing' if no request received within the specified+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 r) => Endpoint -> Name -> Method -> Int -> IO (Maybe (a, Reply r))+hearTimeout endpoint name method timeout = do+  -- traceIO $ "listening for " ++ show method ++ " on " ++ show name+  req <- selectMessageTimeout endpoint timeout $ typedMethodSelector method+  case req of+    Just (caller,rid,args) -> do+      -- traceIO $ "heard " ++ show method ++ " on " ++ show name+      return $ Just (args, reply caller rid)+    Nothing -> do+      -- traceIO $ "did not hear " ++ method ++ " on " ++ show name+      return Nothing+  where+  reply caller rid result = sendMessage endpoint caller $ encode $ Response rid name $ encode result++{-|+Handle all RPCs to invoke the indicated 'Method' on the specified 'Endpoint',+until 'hangup' is called on the returned 'HandleSite'.+-}+handle :: (Serialize a, Serialize r) => Endpoint -> Name -> Method -> (a -> IO r) -> IO HandleSite+handle endpoint name method fn = do+    task <- async handleCall+    return $ HandleSite name task+    where+        handleCall = do+            (args,reply) <- hear endpoint name method+            result <- fn args+            reply result+            handleCall++{-|+A method selector that only matches if the message deserializes into+a type that matches arguments to a call.+-}+typedMethodSelector :: (Serialize a) => Method -> Message -> Maybe (Name,RequestId,a)+typedMethodSelector method msg =+  case decode msg of+    Left _ ->+      Nothing+    Right (Request rid caller rmethod bytes) ->+      if rmethod == method+        then case decode bytes of+          Left _ -> Nothing+          Right args -> Just (caller,rid,args)+        else Nothing

@@ -54,8 +54,9 @@   dispatchMessage,    withTransport,-   withEndpoint,+  withClient,+  withServer,    Binding(..),   withBinding,@@ -300,3 +301,27 @@   if destination == origin     then withConnections transport endpoint destinations fn     else withCompleteNetwork transport destinations endpoint origin fn++{-|+Helper for easily creating clients capable of sending messages on the `Transport`+and receiving messages at the provided `Name`+-}+withClient :: IO Transport -> Name -> (Endpoint -> IO a) -> IO a+withClient transportFactory name clientFn =+  withTransport transportFactory $ \transport -> do+    endpoint <- newEndpoint+    withEndpoint transport endpoint $+      withName endpoint name $+        clientFn endpoint++{-|+Helper for easily creating servers listening messages sent to the specified `Name`+over the provided `Transport`+-}+withServer :: IO Transport -> Name -> (Transport -> Endpoint -> IO a) -> IO a+withServer transportFactory name serverFn =+  withTransport transportFactory $ \transport -> do+    endpoint <- newEndpoint+    withEndpoint transport endpoint $+      withBinding transport endpoint name $+        serverFn transport endpoint

@@ -0,0 +1,106 @@+-----------------------------------------------------------------------------+-- |+-- Module      :  TestRPCTyped+-- Copyright   :  (c) Phil Hargett 2014+-- License     :  MIT (see LICENSE file)+--+-- Maintainer  :  phil@haphazardhouse.net+-- Stability   :  experimental+-- Portability :  non-portable (requires STM)+--+-- (..... module description .....)+--+-----------------------------------------------------------------------------+module TestRPCTyped (+    tests+) where++-- local imports++import Network.Endpoints+import Network.RPC.Typed+import Network.Transport.Memory++import TestUtils++-- external imports++import Control.Concurrent++import qualified Data.Map as M++import Test.Framework+import Test.HUnit+import Test.Framework.Providers.HUnit++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------++_log :: String+_log = "test.rpc"++tests :: [Test.Framework.Test]+tests = [+    testCase "call-typed-one-with-timeout" testOneHandlerWithTimeout,+    testCase "gcall-typed-three-handlers-with-timeout"testGroupCallWithTimeout+  ]++testOneHandlerWithTimeout :: Assertion+testOneHandlerWithTimeout = do+  let name1 = Name "endpoint1"+      name2 = Name "endpoint2"+      longer = (3 * 1000000 :: Int)+      shorter = (1 * 1000000 :: Int)+  withTransport newMemoryTransport $ \transport ->+    withNewEndpoint2 transport $ \endpoint1 endpoint2 ->+      withBinding transport endpoint1 name1 $+        withBinding transport endpoint2 name2 $+          withConnection transport endpoint1 name2 $ do+            -- first call with caller waiting longer than handler+            h1 <- handle endpoint2 name2 "foo" $ \msg -> do+                threadDelay shorter+                return $ msg ++ "!"+            let cs1 = newCallSite endpoint1 name1+            Just result1 <- callWithTimeout cs1 name2 "foo" longer "hello"+            assertEqual "Result not expected value" (Just "hello!") (Just result1)+            hangup h1+            -- now call with handler waiting longer than caller+            h2 <- handle endpoint2 name2 "foo" $ \msg -> do+                threadDelay longer+                return $ msg ++ "!"+            let cs2 = newCallSite endpoint1 name1+            result2 <- (callWithTimeout cs2 name2 "foo" shorter "hello")+            assertEqual "Result not expected value" Nothing (result2 :: Maybe String)+            hangup h2++testGroupCallWithTimeout :: Assertion+testGroupCallWithTimeout = do+  let name1 = Name "endpoint1"+      name2 = Name "endpoint2"+      name3 = Name "endpoint3"+      name4 = Name "endpoint4"+      longest = 750 * 1000 -- three quarters of a second+      longer = 500 * 1000 -- half a second+      shorter = 250 * 1000 -- quarter second+  withTransport newMemoryTransport $ \transport ->+    withNewEndpoint4 transport $ \endpoint1 endpoint2 endpoint3 endpoint4 -> do+      withBinding4 transport (endpoint1,name1) (endpoint2,name2) (endpoint3,name3) (endpoint4,name4) $+        withConnection3 transport endpoint1 name2 name3 name4 $ do+          h2 <- handle endpoint2 name2 "foo" $ \msg -> do+              threadDelay shorter+              return $ if msg == "hello" then "foo" else ""+          h3 <- handle endpoint3 name3 "foo" $ \msg ->  do+              threadDelay shorter+              return $ if msg == "hello" then "bar" else ""+          h4 <- handle endpoint4 name4 "foo" $ \msg ->  do+              threadDelay longest+              return $ if msg == "hello" then "baz" else ""+          let cs = newCallSite endpoint1 name1+          results <- gcallWithTimeout cs [name2,name3,name4] "foo" longer "hello"+          assertEqual "Foo not present in results" (Just $ Just "foo") (M.lookup name2 results)+          assertEqual "Bar not present in results" (Just $ Just "bar") (M.lookup name3 results)+          assertEqual "Baz shouldn't be present in results" (Just $ Nothing) (M.lookup name4 results)+          assertEqual "Unxpected number of results" 3 (M.size results)+          hangup h2+          hangup h3+          hangup h4

@@ -13,6 +13,7 @@      timeBound,     troubleshoot,+    shareTransport      )   where@@ -25,6 +26,7 @@ -- external imports import Control.Concurrent import Control.Concurrent.Async+import Control.Concurrent.STM import Control.Exception  import qualified Network.Socket as NS@@ -118,3 +120,9 @@     finally (do         updateGlobalLogger rootLoggerName (setLevel INFO)         fn) (updateGlobalLogger rootLoggerName (setLevel WARNING))++shareTransport :: IO Transport -> IO (IO Transport)+shareTransport transportFactory  = do+  transport <- transportFactory+  vTransport <- atomically $ newTVar transport+  return $ atomically $ readTVar vTransport

@@ -18,6 +18,7 @@ import qualified TestMailbox as MB import qualified TestMemory as M import qualified TestRPC as R+import qualified TestRPCTyped as RT import qualified TestTCP as T import qualified TestUDP as U @@ -49,3 +50,4 @@   MB.tests   ++ M.tests   ++ R.tests+  ++ RT.tests

@@ -27,7 +27,6 @@  -- external imports -import Control.Applicative import Control.Concurrent.Async  import qualified Data.Map as M@@ -50,6 +49,8 @@     testTransportEndpointSendReceive2SerialServers transport name1 name2,   testCase  (transportLabel ++ "-sendReceive-2-serial-clients") $     testTransportEndpointSendReceive2SerialClients transport name1 name2,+  testCase  (transportLabel ++ "-withClient-withServer") $+    testWithClientWithServer transport name1 name2,   testCase  (transportLabel ++ "-rpc-one-hear-call") $     testTransportOneHearCall transport name1 name2,   testCase  (transportLabel ++ "-rpc-one-call-hear") $@@ -129,6 +130,19 @@             msg <- receiveMessage endpoint2             assertEqual "Received message not same as sent" (Right "hello!") (decode msg)             return ()++testWithClientWithServer :: IO Transport -> Name -> Name -> Assertion+testWithClientWithServer transportFactory name1 name2 = timeLimited $ do+  sharedTransportFactory <- shareTransport transportFactory+  server2 <- async $ withServer sharedTransportFactory name2 $ \_ endpoint2 ->+    receiveMessage endpoint2+  withClient sharedTransportFactory name1 $ \endpoint1 -> do+    transport <- sharedTransportFactory+    withConnection transport endpoint1 name2 $ do+      sendMessage endpoint1 name2 $ encode "hello!"+      msg <- wait server2+      assertEqual "Received message not same as sent" (Right "hello!") (decode msg)+      return ()  testTransportOneHearCall :: IO Transport -> Name -> Name -> Assertion testTransportOneHearCall transportFactory name1 name2 = timeLimited $ do