warp 1.2.2 → 1.3.0
raw patch · 3 files changed
+100/−104 lines, 3 filesdep ~blaze-builder-conduitdep ~conduitdep ~http-types
Dependency ranges changed: blaze-builder-conduit, conduit, http-types, network-conduit, wai
Files
- Network/Wai/Handler/Warp.hs +93/−97
- test/main.hs +1/−1
- warp.cabal +6/−6
Network/Wai/Handler/Warp.hs view
@@ -56,6 +56,7 @@ , T.cancel , T.register , T.initialize+ , socketConnection #if TEST , takeHeaders , parseFirst@@ -95,8 +96,8 @@ import Data.Typeable (Typeable) -import Data.Conduit (ResourceT, runResourceT)-import qualified Data.Conduit as C+import Data.Conduit+import Data.Conduit.Internal (ResumableSource (..)) import qualified Data.Conduit.List as CL import qualified Data.Conduit.Binary as CB import Data.Conduit.Blaze (builderToByteString)@@ -237,62 +238,60 @@ ) `onException` (T.cancel th >> connClose conn >> onClose) -- | Contains a @Source@ and a byte count that is still to be read in.-newtype IsolatedBSSource = IsolatedBSSource (I.IORef (Int, C.Source (ResourceT IO) ByteString))+newtype IsolatedBSSource = IsolatedBSSource (I.IORef (Int, ResumableSource (ResourceT IO) ByteString)) -- | Given an @IsolatedBSSource@ provide a @Source@ that only allows up to the -- specified number of bytes to be passed downstream. All leftovers should be -- retained within the @Source@. If there are not enough bytes available, -- throws a @ConnectionClosedByPeer@ exception.-ibsIsolate :: IsolatedBSSource -> C.Source (ResourceT IO) ByteString-ibsIsolate ibs@(IsolatedBSSource ref) =- C.PipeM pull (return ())- where- pull = do- (count, src) <- liftIO $ I.readIORef ref- if count == 0- -- No more bytes wanted downstream, so we're done.- then return $ C.Done Nothing ()- else do- -- Get the next chunk (if available) and the updated source- (src', mbs) <- src C.$$+ CL.head+ibsIsolate :: IsolatedBSSource -> Source (ResourceT IO) ByteString+ibsIsolate ibs@(IsolatedBSSource ref) = do+ (count, src) <- liftIO $ I.readIORef ref+ if count == 0+ -- No more bytes wanted downstream, so we're done.+ then return ()+ else do+ -- Get the next chunk (if available) and the updated source+ (src', mbs) <- lift $ src $$++ CL.head - -- If no chunk available, then there aren't enough bytes in the- -- stream. Throw a ConnectionClosedByPeer- bs <- maybe (liftIO $ throwIO ConnectionClosedByPeer) return mbs+ -- If no chunk available, then there aren't enough bytes in the+ -- stream. Throw a ConnectionClosedByPeer+ bs <- maybe (liftIO $ throwIO ConnectionClosedByPeer) return mbs - let -- How many of the bytes in this chunk to send downstream- toSend = min count (S.length bs)- -- How many bytes will still remain to be sent downstream- count' = count - toSend- case () of- ()- -- The expected count is greater than the size of the- -- chunk we just read. Send the entire chunk- -- downstream, and then loop on this function for the- -- next chunk.- | count' > 0 -> do- liftIO $ I.writeIORef ref (count', src')- return $ C.HaveOutput (ibsIsolate ibs) (return ()) bs+ let -- How many of the bytes in this chunk to send downstream+ toSend = min count (S.length bs)+ -- How many bytes will still remain to be sent downstream+ count' = count - toSend+ case () of+ ()+ -- The expected count is greater than the size of the+ -- chunk we just read. Send the entire chunk+ -- downstream, and then loop on this function for the+ -- next chunk.+ | count' > 0 -> do+ liftIO $ I.writeIORef ref (count', src')+ yield bs+ ibsIsolate ibs - -- The expected count is the total size of the chunk we- -- just read. Send this chunk downstream, and then- -- terminate the stream.- | count == S.length bs -> do- liftIO $ I.writeIORef ref (count', src')- return $ C.HaveOutput (C.Done Nothing ()) (return ()) bs+ -- The expected count is the total size of the chunk we+ -- just read. Send this chunk downstream, and then+ -- terminate the stream.+ | count == S.length bs -> do+ liftIO $ I.writeIORef ref (count', src')+ yield bs - -- Some of the bytes in this chunk should not be sent- -- downstream. Split up the chunk into the sent and- -- not-sent parts, add the not-sent parts onto the new- -- source, and send the rest of the chunk downstream.- | otherwise -> do- let (x, y) = S.splitAt toSend bs- liftIO $ I.writeIORef ref (count', C.HaveOutput src' (return ()) y)- return $ C.HaveOutput (C.Done Nothing ()) (return ()) x+ -- Some of the bytes in this chunk should not be sent+ -- downstream. Split up the chunk into the sent and+ -- not-sent parts, add the not-sent parts onto the new+ -- source, and send the rest of the chunk downstream.+ | otherwise -> do+ let (x, y) = S.splitAt toSend bs+ liftIO $ I.writeIORef ref (count', fmapResume (yield y >>) src')+ yield x -- | Extract the underlying @Source@ from an @IsolatedBSSource@, which will not -- perform any more isolation.-ibsDone :: IsolatedBSSource -> IO (C.Source (ResourceT IO) ByteString)+ibsDone :: IsolatedBSSource -> IO (ResumableSource (ResourceT IO) ByteString) ibsDone (IsolatedBSSource ref) = fmap snd $ I.readIORef ref serveConnection :: Settings@@ -314,23 +313,24 @@ liftIO $ T.pause th res <- app env - -- flush the rest of the request body- requestBody env C.$$ CL.sinkNull- fromClient' <- liftIO getSource- liftIO $ T.resume th keepAlive <- sendResponse th env conn res++ -- flush the rest of the request body+ requestBody env $$ CL.sinkNull+ ResumableSource fromClient' _ <- liftIO getSource+ when keepAlive $ serveConnection'' fromClient' Just intercept -> do liftIO $ T.pause th- fromClient' <- liftIO getSource+ ResumableSource fromClient' _ <- liftIO getSource intercept fromClient' conn parseRequest :: Connection -> Port -> SockAddr- -> C.Source (ResourceT IO) S.ByteString- -> ResourceT IO (Request, IO (C.Source (ResourceT IO) ByteString))+ -> Source (ResourceT IO) S.ByteString+ -> ResourceT IO (Request, IO (ResumableSource (ResourceT IO) ByteString)) parseRequest conn port remoteHost' src1 = do- (src2, headers') <- src1 C.$$+ takeHeaders+ (src2, headers') <- src1 $$+ takeHeaders parseRequest' conn port headers' remoteHost' src2 -- FIXME come up with good values here@@ -367,8 +367,8 @@ -> Port -> [ByteString] -> SockAddr- -> C.Source (ResourceT IO) S.ByteString -- FIXME was buffered- -> ResourceT IO (Request, IO (C.Source (ResourceT IO) ByteString))+ -> ResumableSource (ResourceT IO) S.ByteString -- FIXME was buffered+ -> ResourceT IO (Request, IO (ResumableSource (ResourceT IO) ByteString)) parseRequest' _ _ [] _ _ = throwIO $ NotEnoughLines [] parseRequest' conn port (firstLine:otherLines) remoteHost' src = do (method, rpath', gets, httpversion) <- parseFirst firstLine@@ -417,24 +417,24 @@ | HaveLen Word chunkedSource :: MonadIO m- => I.IORef (C.Source m ByteString, ChunkState)- -> C.Source m ByteString+ => I.IORef (ResumableSource m ByteString, ChunkState)+ -> Source m ByteString chunkedSource ipair = do (src, mlen) <- liftIO $ I.readIORef ipair go src mlen where go' src front = do- (src', (len, bs)) <- lift $ src C.$$+ front getLen+ (src', (len, bs)) <- lift $ src $$++ front getLen let src'' | S.null bs = src'- | otherwise = C.yield bs >> src'+ | otherwise = fmapResume (yield bs >>) src' go src'' $ HaveLen len go src NeedLen = go' src id go src NeedLenNewline = go' src (CB.take 2 >>) go src (HaveLen 0) = liftIO $ I.writeIORef ipair (src, HaveLen 0) go src (HaveLen len) = do- (src', mbs) <- lift $ src C.$$+ CL.head+ (src', mbs) <- lift $ src $$++ CL.head case mbs of Nothing -> liftIO $ I.writeIORef ipair (src', HaveLen 0) Just bs ->@@ -445,15 +445,15 @@ yield' src' mlen bs GT -> do let (x, y) = S.splitAt (fromIntegral len) bs- let src'' = C.yield y >> src'+ let src'' = fmapResume (yield y >>) src' yield' src'' NeedLenNewline x yield' src mlen bs = do liftIO $ I.writeIORef ipair (src, mlen)- C.yield bs+ yield bs go src mlen - getLen :: Monad m => C.Sink ByteString m (Word, ByteString)+ getLen :: Monad m => Sink ByteString m (Word, ByteString) getLen = do mbs <- CL.head case mbs of@@ -621,31 +621,23 @@ sendResponse' (ResponseSource s hs bodyFlush) | hasBody s req = do let src = CL.sourceList [headers' needsChunked'] `mappend`- (if needsChunked' then body C.$= chunk else body)- src C.$$ builderToByteString C.=$ connSink conn th+ (if needsChunked' then body $= chunk else body)+ src $$ builderToByteString =$ connSink conn th return $ isKeepAlive hs | otherwise = liftIO $ do sendHeader $ headers' False T.tickle th return isPersist where- body = fmap2 (\x -> case x of- C.Flush -> flush- C.Chunk builder -> builder) bodyFlush+ body = mapOutput (\x -> case x of+ Flush -> flush+ Chunk builder -> builder) bodyFlush headers' = headers version s hs -- FIXME perhaps alloca a buffer per thread and reuse that in all -- functions below. Should lessen greatly the GC burden (I hope) needsChunked' = needsChunked hs- chunk :: C.Conduit Builder (ResourceT IO) Builder- chunk = C.NeedInput push close- push x = C.HaveOutput chunk (return ()) (chunkedTransferEncoding x)- close = C.HaveOutput (C.Done Nothing ()) (return ()) chunkedTransferTerminator--fmap2 :: Functor m => (o1 -> o2) -> C.Pipe i o1 m r -> C.Pipe i o2 m r-fmap2 f (C.HaveOutput p c o) = C.HaveOutput (fmap2 f p) c (f o)-fmap2 f (C.NeedInput p c) = C.NeedInput (fmap2 f . p) (fmap2 f c)-fmap2 f (C.PipeM mp c) = C.PipeM (fmap (fmap2 f) mp) c-fmap2 _ (C.Done i x) = C.Done i x+ chunk :: Conduit Builder (ResourceT IO) Builder+ chunk = await >>= maybe (yield chunkedTransferTerminator) (\x -> yield (chunkedTransferEncoding x) >> chunk) parseHeaderNoAttr :: ByteString -> H.Header parseHeaderNoAttr s =@@ -653,31 +645,31 @@ rest' = S.dropWhile (\c -> c == 32 || c == 9) $ S.drop 1 rest in (CI.mk k, rest') -connSource :: Connection -> T.Handle -> C.Source (ResourceT IO) ByteString+connSource :: Connection -> T.Handle -> Source (ResourceT IO) ByteString connSource Connection { connRecv = recv } th = src where- src = C.PipeM (do+ src = do bs <- liftIO recv if S.null bs- then return $ C.Done Nothing ()+ then return () else do when (S.length bs >= 2048) $ liftIO $ T.tickle th- return (C.HaveOutput src (return ()) bs))- (return ())+ yield bs+ src -- | Use 'connSendAll' to send this data while respecting timeout rules.-connSink :: Connection -> T.Handle -> C.Sink B.ByteString (ResourceT IO) ()+connSink :: Connection -> T.Handle -> Sink B.ByteString (ResourceT IO) () connSink Connection { connSendAll = send } th = sink where- sink = C.NeedInput push close+ sink = await >>= maybe close push close = liftIO (T.resume th)- push x = C.PipeM (liftIO $ do- T.resume th- send x- T.pause th- return sink) (liftIO $ T.resume th)+ push x = do+ liftIO $ T.resume th+ liftIO $ send x+ liftIO $ 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 -- also make sure to resume the timeout after the completion of user code@@ -699,7 +691,7 @@ , settingsOnOpen :: IO () -- ^ What to do when a connection is open. Default: do nothing. , settingsOnClose :: IO () -- ^ What to do when a connection is close. Default: do nothing. , settingsTimeout :: Int -- ^ Timeout value in seconds. Default value: 30- , settingsIntercept :: Request -> Maybe (C.Source (ResourceT IO) S.ByteString -> Connection -> ResourceT IO ())+ , settingsIntercept :: Request -> Maybe (Source (ResourceT IO) S.ByteString -> Connection -> ResourceT IO ()) , settingsManager :: Maybe Manager -- ^ Use an existing timeout manager instead of spawning a new one. If used, 'settingsTimeout' is ignored. Default is 'Nothing' } @@ -735,10 +727,11 @@ BSEndoList -- previously parsed lines BSEndo -- bytestrings to be prepended -takeHeaders :: C.Sink ByteString (ResourceT IO) [ByteString]+takeHeaders :: Sink ByteString (ResourceT IO) [ByteString] takeHeaders =- C.NeedInput (push (THStatus 0 id id)) close+ await >>= maybe (throwIO ConnectionClosedByPeer) (push (THStatus 0 id id)) where+ close :: Sink ByteString (ResourceT IO) a close = throwIO IncompleteHeaders push (THStatus len lines prepend) bs@@ -752,7 +745,7 @@ let len' = len + bsLen prepend' = prepend . S.append bs status = THStatus len' lines prepend'- in C.NeedInput (push status) close+ in await >>= maybe close (push status) -- Found a newline, but next line continues as a multiline header Just (end, True) -> let rest = S.drop (end + 1) bs@@ -776,7 +769,7 @@ rest = if start < bsLen then Just (SU.unsafeDrop start bs) else Nothing- in C.Done rest lines'+ in maybe (return ()) leftover rest >> return lines' -- more headers else let len' = len + start@@ -787,7 +780,7 @@ then let bs' = SU.unsafeDrop start bs in push status bs' -- no more bytes in this chunk, ask for more- else C.NeedInput (push status) close+ else await >>= maybe close (push status) where bsLen = S.length bs mnl = do@@ -831,3 +824,6 @@ key = "Server" ver = B.pack $ "Warp/" ++ warpVersion server = (key, ver)++fmapResume :: (Source m o1 -> Source m o2) -> ResumableSource m o1 -> ResumableSource m o2+fmapResume f (ResumableSource src m) = ResumableSource (f src) m
test/main.hs view
@@ -116,7 +116,7 @@ singlePostHello = "POST /hello HTTP/1.1\r\nHost: localhost\r\nContent-length: 5\r\n\r\nHello" main :: IO ()-main = hspecX $ do+main = hspec $ do describe "non-pipelining" $ do it "no body, read" $ runTest 5 readBody $ replicate 5 singleGet it "no body, ignore" $ runTest 5 ignoreBody $ replicate 5 singleGet
warp.cabal view
@@ -1,5 +1,5 @@ Name: warp-Version: 1.2.2+Version: 1.3.0 Synopsis: A fast, light-weight web server for WAI applications. License: MIT License-file: LICENSE@@ -19,15 +19,15 @@ Library Build-Depends: base >= 3 && < 5 , bytestring >= 0.9.1.4- , wai >= 1.2 && < 1.3+ , wai >= 1.3 && < 1.4 , transformers >= 0.2.2 && < 0.4- , conduit >= 0.4 && < 0.5- , network-conduit >= 0.4 && < 0.5- , blaze-builder-conduit >= 0.4 && < 0.5+ , conduit >= 0.5 && < 0.6+ , network-conduit >= 0.5 && < 0.6+ , blaze-builder-conduit >= 0.5 && < 0.6 , lifted-base >= 0.1 && < 0.2 , blaze-builder >= 0.2.1.4 && < 0.4 , simple-sendfile >= 0.2.4 && < 0.3- , http-types >= 0.6 && < 0.7+ , http-types >= 0.7 && < 0.8 , case-insensitive >= 0.2 , unix-compat >= 0.2 , ghc-prim