905595c · curryer · git.iko.soy

git.iko.soy/curryer

rename MessageHandlers to RequestHandlers

add haddock docs add simple key-value client/server examples

commit905595c4765b98cb5106cf20bb81ebbefd97fd64 authorAgentM <agentm@themactionfaction.com> dateFri 18 Dec 2020, 21:19 (6 years ago) parent8f8eeab

4 files changed+38−25

compare with master

unifiedsplit

bench/Basic.hs+3−3 curryer.cabal+12−1 src/Network/RPC/Curryer/Server.hs+15−11 test/Curryer/Test/Basic.hs+8−10

bench/Basic.hs+3−3

@@ -15,7 +15,7 @@ main :: IO ()

1515main = do

1616 --start server shared across benchmarks- gather stats on round-trip requests-responses

1717 portReadyMVar <- newEmptyMVar

18− server <- async (serve benchmarkServerMessageHandlers () localHostAddr 0 (Just portReadyMVar))

18+ server <- async (serve benchmarkServerRequestHandlers () localHostAddr 0 (Just portReadyMVar))

1919 --wait for server to be ready

2020 (SockAddrInet port _) <- takeMVar portReadyMVar

2121 clientConn <- connect [] localHostAddr port

@@ -53,8 +53,8 @@ data WaitMillisecondsResp = WaitMillisecondsResp

5353 deriving (Generic, Show, Eq)

5454 deriving Serialise via WineryVariant WaitMillisecondsResp

5555

56−benchmarkServerMessageHandlers :: MessageHandlers a

57−benchmarkServerMessageHandlers =

56+benchmarkServerRequestHandlers :: RequestHandlers a

57+benchmarkServerRequestHandlers =

