49e7ac4 · typeable/wai

git.iko.soy/typeable/wai

Have stream bodies just return trailers directly

The 'TrailerFunc' approach was needlessly cumbersome since its power can be recovered with a combination of 'try' and '>>='. This way it's more clear what will happen to exceptions in the stream body: if they're uncaught, they'll escape into Warp and we'll reset the stream; if they are caught and the stream body returns trailers, Warp can't know about the exception, so it will necessarily send them and not reset the stream.

commit49e7ac43a90a4a989380c59c2fbbe8bc4976931d authorAndrew Pritchard <awpr@google.com> dateSun 13 Sep 2015, 02:20 (11 years ago) parent01a8df1

2 files changed+28−49

compare with master

unifiedsplit

wai/Network/Wai/HTTP2.hs+21−37 warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+7−12

wai/Network/Wai/HTTP2.hs+21−37

@@ -31,9 +31,7 @@ module Network.Wai.HTTP2

3131 , Responder(..)

3232 , RespondFunc

3333 , Body

34− , BodyOf

3534 , Chunk(..)

36− , TrailerFunc

3735 , Trailers

3836 -- * Server push

3937 , PushFunc

@@ -50,14 +48,13 @@ module Network.Wai.HTTP2

5048 , respondNotFound

5149 , respondWith

5250 -- ** Stream Bodies

53− , SimpleBody

5451 , streamFilePart

5552 , streamBuilder

5653 , streamSimple

5754 ) where

5855

5956import Blaze.ByteString.Builder (Builder)

60−import Control.Exception (Exception, SomeException, throwIO)

57+import Control.Exception (Exception, throwIO)

6158import Control.Monad.Trans.Cont (ContT(..))

6259import Data.ByteString (ByteString)

6360#if __GLASGOW_HASKELL__ < 709

@@ -76,6 +73,7 @@ import Network.Wai.Internal

7673 , Request(requestHeaders)

7774 , Response(..)

7875 , ResponseReceived(..)

76+ , StreamingBody

7977 , adjustForFilePart

8078 , chooseFilePart

8179 , tryGetFileSize

@@ -102,21 +100,12 @@ data Chunk = FileChunk FilePath FilePart | BuilderChunk Builder

102100

103101-- | The streaming body of a response. Equivalent to

104102-- 'Network.Wai.StreamingBody' except that it can also write file ranges and

105−-- return a result of type @a@.

106−type Body a = BodyOf Chunk a

103+-- return the stream's trailers.

104+type Body = (Chunk -> IO ()) -> IO () -> IO Trailers

107105

108−-- | Generalization of 'Body' to arbitrary chunk types.

109−--

110−-- 'Network.Wai.StreamingBody' is identical to @BodyOf Builder ()@.

111−type BodyOf c a = (c -> IO ()) -> IO () -> IO a

112−

113−-- | Create trailers based on the result of a stream body.

114−type TrailerFunc a = Either SomeException a -> Maybe Trailers -- TODO IO?

115−

116−-- | Given to 'Responders'; provide a status, headers, a stream body, and a way

117−-- to produce trailers, and we'll give you a token proving you called the

118−-- 'RespondFunc'.

119−type RespondFunc s a = H.Status -> H.ResponseHeaders -> TrailerFunc a -> Body a -> IO s

106+-- | Given to 'Responders'; provide a status, headers, and a stream body, and

107+-- we'll give you a token proving you called the 'RespondFunc'.

108+type RespondFunc s = H.Status -> H.ResponseHeaders -> Body -> IO s

120109

121110-- | The result of an 'HTTP2Application'; or, alternately, an application

122111-- that's independent of the request. This is a continuation-passing style

@@ -125,9 +114,9 @@ type RespondFunc s a = H.Status -> H.ResponseHeaders -> TrailerFunc a -> Body a

125114--

126115-- The respond function is similar to the one in 'Network.Wai.Application', but

127116-- it only takes a streaming body, the status and headers are curried, and it

128−-- passes on any result value from the stream body.

117+-- also produces trailers for the stream.

