@@ -87,6 +87,7 @@ import Data.Typeable (Typeable)
8787
8888import Control.Monad.Trans.Resource (ResourceT, runResourceT)
8989import qualified Data.Conduit as C
90+import qualified Data.Conduit.Binary as CB
9091import qualified Data.Conduit.List as CL
9192import Control.Exception.Lifted (throwIO)
9293import Blaze.ByteString.Builder.HTTP
@@ -206,9 +207,9 @@ serveConnection settings th onException port app conn remoteHost' = do
206207 (sClose conn))
207208 onException
208209 where
209− fromClient = sourceSocket th bytesPerRead conn
210210 serveConnection' :: ResourceT IO ()
211211 serveConnection' = do
212+ fromClient <- C.bufferSource $ sourceSocket th bytesPerRead conn
212213 (len, env) <- parseRequest port remoteHost' fromClient
213214 case settingsIntercept settings env of
214215 Nothing -> do
@@ -223,7 +224,7 @@ serveConnection settings th onException port app conn remoteHost' = do
223224 intercept conn-}
224225
225226parseRequest :: Port -> SockAddr
226− -> C.Source IO S.ByteString
227+ -> C.BSource IO S.ByteString
227228 -> ResourceT IO (Integer, Request)
228229parseRequest port remoteHost' src = do
229230 headers' <- takeHeaders src
@@ -248,7 +249,7 @@ instance Exception InvalidRequest
248249parseRequest' :: Port
249250 -> [ByteString]
250251 -> SockAddr
251− -> C.Source IO S.ByteString
252+ -> C.BSource IO S.ByteString
252253 -> ResourceT IO (Integer, Request)
253254parseRequest' _ [] _ _ = throwIO $ NotEnoughLines []
254255parseRequest' port (firstLine:otherLines) remoteHost' src = do
@@ -268,6 +269,7 @@ parseRequest' port (firstLine:otherLines) remoteHost' src = do
268269 let serverName' = takeUntil 58 host -- ':'
269270 -- FIXME isolate takes an Integer instead of Int or Int64. If this is a
270271 -- performance penalty, we may need our own version.
272+ rbody <- C.bufferSource $ src C.$= CB.isolate (fromIntegral len)
271273 return (len, Request
272274 { requestMethod = method
273275 , httpVersion = httpversion
@@ -280,7 +282,7 @@ parseRequest' port (firstLine:otherLines) remoteHost' src = do
280282 , requestHeaders = heads
281283 , isSecure = False
282284 , remoteHost = remoteHost'
283− , requestBody = src -- FIXME isolate
285+ , requestBody = rbody
284286 })
285287
286288
@@ -424,7 +426,7 @@ sendResponse th req socket r = sendResponse' r
424426 `mappend` chunkedTransferTerminator
425427 else (headers' False) `mappend` b
426428
427− sendResponse' (ResponseEnumerator res) = error "sendResponse'" {-res enumResponse
429+ sendResponse' (ResponseStream res) = error "sendResponse'" {-res enumResponse
428430 where
429431 enumResponse s hs = response True
430432 where
@@ -462,27 +464,38 @@ parseHeaderNoAttr s =
462464 in (CI.mk k, rest')
463465
464466sourceSocket :: T.Handle -> Int -> Socket -> C.Source IO ByteString
465−sourceSocket th len socket = C.mkSource $ do
466− bs <- liftIO $ Sock.recv socket len
467− liftIO $ T.tickle th
468− return $ if S.null bs then C.EOF else C.Chunks [bs]
467+sourceSocket th len socket = C.Source
468+ { C.sourcePull = do
469+ bs <- liftIO $ Sock.recv socket len
470+ liftIO $ T.tickle th
471+ return $ if S.null bs
472+ then C.SourceResult C.StreamClosed []
473+ else C.SourceResult C.StreamOpen [bs]
474+ , C.sourceClose = return ()
475+ }
469476
470477------ The functions below are not warp-specific and could be split out into a
471478--separate package.
472479
473480sinkSocket :: T.Handle
474481 -> Socket
475− -> C.Sink B.ByteString IO ()
476−sinkSocket th sock =
477− C.Sink $ return $ C.SinkData go
482+ -> C.SinkM B.ByteString IO ()
483+sinkSocket th sock = C.SinkM $ return $ C.SinkData
484+ { C.sinkPush = push
485+ , C.sinkClose = close
486+ }
478487 where
479− go C.EOF = liftIO (T.resume th) >> return (C.SinkResult [] (Just ()))
480− go (C.Chunks []) = return $ C.SinkResult [] Nothing
481− go (C.Chunks xs) = do
488+ close xs = do
489+ liftIO (T.resume th)
490+ liftIO $ Sock.sendMany sock xs
491+ liftIO (T.resume th)
492+ return (C.SinkResult [] ())
493+ push [] = return $ C.Processing
494+ push xs = do
482495 liftIO $ T.resume th
483496 liftIO $ Sock.sendMany sock xs
484497 liftIO $ T.pause th
485− return $ C.SinkResult [] Nothing
498+ return $ C.Processing
486499 -- We pause timeouts before passing control back to user code. This ensures
487500 -- that a timeout will only ever be executed when Warp is in control. We
488501 -- also make sure to resume the timeout after the completion of user code
@@ -526,7 +539,7 @@ defaultSettings = Settings
526539 go' (Just ThreadKilled) = False
527540 go' _ = True
528541
529−takeHeaders :: C.Source IO ByteString -> ResourceT IO [ByteString]
542+takeHeaders :: C.BSource IO ByteString -> ResourceT IO [ByteString]
530543takeHeaders src = do
531544 !x <- forceHead ConnectionClosedByPeer src
532545 takeHeaders' 0 id id x src
@@ -537,7 +550,7 @@ takeHeaders' :: Int
537550 -> ([ByteString] -> [ByteString])
538551 -> ([ByteString] -> [ByteString])
539552 -> ByteString
540− -> C.Source IO ByteString
553+ -> C.BSource IO ByteString
541554 -> ResourceT IO [ByteString]
542555takeHeaders' !len _ _ _ _ | len > maxTotalHeaderLength = throwIO OverLargeHeader
543556takeHeaders' !len !lines !prepend !bs src = do
@@ -581,7 +594,7 @@ takeHeaders' !len !lines !prepend !bs src = do
581594{-# INLINE takeHeaders' #-}
582595
583596forceHead :: InvalidRequest
584− -> C.Source IO ByteString
597+ -> C.BSource IO ByteString
585598 -> ResourceT IO ByteString
586599forceHead err src = do
587600 !mx <- src C.$$ CL.head