@@ -38,13 +38,13 @@ import Network.Wai.Internal (Response(..), ResponseReceived(..), ResponseReceive
38-- This type implements the second argument @Response -> IO ResponseReceived@
38-- This type implements the second argument @Response -> IO ResponseReceived@
39-- with extra arguments.
39-- with extra arguments.
40type Responder = ThreadContinue -> T.Handle -> Stream -> Request ->
40type Responder = ThreadContinue -> T.Handle -> Stream -> Request ->
41− Response -> IO ResponseReceived
41+ TBQueue Sequence -> Response -> IO ResponseReceived
43-- | This function is passed to workers.
43-- | This function is passed to workers.
44-- They also pass 'Response's from 'Http2Application's to this function.
44-- They also pass 'Response's from 'Http2Application's to this function.
45-- This function enqueues commands for the HTTP/2 sender.
45-- This function enqueues commands for the HTTP/2 sender.
46response :: Context -> Manager -> Responder
46response :: 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
48 case rsp of
48 case rsp of
49 ResponseStream _ _ strmbdy -> do
49 ResponseStream _ _ strmbdy -> do
50 -- We must not exit this WAI application.
50 -- We must not exit this WAI application.
@@ -56,9 +56,6 @@ response Context{outputQ} mgr tconf th strm req rsp = do
56 -- After this work, this thread stops to decease
56 -- After this work, this thread stops to decease
57 -- the number of workers.
57 -- the number of workers.
58 setThreadContinue tconf False
58 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
62 tvar <- newTVarIO SyncNone
59 tvar <- newTVarIO SyncNone
63 let out = OResponse strm rsp (Persist sq tvar)
60 let out = OResponse strm rsp (Persist sq tvar)
64 -- Since we must not enqueue an empty queue to the priority
61 -- 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
71 T.tickle th
68 T.tickle th
72 flush = atomically $ writeTBQueue sq SFlush
69 flush = atomically $ writeTBQueue sq SFlush
73 strmbdy push flush
70 strmbdy push flush
74− atomically $ writeTBQueue sq $ SFinish []
76 setThreadContinue tconf True
72 setThreadContinue tconf True
77 let hasBody = requestMethod req /= H.methodHead
73 let hasBody = requestMethod req /= H.methodHead
@@ -95,15 +91,20 @@ worker ctx@Context{inputQ,outputQ} set tm app responder = do
96 go sinfo tcont th = do
92 go sinfo tcont th = do
97 setThreadContinue tcont True
93 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
98 ex <- E.try $ do
97 ex <- E.try $ do
99 T.pause th
98 T.pause th
100 Input strm req <- atomically $ readTQueue inputQ
99 Input strm req <- atomically $ readTQueue inputQ
101 setStreamInfo sinfo strm req
100 setStreamInfo sinfo strm req
102 T.resume th
101 T.resume th
103 T.tickle th
102 T.tickle th
104− app req $ responder tcont th strm req
103+ app req $ responder tcont th strm req sq
105 cont1 <- case ex of
104 cont1 <- case ex of
106− Right ResponseReceived -> return True
105+ Right ResponseReceived -> do
106+ atomically $ writeTBQueue sq $ SFinish []
107 Left e@(SomeException _)
108 Left e@(SomeException _)
108 | Just Break <- E.fromException e -> do
109 | Just Break <- E.fromException e -> do
109 cleanup sinfo Nothing
110 cleanup sinfo Nothing