129118newtype Responder = Responder

130− { runResponder :: forall s. (forall a. RespondFunc s a) -> IO s }

119+ { runResponder :: forall s. RespondFunc s -> IO s }

131120

132121-- | A function given to an 'HTTP2Application' to initiate a server-pushed

133122-- stream. Its argument is the same as the result of an 'HTTP2Application', so

@@ -160,8 +149,8 @@ promiseHeaders p =

160149-- | Create a response body consisting of a single range of a file. Does not

161150-- set Content-Length or Content-Range headers. For that, use

162151-- 'respondFilePart' or 'respondFile'.

163−streamFilePart :: FilePath -> FilePart -> Body ()

164−streamFilePart path part write _ = write $ FileChunk path part

152+streamFilePart :: FilePath -> FilePart -> Body

153+streamFilePart path part write _ = write (FileChunk path part) >> return []

165154

166155-- | Respond with a single range of a file, adding the Accept-Ranges,

167156-- Content-Length and Content-Range headers and changing the status to 206 as

@@ -174,7 +163,7 @@ streamFilePart path part write _ = write $ FileChunk path part

174163respondFilePart :: H.Status -> H.ResponseHeaders -> FilePath -> FilePart -> Responder

175164respondFilePart s h path part = Responder $ \k -> do

176165 let (s', h') = adjustForFilePart s h part

177− k s' h' mempty $ streamFilePart path part

166+ k s' h' $ streamFilePart path part

178167

179168-- | Serve the requested range of the specified file (based on the Range

180169-- header), using the given 'H.Status' and 'H.ResponseHeaders' as a base. If

@@ -196,7 +185,7 @@ respondFileExists s h path size reqHdrs =

196185

197186-- | Respond with a minimal 404 page with the given headers.

198187respondNotFound :: H.ResponseHeaders -> Responder

199−respondNotFound h = Responder $ \k -> k H.notFound404 h' mempty $

188+respondNotFound h = Responder $ \k -> k H.notFound404 h' $

200189 streamBuilder "File not found."

201190 where

202191 contentType = (H.hContentType, "text/plain; charset=utf-8")

@@ -204,8 +193,8 @@ respondNotFound h = Responder $ \k -> k H.notFound404 h' mempty $

204193

205194-- | Construct a 'Responder' that will just call the 'RespondFunc' with the

206195-- given arguments.

207−respond :: H.Status -> H.ResponseHeaders -> TrailerFunc a -> Body a -> Responder

208−respond s h t b = Responder $ \k -> k s h t b

196+respond :: H.Status -> H.ResponseHeaders -> Body -> Responder

197+respond s h b = Responder $ \k -> k s h b

209198

210199-- | Fold the given bracketing action into a 'Responder'. Note the first

211200-- argument is isomorphic to @Codensity IO a@ or @forall s. ContT s IO a@, and

@@ -233,17 +222,12 @@ respondIO :: IO Responder -> Responder

233222respondIO io = Responder $ \k -> io >>= \r -> runResponder r k

234223

235224-- | Create a response body consisting of a single builder.

236−streamBuilder :: Builder -> Body ()

237−streamBuilder builder write _ = write $ BuilderChunk builder

238−

239−-- | Equivalent to 'Body' but only streaming 'Builder's.

240−--

241−-- 'Network.Wai.StreamingBody' is identical to @SimpleBody ()@.

242−type SimpleBody a = BodyOf Builder a

225+streamBuilder :: Builder -> Body

226+streamBuilder builder write _ = write (BuilderChunk builder) >> return []

243227

244228-- | Create a response body of a stream of 'Builder's.

245−streamSimple :: SimpleBody a -> Body a

246−streamSimple body write flush = body (write . BuilderChunk) flush

229+streamSimple :: StreamingBody -> Body

230+streamSimple body write flush = body (write . BuilderChunk) flush >> return []

247231

248232-- | Use a normal WAI 'Response' to send the response. Useful if you're

249233-- sharing code between HTTP\/2 applications and HTTP\/1 applications.

