@@ -2,6 +2,8 @@
22{-# LANGUAGE NamedFieldPuns #-}
33{-# LANGUAGE PatternGuards #-}
44{-# LANGUAGE CPP #-}
5+{-# LANGUAGE ScopedTypeVariables #-}
6+{-# LANGUAGE OverloadedStrings #-}
57
68module Network.Wai.Handler.Warp.HTTP2.Worker (
79 Responder
@@ -17,18 +19,23 @@ import Control.Concurrent.STM
1719import Control.Exception (Exception, SomeException(..), AsyncException(..))
1820import qualified Control.Exception as E
1921import Control.Monad (void, when)
22+import Data.ByteString.Builder (byteString)
2023import Data.Typeable
2124import qualified Network.HTTP.Types as H
2225import Network.HTTP2
2326import Network.HTTP2.Priority
2427import Network.Wai
28+import Network.Wai.Handler.Warp.File
29+import Network.Wai.Handler.Warp.FileInfoCache
2530import Network.Wai.Handler.Warp.HTTP2.EncodeFrame
2631import Network.Wai.Handler.Warp.HTTP2.Manager
2732import Network.Wai.Handler.Warp.HTTP2.Types
33+import Network.Wai.Handler.Warp.Header
2834import Network.Wai.Handler.Warp.IORef
2935import qualified Network.Wai.Handler.Warp.Response as R
3036import qualified Network.Wai.Handler.Warp.Settings as S
3137import qualified Network.Wai.Handler.Warp.Timeout as T
38+import Network.Wai.Handler.Warp.Types
3239import Network.Wai.Internal (Response(..), ResponseReceived(..), ResponseReceived(..))
3340
3441----------------------------------------------------------------
@@ -42,42 +49,88 @@ type Responder = ThreadContinue -> T.Handle -> Stream -> Request ->
4249-- | This function is passed to workers.
4350-- They also pass 'Response's from 'Application's to this function.
4451-- This function enqueues commands for the HTTP/2 sender.
45−response :: Context -> Manager -> Responder
46−response Context{outputQ} mgr tconf th strm req rsp = do
47− case rsp of
48− ResponseStream _ _ strmbdy -> do
49− -- We must not exit this WAI application.
50− -- If the application exits, streaming would be also closed.
51− -- So, this work occupies this thread.
52− --
53− -- We need to increase the number of workers.
54− myThreadId >>= replaceWithAction mgr
55− -- After this work, this thread stops to decease
56− -- the number of workers.
57− setThreadContinue tconf False
58− -- Since 'StreamingBody' is loop, we cannot control it.
59− -- So, let's serialize 'Builder' with a designated queue.
60− sq <- newTBQueueIO 10 -- fixme: hard coding: 10
61− tvar <- newTVarIO SyncNone
62− let out = OResponse strm rsp (Persist sq tvar)
63− -- Since we must not enqueue an empty queue to the priority
64− -- queue, we spawn a thread to ensure that the designated
65− -- queue is not empty.
66− void $ forkIO $ waiter tvar sq outputQ
67− atomically $ writeTVar tvar (SyncNext out)
68− let push b = do
69− atomically $ writeTBQueue sq (SBuilder b)
70− T.tickle th
71− flush = atomically $ writeTBQueue sq SFlush
72− strmbdy push flush
73− atomically $ writeTBQueue sq SFinish
74− _ -> do
75− setThreadContinue tconf True
76− let hasBody = requestMethod req /= H.methodHead
77− && R.hasBody (responseStatus rsp)
78− out = OResponse strm rsp (Oneshot hasBody)
79− enqueueOrSpawnTemporaryWaiter strm outputQ out
80− return ResponseReceived
52+response :: InternalInfo -> S.Settings -> Context -> Manager -> Responder
53+response ii settings Context{outputQ} mgr tconf th strm req rsp
54+ | R.hasBody s0 = case rsp of
55+ ResponseStream _ _ strmbdy
56+ | isHead -> responseNoBody s0 hs0
57+ | otherwise -> responseStreaming strmbdy
58+ ResponseBuilder _ _ b
59+ | isHead -> responseNoBody s0 hs0
60+ | otherwise -> responseBuilderBody s0 hs0 b
61+ ResponseFile _ _ p mp -> responseFileXXX p mp
62+ ResponseRaw _ _ -> error "HTTP/2 does not support ResponseRaw"
63+ | otherwise = responseNoBody s0 hs0
64+ where
65+ isHead = requestMethod req == H.methodHead
66+ s0 = responseStatus rsp
67+ hs0 = responseHeaders rsp
68+ logger = S.settingsLogger settings req
69+
70+ responseNoBody s hs = responseBuilderBody s hs mempty
71+
72+ responseBuilderBody s hs bdy = do
73+ writeIORef (streamLogger strm) (logger s Nothing)
74+ setThreadContinue tconf True
75+ let rsp' = ResponseBuilder s hs bdy
76+ out = OResponse strm rsp' (Oneshot True)
77+ enqueueOrSpawnTemporaryWaiter strm outputQ out
78+ return ResponseReceived
79+
80+ responseFileXXX path Nothing = do
81+ efinfo <- E.try $ fileInfo ii path
82+ case efinfo of
83+ Left (_ex :: E.IOException) -> response404
84+ Right finfo -> case conditionalRequest finfo hs0 (indexRequestHeader (requestHeaders req)) of
85+ WithoutBody s -> responseNoBody s hs0
86+ WithBody s hs beg len -> responseFile2XX s hs path (Just (FilePart beg len (fileInfoSize finfo)))
87+
88+ responseFileXXX path mpart = responseFile2XX s0 hs0 path mpart
89+
90+ responseFile2XX s hs path mpart
91+ | isHead = responseNoBody s hs
92+ | otherwise = do
93+ writeIORef (streamLogger strm) (logger s (filePartByteCount <$> mpart))
94+ setThreadContinue tconf True
95+ let rsp' = ResponseFile s hs path mpart
96+ out = OResponse strm rsp' (Oneshot True)
97+ enqueueOrSpawnTemporaryWaiter strm outputQ out
98+ return ResponseReceived
99+
100+ response404 = responseBuilderBody s hs body
101+ where
102+ s = H.notFound404
103+ hs = R.replaceHeader H.hContentType "text/plain; charset=utf-8" hs0
104+ body = byteString "File not found"
105+
106+ responseStreaming strmbdy = do
107+ writeIORef (streamLogger strm) (logger s0 Nothing)
108+ -- We must not exit this WAI application.
109+ -- If the application exits, streaming would be also closed.
110+ -- So, this work occupies this thread.
111+ --
112+ -- We need to increase the number of workers.
113+ myThreadId >>= replaceWithAction mgr
114+ -- After this work, this thread stops to decease
115+ -- the number of workers.
116+ setThreadContinue tconf False
117+ -- Since 'StreamingBody' is loop, we cannot control it.
118+ -- So, let's serialize 'Builder' with a designated queue.
119+ sq <- newTBQueueIO 10 -- fixme: hard coding: 10
120+ tvar <- newTVarIO SyncNone
121+ let out = OResponse strm rsp (Persist sq tvar)
122+ -- Since we must not enqueue an empty queue to the priority
123+ -- queue, we spawn a thread to ensure that the designated
124+ -- queue is not empty.
125+ void $ forkIO $ waiter tvar sq outputQ
126+ atomically $ writeTVar tvar (SyncNext out)
127+ let push b = do
128+ atomically $ writeTBQueue sq (SBuilder b)
129+ T.tickle th
130+ flush = atomically $ writeTBQueue sq SFlush
131+ _ <- strmbdy push flush
132+ atomically $ writeTBQueue sq SFinish
133+ return ResponseReceived
81134
82135data Break = Break deriving (Show, Typeable)
83136