71ce33f · typeable/wai

git.iko.soy/typeable/wai

on master help sign in

Beginning of conduit 0.5

71ce33fMichael Snoyman14 years ago

wai/wai.cabal+2−2

compare with master

unifiedsplit

@@ -1,5 +1,5 @@

1Name: wai

2−Version: 1.2.0.2

2+Version: 1.2.0.3

3Synopsis: Web Application Interface.

4Description: Provides a common protocol for communication between web applications and web servers.

5License: MIT

@@ -20,7 +20,7 @@ Library

20 Build-Depends: base >= 4 && < 5

21 , bytestring >= 0.9.1.4

22 , blaze-builder >= 0.2.1.4 && < 0.4

23− , conduit >= 0.4 && < 0.5

23+ , conduit >= 0.4 && < 0.6

24 , network >= 2.2.1.5 && < 2.4

25 , http-types >= 0.6 && < 0.7

26 , text >= 0.7 && < 0.12

warp/Network/Wai/Handler/Warp.hs+93−99

@@ -95,8 +95,8 @@ import Data.Word (Word)

96import Data.Typeable (Typeable)

98−import Data.Conduit (ResourceT, runResourceT)

98+import Data.Conduit

99−import qualified Data.Conduit as C

99+import Data.Conduit.Internal (ResumableSource (..))

100import qualified Data.Conduit.List as CL

101import qualified Data.Conduit.Binary as CB

102import Data.Conduit.Blaze (builderToByteString)

@@ -237,62 +237,60 @@ runSettingsConnection set getConn app = do

237 ) `onException` (T.cancel th >> connClose conn >> onClose)

238

239-- | Contains a @Source@ and a byte count that is still to be read in.

240−newtype IsolatedBSSource = IsolatedBSSource (I.IORef (Int, C.Source (ResourceT IO) ByteString))

240+newtype IsolatedBSSource = IsolatedBSSource (I.IORef (Int, ResumableSource (ResourceT IO) ByteString))

241

242-- | Given an @IsolatedBSSource@ provide a @Source@ that only allows up to the

243-- specified number of bytes to be passed downstream. All leftovers should be

244-- retained within the @Source@. If there are not enough bytes available,

245-- throws a @ConnectionClosedByPeer@ exception.

246−ibsIsolate :: IsolatedBSSource -> C.Source (ResourceT IO) ByteString

246+ibsIsolate :: IsolatedBSSource -> Source (ResourceT IO) ByteString

247−ibsIsolate ibs@(IsolatedBSSource ref) =

247+ibsIsolate ibs@(IsolatedBSSource ref) = do

248− C.PipeM pull (return ())

249− where

250− pull = do

251− (count, src) <- liftIO $ I.readIORef ref

248+ (count, src) <- liftIO $ I.readIORef ref

252− if count == 0

249+ if count == 0

253− -- No more bytes wanted downstream, so we're done.

250+ -- No more bytes wanted downstream, so we're done.

254− then return $ C.Done Nothing ()

251+ then return ()

255− else do

252+ else do

256− -- Get the next chunk (if available) and the updated source

253+ -- Get the next chunk (if available) and the updated source