@@ -253,9 +237,9 @@ streamSimple body write flush = body (write . BuilderChunk) flush

253237promoteResponse :: Request -> Response -> Responder

254238promoteResponse req response = case response of

255239 (ResponseBuilder s h b) ->

256− Responder $ \k -> k s h mempty (streamBuilder b)

240+ Responder $ \k -> k s h (streamBuilder b)

257241 (ResponseStream s h body) ->

258− Responder $ \k -> k s h mempty (streamSimple body)

242+ Responder $ \k -> k s h (streamSimple body)

259243 (ResponseRaw _ fallback) -> promoteResponse req fallback

260244 (ResponseFile s h path mpart) -> maybe

261245 (respondFile s h path $ requestHeaders req)

warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+7−12

@@ -15,10 +15,9 @@ import Control.Applicative

1515#endif

1616import Control.Concurrent

1717import Control.Concurrent.STM

18−import Control.Exception (Exception, SomeException(..), AsyncException(..), throwIO)

18+import Control.Exception (Exception, SomeException(..), AsyncException(..))

1919import qualified Control.Exception as E

2020import Control.Monad (void, when)

21−import Data.Maybe (fromMaybe)

2221import Data.Typeable

2322import qualified Network.HTTP.Types as H

2423import Network.HTTP2

@@ -49,13 +48,13 @@ import qualified Network.Wai.Handler.Warp.Timeout as T

4948-- doesn't break that property.

5049--

5150-- This is the argument to a 'Responder'.

52−type Respond = IO () -> Stream -> (forall a. RespondFunc () a)

51+type Respond = IO () -> Stream -> RespondFunc ()

5352

5453-- | This function is passed to workers. They also pass responses from

5554-- 'HTTP2Application's to this function. This function enqueues commands for

5655-- the HTTP/2 sender.

5756response :: Context -> Manager -> ThreadContinue -> Respond

58−response ctx mgr tconf tickle strm s h trailers strmbdy = do

57+response ctx mgr tconf tickle strm s h strmbdy = do

5958 -- TODO(awpr) HEAD requests will still stream.

6059

6160 -- We must not exit this WAI application.

@@ -67,14 +66,14 @@ response ctx mgr tconf tickle strm s h trailers strmbdy = do

6766 -- After this work, this thread stops to decrease the number of workers.

6867 setThreadContinue tconf False

6968

70− runStream ctx OResponse tickle strm s h trailers strmbdy

69+ runStream ctx OResponse tickle strm s h strmbdy

7170

7271-- | Set up a waiter thread and run the stream body with functions to enqueue

7372-- 'Sequence's on the stream's queue.

7473runStream :: Context

7574 -> (Stream -> H.Status -> H.ResponseHeaders -> Aux -> Output)

7675 -> Respond

77−runStream Context{outputQ} mkOutput tickle strm s h trailers strmbdy = do

76+runStream Context{outputQ} mkOutput tickle strm s h strmbdy = do

7877 -- Since 'Body' is loop, we cannot control it.

7978 -- So, let's serialize 'Builder' with a designated queue.

8079 sq <- newTBQueueIO 10 -- fixme: hard coding: 10

@@ -91,12 +90,8 @@ runStream Context{outputQ} mkOutput tickle strm s h trailers strmbdy = do

9190 FileChunk path part -> SFile path part

9291 tickle

9392 flush = atomically $ writeTBQueue sq SFlush

94− -- Run the stream body, use its result to construct the trailers, and

95− -- re-raise any Exception so it will still be handled later.

96− x <- E.try $ strmbdy write flush

97− case (x, trailers x) of

98− (Left exc, Nothing) -> throwIO exc

99− (_, mts) -> atomically $ writeTBQueue sq $ SFinish $ fromMaybe [] mts

93+ trailers <- strmbdy write flush

94+ atomically $ writeTBQueue sq $ SFinish trailers

10095

10196-- | Handle abnormal termination of a stream: mark it as closed, send a reset

10297-- frame, and call the user's 'settingsOnException' handler if applicable.