| ... |
... |
@@ -304,47 +304,46 @@ handleAsyncReq pipe getit mii k = mask $ \restore -> do |
|
304
|
304
|
ack <- newEmptyMVar
|
|
305
|
305
|
|
|
306
|
306
|
-- Block waiting for request
|
|
307
|
|
- mbreq <- atomically $ do
|
|
|
307
|
+ (uq,bs) <- atomically $ do
|
|
308
|
308
|
readTVar (pipeWriteClosed pipe) >>= \case
|
|
309
|
309
|
-- We will not be able to write the reply, so no point in continuiting.
|
|
310
|
310
|
True -> throwSTM PipeWriteClosedException
|
|
311
|
311
|
False -> pure ()
|
|
312
|
|
- isClosed <- readTVar (pipeReadClosed pipe)
|
|
313
|
312
|
taken <- case mii of
|
|
314
|
313
|
Nothing -> tryReadTQueue $ requestsQueue am
|
|
315
|
314
|
Just ii -> tryReadTQueue =<< getMailbox am ii
|
|
316
|
315
|
case taken of
|
|
317
|
|
- Just req@(uq,_) -> pure (Just req)
|
|
318
|
|
- Nothing | isClosed -> pure Nothing -- empty and closed
|
|
319
|
|
- | otherwise -> retry -- wait for msg
|
|
320
|
|
-
|
|
321
|
|
- case mbreq of
|
|
322
|
|
- Nothing -> throwIO PipeReadClosedException -- can't expect reqs
|
|
323
|
|
- Just (uq, bs) -> do
|
|
324
|
|
- let reply p = do
|
|
325
|
|
- b <- atomically $ doneHandling am uq
|
|
326
|
|
- when b $ writePipe pipe (Reply uq (bimap toSerializableException runPut p))
|
|
327
|
|
- putMVar ack ()
|
|
328
|
|
- -- Run the continuation which must reply exactly once with 'writeReply'
|
|
329
|
|
- case runGetOrFail getit bs of
|
|
330
|
|
- Left (_, _, err) -> do
|
|
331
|
|
- let exc = DecodeAsyncReqException err
|
|
332
|
|
- reply (Left (toException exc))
|
|
333
|
|
- throwIO exc
|
|
334
|
|
- Right (_, _, it) ->
|
|
335
|
|
- restore (k (WriteReply reply) (setActive am uq) it)
|
|
336
|
|
-
|
|
337
|
|
- -- If there's an exception before the reply is sent, reply with an error instead.
|
|
338
|
|
- `catch` \(e :: SomeException) -> do
|
|
339
|
|
-
|
|
340
|
|
- tryTakeMVar ack >>= \case
|
|
341
|
|
- Just () ->
|
|
342
|
|
- -- Already replied
|
|
343
|
|
- throwIO e
|
|
344
|
|
- Nothing -> do
|
|
345
|
|
- -- Reply with an error
|
|
346
|
|
- reply (Left e)
|
|
347
|
|
- throwIO e
|
|
|
316
|
+ Just req -> pure req
|
|
|
317
|
+ Nothing -> readTVar (pipeReadClosed pipe) >>= \case
|
|
|
318
|
+ True -> throwSTM PipeReadClosedException -- can't expect reqs
|
|
|
319
|
+ False -> retry
|
|
|
320
|
+
|
|
|
321
|
+ let
|
|
|
322
|
+ reply p = do
|
|
|
323
|
+ b <- atomically $ doneHandling am uq
|
|
|
324
|
+ when b $ writePipe pipe (Reply uq (bimap toSerializableException runPut p))
|
|
|
325
|
+ putMVar ack ()
|
|
|
326
|
+
|
|
|
327
|
+ -- Run the continuation which must reply exactly once with 'writeReply'
|
|
|
328
|
+ case runGetOrFail getit bs of
|
|
|
329
|
+ Left (_, _, err) -> do
|
|
|
330
|
+ let exc = DecodeAsyncReqException err
|
|
|
331
|
+ reply (Left (toException exc))
|
|
|
332
|
+ throwIO exc
|
|
|
333
|
+ Right (_, _, it) ->
|
|
|
334
|
+ restore (k (WriteReply reply) (setActive am uq) it)
|
|
|
335
|
+
|
|
|
336
|
+ -- If there's an exception before the reply is sent, reply with an error instead.
|
|
|
337
|
+ `catch` \(e :: SomeException) -> do
|
|
|
338
|
+
|
|
|
339
|
+ tryTakeMVar ack >>= \case
|
|
|
340
|
+ Just () ->
|
|
|
341
|
+ -- Already replied
|
|
|
342
|
+ throwIO e
|
|
|
343
|
+ Nothing -> do
|
|
|
344
|
+ -- Reply with an error
|
|
|
345
|
+ reply (Left e)
|
|
|
346
|
+ throwIO e
|
|
348
|
347
|
|
|
349
|
348
|
|
|
350
|
349
|
-- | 'handleSyncReq' failed to decode an incoming request payload using the given @'Get' a@
|