diff --git a/Network/Wai/Handler/Warp/Response.hs b/Network/Wai/Handler/Warp/Response.hs --- a/Network/Wai/Handler/Warp/Response.hs +++ b/Network/Wai/Handler/Warp/Response.hs @@ -37,8 +37,8 @@ import qualified Network.HTTP.Types as H import Network.Wai import qualified Network.Wai.Handler.Warp.Date as D -import Network.Wai.Handler.Warp.Header import Network.Wai.Handler.Warp.Buffer (toBlazeBuffer) +import Network.Wai.Handler.Warp.Header import Network.Wai.Handler.Warp.IO (toBufIOWith) import Network.Wai.Handler.Warp.ResponseHeader import Network.Wai.Handler.Warp.RequestHeader (parseByteRanges) @@ -237,7 +237,7 @@ | needsChunked = header <> chunkedTransferEncoding body <> chunkedTransferTerminator | otherwise = header <> body - buffer = connBuffer conn + buffer = connWriteBuffer conn size = connBufferSize conn toBufIOWith buffer size (connSendAll conn) hdrBdy @@ -246,8 +246,8 @@ sendRsp conn ver s hs restore (RspSource withBodyFlush needsChunked th) = withBodyFlush $ \bodyFlush -> restore $ do header <- composeHeaderBuilder ver s hs needsChunked + buffer <- toBlazeBuffer (connWriteBuffer conn) (connBufferSize conn) let src = yield header >> cbody bodyFlush - buffer <- toBlazeBuffer (connBuffer conn) (connBufferSize conn) src $$ unsafeBuilderToByteString (return buffer) =$ connSink conn th where cbody bodyFlush = if needsChunked then body $= chunk else body @@ -305,9 +305,10 @@ sink = await >>= maybe close push close = liftIO (T.resume th) push x = do - liftIO $ T.resume th - liftIO $ send x - liftIO $ T.pause th + liftIO $ do + T.resume th + send x + T.pause th sink -- We pause timeouts before passing control back to user code. This ensures -- that a timeout will only ever be executed when Warp is in control. We diff --git a/Network/Wai/Handler/Warp/Run.hs b/Network/Wai/Handler/Warp/Run.hs --- a/Network/Wai/Handler/Warp/Run.hs +++ b/Network/Wai/Handler/Warp/Run.hs @@ -51,14 +51,16 @@ -- | Default action value for 'Connection'. socketConnection :: Socket -> IO Connection socketConnection s = do - buf <- allocateBuffer bufferSize + readBuf <- allocateBuffer bufferSize + writeBuf <- allocateBuffer bufferSize return Connection { connSendMany = Sock.sendMany s , connSendAll = Sock.sendAll s , connSendFile = defaultSendFile s - , connClose = sClose s >> freeBuffer buf - , connRecv = receive s buf bufferSize - , connBuffer = buf + , connClose = sClose s >> freeBuffer readBuf >> freeBuffer writeBuf + , connRecv = receive s readBuf bufferSize + , connReadBuffer = readBuf + , connWriteBuffer = writeBuf , connBufferSize = bufferSize , connSendFileOverride = Override s } diff --git a/Network/Wai/Handler/Warp/Types.hs b/Network/Wai/Handler/Warp/Types.hs --- a/Network/Wai/Handler/Warp/Types.hs +++ b/Network/Wai/Handler/Warp/Types.hs @@ -74,7 +74,8 @@ , connSendFile :: FilePath -> Integer -> Integer -> IO () -> [ByteString] -> IO () -- ^ filepath, offset, length, hook action, HTTP headers , connClose :: IO () , connRecv :: IO ByteString - , connBuffer :: Buffer + , connReadBuffer :: Buffer + , connWriteBuffer :: Buffer , connBufferSize :: BufSize , connSendFileOverride :: ConnSendFileOverride } diff --git a/test/RunSpec.hs b/test/RunSpec.hs --- a/test/RunSpec.hs +++ b/test/RunSpec.hs @@ -24,9 +24,13 @@ import Test.Hspec import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar) import Control.Exception.Lifted (bracket, try, IOException, onException) -import Data.Streaming.Network (bindPortTCP) +import Data.Streaming.Network (bindPortTCP, getSocketTCP, safeRecv) import Network.Socket (sClose) import qualified Network.HTTP as HTTP +import Data.Conduit (Flush (Chunk), ($=)) +import qualified Data.Conduit.List as CL +import Blaze.ByteString.Builder (fromByteString) +import Network.Socket.ByteString (sendAll) main :: IO () main = hspec spec @@ -135,6 +139,12 @@ singlePostHello :: ByteString singlePostHello = "POST /hello HTTP/1.1\r\nHost: localhost\r\nContent-length: 5\r\n\r\nHello" +singleChunkedPostHello :: [ByteString] +singleChunkedPostHello = + [ "POST /hello HTTP/1.1\r\nHost: localhost\r\nTransfer-Encoding: chunked\r\n\r\n" + , "5\r\nHello\r\n0\r\n" + ] + spec :: Spec spec = do describe "non-pipelining" $ do @@ -148,6 +158,14 @@ [ singlePostHello , singleGet ] + it "chunked body, read" $ runTest 2 readBody $ concat + [ singleChunkedPostHello + , [singleGet] + ] + it "chunked body, ignore" $ runTest 2 ignoreBody $ concat + [ singleChunkedPostHello + , [singleGet] + ] describe "pipelining" $ do it "no body, read" $ runTest 5 readBody [S.concat $ replicate 5 singleGet] it "no body, ignore" $ runTest 5 ignoreBody [S.concat $ replicate 5 singleGet] @@ -159,6 +177,14 @@ [ singlePostHello , singleGet ] + it "chunked body, read" $ runTest 2 readBody $ return $ S.concat + [ S.concat singleChunkedPostHello + , singleGet + ] + it "chunked body, ignore" $ runTest 2 ignoreBody $ return $ S.concat + [ S.concat singleChunkedPostHello + , singleGet + ] describe "no hanging" $ do it "has body, read" $ runTest 1 readBody $ map S.singleton $ S.unpack singlePostHello it "double connect" $ runTest 1 doubleConnect [singlePostHello] @@ -323,3 +349,14 @@ `shouldBe` ["server"] map HTTP.hdrValue (HTTP.retrieveHeaders HTTP.HdrDate res) `shouldBe` ["date"] + + it "streaming echo #249" $ do + let app req = return $ responseSource status200 [] + $ requestBody req $= CL.map (Chunk . fromByteString) + withApp defaultSettings app $ \port -> do + (socket, _addr) <- getSocketTCP "127.0.0.1" port + sendAll socket "POST / HTTP/1.1\r\ntransfer-encoding: chunked\r\n\r\n" + threadDelay 10000 + sendAll socket "5\r\nhello\r\n0\r\n\r\n" + bs <- safeRecv socket 4096 + S.takeWhile (/= 13) bs `shouldBe` "HTTP/1.1 200 OK" diff --git a/warp.cabal b/warp.cabal --- a/warp.cabal +++ b/warp.cabal @@ -1,5 +1,5 @@ Name: warp -Version: 2.1.4.1 +Version: 2.1.5 Synopsis: A fast, light-weight web server for WAI applications. License: MIT License-file: LICENSE