5858 [RequestHandler $ \_ (WaitMillisecondsReq ms) -> do

5959 threadDelay (1000 * ms)

6060 pure WaitMillisecondsResp,

curryer.cabal+12−1

@@ -68,4 +68,15 @@ Benchmark perf

6868 GHC-Options: -Wall -threaded -rtsopts -fprof-auto

6969 HS-Source-Dirs: ./bench

7070

71−

71+Executable SimpleKeyValueServer

72+ Build-Depends: stm-containers, stm, base, curryer, winery

73+ Main-Is: examples/SimpleKeyValueServer.hs

74+ GHC-Options: -Wall -threaded

75+ Default-Language: Haskell2010

76+

77+Executable SimpleKeyValueClient

78+ Build-Depends: stm, base, curryer, winery, optparse-generic

79+ Main-Is: examples/SimpleKeyValueClient.hs

80+ GHC-Options: -Wall -threaded

81+ Default-Language: Haskell2010

82+

src/Network/RPC/Curryer/Server.hs+15−11

@@ -86,6 +86,7 @@ type TimeoutMicroseconds = Int

8686deriving instance Generic Fingerprint

8787deriving via WineryVariant Fingerprint instance Serialise Fingerprint

8888

89+-- | Internal type used to mark envelope types.

8990data MessageType = RequestMessage TimeoutMicroseconds

9091 | ResponseMessage

9192 | TimeoutResponseMessage

@@ -93,15 +94,17 @@ data MessageType = RequestMessage TimeoutMicroseconds

9394 deriving (Generic, Show)

9495 deriving Serialise via WineryVariant MessageType

9596

96−type MessageHandlers serverState = [RequestHandler serverState]

97−

97+-- | A list of `RequestHandler`s.

98+type RequestHandlers serverState = [RequestHandler serverState]

99+

100+-- |

98101data RequestHandler serverState where

99102 -- | create a request handler with a response

100103 RequestHandler :: forall a b serverState. (Serialise a, Serialise b) => (ConnectionState serverState -> a -> IO b) -> RequestHandler serverState

101104 -- | create an asynchronous request handler where the client does not expect nor await a response

102105 AsyncRequestHandler :: forall a serverState. Serialise a => (ConnectionState serverState -> a -> IO ()) -> RequestHandler serverState

103106

104−-- | Passed to

107+-- | Server state sent in via `serve` and passed to `RequestHandler`s.

105108data ConnectionState a = ConnectionState {

106109 connectionServerState :: a,

107110 connectionSocket :: Locking Socket

@@ -133,7 +136,7 @@ instance Serialise UUID where

133136 x -> error $ "invalid schema element " <> show x

134137 decodeCurrent = B.decode . BSL.fromStrict <$> (decodeVarInt >>= getBytes)

135138

136−

139+-- | Errors from remote calls.

137140data ConnectionError = CodecError String -- show of WineryException from exception initiator which cannot otherwise be transmitted over a line due to dependencies on TypeReps

138141 | TimeoutError

139142 | ExceptionError String

@@ -146,7 +149,7 @@ allHostAddrs,localHostAddr :: HostAddr

146149allHostAddrs = (0,0,0,0)

147150localHostAddr = (127,0,0,1)

148151

149−-- Each message is lxggength-prefixed by a 32-bit unsigned length.

152+-- Each message is length-prefixed by a 32-bit unsigned length.

150153envelopeP :: Parser IO Word8 Envelope

151154envelopeP = do

152155 let s = FL.toList

@@ -207,9 +210,10 @@ uuidP = do

207210type NewConnectionHandler msg = IO (Maybe msg)

208211

209212type NewMessageHandler req resp = req -> IO resp

210−

213+

214+-- | Listen for new connections and handle requests which are passed the server state 's'. The MVar SockAddr can be be optionally used to know when the server is ready for processing requests.

211215serve ::

212− MessageHandlers s->

216+ RequestHandlers s->

213217 s ->

214218 HostAddr ->

215219 PortNumber ->

@@ -242,14 +246,16 @@ matchEnvelope envelope dispatchf =

242246 Nothing -> Nothing

243247 Just decoded -> Just (dispatchf, decoded)

244248

249+-- | Called by `serve` to process incoming envelope requests. Never returns, so use `async` to spin it off on another thread.

245250serverEnvelopeHandler ::

246251 Locking Socket

247− -> MessageHandlers s

252+ -> RequestHandlers s

248253 -> s

249254 -> Envelope

250255 -> IO ()

251256serverEnvelopeHandler _ _ _ (Envelope _ TimeoutResponseMessage _ _) = pure ()

252257serverEnvelopeHandler _ _ _ (Envelope _ ExceptionResponseMessage _ _) = pure ()

258+serverEnvelopeHandler _ _ _ (Envelope _ ResponseMessage _ _) = pure ()

253259serverEnvelopeHandler sockLock msgHandlers serverState envelope@(Envelope _ (RequestMessage timeoutms) msgId _) = do

254260 --find first matching handler

255261 let runTimeout :: IO b -> IO (Maybe b)

@@ -291,11 +297,9 @@ serverEnvelopeHandler sockLock msgHandlers serverState envelope@(Envelope _ (Req

291297 let env = Envelope (fingerprint (show exc)) ExceptionResponseMessage msgId (serialise (show exc)) in

292298 sendEnvelope env sockLock

293299 Right () -> pure ()

294−serverEnvelopeHandler _ _ _ (Envelope _ ResponseMessage _ _) = error "server received response message"

295300

296−type EnvelopeHandler = Envelope -> IO ()

297301

298−type AsyncMessageHandler a = a -> IO ()

302+type EnvelopeHandler = Envelope -> IO ()

299303

300304drainSocketMessages :: Socket -> EnvelopeHandler -> IO ()

301305drainSocketMessages sock envelopeHandler = do

test/Curryer/Test/Basic.hs+8−10

@@ -61,8 +61,8 @@ data AsyncHelloReq = AsyncHelloReq String

6161 deriving (Generic, Show)

6262 deriving Serialise via WineryVariant AsyncHelloReq

6363

64−testServerMessageHandlers :: Maybe (MVar String) -> MessageHandlers ()

65−testServerMessageHandlers mAsyncMVar =

64+testServerRequestHandlers :: Maybe (MVar String) -> RequestHandlers ()

65+testServerRequestHandlers mAsyncMVar =

6666 [ RequestHandler $ \_ (AddTwoNumbersReq x y) -> pure (x + y)

6767 , RequestHandler $ \_ (TestCallMeBackReq s) ->

6868 case mAsyncMVar of

@@ -88,7 +88,7 @@ testSimpleCall :: Assertion

8888testSimpleCall = do

8989 readyVar <- newEmptyMVar

9090

91− server <- async (serve (testServerMessageHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

91+ server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

9292 --wait for server to be ready

9393 (SockAddrInet port _) <- takeMVar readyVar

9494 conn <- connect [] localHostAddr port

@@ -107,7 +107,7 @@ testAsyncServerCall = do

107107 let clientAsyncHandlers =

108108 [ClientAsyncRequestHandler (\(AsyncHelloReq s) ->

109109 putMVar receivedAsyncMessageVar s)]

110− server <- async (serve (testServerMessageHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))

110+ server <- async (serve (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))

111111 (SockAddrInet port _) <- takeMVar portReadyVar

112112 conn <- connect clientAsyncHandlers localHostAddr port

113113 Right () <- asyncCall conn (AsyncHelloReq "welcome")

@@ -126,7 +126,7 @@ testAsyncClientCall = do

126126 portReadyVar <- newEmptyMVar

127127 receivedAsyncMessageVar <- newEmptyMVar

128128

129− server <- async (serve (testServerMessageHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))

129+ server <- async (serve (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))

130130 (SockAddrInet port _) <- takeMVar portReadyVar

131131

132132 conn <- connect [] localHostAddr port

@@ -141,7 +141,7 @@ testSyncClientCallTimeout :: Assertion

141141testSyncClientCallTimeout = do

142142 readyVar <- newEmptyMVar

143143

144− server <- async (serve (testServerMessageHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

144+ server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

145145 --wait for server to be ready

146146 (SockAddrInet port _) <- takeMVar readyVar

147147 conn <- connect [] localHostAddr port

@@ -154,7 +154,7 @@ testSyncException :: Assertion

154154testSyncException = do

155155 readyVar <- newEmptyMVar

156156

157− server <- async (serve (testServerMessageHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

157+ server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

158158 (SockAddrInet port _) <- takeMVar readyVar

159159

160160 conn <- connect [] localHostAddr port

@@ -173,10 +173,8 @@ testMultithreadedClient :: Assertion

173173testMultithreadedClient = do

174174 readyVar <- newEmptyMVar

175175

176− server <- async (serve (testServerMessageHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

176+ server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

177177 (SockAddrInet port _) <- takeMVar readyVar

178− {-let clientHandler :: AsyncMessageHandler TestResponse

179− clientHandler _ = error "async handler called"-}

180178 conn <- connect [] localHostAddr port

181179 let bigString = replicate (1024 * 1000) 'x'

182180 replicateM_ 10 $ do