@@ -7,55 +7,70 @@ module Network.Wai.Handler.Warp.HTTP2.Sender (frameSender) where
77#if __GLASGOW_HASKELL__ < 709
88import Control.Applicative
99#endif
10+import Control.Concurrent.MVar (putMVar)
1011import Control.Concurrent.STM
1112import qualified Control.Exception as E
12−import Control.Monad (unless, void, when)
13+import Control.Monad (void, when)
1314import qualified Data.ByteString as BS
15+import qualified Data.ByteString.Builder as B (int32BE)
1416import qualified Data.ByteString.Builder.Extra as B
17+import Data.Monoid ((<>))
1518import Foreign.Ptr
19+import qualified Network.HTTP.Types as H
1620import Network.HTTP2
1721import Network.HTTP2.Priority
18−import Network.Wai
22+import Network.Wai (FilePart(..))
23+import Network.Wai.HTTP2 (Trailers, promiseHeaders)
1924import Network.Wai.Handler.Warp.Buffer
2025import Network.Wai.Handler.Warp.HTTP2.EncodeFrame
2126import Network.Wai.Handler.Warp.HTTP2.HPACK
2227import Network.Wai.Handler.Warp.HTTP2.Types
2328import Network.Wai.Handler.Warp.IORef
2429import qualified Network.Wai.Handler.Warp.Settings as S
25−import qualified Network.Wai.Handler.Warp.Timeout as T
2630import Network.Wai.Handler.Warp.Types
27−import Network.Wai.Internal (Response(..))
28−import qualified System.PosixCompat.Files as P
2931
3032#ifdef WINDOWS
3133import qualified System.IO as IO
3234#else
33−import Network.Wai.Handler.Warp.FdCache
35+import Network.Wai.Handler.Warp.FdCache (getFd)
3436import Network.Wai.Handler.Warp.SendFile (positionRead)
37+import qualified Network.Wai.Handler.Warp.Timeout as T
3538import System.Posix.IO (openFd, OpenFileFlags(..), defaultFileFlags, OpenMode(ReadOnly), closeFd)
36−import System.Posix.Types
39+import System.Posix.Types (Fd)
3740#endif
3841
3942----------------------------------------------------------------
4043
44+-- | The platform-specific type of an open file to stream from. On Windows we
45+-- don't have pread, so this is just a Handle; on Unix platforms with pread,
46+-- this is a file descriptor supplied by the fd cache.
47+#ifdef WINDOWS
48+type OpenFile = IO.Handle
49+#else
50+type OpenFile = Fd
51+#endif
52+
4153data Leftover = LZero
4254 | LOne B.BufferWriter
4355 | LTwo BS.ByteString B.BufferWriter
56+ | LFile OpenFile Integer Integer (IO ())
4457
4558----------------------------------------------------------------
4659
47−unlessClosed :: Context -> Connection -> Stream -> IO () -> IO ()
48−unlessClosed ctx
49− Connection{connSendAll}
60+-- | Run the given action if the stream is not closed; handle any exceptions by
61+-- resetting the stream.
62+unlessClosed :: Connection -> Stream -> IO () -> IO Bool
63+unlessClosed Connection{connSendAll}
5064 strm@Stream{streamState,streamNumber}
5165 body = E.handle resetStream $ do
5266 state <- readIORef streamState
53− unless (isClosed state) body
67+ if (isClosed state) then return False else body >> return True
5468 where
5569 resetStream e = do
56− closed ctx strm (ResetByMe e)
70+ closed strm (ResetByMe e)
5771 let rst = resetFrame InternalError streamNumber
5872 connSendAll rst
73+ return False
5974
6075getWindowSize :: TVar WindowSize -> TVar WindowSize -> IO WindowSize
6176getWindowSize connWindow strmWindow = do
@@ -93,64 +108,98 @@ frameSender ctx@Context{outputQ,connectionWindow}
93108 switch (OFrame frame) = do
94109 connSendAll frame
95110 loop
96− switch (OResponse strm rsp aux) = do
97− unlessClosed ctx conn strm $ do
98− lim <- getWindowSize connectionWindow (streamWindow strm)
99− -- Header frame and Continuation frame
100− let sid = streamNumber strm
101− endOfStream = case aux of
102− Persist{} -> False
103− Oneshot hb -> not hb
104− len <- headerContinue sid rsp endOfStream
105− let total = len + frameHeaderLength
106− case aux of
107− Oneshot True -> do -- hasBody
108− -- Data frame payload
109− (off, _) <- sendHeadersIfNecessary total
110− let payloadOff = off + frameHeaderLength
111− Next datPayloadLen mnext <-
112− fillResponseBodyGetNext conn ii payloadOff lim rsp
113− fillDataHeaderSend strm total datPayloadLen mnext
114− maybeEnqueueNext strm mnext
115− Oneshot False -> do
116− -- "closed" must be before "connSendAll". If not,
117− -- the context would be switched to the receiver,
118− -- resulting the inconsistency of concurrency.
119− closed ctx strm Finished
120− flushN total
121− Persist sq tvar -> do
122− (off, needSend) <- sendHeadersIfNecessary total
123− let payloadOff = off + frameHeaderLength
124− Next datPayloadLen mnext <-
125− fillStreamBodyGetNext conn payloadOff lim sq tvar strm
126− -- If no data was immediately available, avoid sending an
127− -- empty data frame.
128− if datPayloadLen > 0 then
129− fillDataHeaderSend strm total datPayloadLen mnext
130− else
131− when needSend $ flushN off
132− maybeEnqueueNext strm mnext
111+ switch (OResponse strm s h aux) = do
112+ _ <- unlessClosed conn strm $
113+ getWindowSize connectionWindow (streamWindow strm) >>=
114+ sendResponse strm s h aux
133115 loop
134116 switch (ONext strm curr) = do
135− unlessClosed ctx conn strm $ do
117+ _ <- unlessClosed conn strm $ do
136118 lim <- getWindowSize connectionWindow (streamWindow strm)
137119 -- Data frame payload
138120 Next datPayloadLen mnext <- curr lim
139− fillDataHeaderSend strm 0 datPayloadLen mnext
140− maybeEnqueueNext strm mnext
121+ fillDataHeaderSend strm 0 datPayloadLen
122+ dispatchNext strm mnext
123+ loop
124+ switch (OPush oldStrm push mvar strm s h aux) = do
125+ pushed <- unlessClosed conn oldStrm $ do
126+ lim <- getWindowSize connectionWindow (streamWindow strm)
127+ -- Write and send the promise.
128+ builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push
129+ off <- pushContinue (streamNumber oldStrm) (streamNumber strm) builder
130+ flushN $ off + frameHeaderLength
131+ -- TODO(awpr): refactor sendResponse to be able to handle non-zero
132+ -- initial offsets and use that to potentially avoid the extra syscall.
133+ sendResponse strm s h aux lim
134+ putMVar mvar pushed
141135 loop
142136
137+ -- Send the response headers and as much of the response as is immediately
138+ -- available; shared by normal responses and pushed streams.
139+ sendResponse :: Stream -> H.Status -> H.ResponseHeaders -> Aux -> WindowSize -> IO ()
140+ sendResponse strm s h (Persist sq tvar) lim = do
141+ -- Header frame and Continuation frame
142+ let sid = streamNumber strm
143+ builder <- hpackEncodeHeader ctx ii settings s h
144+ len <- headerContinue sid builder False
145+ let total = len + frameHeaderLength
146+ (off, needSend) <- sendHeadersIfNecessary total
147+ let payloadOff = off + frameHeaderLength
148+ Next datPayloadLen mnext <-
149+ fillStreamBodyGetNext ii conn payloadOff lim sq tvar strm
150+ -- If no data was immediately available, avoid sending an
151+ -- empty data frame.
152+ if datPayloadLen > 0 then
153+ fillDataHeaderSend strm total datPayloadLen
154+ else
155+ when needSend $ flushN off
156+ dispatchNext strm mnext
157+
158+ -- Send the stream's trailers and close the stream.
159+ sendTrailers :: Stream -> Trailers -> IO ()
160+ sendTrailers strm trailers = do
161+ -- Trailers always indicate the end of a stream; send them in
162+ -- consecutive header+continuation frames and end the stream. Some
163+ -- clients dislike empty headers frames, so end the stream with an
164+ -- empty data frame instead, as recommended by the spec.
165+ toFlush <- case trailers of
166+ [] -> frameHeaderLength <$ fillFrameHeader FrameData 0
167+ (streamNumber strm)
168+ (setEndStream defaultFlags)
169+ connWriteBuffer
170+ _ -> do
171+ builder <- hpackEncodeCIHeaders ctx trailers
172+ off <- headerContinue (streamNumber strm) builder True
173+ return (off + frameHeaderLength)
174+ -- 'closed' must be before 'flushN'. If not, the context would be
175+ -- switched to the receiver, resulting in the inconsistency of
176+ -- concurrency.
177+ closed strm Finished
178+ flushN toFlush
179+
143180 -- Flush the connection buffer to the socket, where the first 'n' bytes of
144181 -- the buffer are filled.
145182 flushN :: Int -> IO ()
146183 flushN n = bufferIO connWriteBuffer n connSendAll
147184
148− headerContinue sid rsp endOfStream = do
149− builder <- hpackEncodeHeader ctx ii settings rsp
185+ -- A flags value with the end-header flag set iff the argument is B.Done.
186+ maybeEndHeaders B.Done = setEndHeader defaultFlags
187+ maybeEndHeaders _ = defaultFlags
188+
189+ -- Write PUSH_PROMISE and possibly CONTINUATION frames into the connection
190+ -- buffer, using the given builder as their contents; flush them to the
191+ -- socket as necessary.
192+ pushContinue sid newSid builder = do
193+ let builder' = B.int32BE (fromIntegral newSid) <> builder
194+ (len, signal) <- B.runBuilder builder' bufHeaderPayload headerPayloadLim
195+ let flag = maybeEndHeaders signal
196+ fillFrameHeader FramePushPromise len sid flag connWriteBuffer
197+ continue sid len signal
198+
199+ -- Write HEADER and possibly CONTINUATION frames.
200+ headerContinue sid builder endOfStream = do
150201 (len, signal) <- B.runBuilder builder bufHeaderPayload headerPayloadLim
151− let flag0 = case signal of
152− B.Done -> setEndHeader defaultFlags
153− _ -> defaultFlags
202+ let flag0 = maybeEndHeaders signal
154203 flag = if endOfStream then setEndStream flag0 else flag0
155204 fillFrameHeader FrameHeaders len sid flag connWriteBuffer
156205 continue sid len signal
@@ -159,9 +208,7 @@ frameSender ctx@Context{outputQ,connectionWindow}
159208 continue sid len (B.More _ writer) = do
160209 flushN $ len + frameHeaderLength
161210 (len', signal') <- writer bufHeaderPayload headerPayloadLim
162− let flag = case signal' of
163− B.Done -> setEndHeader defaultFlags
164− _ -> defaultFlags
211+ let flag = maybeEndHeaders signal'
165212 fillFrameHeader FrameContinuation len' sid flag connWriteBuffer
166213 continue sid len' signal'
167214 continue sid len (B.Chunk bs writer) = do
@@ -178,16 +225,22 @@ frameSender ctx@Context{outputQ,connectionWindow}
178225 -- True if the connection buffer has room for a 1-byte data frame.
179226 canFitDataFrame total = total + frameHeaderLength < connBufferSize
180227
181− -- Re-enqueue the stream in the output queue if more output is immediately
182− -- available; do nothing otherwise.
183− maybeEnqueueNext :: Stream -> Control DynaNext -> IO ()
184− maybeEnqueueNext strm (CNext next) = do
228+ -- Take the appropriate action based on the given 'Control':
229+ -- - If more output is immediately available, re-enqueue the stream in the
230+ -- output queue.
231+ -- - If the output is over and trailers are available, send them now and
232+ -- end the stream.
233+ -- - If we've drained the queue and handed the stream back to its waiter,
234+ -- do nothing.
235+ --
236+ -- This is done after sending any part of the stream body, so it's shared
237+ -- by 'sendResponse' and @switch (ONext ...)@.
238+ dispatchNext :: Stream -> Control DynaNext -> IO ()
239+ dispatchNext _ CNone = return ()
240+ dispatchNext strm (CFinish trailers) = sendTrailers strm trailers
241+ dispatchNext strm (CNext next) = do
185242 let out = ONext strm next
186243 enqueueOrSpawnTemporaryWaiter strm outputQ out
187− -- If the streaming is not finished, it must already have been
188− -- written to the 'TVar' owned by 'waiter', which will
189− -- put it back into the queue when more output becomes available.
190− maybeEnqueueNext _ _ = return ()
191244
192245
193246 -- Send headers if there is not room for a 1-byte data frame, and return
@@ -199,21 +252,12 @@ frameSender ctx@Context{outputQ,connectionWindow}
199252 flushN total
200253 return (0, False)
201254
202− fillDataHeaderSend strm otherLen datPayloadLen mnext = do
255+ fillDataHeaderSend strm otherLen datPayloadLen = do
203256 -- Data frame header
204257 let sid = streamNumber strm
205258 buf = connWriteBuffer `plusPtr` otherLen
206259 total = otherLen + frameHeaderLength + datPayloadLen
207− flag = case mnext of
208− CFinish -> setEndStream defaultFlags
209− _ -> defaultFlags
210− fillFrameHeader FrameData datPayloadLen sid flag buf
211− -- "closed" must be before "connSendAll". If not,
212− -- the context would be switched to the receiver,
213− -- resulting the inconsistency of concurrency.
214− case mnext of
215− CFinish -> closed ctx strm Finished
216− _ -> return ()
260+ fillFrameHeader FrameData datPayloadLen sid defaultFlags buf
217261 flushN total
218262 atomically $ do
219263 modifyTVar' connectionWindow (subtract datPayloadLen)
@@ -225,130 +269,106 @@ frameSender ctx@Context{outputQ,connectionWindow}
225269
226270----------------------------------------------------------------
227271
228−{-
229−ResponseFile Status ResponseHeaders FilePath (Maybe FilePart)
230−ResponseBuilder Status ResponseHeaders Builder
231−ResponseStream Status ResponseHeaders StreamingBody
232−ResponseRaw (IO ByteString -> (ByteString -> IO ()) -> IO ()) Response
233−-}
234−
235−fillResponseBodyGetNext :: Connection -> InternalInfo -> Int -> WindowSize -> Response -> IO Next
236−fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize}
237− _ off lim (ResponseBuilder _ _ bb) = do
272+fillStreamBodyGetNext :: InternalInfo -> Connection -> Int -> WindowSize
273+ -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next
274+fillStreamBodyGetNext ii Connection{connWriteBuffer,connBufferSize}
275+ off lim sq tvar strm = do
238276 let datBuf = connWriteBuffer `plusPtr` off
239277 room = min (connBufferSize - off) lim
240− (len, signal) <- B.runBuilder bb datBuf room
241− return $ nextForBuilder connWriteBuffer connBufferSize len signal
278+ (leftover, cont, len) <- runStreamBuilder ii datBuf room sq
279+ nextForStream ii connWriteBuffer connBufferSize sq tvar strm leftover cont len
280+
281+----------------------------------------------------------------
282+
283+runStreamBuilder :: InternalInfo -> Buffer -> BufSize -> TBQueue Sequence
284+ -> IO (Leftover, Maybe Trailers, BytesFilled)
285+runStreamBuilder ii buf0 room0 sq = loop buf0 room0 0
286+ where
287+ loop !buf !room !total = do
288+ mbuilder <- atomically $ tryReadTBQueue sq
289+ case mbuilder of
290+ Nothing -> return (LZero, Nothing, total)
291+ Just (SBuilder builder) -> do
292+ (len, signal) <- B.runBuilder builder buf room
293+ let !total' = total + len
294+ case signal of
295+ B.Done -> loop (buf `plusPtr` len) (room - len) total'
296+ B.More _ writer -> return (LOne writer, Nothing, total')
297+ B.Chunk bs writer -> return (LTwo bs writer, Nothing, total')
298+ Just (SFile path part) -> do
299+ (leftover, len) <- runStreamFile ii buf room path part
300+ let !total' = total + len
301+ return (leftover, Nothing, total')
302+ -- TODO if file part is done, go back to loop
303+ Just SFlush -> return (LZero, Nothing, total)
304+ Just (SFinish trailers) -> return (LZero, Just trailers, total)
305+
306+-- | Open the file and start reading into the send buffer.
307+runStreamFile :: InternalInfo -> Buffer -> BufSize -> FilePath -> FilePart
308+ -> IO (Leftover, BytesFilled)
309+
310+-- | Read the given (OS-specific) file representation into the buffer. On
311+-- non-Windows systems this uses pread; on Windows this ignores the position
312+-- because we use the Handle's internal read position instead (because it's not
313+-- potentially shared with other readers).
314+readOpenFile :: OpenFile -> Buffer -> BufSize -> Integer -> IO Int
242315
243316#ifdef WINDOWS
244−fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize}
245− _ off lim (ResponseFile _ _ path mpart) = do
246− let datBuf = connWriteBuffer `plusPtr` off
247− room = min (connBufferSize - off) lim
248− (start, bytes) <- fileStartEnd path mpart
317+runStreamFile _ buf room path part = do
318+ let start = filePartOffset part
319+ bytes = filePartByteCount part
249320 -- fixme: how to close Handle? GC does it at this moment.
250321 h <- IO.openBinaryFile path IO.ReadMode
251322 IO.hSeek h IO.AbsoluteSeek start
252− len <- IO.hGetBufSome h datBuf (mini room bytes)
253− let bytes' = bytes - fromIntegral len
254− return $ nextForFile len connWriteBuffer connBufferSize h bytes' (return ())
323+ fillBufFile buf room h start bytes (return ())
324+
325+readOpenFile h buf room _ = IO.hGetBufSome h buf room
255326#else
256−fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize}
257− ii off lim (ResponseFile _ _ path mpart) = do
327+runStreamFile ii buf room path part = do
328+ let start = filePartOffset part
329+ bytes = filePartByteCount part
258330 (fd, refresh) <- case fdCacher ii of
259331 Just fdcache -> getFd fdcache path
260332 Nothing -> do
261333 fd' <- openFd path ReadOnly Nothing defaultFileFlags{nonBlock=True}
262334 th <- T.register (timeoutManager ii) (closeFd fd')
263335 return (fd', T.tickle th)
264− let datBuf = connWriteBuffer `plusPtr` off
265− room = min (connBufferSize - off) lim
266− (start, bytes) <- fileStartEnd path mpart
267− len <- positionRead fd datBuf (mini room bytes) start
268− refresh
269− let len' = fromIntegral len
270− return $ nextForFile len connWriteBuffer connBufferSize fd (start + len') (bytes - len') refresh
271−#endif
272−
273−fillResponseBodyGetNext _ _ _ _ _ = error "fillResponseBodyGetNext"
336+ fillBufFile buf room fd start bytes refresh
274337
275−fileStartEnd :: FilePath -> Maybe FilePart -> IO (Integer, Integer)
276−fileStartEnd path Nothing = do
277− end <- fromIntegral . P.fileSize <$> P.getFileStatus path
278− return (0, end)
279−fileStartEnd _ (Just part) =
280− return (filePartOffset part, filePartByteCount part)
281−
282−----------------------------------------------------------------
283−
284−fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next
285−fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}
286− off lim sq tvar strm = do
287− let datBuf = connWriteBuffer `plusPtr` off
288− room = min (connBufferSize - off) lim
289− (leftover, cont, len) <- runStreamBuilder datBuf room sq
290− nextForStream connWriteBuffer connBufferSize sq tvar strm leftover cont len
291−
292−----------------------------------------------------------------
293−
294−fillBufBuilder :: Buffer -> BufSize -> Leftover -> DynaNext
295−fillBufBuilder buf0 siz0 leftover lim = do
296− let payloadBuf = buf0 `plusPtr` frameHeaderLength
297− room = min (siz0 - frameHeaderLength) lim
298− case leftover of
299− LZero -> error "fillBufBuilder: LZero"
300− LOne writer -> do
301− (len, signal) <- writer payloadBuf room
302− getNext len signal
303− LTwo bs writer
304− | BS.length bs <= room -> do
305− buf1 <- copy payloadBuf bs
306− let len1 = BS.length bs
307− (len2, signal) <- writer buf1 (room - len1)
308− getNext (len1 + len2) signal
309− | otherwise -> do
310− let (bs1,bs2) = BS.splitAt room bs
311− void $ copy payloadBuf bs1
312− getNext room (B.Chunk bs2 writer)
313− where
314− getNext l s = return $ nextForBuilder buf0 siz0 l s
315−
316−nextForBuilder :: Buffer -> BufSize -> BytesFilled -> B.Next -> Next
317−nextForBuilder _ _ len B.Done
318− = Next len CFinish
319−nextForBuilder buf siz len (B.More _ writer)
320− = Next len (CNext (fillBufBuilder buf siz (LOne writer)))
321−nextForBuilder buf siz len (B.Chunk bs writer)
322− = Next len (CNext (fillBufBuilder buf siz (LTwo bs writer)))
338+readOpenFile = positionRead
339+#endif
323340
324−----------------------------------------------------------------
341+-- | Read as much of the file as is currently available into the buffer, then
342+-- return a 'Leftover' to indicate whether this file chunk has more data to
343+-- send. If this read hit the end of the file range, return 'LZero'; otherwise
344+-- return 'LFile' so this stream will continue reading from the file the next
345+-- time it's pulled from the queue.
346+fillBufFile :: Buffer -> BufSize -> OpenFile -> Integer -> Integer -> (IO ())
347+ -> IO (Leftover, BytesFilled)
348+fillBufFile buf room f start bytes refresh = do
349+ len <- readOpenFile f buf (mini room bytes) start
350+ refresh
351+ let len' = fromIntegral len
352+ leftover = if bytes > len' then
353+ LFile f (start + len') (bytes - len') refresh
354+ else
355+ LZero
356+ return (leftover, len)
325357
326−runStreamBuilder :: Buffer -> BufSize -> TBQueue Sequence
327− -> IO (Leftover, Bool, BytesFilled)
328−runStreamBuilder buf0 room0 sq = loop buf0 room0 0
329− where
330− loop !buf !room !total = do
331− mbuilder <- atomically $ tryReadTBQueue sq
332− case mbuilder of
333− Nothing -> return (LZero, True, total)
334− Just (SBuilder builder) -> do
335− (len, signal) <- B.runBuilder builder buf room
336− let !total' = total + len
337− case signal of
338− B.Done -> loop (buf `plusPtr` len) (room - len) total'
339− B.More _ writer -> return (LOne writer, True, total')
340− B.Chunk bs writer -> return (LTwo bs writer, True, total')
341− Just SFlush -> return (LZero, True, total)
342− Just SFinish -> return (LZero, False, total)
358+mini :: Int -> Integer -> Int
359+mini i n
360+ | fromIntegral i < n = i
361+ | otherwise = fromIntegral n
343362
344−fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext
345−fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do
363+fillBufStream :: InternalInfo -> Buffer -> BufSize -> Leftover
364+ -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext
365+fillBufStream ii buf0 siz0 leftover0 sq tvar strm lim0 = do
346366 let payloadBuf = buf0 `plusPtr` frameHeaderLength
347367 room0 = min (siz0 - frameHeaderLength) lim0
348368 case leftover0 of
349369 LZero -> do
350− (leftover, cont, len) <- runStreamBuilder payloadBuf room0 sq
351− getNext leftover cont len
370+ (leftover, end, len) <- runStreamBuilder ii payloadBuf room0 sq
371+ getNext leftover end len
352372 LOne writer -> write writer payloadBuf room0 0
353373 LTwo bs writer
354374 | BS.length bs <= room0 -> do
@@ -358,71 +378,35 @@ fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do
358378 | otherwise -> do
359379 let (bs1,bs2) = BS.splitAt room0 bs
360380 void $ copy payloadBuf bs1
361− getNext (LTwo bs2 writer) True room0
381+ getNext (LTwo bs2 writer) Nothing room0
382+ LFile fd start bytes refresh -> do
383+ (leftover, len) <- fillBufFile payloadBuf room0 fd start bytes refresh
384+ getNext leftover Nothing len
362385 where
363− getNext = nextForStream buf0 siz0 sq tvar strm
386+ getNext = nextForStream ii buf0 siz0 sq tvar strm
364387 write writer1 buf room sofar = do
365388 (len, signal) <- writer1 buf room
366389 case signal of
367390 B.Done -> do
368− (leftover, cont, extra) <- runStreamBuilder (buf `plusPtr` len) (room - len) sq
391+ (leftover, end, extra) <- runStreamBuilder ii (buf `plusPtr` len) (room - len) sq
369392 let !total = sofar + len + extra
370− getNext leftover cont total
393+ getNext leftover end total
371394 B.More _ writer -> do
372395 let !total = sofar + len
373− getNext (LOne writer) True total
396+ getNext (LOne writer) Nothing total
374397 B.Chunk bs writer -> do
375398 let !total = sofar + len
376− getNext (LTwo bs writer) True total
377−
378−nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync -> Stream
379− -> Leftover -> Bool -> BytesFilled
380− -> IO Next
381−nextForStream _ _ _ tvar _ _ False len = do
382− atomically $ writeTVar tvar SyncFinish
383− return $ Next len CFinish
384−nextForStream buf siz sq tvar strm LZero True len = do
385− let out = ONext strm (fillBufStream buf siz LZero sq tvar strm)
399+ getNext (LTwo bs writer) Nothing total
400+
401+nextForStream :: InternalInfo -> Buffer -> BufSize -> TBQueue Sequence
402+ -> TVar Sync -> Stream -> Leftover -> Maybe Trailers
403+ -> BytesFilled -> IO Next
404+nextForStream _ _ _ _ tvar _ _ (Just trailers) len = do
405+ atomically $ writeTVar tvar $ SyncFinish
406+ return $ Next len $ CFinish trailers
407+nextForStream ii buf siz sq tvar strm LZero Nothing len = do
408+ let out = ONext strm (fillBufStream ii buf siz LZero sq tvar strm)
386409 atomically $ writeTVar tvar $ SyncNext out
387410 return $ Next len CNone
388−nextForStream buf siz sq tvar strm leftover True len =
389− return $ Next len (CNext (fillBufStream buf siz leftover sq tvar strm))
390−
391−----------------------------------------------------------------
392−
393−#ifdef WINDOWS
394−fillBufFile :: Buffer -> BufSize -> IO.Handle -> Integer -> IO () -> DynaNext
395−fillBufFile buf siz h bytes refresh lim = do
396− let payloadBuf = buf `plusPtr` frameHeaderLength
397− room = min (siz - frameHeaderLength) lim
398− len <- IO.hGetBufSome h payloadBuf room
399− refresh
400− let bytes' = bytes - fromIntegral len
401− return $ nextForFile len buf siz h bytes' refresh
402−
403−nextForFile :: BytesFilled -> Buffer -> BufSize -> IO.Handle -> Integer -> IO () -> Next
404−nextForFile 0 _ _ _ _ _ = Next 0 CFinish
405−nextForFile len _ _ _ 0 _ = Next len CFinish
406−nextForFile len buf siz h bytes refresh =
407− Next len (CNext (fillBufFile buf siz h bytes refresh))
408−#else
409−fillBufFile :: Buffer -> BufSize -> Fd -> Integer -> Integer -> IO () -> DynaNext
410−fillBufFile buf siz fd start bytes refresh lim = do
411− let payloadBuf = buf `plusPtr` frameHeaderLength
412− room = min (siz - frameHeaderLength) lim
413− len <- positionRead fd payloadBuf (mini room bytes) start
414− let len' = fromIntegral len
415− refresh
416− return $ nextForFile len buf siz fd (start + len') (bytes - len') refresh
417−
418−nextForFile :: BytesFilled -> Buffer -> BufSize -> Fd -> Integer -> Integer -> IO () -> Next
419−nextForFile 0 _ _ _ _ _ _ = Next 0 CFinish
420−nextForFile len _ _ _ _ 0 _ = Next len CFinish
421−nextForFile len buf siz fd start bytes refresh =
422− Next len (CNext (fillBufFile buf siz fd start bytes refresh))
423−#endif
424−
425−mini :: Int -> Integer -> Int
426−mini i n
427− | fromIntegral i < n = i
428− | otherwise = fromIntegral n
411+nextForStream ii buf siz sq tvar strm leftover Nothing len =
412+ return $ Next len (CNext (fillBufStream ii buf siz leftover sq tvar strm))