@@ -22,7 +22,7 @@ import Data.Typeable
22import qualified Network.HTTP.Types as H
22import qualified Network.HTTP.Types as H
23import Network.HTTP2
23import Network.HTTP2
24import Network.HTTP2.Priority
24import Network.HTTP2.Priority
25−import Network.Wai.HTTP2 (Http2Application, Response, absurd)
25+import Network.Wai.HTTP2 (Http2Application, Responder, Response, absurd)
26import Network.Wai hiding (Response, responseStatus)
26import Network.Wai hiding (Response, responseStatus)
27import Network.Wai.Handler.Warp.HTTP2.EncodeFrame
27import Network.Wai.Handler.Warp.HTTP2.EncodeFrame
28import Network.Wai.Handler.Warp.HTTP2.Manager
28import Network.Wai.Handler.Warp.HTTP2.Manager
@@ -75,6 +75,14 @@ runStream Context{outputQ} mkOutput tickle strm sq (s, h, strmbdy) = do
75 flush = atomically $ writeTBQueue sq SFlush
75 flush = atomically $ writeTBQueue sq SFlush
76 strmbdy push flush
76 strmbdy push flush
78+runResponder :: Responder -> Respond -> IO () -> Stream -> IO ()
79+runResponder responder respond tickle strm = do
80+ -- Since 'Body' is loop, we cannot control it.
81+ -- So, let's serialize 'Builder' with a designated queue.
82+ sq <- newTBQueueIO 10 -- fixme: hard coding: 10
83+ trailers <- responder $ respond tickle strm sq
84+ atomically $ writeTBQueue sq $ SFinish trailers
78cleanupStream :: Context -> S.Settings -> Stream -> Maybe Request -> Maybe SomeException -> IO ()
86cleanupStream :: Context -> S.Settings -> Stream -> Maybe Request -> Maybe SomeException -> IO ()
79cleanupStream ctx@Context{outputQ} set strm req me = do
87cleanupStream ctx@Context{outputQ} set strm req me = do
80 closed ctx strm Killed
88 closed ctx strm Killed
@@ -103,20 +111,17 @@ worker ctx@Context{inputQ} set tm app respond = do
104 go sinfo tcont th = do
112 go sinfo tcont th = do
105 setThreadContinue tcont True
113 setThreadContinue tcont True
106− -- Since 'Body' is loop, we cannot control it.
114+
107− -- So, let's serialize 'Builder' with a designated queue.
108− sq <- newTBQueueIO 10 -- fixme: hard coding: 10
109 ex <- E.try $ do
115 ex <- E.try $ do
110 T.pause th
116 T.pause th
111 Input strm req <- atomically $ readTQueue inputQ
117 Input strm req <- atomically $ readTQueue inputQ
112 setStreamInfo sinfo strm req
118 setStreamInfo sinfo strm req
113 T.resume th
119 T.resume th
114 T.tickle th
120 T.tickle th
121+ let responder = app (req, flip (const absurd))
115− app (req, flip (const absurd)) $ respond tcont (T.tickle th) strm sq
122+ runResponder responder (respond tcont) (T.tickle th) strm
116 cont1 <- case ex of
123 cont1 <- case ex of
117− Right trailers -> do
118− atomically $ writeTBQueue sq $ SFinish trailers
119− return True
124+ Right () -> return True
120 Left e@(SomeException _)
125 Left e@(SomeException _)
121 | Just Break <- E.fromException e -> do
126 | Just Break <- E.fromException e -> do
122 cleanup sinfo Nothing
127 cleanup sinfo Nothing