@@ -38,13 +38,13 @@ import Network.Wai.Internal (Response(..), ResponseReceived(..), ResponseReceive
3838-- This type implements the second argument @Response -> IO ResponseReceived@
3939-- with extra arguments.
4040type Responder = ThreadContinue -> T.Handle -> Stream -> Request ->
41− Response -> IO ResponseReceived
41+ TBQueue Sequence -> Response -> IO ResponseReceived
4242
4343-- | This function is passed to workers.
4444-- They also pass 'Response's from 'Http2Application's to this function.
4545-- This function enqueues commands for the HTTP/2 sender.
4646response :: Context -> Manager -> Responder
47−response Context{outputQ} mgr tconf th strm req rsp = do
47+response Context{outputQ} mgr tconf th strm req sq rsp = do
4848 case rsp of
4949 ResponseStream _ _ strmbdy -> do
5050 -- We must not exit this WAI application.
@@ -56,9 +56,6 @@ response Context{outputQ} mgr tconf th strm req rsp = do
5656 -- After this work, this thread stops to decease
5757 -- the number of workers.
5858 setThreadContinue tconf False
59− -- Since 'StreamingBody' is loop, we cannot control it.
60− -- So, let's serialize 'Builder' with a designated queue.
61− sq <- newTBQueueIO 10 -- fixme: hard coding: 10
6259 tvar <- newTVarIO SyncNone
6360 let out = OResponse strm rsp (Persist sq tvar)
6461 -- Since we must not enqueue an empty queue to the priority
@@ -71,7 +68,6 @@ response Context{outputQ} mgr tconf th strm req rsp = do
7168 T.tickle th
7269 flush = atomically $ writeTBQueue sq SFlush
7370 strmbdy push flush
74− atomically $ writeTBQueue sq $ SFinish []
7571 _ -> do
7672 setThreadContinue tconf True
7773 let hasBody = requestMethod req /= H.methodHead
@@ -95,15 +91,20 @@ worker ctx@Context{inputQ,outputQ} set tm app responder = do
9591 where
9692 go sinfo tcont th = do
9793 setThreadContinue tcont True
94+ -- Since 'StreamingBody' is loop, we cannot control it.
95+ -- So, let's serialize 'Builder' with a designated queue.
96+ sq <- newTBQueueIO 10 -- fixme: hard coding: 10
9897 ex <- E.try $ do
9998 T.pause th
10099 Input strm req <- atomically $ readTQueue inputQ
101100 setStreamInfo sinfo strm req
102101 T.resume th
103102 T.tickle th
104− app req $ responder tcont th strm req
103+ app req $ responder tcont th strm req sq
105104 cont1 <- case ex of
106− Right ResponseReceived -> return True
105+ Right ResponseReceived -> do
106+ atomically $ writeTBQueue sq $ SFinish []
107+ return True
107108 Left e@(SomeException _)
108109 | Just Break <- E.fromException e -> do
109110 cleanup sinfo Nothing