257− (src', mbs) <- src C.$$+ CL.head

254+ (src', mbs) <- lift $ src $$++ CL.head

258−

255+

259− -- If no chunk available, then there aren't enough bytes in the

256+ -- If no chunk available, then there aren't enough bytes in the

260− -- stream. Throw a ConnectionClosedByPeer

257+ -- stream. Throw a ConnectionClosedByPeer

261− bs <- maybe (liftIO $ throwIO ConnectionClosedByPeer) return mbs

258+ bs <- maybe (liftIO $ throwIO ConnectionClosedByPeer) return mbs

262−

259+

263− let -- How many of the bytes in this chunk to send downstream

260+ let -- How many of the bytes in this chunk to send downstream

264− toSend = min count (S.length bs)

261+ toSend = min count (S.length bs)

265− -- How many bytes will still remain to be sent downstream

262+ -- How many bytes will still remain to be sent downstream

266− count' = count - toSend

263+ count' = count - toSend

267− case () of

264+ case () of

268− ()

265+ ()

269− -- The expected count is greater than the size of the

266+ -- The expected count is greater than the size of the

270− -- chunk we just read. Send the entire chunk

267+ -- chunk we just read. Send the entire chunk

271− -- downstream, and then loop on this function for the

268+ -- downstream, and then loop on this function for the

272− -- next chunk.

269+ -- next chunk.

273− | count' > 0 -> do

270+ | count' > 0 -> do

274− liftIO $ I.writeIORef ref (count', src')

271+ liftIO $ I.writeIORef ref (count', src')

272+ yield bs

275− return $ C.HaveOutput (ibsIsolate ibs) (return ()) bs

273+ ibsIsolate ibs

276−

274+

277− -- The expected count is the total size of the chunk we

275+ -- The expected count is the total size of the chunk we

278− -- just read. Send this chunk downstream, and then

276+ -- just read. Send this chunk downstream, and then

279− -- terminate the stream.

277+ -- terminate the stream.

280− | count == S.length bs -> do

278+ | count == S.length bs -> do

281− liftIO $ I.writeIORef ref (count', src')

279+ liftIO $ I.writeIORef ref (count', src')

282− return $ C.HaveOutput (C.Done Nothing ()) (return ()) bs

280+ yield bs

283−

281+

284− -- Some of the bytes in this chunk should not be sent

282+ -- Some of the bytes in this chunk should not be sent

285− -- downstream. Split up the chunk into the sent and

283+ -- downstream. Split up the chunk into the sent and

286− -- not-sent parts, add the not-sent parts onto the new

284+ -- not-sent parts, add the not-sent parts onto the new

287− -- source, and send the rest of the chunk downstream.

285+ -- source, and send the rest of the chunk downstream.

288− | otherwise -> do

286+ | otherwise -> do

289− let (x, y) = S.splitAt toSend bs

287+ let (x, y) = S.splitAt toSend bs

290− liftIO $ I.writeIORef ref (count', C.HaveOutput src' (return ()) y)

288+ liftIO $ I.writeIORef ref (count', fmapResume (yield y >>) src')

291− return $ C.HaveOutput (C.Done Nothing ()) (return ()) x

289+ yield x

292

290

293-- | Extract the underlying @Source@ from an @IsolatedBSSource@, which will not

291-- | Extract the underlying @Source@ from an @IsolatedBSSource@, which will not

294-- perform any more isolation.

292-- perform any more isolation.

295−ibsDone :: IsolatedBSSource -> IO (C.Source (ResourceT IO) ByteString)

293+ibsDone :: IsolatedBSSource -> IO (ResumableSource (ResourceT IO) ByteString)

296ibsDone (IsolatedBSSource ref) = fmap snd $ I.readIORef ref

294ibsDone (IsolatedBSSource ref) = fmap snd $ I.readIORef ref

297

295

298serveConnection :: Settings

296serveConnection :: Settings

@@ -315,22 +313,22 @@ serveConnection settings th port app conn remoteHost' =

315 res <- app env

313 res <- app env

316

314

317 -- flush the rest of the request body

315 -- flush the rest of the request body

318− requestBody env C.$$ CL.sinkNull

316+ requestBody env $$ CL.sinkNull

319− fromClient' <- liftIO getSource

317+ ResumableSource fromClient' _ <- liftIO getSource

320

318

321 liftIO $ T.resume th

319 liftIO $ T.resume th

322 keepAlive <- sendResponse th env conn res

320 keepAlive <- sendResponse th env conn res

323 when keepAlive $ serveConnection'' fromClient'

321 when keepAlive $ serveConnection'' fromClient'

324 Just intercept -> do

322 Just intercept -> do

325 liftIO $ T.pause th

323 liftIO $ T.pause th

326− fromClient' <- liftIO getSource

324+ ResumableSource fromClient' _ <- liftIO getSource

327 intercept fromClient' conn

325 intercept fromClient' conn

328

326

329parseRequest :: Connection -> Port -> SockAddr

327parseRequest :: Connection -> Port -> SockAddr

330− -> C.Source (ResourceT IO) S.ByteString

328+ -> Source (ResourceT IO) S.ByteString

331− -> ResourceT IO (Request, IO (C.Source (ResourceT IO) ByteString))

329+ -> ResourceT IO (Request, IO (ResumableSource (ResourceT IO) ByteString))

332parseRequest conn port remoteHost' src1 = do

330parseRequest conn port remoteHost' src1 = do

333− (src2, headers') <- src1 C.$$+ takeHeaders

331+ (src2, headers') <- src1 $$+ takeHeaders

334 parseRequest' conn port headers' remoteHost' src2

332 parseRequest' conn port headers' remoteHost' src2

335

333

336-- FIXME come up with good values here

334-- FIXME come up with good values here

@@ -367,8 +365,8 @@ parseRequest' :: Connection

367 -> Port

365 -> Port

368 -> [ByteString]

366 -> [ByteString]

369 -> SockAddr

367 -> SockAddr

370− -> C.Source (ResourceT IO) S.ByteString -- FIXME was buffered

368+ -> ResumableSource (ResourceT IO) S.ByteString -- FIXME was buffered

371− -> ResourceT IO (Request, IO (C.Source (ResourceT IO) ByteString))

369+ -> ResourceT IO (Request, IO (ResumableSource (ResourceT IO) ByteString))

372parseRequest' _ _ [] _ _ = throwIO $ NotEnoughLines []

370parseRequest' _ _ [] _ _ = throwIO $ NotEnoughLines []

373parseRequest' conn port (firstLine:otherLines) remoteHost' src = do

371parseRequest' conn port (firstLine:otherLines) remoteHost' src = do

374 (method, rpath', gets, httpversion) <- parseFirst firstLine

372 (method, rpath', gets, httpversion) <- parseFirst firstLine

@@ -417,24 +415,24 @@ data ChunkState = NeedLen

417 | HaveLen Word

415 | HaveLen Word

418

416

419chunkedSource :: MonadIO m

417chunkedSource :: MonadIO m

420− => I.IORef (C.Source m ByteString, ChunkState)

418+ => I.IORef (ResumableSource m ByteString, ChunkState)

421− -> C.Source m ByteString

419+ -> Source m ByteString

422chunkedSource ipair = do

420chunkedSource ipair = do

423 (src, mlen) <- liftIO $ I.readIORef ipair

421 (src, mlen) <- liftIO $ I.readIORef ipair

424 go src mlen

422 go src mlen

425 where

423 where

426 go' src front = do

424 go' src front = do

427− (src', (len, bs)) <- lift $ src C.$$+ front getLen

425+ (src', (len, bs)) <- lift $ src $$++ front getLen

428 let src''

426 let src''

429 | S.null bs = src'

427 | S.null bs = src'

430− | otherwise = C.yield bs >> src'

428+ | otherwise = fmapResume (yield bs >>) src'

431 go src'' $ HaveLen len

429 go src'' $ HaveLen len

432

430

433 go src NeedLen = go' src id

431 go src NeedLen = go' src id

434 go src NeedLenNewline = go' src (CB.take 2 >>)

432 go src NeedLenNewline = go' src (CB.take 2 >>)

435 go src (HaveLen 0) = liftIO $ I.writeIORef ipair (src, HaveLen 0)

433 go src (HaveLen 0) = liftIO $ I.writeIORef ipair (src, HaveLen 0)

436 go src (HaveLen len) = do

434 go src (HaveLen len) = do

437− (src', mbs) <- lift $ src C.$$+ CL.head

435+ (src', mbs) <- lift $ src $$++ CL.head

438 case mbs of

436 case mbs of

439 Nothing -> liftIO $ I.writeIORef ipair (src', HaveLen 0)

437 Nothing -> liftIO $ I.writeIORef ipair (src', HaveLen 0)

440 Just bs ->

438 Just bs ->

@@ -445,15 +443,15 @@ chunkedSource ipair = do

445 yield' src' mlen bs

443 yield' src' mlen bs

446 GT -> do

444 GT -> do

447 let (x, y) = S.splitAt (fromIntegral len) bs

445 let (x, y) = S.splitAt (fromIntegral len) bs

448− let src'' = C.yield y >> src'

446+ let src'' = fmapResume (yield y >>) src'

449 yield' src'' NeedLenNewline x

447 yield' src'' NeedLenNewline x

450

448

451 yield' src mlen bs = do

449 yield' src mlen bs = do

452 liftIO $ I.writeIORef ipair (src, mlen)

450 liftIO $ I.writeIORef ipair (src, mlen)

453− C.yield bs

451+ yield bs

454 go src mlen

452 go src mlen

455

453

456− getLen :: Monad m => C.Sink ByteString m (Word, ByteString)

454+ getLen :: Monad m => Sink ByteString m (Word, ByteString)

457 getLen = do

455 getLen = do

458 mbs <- CL.head

456 mbs <- CL.head

459 case mbs of

457 case mbs of

@@ -621,31 +619,23 @@ sendResponse th req conn r = sendResponse' r

621 sendResponse' (ResponseSource s hs bodyFlush)

619 sendResponse' (ResponseSource s hs bodyFlush)

622 | hasBody s req = do

620 | hasBody s req = do

623 let src = CL.sourceList [headers' needsChunked'] `mappend`

621 let src = CL.sourceList [headers' needsChunked'] `mappend`

624− (if needsChunked' then body C.$= chunk else body)

622+ (if needsChunked' then body $= chunk else body)

625− src C.$$ builderToByteString C.=$ connSink conn th

623+ src $$ builderToByteString =$ connSink conn th

626 return $ isKeepAlive hs

624 return $ isKeepAlive hs

627 | otherwise = liftIO $ do

625 | otherwise = liftIO $ do

628 sendHeader $ headers' False

626 sendHeader $ headers' False

629 T.tickle th

627 T.tickle th

630 return isPersist

628 return isPersist

631 where

629 where

632− body = fmap2 (\x -> case x of

630+ body = mapOutput (\x -> case x of

633− C.Flush -> flush

631+ Flush -> flush

634− C.Chunk builder -> builder) bodyFlush

632+ Chunk builder -> builder) bodyFlush

635 headers' = headers version s hs

633 headers' = headers version s hs

636 -- FIXME perhaps alloca a buffer per thread and reuse that in all

634 -- FIXME perhaps alloca a buffer per thread and reuse that in all

637 -- functions below. Should lessen greatly the GC burden (I hope)

635 -- functions below. Should lessen greatly the GC burden (I hope)

638 needsChunked' = needsChunked hs

636 needsChunked' = needsChunked hs

639− chunk :: C.Conduit Builder (ResourceT IO) Builder

637+ chunk :: Conduit Builder (ResourceT IO) Builder

640− chunk = C.NeedInput push close

638+ chunk = await >>= maybe (yield chunkedTransferTerminator) (\x -> yield (chunkedTransferEncoding x) >> chunk)

641− push x = C.HaveOutput chunk (return ()) (chunkedTransferEncoding x)

642− close = C.HaveOutput (C.Done Nothing ()) (return ()) chunkedTransferTerminator

643−

644−fmap2 :: Functor m => (o1 -> o2) -> C.Pipe i o1 m r -> C.Pipe i o2 m r

645−fmap2 f (C.HaveOutput p c o) = C.HaveOutput (fmap2 f p) c (f o)

646−fmap2 f (C.NeedInput p c) = C.NeedInput (fmap2 f . p) (fmap2 f c)

647−fmap2 f (C.PipeM mp c) = C.PipeM (fmap (fmap2 f) mp) c

648−fmap2 _ (C.Done i x) = C.Done i x

649

639

650parseHeaderNoAttr :: ByteString -> H.Header

640parseHeaderNoAttr :: ByteString -> H.Header

651parseHeaderNoAttr s =

641parseHeaderNoAttr s =

@@ -653,31 +643,31 @@ parseHeaderNoAttr s =

653 rest' = S.dropWhile (\c -> c == 32 || c == 9) $ S.drop 1 rest

643 rest' = S.dropWhile (\c -> c == 32 || c == 9) $ S.drop 1 rest

654 in (CI.mk k, rest')

644 in (CI.mk k, rest')

655

645

656−connSource :: Connection -> T.Handle -> C.Source (ResourceT IO) ByteString

646+connSource :: Connection -> T.Handle -> Source (ResourceT IO) ByteString

657connSource Connection { connRecv = recv } th =

647connSource Connection { connRecv = recv } th =

658 src

648 src

659 where

649 where

660− src = C.PipeM (do

650+ src = do

661 bs <- liftIO recv

651 bs <- liftIO recv

662 if S.null bs

652 if S.null bs

663− then return $ C.Done Nothing ()

653+ then return ()

664 else do

654 else do

665 when (S.length bs >= 2048) $ liftIO $ T.tickle th

655 when (S.length bs >= 2048) $ liftIO $ T.tickle th

666− return (C.HaveOutput src (return ()) bs))

656+ yield bs

667− (return ())

657+ src

668

658

669-- | Use 'connSendAll' to send this data while respecting timeout rules.

659-- | Use 'connSendAll' to send this data while respecting timeout rules.

670−connSink :: Connection -> T.Handle -> C.Sink B.ByteString (ResourceT IO) ()

660+connSink :: Connection -> T.Handle -> Sink B.ByteString (ResourceT IO) ()

671connSink Connection { connSendAll = send } th =

661connSink Connection { connSendAll = send } th =

672 sink

662 sink

673 where

663 where

674− sink = C.NeedInput push close

664+ sink = await >>= maybe close push

675 close = liftIO (T.resume th)

665 close = liftIO (T.resume th)

676− push x = C.PipeM (liftIO $ do

666+ push x = do

677− T.resume th

667+ liftIO $ T.resume th

678− send x

668+ liftIO $ send x

679− T.pause th

669+ liftIO $ T.pause th

680− return sink) (liftIO $ T.resume th)

670+ sink

681 -- We pause timeouts before passing control back to user code. This ensures

671 -- We pause timeouts before passing control back to user code. This ensures

682 -- that a timeout will only ever be executed when Warp is in control. We

672 -- that a timeout will only ever be executed when Warp is in control. We

683 -- also make sure to resume the timeout after the completion of user code

673 -- also make sure to resume the timeout after the completion of user code

@@ -699,7 +689,7 @@ data Settings = Settings

699 , settingsOnOpen :: IO () -- ^ What to do when a connection is open. Default: do nothing.

689 , settingsOnOpen :: IO () -- ^ What to do when a connection is open. Default: do nothing.

700 , settingsOnClose :: IO () -- ^ What to do when a connection is close. Default: do nothing.

690 , settingsOnClose :: IO () -- ^ What to do when a connection is close. Default: do nothing.

701 , settingsTimeout :: Int -- ^ Timeout value in seconds. Default value: 30

691 , settingsTimeout :: Int -- ^ Timeout value in seconds. Default value: 30

702− , settingsIntercept :: Request -> Maybe (C.Source (ResourceT IO) S.ByteString -> Connection -> ResourceT IO ())

692+ , settingsIntercept :: Request -> Maybe (Source (ResourceT IO) S.ByteString -> Connection -> ResourceT IO ())

703 , settingsManager :: Maybe Manager -- ^ Use an existing timeout manager instead of spawning a new one. If used, 'settingsTimeout' is ignored. Default is 'Nothing'

693 , settingsManager :: Maybe Manager -- ^ Use an existing timeout manager instead of spawning a new one. If used, 'settingsTimeout' is ignored. Default is 'Nothing'

704 }

694 }

705

695

@@ -735,10 +725,11 @@ data THStatus = THStatus

735 BSEndoList -- previously parsed lines

725 BSEndoList -- previously parsed lines

736 BSEndo -- bytestrings to be prepended

726 BSEndo -- bytestrings to be prepended

737

727

738−takeHeaders :: C.Sink ByteString (ResourceT IO) [ByteString]

728+takeHeaders :: Sink ByteString (ResourceT IO) [ByteString]

739takeHeaders =

729takeHeaders =

740− C.NeedInput (push (THStatus 0 id id)) close

730+ await >>= maybe close (push (THStatus 0 id id))

741 where

731 where

732+ close :: Sink ByteString (ResourceT IO) a

742 close = throwIO IncompleteHeaders

733 close = throwIO IncompleteHeaders

743

734

744 push (THStatus len lines prepend) bs

735 push (THStatus len lines prepend) bs

@@ -752,7 +743,7 @@ takeHeaders =

752 let len' = len + bsLen

743 let len' = len + bsLen

753 prepend' = prepend . S.append bs

744 prepend' = prepend . S.append bs

754 status = THStatus len' lines prepend'

745 status = THStatus len' lines prepend'

755− in C.NeedInput (push status) close

746+ in await >>= maybe close (push status)

756 -- Found a newline, but next line continues as a multiline header

747 -- Found a newline, but next line continues as a multiline header

757 Just (end, True) ->

748 Just (end, True) ->

758 let rest = S.drop (end + 1) bs

749 let rest = S.drop (end + 1) bs

@@ -776,7 +767,7 @@ takeHeaders =

776 rest = if start < bsLen

767 rest = if start < bsLen

777 then Just (SU.unsafeDrop start bs)

768 then Just (SU.unsafeDrop start bs)

778 else Nothing

769 else Nothing

779− in C.Done rest lines'

770+ in maybe (return ()) leftover rest >> return lines'

780 -- more headers

771 -- more headers

781 else

772 else

782 let len' = len + start

773 let len' = len + start

@@ -787,7 +778,7 @@ takeHeaders =

787 then let bs' = SU.unsafeDrop start bs

778 then let bs' = SU.unsafeDrop start bs

788 in push status bs'

779 in push status bs'

789 -- no more bytes in this chunk, ask for more

780 -- no more bytes in this chunk, ask for more

790− else C.NeedInput (push status) close

781+ else await >>= maybe close (push status)

791 where

782 where

792 bsLen = S.length bs

783 bsLen = S.length bs

793 mnl = do

784 mnl = do

@@ -831,3 +822,6 @@ serverHeader hdrs = case lookup key hdrs of

831 key = "Server"

822 key = "Server"

832 ver = B.pack $ "Warp/" ++ warpVersion

823 ver = B.pack $ "Warp/" ++ warpVersion

833 server = (key, ver)

824 server = (key, ver)

825+

826+fmapResume :: (Source m o1 -> Source m o2) -> ResumableSource m o1 -> ResumableSource m o2

827+fmapResume f (ResumableSource src m) = ResumableSource (f src) m

warp/warp.cabal+4−4

@@ -1,5 +1,5 @@

1Name: warp

2−Version: 1.2.1.1

2+Version: 1.2.1.2

3Synopsis: A fast, light-weight web server for WAI applications.

4License: MIT

5License-file: LICENSE

@@ -21,9 +21,9 @@ Library

21 , bytestring >= 0.9.1.4

22 , wai >= 1.2 && < 1.3

23 , transformers >= 0.2.2 && < 0.4

24− , conduit >= 0.4 && < 0.5

24+ , conduit >= 0.5 && < 0.6

25− , network-conduit >= 0.4 && < 0.5

25+ , network-conduit >= 0.5 && < 0.6

26− , blaze-builder-conduit >= 0.4 && < 0.5

26+ , blaze-builder-conduit >= 0.5 && < 0.6

27 , lifted-base >= 0.1 && < 0.2

28 , blaze-builder >= 0.2.1.4 && < 0.4

29 , simple-sendfile >= 0.2.2 && < 0.3