conduit-extra 1.1.3 → 1.3.8
raw patch · 30 files changed
Files
- ChangeLog.md +163/−0
- Data/Conduit/Attoparsec.hs +38/−29
- Data/Conduit/Binary.hs +157/−149
- Data/Conduit/Blaze.hs +0/−136
- Data/Conduit/ByteString/Builder.hs +34/−0
- Data/Conduit/Filesystem.hs +5/−62
- Data/Conduit/Foldl.hs +26/−0
- Data/Conduit/Lazy.hs +35/−21
- Data/Conduit/Network.hs +57/−25
- Data/Conduit/Network/UDP.hs +7/−8
- Data/Conduit/Network/Unix.hs +3/−12
- Data/Conduit/Process.hs +161/−12
- Data/Conduit/Process/Typed.hs +107/−0
- Data/Conduit/Text.hs +78/−100
- Data/Conduit/Zlib.hs +126/−47
- README.md +5/−0
- bench/blaze.hs +65/−0
- conduit-extra.cabal +41/−16
- test/Data/Conduit/AttoparsecSpec.hs +85/−17
- test/Data/Conduit/BinarySpec.hs +169/−25
- test/Data/Conduit/BlazeSpec.hs +0/−63
- test/Data/Conduit/ByteString/BuilderSpec.hs +61/−0
- test/Data/Conduit/ExtraSpec.hs +11/−11
- test/Data/Conduit/FilesystemSpec.hs +21/−21
- test/Data/Conduit/LazySpec.hs +0/−1
- test/Data/Conduit/NetworkSpec.hs +34/−7
- test/Data/Conduit/Process/TypedSpec.hs +31/−0
- test/Data/Conduit/ProcessSpec.hs +55/−5
- test/Data/Conduit/TextSpec.hs +70/−72
- test/Data/Conduit/ZlibSpec.hs +63/−20
+ ChangeLog.md view
@@ -0,0 +1,163 @@+# ChangeLog for conduit-extra++## 1.3.8++* Gracefully handle when a subprocess started using `Data.Conduit.Process.sourceProcessWithStreams` closes its stdin. Fixes [#523](https://github.com/snoyberg/conduit/issues/523)++## 1.3.7++* Allow Data.Conduit.Network.Unix on Windows [#518](https://github.com/snoyberg/conduit/pull/518)++## 1.3.6++* Add support for `transformers-0.6`++## 1.3.5++* Add `createSinkClose`++## 1.3.4++* Use `MonadUnliftIO`-generalized versions of `withProcess`-style functions now provided by `typed-process`++## 1.3.3++* Disable buffering in process modules [#402](https://github.com/snoyberg/conduit/issues/402)++## 1.3.2++* Expose `BuilderInput` and `FlushInput`.++## 1.3.1.1++* Attempt to fix a test suite failure [#385](https://github.com/snoyberg/conduit/issues/385)++## 1.3.1++* Switched `gzip` to use zlib's default compression level.++## 1.3.0++* Switch over to unliftio+* Upgrade to conduit 1.3.0++## 1.2.3.2++* Fix withSinkFileBuilder [#344](https://github.com/snoyberg/conduit/pull/344)++## 1.2.3.1++* Fix typo in implementation of `withProcess_`++## 1.2.3++* Added `withLoggedProcess_`++## 1.2.2.1++* Add missing `hClose` to `withSinkFileCautious`++## 1.2.2++* `sinkHandleBuilder`, `sinkHandleFlush`, `BuilderInput`, and `FlushInput`+ [#336](https://github.com/snoyberg/conduit/pull/336)+* `withSinkFileCautious`++## 1.2.1++* `Data.Conduit.Process.Typed`+* `withSourceFile`, `withSinkFile`, and `withSinkFileBuilder`++## 1.2.0++* Added the `posOffset` field to the+ `Data.Conduit.Attoparsec.Position` data type+ [#331](https://github.com/snoyberg/conduit/issues/331).++## 1.1.17++* Speed up `sinkHandle` by not flushing after every output operation.+ [#322](https://github.com/snoyberg/conduit/issues/322)++## 1.1.16++* Add `Data.Conduit.Foldl` adapter module for the `foldl`+ package. [#312](https://github.com/snoyberg/conduit/pull/312)++## 1.1.15++* `sinkTempFile` and `sinkSystemTempFile`++## 1.1.14++* `sinkFileCautious`++## 1.1.13.3++* `withCheckedProcessCleanup` properly closes opened `Handle`s+ [#280](https://github.com/snoyberg/conduit/issues/280)++## 1.1.13.2++* Fix alignment issues on non-X86 archs++## 1.1.13.1++* Fix an incorrect comment++## 1.1.13++* Add `sinkStorable` and `sinkStorableEx`++## 1.1.12.1++* Fix build for GHC `<= 7.8` [#260](https://github.com/snoyberg/conduit/issues/260)+* Fix accidentally breaking change in `sourceProcessWithConsumer` type signature++## 1.1.12++* Add sourceProcessWithStreams [#258](https://github.com/snoyberg/conduit/pull/258)++## 1.1.11++* `withCheckedProcessCleanup`++## 1.1.10.1++* Fix a leftovers bug in helperDecompress #254++## 1.1.10++* `multiple` combinator for `Data.Conduit.Zlib` [#254](https://github.com/snoyberg/conduit/issues/254)++## 1.1.9.3++* Some typo fixes in docs++## 1.1.9++* detectUtf [#217](https://github.com/snoyberg/conduit/pull/217)++## 1.1.8++* Adding buffer size to sourceHandleRange [#213](https://github.com/snoyberg/conduit/pull/213)++## 1.1.7.3++* Make Binary.lines O(n) instead of O(n^2) [#209](https://github.com/snoyberg/conduit/pull/209)++## 1.1.7.2++* Fix for: Decompressing a specific amount of zlib data "eats" following data [#20](https://github.com/fpco/streaming-commons/issues/20)++## 1.1.7++Add `Data.Conduit.ByteString.Builder`++## 1.1.6++Generalized return type in `runGeneralTCPServer`.++## 1.1.5++Added `sinkParserEither` ([pull request #189](https://github.com/snoyberg/conduit/pull/189))
Data/Conduit/Attoparsec.hs view
@@ -13,6 +13,7 @@ module Data.Conduit.Attoparsec ( -- * Sink sinkParser+ , sinkParserEither -- * Conduit , conduitParser , conduitParserEither@@ -37,7 +38,7 @@ import qualified Data.Attoparsec.Text import qualified Data.Attoparsec.Types as A import Data.Conduit-import Control.Monad.Trans.Resource (MonadThrow, monadThrow)+import Control.Monad.Trans.Resource (MonadThrow, throwM) -- | The context and message from a 'A.Fail' value. data ParseError = ParseError@@ -52,11 +53,13 @@ data Position = Position { posLine :: {-# UNPACK #-} !Int , posCol :: {-# UNPACK #-} !Int+ , posOffset :: {-# UNPACK #-} !Int+ -- ^ @since 1.2.0 } deriving (Eq, Ord) instance Show Position where- show (Position l c) = show l ++ ':' : show c+ show (Position l c off) = show l ++ ':' : show c ++ " (" ++ show off ++ ")" data PositionRange = PositionRange { posRangeStart :: {-# UNPACK #-} !Position@@ -73,7 +76,6 @@ feedA :: A.IResult a b -> a -> A.IResult a b empty :: a isNull :: a -> Bool- notEmpty :: [a] -> [a] getLinesCols :: a -> Position -- | Return the beginning of the first input with the length of@@ -86,11 +88,11 @@ feedA = Data.Attoparsec.ByteString.feed empty = B.empty isNull = B.null- notEmpty = filter (not . B.null)- getLinesCols = B.foldl' f (Position 0 0)+ getLinesCols = B.foldl' f (Position 0 0 0) where- f (Position l c) ch | ch == 10 = Position (l + 1) 0- | otherwise = Position l (c + 1)+ f (Position l c o) ch+ | ch == 10 = Position (l + 1) 0 (o + 1)+ | otherwise = Position l (c + 1) (o + 1) stripFromEnd b1 b2 = B.take (B.length b1 - B.length b2) b1 instance AttoparsecInput T.Text where@@ -98,23 +100,29 @@ feedA = Data.Attoparsec.Text.feed empty = T.empty isNull = T.null- notEmpty = filter (not . T.null)- getLinesCols = T.foldl' f (Position 0 0)+ getLinesCols = T.foldl' f (Position 0 0 0) where- f (Position l c) ch | ch == '\n' = Position (l + 1) 0- | otherwise = Position l (c + 1)+ f (Position l c o) ch+ | ch == '\n' = Position (l + 1) 0 (o + 1)+ | otherwise = Position l (c + 1) (o + 1) stripFromEnd (TI.Text arr1 off1 len1) (TI.Text _ _ len2) =- TI.textP arr1 off1 (len1 - len2)+ TI.text arr1 off1 (len1 - len2) -- | Convert an Attoparsec 'A.Parser' into a 'Sink'. The parser will -- be streamed bytes until it returns 'A.Done' or 'A.Fail'. ----- If parsing fails, a 'ParseError' will be thrown with 'monadThrow'.+-- If parsing fails, a 'ParseError' will be thrown with 'throwM'. -- -- Since 0.5.0-sinkParser :: (AttoparsecInput a, MonadThrow m) => A.Parser a b -> Consumer a m b-sinkParser = fmap snd . sinkParserPosErr (Position 1 1)+sinkParser :: (AttoparsecInput a, MonadThrow m) => A.Parser a b -> ConduitT a o m b+sinkParser = fmap snd . sinkParserPosErr (Position 1 1 0) +-- | Same as 'sinkParser', but we return an 'Either' type instead+-- of raising an exception.+--+-- Since 1.1.5+sinkParserEither :: (AttoparsecInput a, Monad m) => A.Parser a b -> ConduitT a o m (Either ParseError b)+sinkParserEither = (fmap.fmap) snd . sinkParserPos (Position 1 1 0) -- | Consume a stream of parsed tokens, returning both the token and@@ -122,9 +130,9 @@ -- on bad input. -- -- Since 0.5.0-conduitParser :: (AttoparsecInput a, MonadThrow m) => A.Parser a b -> Conduit a m (PositionRange, b)+conduitParser :: (AttoparsecInput a, MonadThrow m) => A.Parser a b -> ConduitT a (PositionRange, b) m () conduitParser parser =- conduit $ Position 1 0+ conduit $ Position 1 1 0 where conduit !pos = await >>= maybe (return ()) go where@@ -136,11 +144,11 @@ {-# SPECIALIZE conduitParser :: MonadThrow m => A.Parser T.Text b- -> Conduit T.Text m (PositionRange, b) #-}+ -> ConduitT T.Text (PositionRange, b) m () #-} {-# SPECIALIZE conduitParser :: MonadThrow m => A.Parser B.ByteString b- -> Conduit B.ByteString m (PositionRange, b) #-}+ -> ConduitT B.ByteString (PositionRange, b) m () #-} @@ -149,9 +157,9 @@ conduitParserEither :: (Monad m, AttoparsecInput a) => A.Parser a b- -> Conduit a m (Either ParseError (PositionRange, b))+ -> ConduitT a (Either ParseError (PositionRange, b)) m () conduitParserEither parser =- conduit $ Position 1 0+ conduit $ Position 1 1 0 where conduit !pos = await >>= maybe (return ()) go where@@ -166,11 +174,11 @@ {-# SPECIALIZE conduitParserEither :: Monad m => A.Parser T.Text b- -> Conduit T.Text m (Either ParseError (PositionRange, b)) #-}+ -> ConduitT T.Text (Either ParseError (PositionRange, b)) m () #-} {-# SPECIALIZE conduitParserEither :: Monad m => A.Parser B.ByteString b- -> Conduit B.ByteString m (Either ParseError (PositionRange, b)) #-}+ -> ConduitT B.ByteString (Either ParseError (PositionRange, b)) m () #-} @@ -179,10 +187,10 @@ :: (AttoparsecInput a, MonadThrow m) => Position -> A.Parser a b- -> Consumer a m (Position, b)+ -> ConduitT a o m (Position, b) sinkParserPosErr pos0 p = sinkParserPos pos0 p >>= f where- f (Left e) = monadThrow e+ f (Left e) = throwM e f (Right a) = return a {-# INLINE sinkParserPosErr #-} @@ -191,7 +199,7 @@ :: (AttoparsecInput a, Monad m) => Position -> A.Parser a b- -> Consumer a m (Either ParseError (Position, b))+ -> ConduitT a o m (Either ParseError (Position, b)) sinkParserPos pos0 p = sink empty pos0 (parseA p) where sink prev pos parser = await >>= maybe close push@@ -225,10 +233,11 @@ pos' = addLinesCols prev pos addLinesCols :: AttoparsecInput a => a -> Position -> Position- addLinesCols x (Position lines cols) =- lines' `seq` cols' `seq` Position lines' cols'+ addLinesCols x (Position lines cols off) =+ lines' `seq` cols' `seq` off' `seq` Position lines' cols' off' where- Position dlines dcols = getLinesCols x+ Position dlines dcols doff = getLinesCols x lines' = lines + dlines cols' = (if dlines > 0 then 1 else cols) + dcols+ off' = off + doff {-# INLINE sinkParserPos #-}
Data/Conduit/Binary.hs view
@@ -1,5 +1,13 @@ {-# LANGUAGE CPP, RankNTypes #-}--- | Functions for interacting with bytes.+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE ScopedTypeVariables #-}+-- | /NOTE/ It is recommended to start using "Data.Conduit.Combinators" instead+-- of this module.+--+-- Functions for interacting with bytes.+--+-- For many purposes, it's recommended to use the conduit-combinators library,+-- which provides a more complete set of functions. module Data.Conduit.Binary ( -- * Files and @Handle@s @@ -8,16 +16,26 @@ -- order to run such code, you will need to use @runResourceT@. -- ** Sources- sourceFile- , sourceHandle- , sourceHandleUnsafe- , sourceIOHandle+ CC.sourceFile+ , CC.sourceHandle+ , CC.sourceHandleUnsafe+ , CC.sourceIOHandle , sourceFileRange , sourceHandleRange+ , sourceHandleRangeWithBuffer+ , CC.withSourceFile -- ** Sinks- , sinkFile- , sinkHandle- , sinkIOHandle+ , CC.sinkFile+ , CC.sinkFileCautious+ , CC.sinkTempFile+ , CC.sinkSystemTempFile+ , CC.sinkHandle+ , CC.sinkIOHandle+ , CC.sinkHandleBuilder+ , CC.sinkHandleFlush+ , CC.withSinkFile+ , CC.withSinkFileBuilder+ , CC.withSinkFileCautious -- ** Conduits , conduitFile , conduitHandle@@ -32,128 +50,46 @@ , sinkCacheLength , sinkLbs , mapM_+ -- *** Storable+ , sinkStorable+ , sinkStorableEx -- ** Conduits , isolate , takeWhile , Data.Conduit.Binary.lines ) where -import qualified Data.Streaming.FileRead as FR+import qualified Data.Conduit.Combinators as CC import Prelude hiding (head, take, drop, takeWhile, dropWhile, mapM_) import qualified Data.ByteString as S+import Data.ByteString.Unsafe (unsafeUseAsCString) import qualified Data.ByteString.Lazy as L import Data.Conduit import Data.Conduit.List (sourceList, consume) import Control.Exception (assert, finally)-import Control.Monad (unless, when)+import Control.Monad (unless) import Control.Monad.IO.Class (liftIO, MonadIO)-import Control.Monad.Trans.Resource (allocate, release)+import Control.Monad.Trans.Resource (allocate, release, MonadThrow (..)) import Control.Monad.Trans.Class (lift) import qualified System.IO as IO import Data.Word (Word8, Word64)+#if (__GLASGOW_HASKELL__ < 710) import Control.Applicative ((<$>))+#endif import System.Directory (getTemporaryDirectory, removeFile) import Data.ByteString.Lazy.Internal (defaultChunkSize)-import Data.ByteString.Internal (ByteString (PS), inlinePerformIO)+import Data.ByteString.Internal (ByteString (PS), accursedUnutterablePerformIO) import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr) import Foreign.ForeignPtr (touchForeignPtr)-import Foreign.Ptr (plusPtr)-import Foreign.Storable (peek)-import GHC.ForeignPtr (mallocPlainForeignPtrBytes)+import Foreign.Ptr (plusPtr, castPtr)+import Foreign.Storable (Storable, peek, sizeOf) import Control.Monad.Trans.Resource (MonadResource)---- | Stream the contents of a file as binary data.------ Since 0.3.0-sourceFile :: MonadResource m- => FilePath- -> Producer m S.ByteString-sourceFile fp =- bracketP- (FR.openFile fp)- FR.closeFile- loop- where- loop h = do- bs <- liftIO $ FR.readChunk h- unless (S.null bs) $ do- yield bs- loop h---- | Stream the contents of a 'IO.Handle' as binary data. Note that this--- function will /not/ automatically close the @Handle@ when processing--- completes, since it did not acquire the @Handle@ in the first place.------ Since 0.3.0-sourceHandle :: MonadIO m- => IO.Handle- -> Producer m S.ByteString-sourceHandle h =- loop- where- loop = do- bs <- liftIO (S.hGetSome h defaultChunkSize)- if S.null bs- then return ()- else yield bs >> loop---- | Same as @sourceHandle@, but instead of allocating a new buffer for each--- incoming chunk of data, reuses the same buffer. Therefore, the @ByteString@s--- yielded by this function are not referentially transparent between two--- different @yield@s.------ This function will be slightly more efficient than @sourceHandle@ by--- avoiding allocations and reducing garbage collections, but should only be--- used if you can guarantee that you do not reuse a @ByteString@ (or any slice--- thereof) between two calls to @await@.------ Since 1.0.12-sourceHandleUnsafe :: MonadIO m => IO.Handle -> Source m ByteString-sourceHandleUnsafe handle = do- fptr <- liftIO $ mallocPlainForeignPtrBytes defaultChunkSize- let ptr = unsafeForeignPtrToPtr fptr- loop = do- count <- liftIO $ IO.hGetBuf handle ptr defaultChunkSize- when (count > 0) $ do- yield (PS fptr 0 count)- loop-- loop-- liftIO $ touchForeignPtr fptr---- | An alternative to 'sourceHandle'.--- Instead of taking a pre-opened 'IO.Handle', it takes an action that opens--- a 'IO.Handle' (in read mode), so that it can open it only when needed--- and closed it as soon as possible.------ Since 0.3.0-sourceIOHandle :: MonadResource m- => IO IO.Handle- -> Producer m S.ByteString-sourceIOHandle alloc = bracketP alloc IO.hClose sourceHandle---- | Stream all incoming data to the given 'IO.Handle'. Note that this function--- will /not/ automatically close the @Handle@ when processing completes.------ Since 0.3.0-sinkHandle :: MonadIO m- => IO.Handle- -> Consumer S.ByteString m ()-sinkHandle h = awaitForever $ \bs -> liftIO $ do- S.hPut h bs- IO.hFlush h---- | An alternative to 'sinkHandle'.--- Instead of taking a pre-opened 'IO.Handle', it takes an action that opens--- a 'IO.Handle' (in write mode), so that it can open it only when needed--- and close it as soon as possible.------ Since 0.3.0-sinkIOHandle :: MonadResource m- => IO IO.Handle- -> Consumer S.ByteString m ()-sinkIOHandle alloc = bracketP alloc IO.hClose sinkHandle+import Control.Exception (Exception)+import Data.Typeable (Typeable)+import Foreign.Ptr (Ptr)+#ifndef ALLOW_UNALIGNED_ACCESS+import Foreign.Marshal (alloca, copyBytes)+#endif -- | Stream the contents of a file as binary data, starting from a certain -- offset and only consuming up to a certain number of bytes.@@ -163,7 +99,7 @@ => FilePath -> Maybe Integer -- ^ Offset -> Maybe Integer -- ^ Maximum count- -> Producer m S.ByteString+ -> ConduitT i S.ByteString m () sourceFileRange fp offset count = bracketP (IO.openBinaryFile fp IO.ReadMode) IO.hClose@@ -177,8 +113,22 @@ => IO.Handle -> Maybe Integer -- ^ Offset -> Maybe Integer -- ^ Maximum count- -> Producer m S.ByteString-sourceHandleRange handle offset count = do+ -> ConduitT i S.ByteString m ()+sourceHandleRange handle offset count =+ sourceHandleRangeWithBuffer handle offset count defaultChunkSize++-- | Stream the contents of a handle as binary data, starting from a certain+-- offset and only consuming up to a certain number of bytes. This function+-- consumes chunks as specified by the buffer size.+--+-- Since 1.1.8+sourceHandleRangeWithBuffer :: MonadIO m+ => IO.Handle+ -> Maybe Integer -- ^ Offset+ -> Maybe Integer -- ^ Maximum count+ -> Int -- ^ Buffer size+ -> ConduitT i S.ByteString m ()+sourceHandleRangeWithBuffer handle offset count buffer = do case offset of Nothing -> return () Just off -> liftIO $ IO.hSeek handle IO.AbsoluteSeek off@@ -187,7 +137,7 @@ Just c -> pullLimited (fromInteger c) where pullUnlimited = do- bs <- liftIO $ S.hGetSome handle 4096+ bs <- liftIO $ S.hGetSome handle buffer if S.null bs then return () else do@@ -195,7 +145,7 @@ pullUnlimited pullLimited c = do- bs <- liftIO $ S.hGetSome handle (min c 4096)+ bs <- liftIO $ S.hGetSome handle (min c buffer) let c' = c - S.length bs assert (c' >= 0) $ if S.null bs@@ -204,21 +154,13 @@ yield bs pullLimited c' --- | Stream all incoming data to the given file.------ Since 0.3.0-sinkFile :: MonadResource m- => FilePath- -> Consumer S.ByteString m ()-sinkFile fp = sinkIOHandle (IO.openBinaryFile fp IO.WriteMode)- -- | Stream the contents of the input to a file, and also send it along the -- pipeline. Similar in concept to the Unix command @tee@. -- -- Since 0.3.0 conduitFile :: MonadResource m => FilePath- -> Conduit S.ByteString m S.ByteString+ -> ConduitT S.ByteString S.ByteString m () conduitFile fp = bracketP (IO.openBinaryFile fp IO.WriteMode) IO.hClose@@ -229,17 +171,17 @@ -- does not close the handle on completion. Related to: @conduitFile@. -- -- Since 1.0.9-conduitHandle :: MonadIO m => IO.Handle -> Conduit S.ByteString m S.ByteString+conduitHandle :: MonadIO m => IO.Handle -> ConduitT S.ByteString S.ByteString m () conduitHandle h = awaitForever $ \bs -> liftIO (S.hPut h bs) >> yield bs --- | Ensure that only up to the given number of bytes are consume by the inner+-- | Ensure that only up to the given number of bytes are consumed by the inner -- sink. Note that this does /not/ ensure that all of those bytes are in fact -- consumed. -- -- Since 0.3.0 isolate :: Monad m => Int- -> Conduit S.ByteString m S.ByteString+ -> ConduitT S.ByteString S.ByteString m () isolate = loop where@@ -259,7 +201,7 @@ -- | Return the next byte from the stream, if available. -- -- Since 0.3.0-head :: Monad m => Consumer S.ByteString m (Maybe Word8)+head :: Monad m => ConduitT S.ByteString o m (Maybe Word8) head = do mbs <- await case mbs of@@ -272,7 +214,7 @@ -- | Return all bytes while the predicate returns @True@. -- -- Since 0.3.0-takeWhile :: Monad m => (Word8 -> Bool) -> Conduit S.ByteString m S.ByteString+takeWhile :: Monad m => (Word8 -> Bool) -> ConduitT S.ByteString S.ByteString m () takeWhile p = loop where@@ -288,7 +230,7 @@ -- | Ignore all bytes while the predicate returns @True@. -- -- Since 0.3.0-dropWhile :: Monad m => (Word8 -> Bool) -> Consumer S.ByteString m ()+dropWhile :: Monad m => (Word8 -> Bool) -> ConduitT S.ByteString o m () dropWhile p = loop where@@ -303,7 +245,7 @@ -- | Take the given number of bytes, if available. -- -- Since 0.3.0-take :: Monad m => Int -> Consumer S.ByteString m L.ByteString+take :: Monad m => Int -> ConduitT S.ByteString o m L.ByteString take 0 = return L.empty take n0 = go n0 id where@@ -321,7 +263,7 @@ -- | Drop up to the given number of bytes. -- -- Since 0.5.0-drop :: Monad m => Int -> Consumer S.ByteString m ()+drop :: Monad m => Int -> ConduitT S.ByteString o m () drop 0 = return () drop n0 = go n0 where@@ -340,29 +282,27 @@ -- (10), and strip it from the output. -- -- Since 0.3.0-lines :: Monad m => Conduit S.ByteString m S.ByteString+lines :: Monad m => ConduitT S.ByteString S.ByteString m () lines =- loop id+ loop [] where- loop front = await >>= maybe (finish front) (go front)+ loop acc = await >>= maybe (finish acc) (go acc) - finish front =- let final = front S.empty+ finish acc =+ let final = S.concat $ reverse acc in unless (S.null final) (yield final) - go sofar more =+ go acc more = case S.uncons second of- Just (_, second') -> yield (sofar first) >> go id second'- Nothing ->- let rest = sofar more- in loop $ S.append rest+ Just (_, second') -> yield (S.concat $ reverse $ first:acc) >> go [] second'+ Nothing -> loop $ more:acc where- (first, second) = S.breakByte 10 more+ (first, second) = S.break (== 10) more -- | Stream the chunks from a lazy bytestring. -- -- Since 0.5.0-sourceLbs :: Monad m => L.ByteString -> Producer m S.ByteString+sourceLbs :: Monad m => L.ByteString -> ConduitT i S.ByteString m () sourceLbs = sourceList . L.toChunks -- | Stream the input data into a temp file and count the number of bytes@@ -373,7 +313,7 @@ -- -- Since 1.0.5 sinkCacheLength :: (MonadResource m1, MonadResource m2)- => Sink S.ByteString m1 (Word64, Source m2 S.ByteString)+ => ConduitT S.ByteString o m1 (Word64, ConduitT i S.ByteString m2 ()) sinkCacheLength = do tmpdir <- liftIO getTemporaryDirectory (releaseKey, (fp, h)) <- allocate@@ -381,9 +321,9 @@ (\(fp, h) -> IO.hClose h `finally` removeFile fp) len <- sinkHandleLen h liftIO $ IO.hClose h- return (len, sourceFile fp >> release releaseKey)+ return (len, CC.sourceFile fp >> release releaseKey) where- sinkHandleLen :: MonadResource m => IO.Handle -> Sink S.ByteString m Word64+ sinkHandleLen :: MonadResource m => IO.Handle -> ConduitT S.ByteString o m Word64 sinkHandleLen h = loop 0 where@@ -398,7 +338,7 @@ -- is performed, but rather all content is read into memory strictly. -- -- Since 1.0.5-sinkLbs :: Monad m => Sink S.ByteString m L.ByteString+sinkLbs :: Monad m => ConduitT S.ByteString o m L.ByteString sinkLbs = fmap L.fromChunks consume mapM_BS :: Monad m => (Word8 -> m ()) -> S.ByteString -> m ()@@ -406,9 +346,9 @@ let start = unsafeForeignPtrToPtr fptr `plusPtr` offset end = start `plusPtr` len loop ptr- | ptr >= end = inlinePerformIO (touchForeignPtr fptr) `seq` return ()+ | ptr >= end = accursedUnutterablePerformIO (touchForeignPtr fptr) `seq` return () | otherwise = do- f (inlinePerformIO (peek ptr))+ f (accursedUnutterablePerformIO (peek ptr)) loop (ptr `plusPtr` 1) loop start {-# INLINE mapM_BS #-}@@ -416,6 +356,74 @@ -- | Perform a computation on each @Word8@ in a stream. -- -- Since 1.0.10-mapM_ :: Monad m => (Word8 -> m ()) -> Consumer S.ByteString m ()+mapM_ :: Monad m => (Word8 -> m ()) -> ConduitT S.ByteString o m () mapM_ f = awaitForever (lift . mapM_BS f) {-# INLINE mapM_ #-}++-- | Consume some instance of @Storable@ from the incoming byte stream. In the+-- event of insufficient bytes in the stream, returns a @Nothing@ and returns+-- all unused input as leftovers.+--+-- @since 1.1.13+sinkStorable :: (Monad m, Storable a) => ConduitT S.ByteString o m (Maybe a)+sinkStorable = sinkStorableHelper Just (return Nothing)++-- | Same as 'sinkStorable', but throws a 'SinkStorableInsufficientBytes'+-- exception (via 'throwM') in the event of insufficient bytes. This can be+-- more efficient to use than 'sinkStorable' as it avoids the need to+-- construct/deconstruct a @Maybe@ wrapper in the success case.+--+-- @since 1.1.13+sinkStorableEx :: (MonadThrow m, Storable a) => ConduitT S.ByteString o m a+sinkStorableEx = sinkStorableHelper id (throwM SinkStorableInsufficientBytes)++sinkStorableHelper :: forall m a b o. (Monad m, Storable a)+ => (a -> b)+ -> (ConduitT S.ByteString o m b)+ -> ConduitT S.ByteString o m b+sinkStorableHelper wrap failure = do+ start+ where+ size = sizeOf (undefined :: a)++ -- try the optimal case: next chunk has all the data we need+ start = do+ mbs <- await+ case mbs of+ Nothing -> failure+ Just bs+ | S.null bs -> start+ | otherwise ->+ case compare (S.length bs) size of+ LT -> do+ -- looks like we're stuck concating+ leftover bs+ lbs <- take size+ let bs' = S.concat $ L.toChunks lbs+ case compare (S.length bs') size of+ LT -> do+ leftover bs'+ failure+ EQ -> process bs'+ GT -> assert False (process bs')+ EQ -> process bs+ GT -> do+ let (x, y) = S.splitAt size bs+ leftover y+ process x++ -- Given a bytestring of exactly the correct size, grab the value+ process bs = return $! wrap $! accursedUnutterablePerformIO $!+ unsafeUseAsCString bs (safePeek undefined . castPtr)++ safePeek :: a -> Ptr a -> IO a+#ifdef ALLOW_UNALIGNED_ACCESS+ safePeek _ = peek+#else+ safePeek val ptr = alloca (\t -> copyBytes t ptr (sizeOf val) >> peek t)+#endif+{-# INLINE sinkStorableHelper #-}++data SinkStorableException = SinkStorableInsufficientBytes+ deriving (Show, Typeable)+instance Exception SinkStorableException
− Data/Conduit/Blaze.hs
@@ -1,136 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE RankNTypes #-}--- | Convert a stream of blaze-builder @Builder@s into a stream of @ByteString@s.------ Adapted from blaze-builder-enumerator, written by myself and Simon Meier.------ Note that the functions here can work in any monad built on top of @IO@ or--- @ST@.-module Data.Conduit.Blaze- (-- -- * Conduits from builders to bytestrings- builderToByteString- , unsafeBuilderToByteString- , builderToByteStringWith-- -- ** Flush- , builderToByteStringFlush- , builderToByteStringWithFlush-- -- * Buffers- , Buffer-- -- ** Status information- , freeSize- , sliceSize- , bufferSize-- -- ** Creation and modification- , allocBuffer- , reuseBuffer- , nextSlice-- -- ** Conversion to bytestings- , unsafeFreezeBuffer- , unsafeFreezeNonEmptyBuffer-- -- * Buffer allocation strategies- , BufferAllocStrategy- , allNewBuffersStrategy- , reuseBufferStrategy- ) where--import Data.Conduit-import Control.Monad (unless, liftM)-import Control.Monad.Trans.Class (lift, MonadTrans)--import qualified Data.ByteString as S--import Blaze.ByteString.Builder.Internal-import Blaze.ByteString.Builder.Internal.Types-import Blaze.ByteString.Builder.Internal.Buffer-import Control.Monad.Primitive (PrimMonad, unsafePrimToPrim)-import Control.Monad.Base (MonadBase, liftBase)-import Data.Streaming.Blaze--unsafeLiftIO :: (MonadBase base m, PrimMonad base) => IO a -> m a-unsafeLiftIO = liftBase . unsafePrimToPrim---- | Incrementally execute builders and pass on the filled chunks as--- bytestrings.-builderToByteString :: (MonadBase base m, PrimMonad base) => Conduit Builder m S.ByteString-builderToByteString =- builderToByteStringWith defaultStrategy---- |------ Since 0.0.2-builderToByteStringFlush :: (MonadBase base m, PrimMonad base) => Conduit (Flush Builder) m (Flush S.ByteString)-builderToByteStringFlush =- builderToByteStringWithFlush defaultStrategy---- | Incrementally execute builders on the given buffer and pass on the filled--- chunks as bytestrings. Note that, if the given buffer is too small for the--- execution of a build step, a larger one will be allocated.------ WARNING: This conduit yields bytestrings that are NOT--- referentially transparent. Their content will be overwritten as soon--- as control is returned from the inner sink!-unsafeBuilderToByteString :: (MonadBase base m, PrimMonad base)- => IO Buffer -- action yielding the inital buffer.- -> Conduit Builder m S.ByteString-unsafeBuilderToByteString = builderToByteStringWith . reuseBufferStrategy----- | A conduit that incrementally executes builders and passes on the--- filled chunks as bytestrings to an inner sink.------ INV: All bytestrings passed to the inner sink are non-empty.-builderToByteStringWith :: (MonadBase base m, PrimMonad base)- => BufferAllocStrategy- -> Conduit Builder m S.ByteString-builderToByteStringWith =- helper (liftM (fmap Chunk) await) yield'- where- yield' Flush = return ()- yield' (Chunk bs) = yield bs---- |------ Since 0.0.2-builderToByteStringWithFlush- :: (MonadBase base m, PrimMonad base)- => BufferAllocStrategy- -> Conduit (Flush Builder) m (Flush S.ByteString)-builderToByteStringWithFlush = helper await yield--helper :: (MonadBase base m, PrimMonad base, Monad (t m), MonadTrans t)- => t m (Maybe (Flush Builder))- -> (Flush S.ByteString -> t m ())- -> BufferAllocStrategy- -> t m ()-helper await' yield' strat = do- (recv, finish) <- lift $ unsafeLiftIO $ newBlazeRecv strat- let loop = await' >>= maybe finish' cont- finish' = do- mbs <- lift $ unsafeLiftIO finish- maybe (return ()) (yield' . Chunk) mbs- cont fbuilder = do- let builder =- case fbuilder of- Flush -> flush- Chunk b -> b- popper <- lift $ unsafeLiftIO $ recv builder- let cont' = do- bs <- lift $ unsafeLiftIO popper- unless (S.null bs) $ do- yield' (Chunk bs)- cont'- cont'- case fbuilder of- Flush -> yield' Flush- Chunk _ -> return ()- loop- loop
+ Data/Conduit/ByteString/Builder.hs view
@@ -0,0 +1,34 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE RankNTypes #-}+-- | Convert a stream of blaze-builder @Builder@s into a stream of @ByteString@s.+--+-- Works with both blaze-builder < 0.4's @Builder@s and+-- 'Data.ByteString.Builder.Builder'.+--+-- Adapted from blaze-builder-enumerator, written by myself and Simon Meier.+--+-- Note that the functions here can work in any monad built on top of @IO@ or+-- @ST@.+--+-- Since 1.1.7.0+--+module Data.Conduit.ByteString.Builder+ (++ -- * Conduits from builders to bytestrings+ CC.builderToByteString+ , CC.unsafeBuilderToByteString+ , CC.builderToByteStringWith++ -- ** Flush+ , CC.builderToByteStringFlush+ , CC.builderToByteStringWithFlush++ -- * Buffer allocation strategies+ , CC.BufferAllocStrategy+ , CC.allNewBuffersStrategy+ , CC.reuseBufferStrategy+ ) where++import qualified Data.Conduit.Combinators as CC
Data/Conduit/Filesystem.hs view
@@ -1,66 +1,9 @@ {-# LANGUAGE RankNTypes #-}+-- | /NOTE/ It is recommended to start using "Data.Conduit.Combinators" instead+-- of this module. module Data.Conduit.Filesystem- ( sourceDirectory- , sourceDirectoryDeep+ ( CC.sourceDirectory+ , CC.sourceDirectoryDeep ) where -import Data.Conduit-import Control.Monad.Trans.Resource (MonadResource)-import Control.Monad.IO.Class (liftIO)-import System.FilePath ((</>))-import qualified Data.Streaming.Filesystem as F---- | Stream the contents of the given directory, without traversing deeply.------ This function will return /all/ of the contents of the directory, whether--- they be files, directories, etc.------ Note that the generated filepaths will be the complete path, not just the--- filename. In other words, if you have a directory @foo@ containing files--- @bar@ and @baz@, and you use @sourceDirectory@ on @foo@, the results will be--- @foo/bar@ and @foo/baz@.------ Since 1.1.0-sourceDirectory :: MonadResource m => FilePath -> Producer m FilePath-sourceDirectory dir =- bracketP (F.openDirStream dir) F.closeDirStream go- where- go ds =- loop- where- loop = do- mfp <- liftIO $ F.readDirStream ds- case mfp of- Nothing -> return ()- Just fp -> do- yield $ dir </> fp- loop---- | Deeply stream the contents of the given directory.------ This works the same as @sourceDirectory@, but will not return directories at--- all. This function also takes an extra parameter to indicate whether--- symlinks will be followed.------ Since 1.1.0-sourceDirectoryDeep :: MonadResource m- => Bool -- ^ Follow directory symlinks- -> FilePath -- ^ Root directory- -> Producer m FilePath-sourceDirectoryDeep followSymlinks =- start- where- start :: MonadResource m => FilePath -> Producer m FilePath- start dir = sourceDirectory dir =$= awaitForever go-- go :: MonadResource m => FilePath -> Producer m FilePath- go fp = do- ft <- liftIO $ F.getFileType fp- case ft of- F.FTFile -> yield fp- F.FTFileSym -> yield fp- F.FTDirectory -> start fp- F.FTDirectorySym- | followSymlinks -> start fp- | otherwise -> return ()- F.FTOther -> return ()+import qualified Data.Conduit.Combinators as CC
+ Data/Conduit/Foldl.hs view
@@ -0,0 +1,26 @@+{-# LANGUAGE RankNTypes #-}+-- | Adapter module to work with the <https://hackage.haskell.org/package/foldl foldl> package.+--+-- @since 1.1.16+module Data.Conduit.Foldl where++import Data.Conduit+import Control.Monad.Trans.Class (lift)+import qualified Data.Conduit.List as CL++-- | Convert a left fold into a 'Consumer'. This function is intended+-- to be used with @purely@ from the+-- <https://hackage.haskell.org/package/foldl foldl> package.+--+-- @since 1.1.16+sinkFold :: Monad m => (x -> a -> x) -> x -> (x -> b) -> ConduitT a o m b+sinkFold combine seed extract = fmap extract (CL.fold combine seed)++-- | Convert a monadic left fold into a 'Consumer'. This function is+-- intended to be used with @impurely@ from the+-- <https://hackage.haskell.org/package/foldl foldl> package.+--+-- @since 1.1.16+sinkFoldM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> ConduitT a o m b+sinkFoldM combine seed extract =+ lift . extract =<< CL.foldM combine =<< lift seed
Data/Conduit/Lazy.hs view
@@ -1,5 +1,9 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE Trustworthy #-}+{-# LANGUAGE ScopedTypeVariables #-}++{-# OPTIONS_GHC -fno-warn-deprecations #-} -- Suppress warnings around Control.Monad.Trans.Error -- | Use lazy I\/O for consuming the contents of a source. Warning: All normal -- warnings of lazy I\/O apply. In particular, if you are using this with a -- @ResourceT@ transformer, you must force the list to be evaluated before@@ -10,17 +14,18 @@ ) where import Data.Conduit-import Data.Conduit.Internal (Pipe (..), unConduitM)+import Data.Conduit.Internal (Pipe (..), ConduitT (..)) import System.IO.Unsafe (unsafeInterleaveIO) -import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_) import Control.Monad.Trans.Class (lift)-import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.IO.Unlift (MonadIO, liftIO, MonadUnliftIO, withUnliftIO, unliftIO) import Control.Monad.Trans.Identity ( IdentityT)+#if !MIN_VERSION_transformers(0,6,0)+import Control.Monad.Trans.Error ( ErrorT, Error) import Control.Monad.Trans.List ( ListT )+#endif import Control.Monad.Trans.Maybe ( MaybeT )-import Control.Monad.Trans.Error ( ErrorT, Error) import Control.Monad.Trans.Reader ( ReaderT ) import Control.Monad.Trans.State ( StateT ) import Control.Monad.Trans.Writer ( WriterT )@@ -30,7 +35,9 @@ import qualified Control.Monad.Trans.State.Strict as Strict ( StateT ) import qualified Control.Monad.Trans.Writer.Strict as Strict ( WriterT ) +#if (__GLASGOW_HASKELL__ < 710) import Data.Monoid (Monoid)+#endif import Control.Monad.ST (ST) import qualified Control.Monad.ST.Lazy as Lazy import Data.Functor.Identity (Identity)@@ -43,21 +50,26 @@ -- state has been closed. -- -- Since 0.3.0-lazyConsume :: (MonadBaseControl IO m, MonadActive m) => Source m a -> m [a]-lazyConsume =- go . unConduitM- where- go (Done _) = return []- go (HaveOutput src _ x) = do- xs <- liftBaseOp_ unsafeInterleaveIO $ go src- return $ x : xs- go (PipeM msrc) = liftBaseOp_ unsafeInterleaveIO $ do- a <- monadActive- if a- then msrc >>= go- else return []- go (NeedInput _ c) = go (c ())- go (Leftover p _) = go p+lazyConsume+ :: forall m a.+ (MonadUnliftIO m, MonadActive m)+ => Source m a+ -> m [a]+lazyConsume (ConduitT f0) =+ withUnliftIO $ \u ->+ let go :: Pipe () () a () m () -> IO [a]+ go (Done _) = return []+ go (HaveOutput src x) = do+ xs <- unsafeInterleaveIO $ go src+ return $ x : xs+ go (PipeM msrc) = unsafeInterleaveIO $ do+ a <- unliftIO u monadActive+ if a+ then unliftIO u msrc >>= go+ else return []+ go (NeedInput _ c) = go (c ())+ go (Leftover p _) = go p+ in go (f0 Done) -- | Determine if some monad is still active. This is intended to prevent usage -- of a monadic state after it has been closed. This is necessary for such@@ -90,9 +102,11 @@ #define GO(T) instance MonadActive m => MonadActive (T m) where monadActive = lift monadActive #define GOX(X, T) instance (X, MonadActive m) => MonadActive (T m) where monadActive = lift monadActive GO(IdentityT)+#if !MIN_VERSION_transformers(0,6,0)+GOX(Error e, ErrorT e) GO(ListT)+#endif GO(MaybeT)-GOX(Error e, ErrorT e) GO(ReaderT r) GO(StateT s) GOX(Monoid w, WriterT w)@@ -105,5 +119,5 @@ instance MonadActive m => MonadActive (Pipe l i o u m) where monadActive = lift monadActive-instance MonadActive m => MonadActive (ConduitM i o m) where+instance MonadActive m => MonadActive (ConduitT i o m) where monadActive = lift monadActive
Data/Conduit/Network.hs view
@@ -17,6 +17,7 @@ , serverSettings , SN.runTCPServer , SN.runTCPServerWithHandle+ , forkTCPServer , runGeneralTCPServer -- ** Client , SN.ClientSettings@@ -37,29 +38,26 @@ , SN.HostPreference ) where -import Prelude hiding (catch)+import Prelude import Data.Conduit-import qualified Network.Socket as NS import Network.Socket (Socket)-import Network.Socket.ByteString (sendAll, recv)+import Network.Socket.ByteString (sendAll) import Data.ByteString (ByteString) import qualified GHC.Conc as Conc (yield) import qualified Data.ByteString as S-import qualified Data.ByteString.Char8 as S8 import Control.Monad.IO.Class (MonadIO (liftIO))-import Control.Exception (throwIO, SomeException, try, finally, bracket, IOException, catch)-import Control.Monad (forever, unless, void)-import Control.Monad.Trans.Control (MonadBaseControl, control, liftBaseWith)+import Control.Monad (unless) import Control.Monad.Trans.Class (lift)-import Control.Concurrent (forkIO, threadDelay, newEmptyMVar, putMVar, takeMVar)+import Control.Concurrent (forkIO, newEmptyMVar, putMVar, takeMVar, MVar, ThreadId) import qualified Data.Streaming.Network as SN+import Control.Monad.IO.Unlift (MonadUnliftIO, withRunInIO) -- | Stream data from the socket. -- -- This function does /not/ automatically close the socket. -- -- Since 0.0.0-sourceSocket :: MonadIO m => Socket -> Producer m ByteString+sourceSocket :: MonadIO m => Socket -> ConduitT i ByteString m () sourceSocket socket = loop where@@ -74,16 +72,19 @@ -- This function does /not/ automatically close the socket. -- -- Since 0.0.0-sinkSocket :: MonadIO m => Socket -> Consumer ByteString m ()+sinkSocket :: MonadIO m => Socket -> ConduitT ByteString o m () sinkSocket socket = loop where loop = await >>= maybe (return ()) (\bs -> lift (liftIO $ sendAll socket bs) >> loop) +serverSettings :: Int -> SN.HostPreference -> SN.ServerSettings serverSettings = SN.serverSettingsTCP++clientSettings :: Int -> ByteString -> SN.ClientSettings clientSettings = SN.clientSettingsTCP -appSource :: (SN.HasReadWrite ad, MonadIO m) => ad -> Producer m ByteString+appSource :: (SN.HasReadWrite ad, MonadIO m) => ad -> ConduitT i ByteString m () appSource ad = loop where@@ -94,13 +95,42 @@ yield bs loop -appSink :: (SN.HasReadWrite ad, MonadIO m) => ad -> Consumer ByteString m ()+appSink :: (SN.HasReadWrite ad, MonadIO m) => ad -> ConduitT ByteString o m () appSink ad = awaitForever $ \d -> liftIO $ SN.appWrite ad d >> Conc.yield +addBoundSignal::MVar ()-> SN.ServerSettings -> SN.ServerSettings+addBoundSignal isBound set = SN.setAfterBind ( \socket -> originalAfterBind socket >> signalBound socket) set+ where originalAfterBind :: Socket -> IO ()+ originalAfterBind = SN.getAfterBind set+ signalBound :: Socket -> IO ()+ signalBound _socket = putMVar isBound ()++-- | Fork a TCP Server+--+-- Will fork the runGeneralTCPServer function but will only return from+-- this call when the server is bound to the port and accepting incoming+-- connections. Will return the thread id of the server+--+-- Since 1.1.4+forkTCPServer+ :: MonadUnliftIO m+ => SN.ServerSettings+ -> (SN.AppData -> m ())+ -> m ThreadId+forkTCPServer set f =+ withRunInIO $ \run -> do+ isBound <- newEmptyMVar+ let setWithWaitForBind = addBoundSignal isBound set+ threadId <- forkIO . run $ runGeneralTCPServer setWithWaitForBind f+ takeMVar isBound+ return threadId+++ -- | Run a general TCP server -- -- Same as 'SN.runTCPServer', except monad can be any instance of--- 'MonadBaseControl' 'IO'.+-- 'MonadUnliftIO'. -- -- Note that any changes to the monadic state performed by individual -- client handlers will be discarded. If you have mutable state you want@@ -108,21 +138,23 @@ -- variables. -- -- Since 1.1.3-runGeneralTCPServer :: MonadBaseControl IO m- => SN.ServerSettings- -> (SN.AppData -> m ())- -> m ()-runGeneralTCPServer set f = liftBaseWith $ \run ->- SN.runTCPServer set $ void . run . f+runGeneralTCPServer+ :: MonadUnliftIO m+ => SN.ServerSettings+ -> (SN.AppData -> m ())+ -> m a+runGeneralTCPServer set f = withRunInIO $ \run ->+ SN.runTCPServer set $ run . f -- | Run a general TCP client ----- Same as 'SN.runTCPClient', except monad can be any instance of 'MonadBaseControl' 'IO'.+-- Same as 'SN.runTCPClient', except monad can be any instance of 'MonadUnliftIO'. -- -- Since 1.1.3-runGeneralTCPClient :: MonadBaseControl IO m- => SN.ClientSettings- -> (SN.AppData -> m a)- -> m a-runGeneralTCPClient set f = control $ \run ->+runGeneralTCPClient+ :: MonadUnliftIO m+ => SN.ClientSettings+ -> (SN.AppData -> m a)+ -> m a+runGeneralTCPClient set f = withRunInIO $ \run -> SN.runTCPClient set $ run . f
Data/Conduit/Network/UDP.hs view
@@ -13,8 +13,7 @@ ) where import Data.Conduit-import Network.Socket (AddrInfo, SockAddr, Socket)-import qualified Network.Socket as NS+import Network.Socket (Socket) import Network.Socket.ByteString (recvFrom, send, sendAll, sendTo, sendAllTo) import Data.ByteString (ByteString) import Control.Monad.IO.Class (MonadIO (liftIO))@@ -28,7 +27,7 @@ -- contains the message payload and the origin address. -- -- This function does /not/ automatically close the socket.-sourceSocket :: MonadIO m => Socket -> Int -> Producer m SN.Message+sourceSocket :: MonadIO m => Socket -> Int -> ConduitT i SN.Message m () sourceSocket socket len = loop where loop = do@@ -40,7 +39,7 @@ -- The payload is sent using @send@, so some of it might be lost. -- -- This function does /not/ automatically close the socket.-sinkSocket :: MonadIO m => Socket -> Consumer ByteString m ()+sinkSocket :: MonadIO m => Socket -> ConduitT ByteString o m () sinkSocket = sinkSocketHelper (\sock bs -> void $ send sock bs) -- | Stream messages to the connected socket.@@ -48,7 +47,7 @@ -- The payload is sent using @sendAll@, so it might end up in multiple packets. -- -- This function does /not/ automatically close the socket.-sinkAllSocket :: MonadIO m => Socket -> Consumer ByteString m ()+sinkAllSocket :: MonadIO m => Socket -> ConduitT ByteString o m () sinkAllSocket = sinkSocketHelper sendAll -- | Stream messages to the socket.@@ -58,7 +57,7 @@ -- lost. -- -- This function does /not/ automatically close the socket.-sinkToSocket :: MonadIO m => Socket -> Consumer SN.Message m ()+sinkToSocket :: MonadIO m => Socket -> ConduitT SN.Message o m () sinkToSocket = sinkSocketHelper (\sock (SN.Message bs addr) -> void $ sendTo sock bs addr) -- | Stream messages to the socket.@@ -68,13 +67,13 @@ -- multiple packets. -- -- This function does /not/ automatically close the socket.-sinkAllToSocket :: MonadIO m => Socket -> Consumer SN.Message m ()+sinkAllToSocket :: MonadIO m => Socket -> ConduitT SN.Message o m () sinkAllToSocket = sinkSocketHelper (\sock (SN.Message bs addr) -> sendAllTo sock bs addr) -- Internal sinkSocketHelper :: MonadIO m => (Socket -> a -> IO ()) -> Socket- -> Consumer a m ()+ -> ConduitT a o m () sinkSocketHelper act socket = loop where loop = await >>= maybe
Data/Conduit/Network/Unix.hs view
@@ -26,20 +26,11 @@ , SN.setAfterBind ) where -import Data.Conduit-import Network.Socket (Socket)-import qualified Network.Socket as NS import Data.Conduit.Network (appSource, appSink, sourceSocket, sinkSocket) import qualified Data.Streaming.Network as SN-import Control.Monad.IO.Class (MonadIO (liftIO))-import Control.Exception (throwIO, SomeException, try, finally, bracket,- bracketOnError, catch)-import Control.Monad (forever, void)-import Control.Monad.Trans.Control (control)-import Control.Concurrent (forkIO)-import System.Directory (removeFile)-import System.IO.Error (isDoesNotExistError)-import Control.Monad.Trans.Resource (MonadBaseControl) +clientSettings :: FilePath -> SN.ClientSettingsUnix clientSettings = SN.clientSettingsUnix++serverSettings :: FilePath -> SN.ServerSettingsUnix serverSettings = SN.serverSettingsUnix
Data/Conduit/Process.hs view
@@ -1,10 +1,14 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE CPP #-} {-# OPTIONS_GHC -fno-warn-orphans #-}--- | A full tutorial for this module is available on FP School of Haskell:--- <https://www.fpcomplete.com/user/snoyberg/library-documentation/data-conduit-process>.+-- | A full tutorial for this module is available at:+-- <https://github.com/snoyberg/conduit/blob/master/PROCESS.md>. --+-- Some utilities in this module require the threaded runtime because they use+-- 'System.Process.waitForProcess' internally.+-- -- Note that this is a very thin layer around the @Data.Streaming.Process@ module. In particular, it: -- -- * Provides orphan instances for conduit@@ -14,6 +18,12 @@ ( -- * Functions sourceCmdWithConsumer , sourceProcessWithConsumer+ , sourceCmdWithStreams+ , sourceProcessWithStreams+ , withCheckedProcessCleanup+ -- * InputSource types+ , FlushInput(..)+ , BuilderInput(..) -- * Reexport , module Data.Streaming.Process ) where@@ -21,32 +31,71 @@ import Data.Streaming.Process import Data.Streaming.Process.Internal import System.Exit (ExitCode (..))-import Control.Monad.IO.Class (MonadIO, liftIO)-import System.IO (hClose)+import Control.Monad.IO.Unlift (MonadIO, liftIO, MonadUnliftIO, withRunInIO, withUnliftIO, unliftIO)+import System.IO (hClose, BufferMode (NoBuffering), hSetBuffering) import Data.Conduit-import Data.Conduit.Binary (sourceHandle, sinkHandle)+import Data.Functor (($>))+import Data.Conduit.Binary (sourceHandle, sinkHandle, sinkHandleBuilder, sinkHandleFlush) import Data.ByteString (ByteString)+import Data.ByteString.Builder (Builder)+import Control.Concurrent.Async (runConcurrently, Concurrently(..))+import Control.Exception (onException, throwIO, finally, bracket, catch)+import System.IO.Error (ioeGetErrorType, isResourceVanishedErrorType)+#if (__GLASGOW_HASKELL__ < 710)+import Control.Applicative ((<$>), (<*>))+#endif instance (r ~ (), MonadIO m, i ~ ByteString) => InputSource (ConduitM i o m r) where- isStdStream = (\(Just h) -> return $ sinkHandle h, Just CreatePipe)+ isStdStream = (\(Just h) -> hSetBuffering h NoBuffering $> sinkHandle h, Just CreatePipe) instance (r ~ (), r' ~ (), MonadIO m, MonadIO n, i ~ ByteString) => InputSource (ConduitM i o m r, n r') where- isStdStream = (\(Just h) -> return (sinkHandle h, liftIO $ hClose h), Just CreatePipe)+ isStdStream = (\(Just h) -> hSetBuffering h NoBuffering $> (sinkHandle h, liftIO $ hClose h), Just CreatePipe) +-- | Wrapper for input source which accepts 'Data.ByteString.Builder.Builder's.+-- You can pass 'Data.ByteString.Builder.Extra.flush' to flush the input. Note+-- that the pipe will /not/ automatically close when the processing completes.+--+-- @since 1.3.2+newtype BuilderInput o m r = BuilderInput (ConduitM Builder o m r)++-- | Wrapper for input source which accepts @Flush@es. Note that the pipe+-- will /not/ automatically close then processing completes.+--+-- @since 1.3.2+newtype FlushInput o m r = FlushInput (ConduitM (Flush ByteString) o m r)++instance (MonadIO m, r ~ ()) => InputSource (BuilderInput o m r) where+ isStdStream = (\(Just h) -> return $ BuilderInput $ sinkHandleBuilder h, Just CreatePipe)+instance (MonadIO m, MonadIO n, r ~ (), r' ~ ()) => InputSource (BuilderInput o m r, n r') where+ isStdStream = (\(Just h) -> return (BuilderInput $ sinkHandleBuilder h, liftIO $ hClose h), Just CreatePipe)+instance (MonadIO m, r ~ ()) => InputSource (FlushInput o m r) where+ isStdStream = (\(Just h) -> return $ FlushInput $ sinkHandleFlush h, Just CreatePipe)+instance (MonadIO m, MonadIO n, r ~ (), r' ~ ()) => InputSource (FlushInput o m r, n r') where+ isStdStream = (\(Just h) -> return (FlushInput $ sinkHandleFlush h, liftIO $ hClose h), Just CreatePipe)+ instance (r ~ (), MonadIO m, o ~ ByteString) => OutputSink (ConduitM i o m r) where- osStdStream = (\(Just h) -> return $ sourceHandle h, Just CreatePipe)+ osStdStream = (\(Just h) -> hSetBuffering h NoBuffering $> sourceHandle h, Just CreatePipe) instance (r ~ (), r' ~ (), MonadIO m, MonadIO n, o ~ ByteString) => OutputSink (ConduitM i o m r, n r') where- osStdStream = (\(Just h) -> return (sourceHandle h, liftIO $ hClose h), Just CreatePipe)+ osStdStream = (\(Just h) -> hSetBuffering h NoBuffering $> (sourceHandle h, liftIO $ hClose h), Just CreatePipe) -- | Given a @CreateProcess@, run the process, with its output being used as a -- @Source@ to feed the provided @Consumer@. Once the process has completed, -- return a tuple of the @ExitCode@ from the process and the output collected -- from the @Consumer@. --+-- Note that, if an exception is raised by the consumer, the process is /not/+-- terminated. This behavior is different from 'sourceProcessWithStreams' due+-- to historical reasons.+--+-- Requires the threaded runtime.+-- -- Since 1.1.2-sourceProcessWithConsumer :: MonadIO m => CreateProcess -> Consumer ByteString m a -> m (ExitCode, a)+sourceProcessWithConsumer :: MonadIO m+ => CreateProcess+ -> ConduitT ByteString Void m a -- ^ stdout+ -> m (ExitCode, a) sourceProcessWithConsumer cp consumer = do (ClosedStream, (source, close), ClosedStream, cph) <- streamingProcess cp- res <- source $$ consumer+ res <- runConduit $ source .| consumer close ec <- waitForStreamingProcess cph return (ec, res)@@ -54,6 +103,106 @@ -- | Like @sourceProcessWithConsumer@ but providing the command to be run as -- a @String@. --+-- Requires the threaded runtime.+-- -- Since 1.1.2-sourceCmdWithConsumer :: MonadIO m => String -> Consumer ByteString m a -> m (ExitCode, a)+sourceCmdWithConsumer :: MonadIO m+ => String -- ^command+ -> ConduitT ByteString Void m a -- ^stdout+ -> m (ExitCode, a) sourceCmdWithConsumer cmd = sourceProcessWithConsumer (shell cmd)+++-- | Given a @CreateProcess@, run the process+-- and feed the provided @Producer@+-- to the stdin @Sink@ of the process.+-- Use the process outputs (stdout, stderr) as @Source@s+-- and feed it to the provided @Consumer@s.+-- Once the process has completed,+-- return a tuple of the @ExitCode@ from the process+-- and the results collected from the @Consumer@s.+--+-- If an exception is raised by any of the streams,+-- the process is terminated.+--+-- IO is required because the streams are run concurrently+-- using the <https://hackage.haskell.org/package/async async> package+--+-- Requires the threaded runtime.+--+-- @since 1.1.12+sourceProcessWithStreams+ :: MonadUnliftIO m+ => CreateProcess+ -> ConduitT () ByteString m () -- ^stdin+ -> ConduitT ByteString Void m a -- ^stdout+ -> ConduitT ByteString Void m b -- ^stderr+ -> m (ExitCode, a, b)+sourceProcessWithStreams cp producerStdin consumerStdout consumerStderr =+ withUnliftIO $ \u -> do+ ( (sinkStdin, closeStdin)+ , (sourceStdout, closeStdout)+ , (sourceStderr, closeStderr)+ , sph) <- streamingProcess cp+ let safeSinkStdin = sinkStdin `catchC` ignoreStdinClosed+ safeCloseStdin = closeStdin `catch` ignoreStdinClosed+ (_, resStdout, resStderr) <-+ runConcurrently (+ (,,)+ <$> Concurrently ((unliftIO u $ runConduit $ producerStdin .| safeSinkStdin) `finally` safeCloseStdin)+ <*> Concurrently (unliftIO u $ runConduit $ sourceStdout .| consumerStdout)+ <*> Concurrently (unliftIO u $ runConduit $ sourceStderr .| consumerStderr))+ `finally` (closeStdout >> closeStderr)+ `onException` terminateStreamingProcess sph+ ec <- waitForStreamingProcess sph+ return (ec, resStdout, resStderr)+ where+ ignoreStdinClosed :: forall m. (MonadIO m) => IOError -> m ()+ ignoreStdinClosed e =+ if isResourceVanishedErrorType (ioeGetErrorType e)+ then pure ()+ else liftIO (throwIO e)++-- | Like @sourceProcessWithStreams@ but providing the command to be run as+-- a @String@.+--+-- Requires the threaded runtime.+--+-- @since 1.1.12+sourceCmdWithStreams+ :: MonadUnliftIO m+ => String -- ^command+ -> ConduitT () ByteString m () -- ^stdin+ -> ConduitT ByteString Void m a -- ^stdout+ -> ConduitT ByteString Void m b -- ^stderr+ -> m (ExitCode, a, b)+sourceCmdWithStreams cmd = sourceProcessWithStreams (shell cmd)++-- | Same as 'withCheckedProcess', but kills the child process in the case of+-- an exception being thrown by the provided callback function.+--+-- Requires the threaded runtime.+--+-- @since 1.1.11+withCheckedProcessCleanup+ :: ( InputSource stdin+ , OutputSink stderr+ , OutputSink stdout+ , MonadUnliftIO m+ )+ => CreateProcess+ -> (stdin -> stdout -> stderr -> m b)+ -> m b+withCheckedProcessCleanup cp f = withRunInIO $ \run -> bracket+ (streamingProcess cp)+ (\(_, _, _, sph) -> closeStreamingProcessHandle sph)+ $ \(x, y, z, sph) -> do+ res <- run (f x y z) `onException` terminateStreamingProcess sph+ ec <- waitForStreamingProcess sph+ if ec == ExitSuccess+ then return res+ else throwIO $ ProcessExitedUnsuccessfully cp ec+++terminateStreamingProcess :: MonadIO m => StreamingProcessHandle -> m ()+terminateStreamingProcess = liftIO . terminateProcess . streamingProcessHandleRaw
+ Data/Conduit/Process/Typed.hs view
@@ -0,0 +1,107 @@+{-# LANGUAGE DataKinds #-}++-- | The "System.Process.Typed" module from @typed-process@, but with+-- added conduit helpers.+module Data.Conduit.Process.Typed+ ( -- * Conduit specific stuff+ createSink+ , createSinkClose+ , createSource+ -- * Running a process with logging+ , withLoggedProcess_+ -- * Reexports+ , module System.Process.Typed+ ) where++import System.Process.Typed+import qualified System.Process.Typed as P+import Data.Conduit (ConduitM, (.|), runConduit)+import qualified Data.Conduit.Binary as CB+import Control.Monad.IO.Unlift+import qualified Data.ByteString as S+import qualified Data.Conduit.List as CL+import qualified Data.ByteString.Lazy as BL+import Data.IORef (IORef, newIORef, readIORef, modifyIORef)+import Control.Exception (throwIO, catch)+import Control.Concurrent.Async (concurrently)+import System.IO (hSetBuffering, BufferMode (NoBuffering), hClose)++-- | Provide input to a process by writing to a conduit. The sink provided here+-- will leave the pipe to the child open after the stream ends. This allows the+-- sink to be used multiple times, but may result in surprising behavior. You+-- may prefer 'createSinkClose', see+-- <https://github.com/snoyberg/conduit/issues/434>.+--+-- @since 1.2.1+createSink :: MonadIO m => StreamSpec 'STInput (ConduitM S.ByteString o m ())+createSink =+ (\h -> liftIO (hSetBuffering h NoBuffering) >> CB.sinkHandle h)+ `fmap` createPipe++-- | Like 'createSink', but closes the pipe to the child process as soon as it+-- runs out of data.+--+-- @since 1.3.5+createSinkClose :: MonadIO m => StreamSpec 'STInput (ConduitM S.ByteString o m ())+createSinkClose =+ (\h -> liftIO (hSetBuffering h NoBuffering) >> CB.sinkHandle h >> liftIO (hClose h))+ `fmap` createPipe++-- | Read output from a process by read from a conduit.+--+-- @since 1.2.1+createSource :: MonadIO m => StreamSpec 'STOutput (ConduitM i S.ByteString m ())+createSource =+ (\h -> liftIO (hSetBuffering h NoBuffering) >> CB.sourceHandle h)+ `fmap` createPipe++-- | Internal function: like 'createSource', but stick all chunks into+-- the 'IORef'.+createSourceLogged+ :: MonadIO m+ => IORef ([S.ByteString] -> [S.ByteString])+ -> StreamSpec 'STOutput (ConduitM i S.ByteString m ())+createSourceLogged ref =+ -- We do not add a cleanup action to close the handle, since in+ -- withLoggedProcess_ we attempt to read from the handle twice+ (\h ->+ ( CB.sourceHandle h+ .| CL.iterM (\bs -> liftIO $ modifyIORef ref (. (bs:))))+ )+ `fmap` createPipe++-- | Run a process, throwing an exception on a failure exit code. This+-- will store all output from stdout and stderr in memory for better+-- error messages. Note that this will require unbounded memory usage,+-- so caveat emptor.+--+-- This will ignore any previous settings for the stdout and stderr+-- streams, and instead force them to use 'createSource'.+--+-- @since 1.2.3+withLoggedProcess_+ :: MonadUnliftIO m+ => ProcessConfig stdin stdoutIgnored stderrIgnored+ -> (Process stdin (ConduitM () S.ByteString m ()) (ConduitM () S.ByteString m ()) -> m a)+ -> m a+withLoggedProcess_ pc inner = withUnliftIO $ \u -> do+ stdoutBuffer <- newIORef id+ stderrBuffer <- newIORef id+ let pc' = setStdout (createSourceLogged stdoutBuffer)+ $ setStderr (createSourceLogged stderrBuffer) pc+ -- withProcessWait vs Term doesn't actually matter here, since we+ -- call checkExitCode inside regardless. But still, Wait is the+ -- safer function to use in general.+ P.withProcessWait pc' $ \p -> do+ a <- unliftIO u $ inner p+ let drain src = unliftIO u (runConduit (src .| CL.sinkNull))+ ((), ()) <- drain (getStdout p) `concurrently`+ drain (getStderr p)+ checkExitCode p `catch` \ece -> do+ stdout <- readIORef stdoutBuffer+ stderr <- readIORef stderrBuffer+ throwIO ece+ { eceStdout = BL.fromChunks $ stdout []+ , eceStderr = BL.fromChunks $ stderr []+ }+ return a
Data/Conduit/Text.hs view
@@ -1,11 +1,16 @@ {-# LANGUAGE DeriveDataTypeable, RankNTypes #-}--- |+-- | /NOTE/ It is recommended to start using "Data.Conduit.Combinators" instead+-- of this module.+-- -- Copyright: 2011 Michael Snoyman, 2010-2011 John Millikin -- License: MIT -- -- Handle streams of text. -- -- Parts of this code were taken from enumerator and adapted for conduits.+--+-- For many purposes, it's recommended to use the conduit-combinators library,+-- which provides a more complete set of functions. module Data.Conduit.Text ( @@ -29,12 +34,12 @@ , drop , foldLines , withLine- , Data.Conduit.Text.decodeUtf8- , decodeUtf8Lenient- , encodeUtf8+ , CC.decodeUtf8+ , CC.decodeUtf8Lenient+ , CC.encodeUtf8+ , detectUtf ) where -import qualified Prelude import Prelude hiding (head, drop, takeWhile, lines, zip, zip3, zipWith, zipWith3, take, dropWhile) import qualified Control.Exception as Exc@@ -48,8 +53,9 @@ import Data.Conduit import qualified Data.Conduit.List as CL+import qualified Data.Conduit.Combinators as CC import Control.Monad.Trans.Class (lift)-import Control.Monad.Trans.Resource (MonadThrow, monadThrow)+import Control.Monad.Trans.Resource (MonadThrow, throwM) import Control.Monad (unless) import Data.Streaming.Text @@ -57,7 +63,7 @@ -- -- Since 0.3.0 data Codec = Codec- { codecName :: T.Text+ { _codecName :: T.Text , codecEncode :: T.Text -> (B.ByteString, Maybe (TextException, T.Text))@@ -70,43 +76,47 @@ | NewCodec T.Text (T.Text -> B.ByteString) (B.ByteString -> DecodeResult) instance Show Codec where- showsPrec d c = showParen (d > 10) $- showString "Codec " . shows (codecName c)-+ showsPrec d c =+ let (cnst, name) = case c of+ Codec t _ _ -> ("Codec ", t)+ NewCodec t _ _ -> ("NewCodec ", t)+ in showParen (d > 10) $ showString cnst . shows name -- | Emit each line separately -- -- Since 0.4.1-lines :: Monad m => Conduit T.Text m T.Text+lines :: Monad m => ConduitT T.Text T.Text m () lines =- awaitText T.empty+ awaitText id where- awaitText buf = await >>= maybe (finish buf) (process buf)-- finish buf = unless (T.null buf) (yield buf)+ awaitText front = await >>= maybe (finish front) (process front) - process buf text = yieldLines $ buf `T.append` text+ finish front =+ let t = T.concat $ front []+ in unless (T.null t) (yield t) - yieldLines buf =- let (line, rest) = T.break (== '\n') buf+ process front text =+ let (line, rest) = T.break (== '\n') text in case T.uncons rest of- Just (_, rest') -> yield line >> yieldLines rest'- _ -> awaitText line+ Just (_, rest') -> do+ yield (T.concat $ front [line])+ process id rest'+ Nothing -> Exc.assert (line == text) $ awaitText $ front . (line:) -- | Variant of the lines function with an integer parameter. -- The text length of any emitted line--- never exceeds the value of the paramater. Whenever+-- never exceeds the value of the parameter. Whenever -- this is about to happen a LengthExceeded exception -- is thrown. This function should be used instead -- of the lines function whenever we are dealing with -- user input (e.g. a file upload) because we can't be sure that -- user input won't have extraordinarily large lines which would -- require large amounts of memory if consumed.-linesBounded :: MonadThrow m => Int -> Conduit T.Text m T.Text+linesBounded :: MonadThrow m => Int -> ConduitT T.Text T.Text m () linesBounded maxLineLen = awaitText 0 T.empty where@@ -118,7 +128,7 @@ let (line, rest) = T.break (== '\n') text len' = len + T.length line in if len' > maxLineLen- then lift $ monadThrow (LengthExceeded maxLineLen)+ then lift $ throwM (LengthExceeded maxLineLen) else case T.uncons rest of Just (_, rest') -> yield (buf `T.append` line) >> process 0 T.empty rest'@@ -131,20 +141,20 @@ -- not capable of representing an input character, an exception will be thrown. -- -- Since 0.3.0-encode :: MonadThrow m => Codec -> Conduit T.Text m B.ByteString+encode :: MonadThrow m => Codec -> ConduitT T.Text B.ByteString m () encode (NewCodec _ enc _) = CL.map enc encode codec = CL.mapM $ \t -> do let (bs, mexc) = codecEncode codec t- maybe (return bs) (monadThrow . fst) mexc+ maybe (return bs) (throwM . fst) mexc decodeNew :: Monad m- => (Int -> B.ByteString -> T.Text -> B.ByteString -> Conduit B.ByteString m T.Text)+ => (Int -> B.ByteString -> T.Text -> B.ByteString -> ConduitT B.ByteString T.Text m ()) -> t -> Int -> (B.ByteString -> DecodeResult)- -> Conduit B.ByteString m T.Text-decodeNew onFailure name =+ -> ConduitT B.ByteString T.Text m ()+decodeNew onFailure _name = loop where loop consumed dec =@@ -167,28 +177,11 @@ in consumed' `seq` next DecodeResultFailure t rest -> onFailure consumed bs t rest --- | Decode a stream of UTF8 data, and replace invalid bytes with the Unicode--- replacement character.------ Since 1.1.1-decodeUtf8Lenient :: Monad m => Conduit B.ByteString m T.Text-decodeUtf8Lenient =- decodeNew onFailure "UTF8-lenient" 0 Data.Streaming.Text.decodeUtf8- where- onFailure _consumed _bs t rest = do- unless (T.null t) (yield t)- case B.uncons rest of- Nothing -> return ()- Just (_, rest') -> do- unless (B.null rest') (leftover rest')- yield $ T.singleton '\xFFFD'- decodeUtf8Lenient- -- | Convert bytes into text, using the provided codec. If the codec is -- not capable of decoding an input byte sequence, an exception will be thrown. -- -- Since 0.3.0-decode :: MonadThrow m => Codec -> Conduit B.ByteString m T.Text+decode :: MonadThrow m => Codec -> ConduitT B.ByteString T.Text m () decode (NewCodec name _ start) = decodeNew onFailure name 0 start where@@ -196,7 +189,7 @@ unless (T.null t) (yield t) leftover rest -- rest will never be null, no need to check let consumed' = consumed + B.length bs - B.length rest- monadThrow $ NewDecodeException name consumed' (B.take 4 rest)+ throwM $ NewDecodeException name consumed' (B.take 4 rest) {-# INLINE onFailure #-} decode codec = loop id@@ -206,11 +199,11 @@ finish front = case B.uncons $ front B.empty of Nothing -> return ()- Just (w, _) -> lift $ monadThrow $ DecodeException codec w+ Just (w, _) -> lift $ throwM $ DecodeException codec w go front bs' = case extra of- Left (exc, _) -> lift $ monadThrow exc+ Left (exc, _) -> lift $ throwM exc Right bs'' -> yield text >> loop (B.append bs'') where (text, extra) = codecDecode codec bs@@ -312,7 +305,7 @@ -- Since 1.0.8 takeWhile :: Monad m => (Char -> Bool)- -> Conduit T.Text m T.Text+ -> ConduitT T.Text T.Text m () takeWhile p = loop where@@ -328,7 +321,7 @@ -- Since 1.0.8 dropWhile :: Monad m => (Char -> Bool)- -> Consumer T.Text m ()+ -> ConduitT T.Text o m () dropWhile p = loop where@@ -342,7 +335,7 @@ -- | -- -- Since 1.0.8-take :: Monad m => Int -> Conduit T.Text m T.Text+take :: Monad m => Int -> ConduitT T.Text T.Text m () take = loop where@@ -359,7 +352,7 @@ -- | -- -- Since 1.0.8-drop :: Monad m => Int -> Consumer T.Text m ()+drop :: Monad m => Int -> ConduitT T.Text o m () drop = loop where@@ -377,15 +370,15 @@ foldLines :: Monad m => (a -> ConduitM T.Text o m a) -> a- -> ConduitM T.Text o m a+ -> ConduitT T.Text o m a foldLines f = start where start a = CL.peek >>= maybe (return a) (const $ loop $ f a) loop consumer = do- a <- takeWhile (/= '\n') =$= do- a <- CL.map (T.filter (/= '\r')) =$= consumer+ a <- takeWhile (/= '\n') .| do+ a <- CL.map (T.filter (/= '\r')) .| consumer CL.sinkNull return a drop 1@@ -395,62 +388,47 @@ -- -- Since 1.0.8 withLine :: Monad m- => Sink T.Text m a- -> Consumer T.Text m (Maybe a)+ => ConduitT T.Text Void m a+ -> ConduitT T.Text o m (Maybe a) withLine consumer = toConsumer $ do mx <- CL.peek case mx of Nothing -> return Nothing Just _ -> do- x <- takeWhile (/= '\n') =$ do- x <- CL.map (T.filter (/= '\r')) =$ consumer+ x <- takeWhile (/= '\n') .| do+ x <- CL.map (T.filter (/= '\r')) .| consumer CL.sinkNull return x drop 1 return $ Just x --- | Decode a stream of UTF8-encoded bytes into a stream of text, throwing an--- exception on invalid input.+-- | Automatically determine which UTF variant is being used. This function+-- checks for BOMs, removing them as necessary. It defaults to assuming UTF-8. ----- Since 1.0.15-decodeUtf8 :: MonadThrow m => Conduit B.ByteString m T.Text-decodeUtf8 = decode utf8- {- no meaningful performance advantage- CI.ConduitM (loop 0 decodeUtf8)+-- Since 1.1.9+detectUtf :: MonadThrow m => ConduitT B.ByteString T.Text m ()+detectUtf =+ go id where- loop consumed dec =- CI.NeedInput go finish- where- finish () =- case dec B.empty of- DecodeResultSuccess _ _ -> return ()- DecodeResultFailure t rest -> onFailure B.empty t rest- {-# INLINE finish #-}+ go front = await >>= maybe (close front) (push front) - go bs | B.null bs = CI.NeedInput go finish- go bs =- case dec bs of- DecodeResultSuccess t dec' -> do- let consumed' = consumed + B.length bs- next' = loop consumed' dec'- next- | T.null t = next'- | otherwise = CI.HaveOutput next' (return ()) t- in consumed' `seq` next- DecodeResultFailure t rest -> onFailure bs t rest+ push front bs'+ | B.length bs < 4 = go $ B.append bs+ | otherwise = leftDecode bs+ where bs = front bs' - onFailure bs t rest = do- unless (T.null t) (CI.yield t)- unless (B.null rest) (CI.leftover rest)- let consumed' = consumed + B.length bs - B.length rest- monadThrow $ NewDecodeException (T.pack "UTF-8") consumed' (B.take 4 rest)- {-# INLINE onFailure #-}- -}-{-# INLINE decodeUtf8 #-}+ close front = leftDecode $ front B.empty --- | Encode a stream of text into a stream of bytes.------ Since 1.0.15-encodeUtf8 :: Monad m => Conduit T.Text m B.ByteString-encodeUtf8 = CL.map TE.encodeUtf8-{-# INLINE encodeUtf8 #-}+ leftDecode bs = leftover bsOut >> decode codec+ where+ bsOut = B.append (B.drop toDrop x) y+ (x, y) = B.splitAt 4 bs+ (toDrop, codec) =+ case B.unpack x of+ [0x00, 0x00, 0xFE, 0xFF] -> (4, utf32_be)+ [0xFF, 0xFE, 0x00, 0x00] -> (4, utf32_le)+ 0xFE : 0xFF: _ -> (2, utf16_be)+ 0xFF : 0xFE: _ -> (2, utf16_le)+ 0xEF : 0xBB: 0xBF : _ -> (3, utf8)+ _ -> (0, utf8) -- Assuming UTF-8+{-# INLINE detectUtf #-}
Data/Conduit/Zlib.hs view
@@ -8,6 +8,8 @@ compress, decompress, gzip, ungzip, -- * Flushing compressFlush, decompressFlush,+ -- * Decompression combinators+ multiple, -- * Re-exported from zlib-bindings WindowBits (..), defaultWindowBits ) where@@ -19,19 +21,19 @@ import Control.Monad (unless, liftM) import Control.Monad.Trans.Class (lift, MonadTrans) import Control.Monad.Primitive (PrimMonad, unsafePrimToPrim)-import Control.Monad.Base (MonadBase, liftBase)-import Control.Monad.Trans.Resource (MonadThrow, monadThrow)+import Control.Monad.Trans.Resource (MonadThrow, throwM)+import Data.Function (fix) -- | Gzip compression with default parameters.-gzip :: (MonadThrow m, MonadBase base m, PrimMonad base) => Conduit ByteString m ByteString-gzip = compress 1 (WindowBits 31)+gzip :: (MonadThrow m, PrimMonad m) => ConduitT ByteString ByteString m ()+gzip = compress (-1) (WindowBits 31) -- | Gzip decompression with default parameters.-ungzip :: (MonadBase base m, PrimMonad base, MonadThrow m) => Conduit ByteString m ByteString+ungzip :: (PrimMonad m, MonadThrow m) => ConduitT ByteString ByteString m () ungzip = decompress (WindowBits 31) -unsafeLiftIO :: (MonadBase base m, PrimMonad base, MonadThrow m) => IO a -> m a-unsafeLiftIO = liftBase . unsafePrimToPrim+unsafeLiftIO :: (PrimMonad m, MonadThrow m) => IO a -> m a+unsafeLiftIO = unsafePrimToPrim -- | -- Decompress (inflate) a stream of 'ByteString's. For example:@@ -39,67 +41,110 @@ -- > sourceFile "test.z" $= decompress defaultWindowBits $$ sinkFile "test" decompress- :: (MonadBase base m, PrimMonad base, MonadThrow m)+ :: (PrimMonad m, MonadThrow m) => WindowBits -- ^ Zlib parameter (see the zlib-bindings package as well as the zlib C library)- -> Conduit ByteString m ByteString+ -> ConduitT ByteString ByteString m () decompress =- helperDecompress (liftM (fmap Chunk) await) yield'+ helperDecompress (liftM (fmap Chunk) await) yield' leftover where yield' Flush = return () yield' (Chunk bs) = yield bs -- | Same as 'decompress', but allows you to explicitly flush the stream. decompressFlush- :: (MonadBase base m, PrimMonad base, MonadThrow m)+ :: (PrimMonad m, MonadThrow m) => WindowBits -- ^ Zlib parameter (see the zlib-bindings package as well as the zlib C library)- -> Conduit (Flush ByteString) m (Flush ByteString)-decompressFlush = helperDecompress await yield+ -> ConduitT (Flush ByteString) (Flush ByteString) m ()+decompressFlush = helperDecompress await yield (leftover . Chunk) -helperDecompress :: (Monad (t m), MonadBase base m, PrimMonad base, MonadThrow m, MonadTrans t)+helperDecompress :: (Monad (t m), PrimMonad m, MonadThrow m, MonadTrans t) => t m (Maybe (Flush ByteString)) -> (Flush ByteString -> t m ())+ -> (ByteString -> t m ()) -> WindowBits -> t m ()-helperDecompress await' yield' config =- await' >>= maybe (return ()) start- where- start input = do- inf <- lift $ unsafeLiftIO $ initInflate config- push inf input+helperDecompress await' yield' leftover' config = do+ -- Initialize the stateful inflater, which will be used below+ -- This inflater is never exposed outside of this function+ inf <- lift $ unsafeLiftIO $ initInflate config - continue inf = await' >>= maybe (close inf) (push inf)+ -- Some helper functions used by the main feeder loop below - goPopper popper = do- mbs <- lift $ unsafeLiftIO popper- case mbs of- PRDone -> return ()- PRNext bs -> yield' (Chunk bs) >> goPopper popper- PRError e -> lift $ monadThrow e+ let -- Flush any remaining inflated bytes downstream+ flush = do+ chunk <- lift $ unsafeLiftIO $ flushInflate inf+ unless (S.null chunk) $ yield' $ Chunk chunk - push inf (Chunk x) = do- popper <- lift $ unsafeLiftIO $ feedInflate inf x- goPopper popper- continue inf+ -- Get any input which is unused by the inflater+ getUnused = lift $ unsafeLiftIO $ getUnusedInflate inf - push inf Flush = do- chunk <- lift $ unsafeLiftIO $ flushInflate inf- unless (S.null chunk) $ yield' $ Chunk chunk- yield' Flush- continue inf+ -- If there is any unused data, return it as leftovers to the stream+ unused = do+ rem' <- getUnused+ unless (S.null rem') $ leftover' rem' - close inf = do- chunk <- lift $ unsafeLiftIO $ finishInflate inf- unless (S.null chunk) $ yield' $ Chunk chunk+ -- Main loop: feed data from upstream into the inflater+ fix $ \feeder -> do+ mnext <- await'+ case mnext of+ -- No more data is available from upstream+ Nothing -> do+ -- Flush any remaining uncompressed data+ flush+ -- Return the rest of the unconsumed data as leftovers+ unused+ -- Another chunk of compressed data arrived+ Just (Chunk x) -> do+ -- Feed the compressed data into the inflater, returning a+ -- "popper" which will return chunks of decompressed data+ popper <- lift $ unsafeLiftIO $ feedInflate inf x + -- Loop over the popper grabbing decompressed chunks and+ -- yielding them downstream+ fix $ \pop -> do+ mbs <- lift $ unsafeLiftIO popper+ case mbs of+ -- No more data from this popper+ PRDone -> do+ rem' <- getUnused+ if S.null rem'+ -- No data was unused by the inflater, so let's+ -- fill it up again and get more data out of it+ then feeder+ -- In this case, there is some unconsumed data,+ -- meaning the compressed stream is complete.+ -- At this point, we need to stop feeding,+ -- return the unconsumed data as leftovers, and+ -- flush any remaining content (which should be+ -- nothing)+ else do+ flush+ leftover' rem'+ -- Another chunk available, yield it downstream and+ -- loop again+ PRNext bs -> do+ yield' (Chunk bs)+ pop+ -- An error occurred inside zlib, throw it+ PRError e -> lift $ throwM e+ -- We've been asked to flush the stream+ Just Flush -> do+ -- Get any uncompressed data waiting for us+ flush+ -- Put a Flush in the stream+ yield' Flush+ -- Feed in more data+ feeder+ -- | -- Compress (deflate) a stream of 'ByteString's. The 'WindowBits' also control -- the format (zlib vs. gzip). compress- :: (MonadBase base m, PrimMonad base, MonadThrow m)+ :: (PrimMonad m, MonadThrow m) => Int -- ^ Compression level -> WindowBits -- ^ Zlib parameter (see the zlib-bindings package as well as the zlib C library)- -> Conduit ByteString m ByteString+ -> ConduitT ByteString ByteString m () compress = helperCompress (liftM (fmap Chunk) await) yield' where@@ -108,13 +153,13 @@ -- | Same as 'compress', but allows you to explicitly flush the stream. compressFlush- :: (MonadBase base m, PrimMonad base, MonadThrow m)+ :: (PrimMonad m, MonadThrow m) => Int -- ^ Compression level -> WindowBits -- ^ Zlib parameter (see the zlib-bindings package as well as the zlib C library)- -> Conduit (Flush ByteString) m (Flush ByteString)+ -> ConduitT (Flush ByteString) (Flush ByteString) m () compressFlush = helperCompress await yield -helperCompress :: (Monad (t m), MonadBase base m, PrimMonad base, MonadThrow m, MonadTrans t)+helperCompress :: (Monad (t m), PrimMonad m, MonadThrow m, MonadTrans t) => t m (Maybe (Flush ByteString)) -> (Flush ByteString -> t m ()) -> Int@@ -134,7 +179,7 @@ case mbs of PRDone -> return () PRNext bs -> yield' (Chunk bs) >> goPopper popper- PRError e -> lift $ monadThrow e+ PRError e -> lift $ throwM e push def (Chunk x) = do popper <- lift $ unsafeLiftIO $ feedDeflate def x@@ -146,7 +191,7 @@ case mchunk of PRDone -> return () PRNext x -> yield' $ Chunk x- PRError e -> lift $ monadThrow e+ PRError e -> lift $ throwM e yield' Flush continue def @@ -155,4 +200,38 @@ case mchunk of PRDone -> return () PRNext chunk -> yield' (Chunk chunk) >> close def- PRError e -> lift $ monadThrow e+ PRError e -> lift $ throwM e++-- | The standard 'decompress' and 'ungzip' functions will only decompress a+-- single compressed entity from the stream. This combinator will exhaust the+-- stream completely of all individual compressed entities. This is useful for+-- cases where you have a concatenated archive, e.g. @cat file1.gz file2.gz >+-- combined.gz@.+--+-- Usage:+--+-- > sourceFile "combined.gz" $$ multiple ungzip =$ consume+--+-- This combinator will not fail on an empty stream. If you want to ensure that+-- at least one compressed entity in the stream exists, consider a usage such+-- as:+--+-- > sourceFile "combined.gz" $$ (ungzip >> multiple ungzip) =$ consume+--+-- @since 1.1.10+multiple :: Monad m+ => ConduitT ByteString a m ()+ -> ConduitT ByteString a m ()+multiple inner =+ loop+ where+ loop = do+ mbs <- await+ case mbs of+ Nothing -> return ()+ Just bs+ | S.null bs -> loop+ | otherwise -> do+ leftover bs+ inner+ loop
+ README.md view
@@ -0,0 +1,5 @@+## conduit-extra++For more information about conduit in general, and how this package in+particular fits into the ecosystem, see [the conduit+homepage](https://github.com/snoyberg/conduit#readme).
+ bench/blaze.hs view
@@ -0,0 +1,65 @@+{-# LANGUAGE OverloadedStrings #-}+import Data.Conduit+import qualified Data.Conduit.List as CL+import Data.Conduit.ByteString.Builder+import Gauge.Main+import Data.Monoid+import Data.ByteString.Builder++count :: Int+count = 100000++single :: Builder+single = shortByteString "Hello World!\n"++oneBuilderLeft :: Builder+oneBuilderLeft =+ loop count mempty+ where+ loop 0 b = b+ loop i b = loop (i - 1) (b <> single)++oneBuilderRight :: Builder+oneBuilderRight =+ loop count mempty+ where+ loop 0 b = b+ loop i b = loop (i - 1) (b <> single)++builderSource :: Monad m => ConduitT i Builder m ()+builderSource = CL.replicate count single++oneBSBuilderLeft :: Builder+oneBSBuilderLeft =+ loop count mempty+ where+ loop 0 b = b+ loop i b = loop (i - 1) (b <> single)++oneBSBuilderRight :: Builder+oneBSBuilderRight =+ loop count mempty+ where+ loop 0 b = b+ loop i b = loop (i - 1) (b <> single)++builderBSSource :: Monad m => ConduitT i Builder m ()+builderBSSource = CL.replicate count single++main :: IO ()+main = defaultMain+ [ bench "conduit, strict, safe" $ whnfIO $ runConduit $+ builderSource .| builderToByteString .| CL.sinkNull+ , bench "conduit, strict, unsafe" $ whnfIO $ runConduit $+ builderSource .| unsafeBuilderToByteString .| CL.sinkNull++ , bench "one builder, left" $ nf toLazyByteString oneBuilderLeft+ , bench "one builder, right" $ nf toLazyByteString oneBuilderRight+ , bench "conduit, lazy" $ flip nf builderSource $ \src ->+ toLazyByteString $ runConduitPure $ src .| CL.fold (<>) mempty++ , bench "one bs builder, left" $ nf toLazyByteString oneBSBuilderLeft+ , bench "one bs builder, right" $ nf toLazyByteString oneBSBuilderRight+ , bench "conduit BS, lazy" $ flip nf builderBSSource $ \src ->+ toLazyByteString $ runConduitPure $ src .| CL.fold (<>) mempty+ ]
conduit-extra.cabal view
@@ -1,5 +1,6 @@+Cabal-version: >=1.10 Name: conduit-extra-Version: 1.1.3+Version: 1.3.8 Synopsis: Batteries included conduit: adapters for common libraries. Description: The conduit package itself maintains relative small dependencies. The purpose of this package is to collect commonly used utility functions wrapping other library dependencies, without depending on heavier-weight dependencies. The basic idea is that this package should only depend on haskell-platform packages and conduit.@@ -9,40 +10,43 @@ Maintainer: michael@snoyman.com Category: Data, Conduit Build-type: Simple-Cabal-version: >=1.8 Homepage: http://github.com/snoyberg/conduit extra-source-files: test/random test/filesystem/*.txt test/filesystem/bin/*.txt+ ChangeLog.md+ README.md Library+ default-language: Haskell2010 Exposed-modules: Data.Conduit.Attoparsec Data.Conduit.Binary- Data.Conduit.Blaze+ Data.Conduit.ByteString.Builder Data.Conduit.Filesystem+ Data.Conduit.Foldl Data.Conduit.Lazy Data.Conduit.Network Data.Conduit.Network.UDP+ Data.Conduit.Network.Unix Data.Conduit.Process+ Data.Conduit.Process.Typed Data.Conduit.Text Data.Conduit.Zlib- if !os(windows)- Exposed-modules: Data.Conduit.Network.Unix - Build-depends: base >= 4 && < 5- , conduit >= 1.1 && < 1.2+ if arch(x86_64) || arch(i386)+ -- These architectures are able to perform unaligned memory accesses+ cpp-options: -DALLOW_UNALIGNED_ACCESS - -- No version bounds necessary, since they're inherited- -- from conduit.- , bytestring- , monad-control+ Build-depends: base >= 4.12 && < 5+ , conduit >= 1.3 && < 1.4++ , bytestring >= 0.10.2 , text , transformers- , transformers-base + , async , attoparsec >= 0.10- , blaze-builder >= 0.3 , directory , filepath , network >= 2.3@@ -50,12 +54,15 @@ , process , resourcet >= 1.1 , stm- , streaming-commons >= 0.1.4+ , streaming-commons >= 0.2.3.0+ , unliftio-core+ , typed-process >= 0.2.6 ghc-options: -Wall test-suite test hs-source-dirs: test+ default-language: Haskell2010 main-is: Spec.hs type: exitcode-stdio-1.0 ghc-options: -threaded@@ -67,28 +74,46 @@ , async , attoparsec- , blaze-builder , bytestring , exceptions , process , resourcet+ , QuickCheck , stm+ , streaming-commons , text , transformers , transformers-base+ , directory+ , filepath+ build-tool-depends: hspec-discover:hspec-discover ghc-options: -Wall if os(windows) cpp-options: -DWINDOWS other-modules: Data.Conduit.AttoparsecSpec Data.Conduit.BinarySpec- Data.Conduit.BlazeSpec+ Data.Conduit.ByteString.BuilderSpec Data.Conduit.ExtraSpec Data.Conduit.FilesystemSpec Data.Conduit.LazySpec Data.Conduit.NetworkSpec Data.Conduit.ProcessSpec+ Data.Conduit.Process.TypedSpec Data.Conduit.TextSpec Data.Conduit.ZlibSpec++benchmark blaze+ default-language: Haskell2010+ type: exitcode-stdio-1.0+ hs-source-dirs: bench+ build-depends: base+ , conduit+ , conduit-extra+ , gauge+ , bytestring+ , transformers+ main-is: blaze.hs+ ghc-options: -Wall -O2 -rtsopts source-repository head type: git
test/Data/Conduit/AttoparsecSpec.hs view
@@ -1,5 +1,7 @@+{-# LANGUAGE BangPatterns #-} {-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TupleSections #-} {-# OPTIONS_GHC -fno-warn-incomplete-patterns #-} module Data.Conduit.AttoparsecSpec (spec) where import Control.Exception (fromException)@@ -7,7 +9,6 @@ import Control.Applicative ((<*), (<|>)) import Control.Monad-import Control.Monad.Trans.Resource (runExceptionT) import qualified Data.Attoparsec.ByteString.Char8 import qualified Data.Attoparsec.Text import Data.Conduit@@ -21,88 +22,155 @@ let input = ["aaa\na", "aaa\n\n", "aaa", "aab\n\naaaa"] badLine = 4 badCol = 6+ badOff = 15 parser = Data.Attoparsec.Text.endOfInput <|> (Data.Attoparsec.Text.notChar 'b' >> parser) sink = sinkParser parser- ea <- runExceptionT $ CL.sourceList input $$ sink+ sink' = sinkParserEither parser+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink'+ case ea' of+ Left pe ->+ errorPosition pe `shouldBe` Position badLine badCol badOff it "works for bytestring" $ do let input = ["aaa\na", "aaa\n\n", "aaa", "aab\n\naaaa"] badLine = 4 badCol = 6+ badOff = 15 parser = Data.Attoparsec.ByteString.Char8.endOfInput <|> (Data.Attoparsec.ByteString.Char8.notChar 'b' >> parser) sink = sinkParser parser- ea <- runExceptionT $ CL.sourceList input $$ sink+ sink' = sinkParserEither parser+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink'+ case ea' of+ Left pe ->+ errorPosition pe `shouldBe` Position badLine badCol badOff it "works in last chunk" $ do let input = ["aaa\na", "aaa\n\n", "aaa", "aab\n\naaaa"] badLine = 6 badCol = 5+ badOff = 22 parser = Data.Attoparsec.Text.char 'c' <|> (Data.Attoparsec.Text.anyChar >> parser) sink = sinkParser parser- ea <- runExceptionT $ CL.sourceList input $$ sink+ sink' = sinkParserEither parser+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink'+ case ea' of+ Left pe ->+ errorPosition pe `shouldBe` Position badLine badCol badOff it "works in last chunk" $ do let input = ["aaa\na", "aaa\n\n", "aaa", "aa\n\naaaab"] badLine = 6 badCol = 6+ badOff = 22 parser = Data.Attoparsec.Text.string "bc" <|> (Data.Attoparsec.Text.anyChar >> parser) sink = sinkParser parser- ea <- runExceptionT $ CL.sourceList input $$ sink+ sink' = sinkParserEither parser+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink'+ case ea' of+ Left pe ->+ errorPosition pe `shouldBe` Position badLine badCol badOff it "works after new line in text" $ do let input = ["aaa\n", "aaa\n\n", "aaa", "aa\nb\naaaa"] badLine = 5 badCol = 1+ badOff = 15 parser = Data.Attoparsec.Text.endOfInput <|> (Data.Attoparsec.Text.notChar 'b' >> parser) sink = sinkParser parser- ea <- runExceptionT $ CL.sourceList input $$ sink+ sink' = sinkParserEither parser+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink'+ case ea' of+ Left pe ->+ errorPosition pe `shouldBe` Position badLine badCol badOff it "works after new line in bytestring" $ do let input = ["aaa\n", "aaa\n\n", "aaa", "aa\nb\naaaa"] badLine = 5 badCol = 1+ badOff = 15 parser = Data.Attoparsec.ByteString.Char8.endOfInput <|> (Data.Attoparsec.ByteString.Char8.notChar 'b' >> parser) sink = sinkParser parser- ea <- runExceptionT $ CL.sourceList input $$ sink+ sink' = sinkParserEither parser+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink'+ case ea' of+ Left pe ->+ errorPosition pe `shouldBe` Position badLine badCol badOff+ it "works for first line" $ do+ let input = ["aab\na", "aaa\n\n", "aaa", "aab\n\naaaa"]+ badLine = 1+ badCol = 3+ badOff = 2+ parser = Data.Attoparsec.Text.endOfInput <|> (Data.Attoparsec.Text.notChar 'b' >> parser)+ sink = sinkParser parser+ sink' = sinkParserEither parser+ ea = runConduit $ CL.sourceList input .| sink+ case ea of+ Left e ->+ case fromException e of+ Just pe -> do+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink'+ case ea' of+ Left pe ->+ errorPosition pe `shouldBe` Position badLine badCol badOff describe "conduitParser" $ do it "parses a repeated stream" $ do let input = ["aaa\n", "aaa\naaa\n", "aaa\n"] parser = Data.Attoparsec.Text.string "aaa" <* Data.Attoparsec.Text.endOfLine- sink = conduitParserEither parser =$= CL.consume- (Right ea) <- runExceptionT $ CL.sourceList input $$ sink+ sink = conduitParserEither parser .| CL.consume+ (Right !ea) = runConduit $ CL.sourceList input .| sink let chk a = case a of Left{} -> False Right (_, xs) -> xs == "aaa"- chkp 1 = (PositionRange (Position 1 0) (Position 2 1))- chkp l = (PositionRange (Position l 1) (Position (l+1) 1))+ chkp l = PositionRange (Position l 1 ((l - 1) * 4)) (Position (l+1) 1 (l * 4)) forM_ ea $ \ a -> a `shouldSatisfy` chk :: Expectation forM_ (zip ea [1..]) $ \ (Right (pos, _), l) -> pos `shouldBe` chkp l length ea `shouldBe` 4 + it "positions on first line" $ do+ results <- runConduit $ yield "hihihi\nhihi"+ .| conduitParser (Data.Attoparsec.Text.string "\n" <|> Data.Attoparsec.Text.string "hi")+ .| CL.consume+ let f (a, b, c, d, e, f', g) = (PositionRange (Position a b c) (Position d e f'), g)+ results `shouldBe` map f+ [ (1, 1, 0, 1, 3, 2, "hi")+ , (1, 3, 2, 1, 5, 4, "hi")+ , (1, 5, 4, 1, 7, 6, "hi") + , (1, 7, 6, 2, 1, 7, "\n")++ , (2, 1, 7, 2, 3, 9, "hi")+ , (2, 3, 9, 2, 5, 11, "hi")+ ]
test/Data/Conduit/BinarySpec.hs view
@@ -1,6 +1,9 @@+{-# LANGUAGE GADTs #-}+{-# LANGUAGE RankNTypes #-} {-# LANGUAGE OverloadedStrings #-} module Data.Conduit.BinarySpec (spec) where +import Data.Conduit (runConduit, runConduitRes, (.|), runConduitPure, ConduitT, Void) import qualified Data.Conduit.Binary as CB import qualified Data.Conduit as C import qualified Data.Conduit.List as CL@@ -15,6 +18,17 @@ import qualified Data.ByteString as S import qualified Data.ByteString.Char8 as S8 import Data.Functor.Identity+import Test.QuickCheck.Arbitrary (Arbitrary, arbitrary)+import Test.QuickCheck.Gen (Gen, oneof)+import Data.Word (Word8)+import Foreign.Storable (Storable, sizeOf, pokeByteOff, alignment)+import Data.Typeable (Typeable)+import Data.ByteString.Internal (createAndTrim')+import Foreign.Ptr (alignPtr, minusPtr)+import System.Directory (doesFileExist)+import System.IO.Unsafe (unsafePerformIO)+import Control.Applicative ((<$>))+import qualified Data.ByteString.Builder as BB spec :: Spec spec = describe "Data.Conduit.Binary" $ do@@ -22,24 +36,34 @@ describe "file access" $ do it "read" $ do bs <- S.readFile "conduit-extra.cabal"- bss <- runResourceT $ CB.sourceFile "conduit-extra.cabal" C.$$ CL.consume+ bss <- runConduitRes $ CB.sourceFile "conduit-extra.cabal" .| CL.consume S.concat bss `shouldBe` bs it "read range" $ do S.writeFile "tmp" "0123456789"- bss <- runResourceT $ CB.sourceFileRange "tmp" (Just 2) (Just 3) C.$$ CL.consume+ bss <- runConduitRes $ CB.sourceFileRange "tmp" (Just 2) (Just 3) .| CL.consume S.concat bss `shouldBe` "234" it "write" $ do- runResourceT $ CB.sourceFile "conduit-extra.cabal" C.$$ CB.sinkFile "tmp"+ runConduitRes $ CB.sourceFile "conduit-extra.cabal" .| CB.sinkFile "tmp" bs1 <- S.readFile "conduit-extra.cabal" bs2 <- S.readFile "tmp" bs2 `shouldBe` bs1 ++ it "write builder (withSinkFileBuilder)" $ do+ CB.withSinkFileBuilder "tmp" $ \sink ->+ CB.withSourceFile "conduit-extra.cabal" $ \src ->+ runConduit $ src .| CL.map BB.byteString .| sink+ bs1 <- S.readFile "conduit-extra.cabal"+ bs2 <- S.readFile "tmp"+ bs2 `shouldBe` bs1+ it "conduit" $ do- runResourceT $ CB.sourceFile "conduit-extra.cabal"- C.$= CB.conduitFile "tmp"- C.$$ CB.sinkFile "tmp2"+ runConduitRes+ $ CB.sourceFile "conduit-extra.cabal"+ .| CB.conduitFile "tmp"+ .| CB.sinkFile "tmp2" bs1 <- S.readFile "conduit-extra.cabal" bs2 <- S.readFile "tmp" bs3 <- S.readFile "tmp2"@@ -47,14 +71,15 @@ bs3 `shouldBe` bs1 describe "binary isolate" $ do it "works" $ do- bss <- runResourceT $ CL.sourceList (replicate 1000 "X")- C.$= CB.isolate 6- C.$$ CL.consume+ bss <- runConduitRes+ $ CL.sourceList (replicate 1000 "X")+ .| CB.isolate 6+ .| CL.consume S.concat bss `shouldBe` "XXXXXX" describe "properly using binary file reading" $ do it "sourceFile" $ do- x <- runResourceT $ CB.sourceFile "test/random" C.$$ CL.consume+ x <- runConduitRes $ CB.sourceFile "test/random" .| CL.consume lbs <- L.readFile "test/random" L.fromChunks x `shouldBe` lbs @@ -69,19 +94,18 @@ prop "works" $ \bss' -> let bss = map S.pack bss'- in runIdentity $- CL.sourceList bss C.$$ go (L.fromChunks bss)+ in runConduitPure $ CL.sourceList bss .| go (L.fromChunks bss) describe "binary takeWhile" $ do prop "works" $ \bss' -> let bss = map S.pack bss' in runIdentity $ do- bss2 <- CL.sourceList bss C.$$ CB.takeWhile (>= 5) C.=$ CL.consume+ bss2 <- runConduit $ CL.sourceList bss .| CB.takeWhile (>= 5) .| CL.consume return $ L.fromChunks bss2 == L.takeWhile (>= 5) (L.fromChunks bss) prop "leftovers present" $ \bss' -> let bss = map S.pack bss' in runIdentity $ do- result <- CL.sourceList bss C.$$ do- x <- CB.takeWhile (>= 5) C.=$ CL.consume+ result <- runConduit $ CL.sourceList bss .| do+ x <- CB.takeWhile (>= 5) .| CL.consume y <- CL.consume return (S.concat x, S.concat y) let expected = S.span (>= 5) $ S.concat bss@@ -93,13 +117,13 @@ prop "works" $ \bss' -> let bss = map S.pack bss' in runIdentity $ do- bss2 <- CL.sourceList bss C.$$ do+ bss2 <- runConduit $ CL.sourceList bss .| do CB.dropWhile (< 5) CL.consume return $ L.fromChunks bss2 == L.dropWhile (< 5) (L.fromChunks bss) describe "binary take" $ do- let go n l = CL.sourceList l C.$$ do+ let go n l = runConduit $ CL.sourceList l .| do a <- CB.take n b <- CL.consume return (a, b)@@ -133,25 +157,52 @@ let bss = map S.pack bss' bs = S.concat bss src = CL.sourceList bss- res <- src C.$$ CB.lines C.=$ CL.consume+ res <- runConduit $ src .| CB.lines .| CL.consume return $ S8.lines bs == res describe "sinkCacheLength" $ do it' "works" $ runResourceT $ do lbs <- liftIO $ L.readFile "test/Data/Conduit/BinarySpec.hs"- (len, src) <- CB.sourceLbs lbs C.$$ CB.sinkCacheLength- lbs' <- src C.$$ CB.sinkLbs+ (len, src) <- runConduit $ CB.sourceLbs lbs .| CB.sinkCacheLength+ lbs' <- runConduit $ src .| CB.sinkLbs liftIO $ do fromIntegral len `shouldBe` L.length lbs lbs' `shouldBe` lbs fromIntegral len `shouldBe` L.length lbs' + describe "sinkFileCautious" $ do+ it' "success" $ do+ runConduitRes $ CB.sourceFile "conduit-extra.cabal" .| CB.sinkFileCautious "tmp"+ bs1 <- S.readFile "conduit-extra.cabal"+ bs2 <- S.readFile "tmp"+ bs2 `shouldBe` bs1+ it' "failure" $ do+ let bs1 = "This is the original content"+ S.writeFile "tmp" bs1+ runConduitRes+ ( (CB.sourceFile "conduit-extra.cabal" >> error "FIXME")+ .| CB.sinkFileCautious "tmp")+ `shouldThrow` anyException+ bs2 <- S.readFile "tmp"+ bs2 `shouldBe` bs1++ it "sinkSystemTempFile" $ do+ let bs = "Hello World!"+ fp <- runResourceT $ do+ fp <- runConduit $ C.yield bs .| CB.sinkSystemTempFile "temp-file-test"+ liftIO $ do+ actual <- S.readFile fp+ actual `shouldBe` bs+ return fp+ exists <- doesFileExist fp+ exists `shouldBe` False+ describe "Data.Conduit.Binary.mapM_" $ do prop "telling works" $ \bytes -> let lbs = L.pack bytes src = CB.sourceLbs lbs sink = CB.mapM_ (tell . return . S.singleton)- bss = execWriter $ src C.$$ sink+ bss = execWriter $ runConduit $ src .| sink in L.fromChunks bss == lbs describe "exception handling" $ do@@ -163,7 +214,7 @@ onErr :: MonadIO m => IOException -> m () onErr _ = liftIO $ I.modifyIORef ref (+ 1) contents <- L.readFile "conduit-extra.cabal"- res <- runResourceT $ src C.$$ CB.sinkLbs+ res <- runConduitRes $ src .| CB.sinkLbs res `shouldBe` contents errCount <- I.readIORef ref errCount `shouldBe` (1 :: Int)@@ -174,7 +225,7 @@ res2 <- C.tryC $ CB.sourceFile "conduit-extra.cabal" liftIO $ I.writeIORef ref (res1, res2) contents <- L.readFile "conduit-extra.cabal"- res <- runResourceT $ src C.$$ CB.sinkLbs+ res <- runConduitRes $ src .| CB.sinkLbs res `shouldBe` contents exc <- I.readIORef ref case exc :: (Either IOException (), Either IOException ()) of@@ -184,10 +235,103 @@ describe "normalFuseLeft" $ do it "does not double close conduit" $ do- x <- runResourceT $ do+ x <- runConduitRes $ let src = CL.sourceList ["foobarbazbin"]- src C.$= CB.isolate 10 C.$$ CL.head+ in src .| CB.isolate 10 .| CL.head x `shouldBe` Just "foobarbazb"++ describe "Storable" $ do+ let test name func = describe name $ do+ let test' size =+ prop ("chunk size " ++ show size) $ \stores0 -> do+ let src =+ loop (someStorables stores0)+ where+ loop bs+ | S.null bs = return ()+ | otherwise = do+ let (x, y) = S.splitAt size bs+ C.yield x+ loop y++ sink :: [SomeStorable]+ -> ConduitT S.ByteString Void IO ()+ sink [] = do+ mw <- CB.head+ case mw of+ Nothing -> return ()+ Just _ -> error "trailing bytes"+ sink (next:rest) = do+ withSomeStorable next checkOne+ sink rest++ checkOne :: (Storable a, Eq a, Show a)+ => a+ -> ConduitT S.ByteString Void IO ()+ checkOne expected = do+ mactual <-+ if func+ then CB.sinkStorable+ else fmap Just CB.sinkStorableEx+ actual <-+ case mactual of+ Nothing -> error "got Nothing"+ Just actual -> return actual+ liftIO $ actual `shouldBe` expected++ runConduit $ src .| sink stores0 :: IO ()+ mapM_ test' [1, 5, 10, 100]++ test "sink Maybe" True+ test "sink exception" False++ it' "insufficient bytes are leftovers, one chunk" $ do+ let src = C.yield $ S.singleton 1+ runConduit $ src .| do+ mactual <- CB.sinkStorable+ liftIO $ mactual `shouldBe` (Nothing :: Maybe Int)+ lbs <- CB.sinkLbs+ liftIO $ lbs `shouldBe` L.singleton 1++ it' "insufficient bytes are leftovers, multiple chunks" $ do+ let src = do+ C.yield $ S.singleton 1+ C.yield $ S.singleton 2+ runConduit $ src .| do+ mactual <- CB.sinkStorable+ liftIO $ mactual `shouldBe` (Nothing :: Maybe Int)+ lbs <- CB.sinkLbs+ liftIO $ lbs `shouldBe` L.pack [1, 2]++data SomeStorable where+ SomeStorable :: (Storable a, Eq a, Show a, Typeable a) => a -> SomeStorable+instance Show SomeStorable where+ show (SomeStorable x) = show x+instance Arbitrary SomeStorable where+ arbitrary = oneof+ [ SomeStorable <$> (arbitrary :: Gen Int)+ , SomeStorable <$> (arbitrary :: Gen Word8)+ , SomeStorable <$> (arbitrary :: Gen Double)+ ]++withSomeStorable :: SomeStorable+ -> (forall a. (Storable a, Eq a, Show a) => a -> b)+ -> b+withSomeStorable (SomeStorable x) f = f x++someStorable :: SomeStorable -> S.ByteString+someStorable store =+ fst $ unsafePerformIO $ createAndTrim' (size + align) start+ where+ size = withSomeStorable store sizeOf+ align = withSomeStorable store alignment+ start ptr = do+ let off = minusPtr ptr (alignPtr ptr align)+ withSomeStorable store (pokeByteOff ptr off)+ return (off, size, ())++someStorables :: [SomeStorable] -> S.ByteString+someStorables = S.concat . map someStorable it' :: String -> IO () -> Spec it' = it
− test/Data/Conduit/BlazeSpec.hs
@@ -1,63 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE CPP #-}-module Data.Conduit.BlazeSpec (spec) where--import Test.Hspec-import Test.Hspec.QuickCheck (prop)--import qualified Data.Conduit as C-import qualified Data.Conduit.List as CL-import Data.ByteString.Char8 ()-import Data.Conduit.Blaze (builderToByteString, builderToByteStringFlush)-import Control.Monad.ST (runST)-import Data.Monoid-import qualified Data.ByteString as S-import Blaze.ByteString.Builder (fromByteString, toLazyByteString, insertLazyByteString, flush)-import qualified Data.ByteString.Lazy as L-import Data.ByteString.Lazy.Char8 ()--spec :: Spec-spec =- describe "Data.Conduit.Blaze" $ do- prop "idempotent to toLazyByteString" $ \bss' -> runST $ do- let bss = map S.pack bss'- let builders = map fromByteString bss- let lbs = toLazyByteString $ mconcat builders- let src = mconcat $ map (CL.sourceList . return) builders- outBss <- src C.$= builderToByteString C.$$ CL.consume- return $ lbs == L.fromChunks outBss-- it "works for large input" $ do- let builders = replicate 10000 (fromByteString "hello world!")- let lbs = toLazyByteString $ mconcat builders- let src = mconcat $ map (CL.sourceList . return) builders- outBss <- src C.$= builderToByteString C.$$ CL.consume- lbs `shouldBe` L.fromChunks outBss-- it "works for lazy bytestring insertion" $ do- let builders = replicate 10000 (insertLazyByteString "hello world!")- let lbs = toLazyByteString $ mconcat builders- let src = mconcat $ map (CL.sourceList . return) builders- outBss <- src C.$= builderToByteString C.$$ CL.consume- lbs `shouldBe` L.fromChunks outBss-- it "flush shouldn't bring in empty strings." $ do- let dat = ["hello", "world"]- src = CL.sourceList dat C.$= CL.map ((`mappend` flush) . fromByteString)- out <- src C.$= builderToByteString C.$$ CL.consume- dat `shouldBe` out-- prop "flushing" $ \bss' -> runST $ do- let bss = concatMap (\bs -> [C.Chunk $ S.pack bs, C.Flush]) $ filter (not . null) bss'- let src = CL.sourceList $ map (fmap fromByteString) bss- outBss <- src C.$= builderToByteStringFlush C.$$ CL.consume- if bss == outBss then return () else error (show (bss, outBss))- return $ bss == outBss- it "large flush input" $ do- let lbs = L.pack $ concat $ replicate 100000 [0..255]- chunks = map (C.Chunk . fromByteString) (L.toChunks lbs)- src = CL.sourceList chunks- bss <- src C.$$ builderToByteStringFlush C.=$ CL.consume- let unFlush (C.Chunk x) = [x]- unFlush C.Flush = []- L.fromChunks (concatMap unFlush bss) `shouldBe` lbs
+ test/Data/Conduit/ByteString/BuilderSpec.hs view
@@ -0,0 +1,61 @@+{-# LANGUAGE OverloadedStrings #-}+module Data.Conduit.ByteString.BuilderSpec (spec) where++import Test.Hspec+import Test.Hspec.QuickCheck (prop)++import Data.Conduit ((.|), runConduit, Flush (..))+import qualified Data.Conduit.List as CL+import Data.Conduit.ByteString.Builder (builderToByteString, builderToByteStringFlush)+import Control.Monad.ST (runST)+import qualified Data.ByteString as S+import Data.ByteString.Builder (byteString, toLazyByteString)+import Data.ByteString.Builder.Internal (lazyByteStringInsert, flush)+import qualified Data.ByteString.Lazy as L++spec :: Spec+spec =+ describe "Data.Conduit.ByteString.Builder" $ do+ prop "idempotent to toLazyByteString" $ \bss' -> runST $ do+ let bss = map S.pack bss'+ let builders = map byteString bss+ let lbs = toLazyByteString $ mconcat builders+ let src = mconcat $ map (CL.sourceList . return) builders+ outBss <- runConduit $ src .| builderToByteString .| CL.consume+ return $ lbs == L.fromChunks outBss++ it "works for large input" $ do+ let builders = replicate 10000 (byteString "hello world!")+ let lbs = toLazyByteString $ mconcat builders+ let src = mconcat $ map (CL.sourceList . return) builders+ outBss <- runConduit $ src .| builderToByteString .| CL.consume+ lbs `shouldBe` L.fromChunks outBss++ it "works for lazy bytestring insertion" $ do+ let builders = replicate 10000 (lazyByteStringInsert "hello world!")+ let lbs = toLazyByteString $ mconcat builders+ let src = mconcat $ map (CL.sourceList . return) builders+ outBss <- runConduit $ src .| builderToByteString .| CL.consume+ lbs `shouldBe` L.fromChunks outBss++ it "flush shouldn't bring in empty strings." $ do+ let dat = ["hello", "world"]+ src = CL.sourceList dat .| CL.map ((`mappend` flush) . byteString)+ out <- runConduit $ src .| builderToByteString .| CL.consume+ dat `shouldBe` out++ prop "flushing" $ \bss' -> runST $ do+ let bss = concatMap (\bs -> [Chunk $ S.pack bs, Flush]) $ filter (not . null) bss'+ let chunks = map (fmap byteString) bss+ let src = CL.sourceList chunks+ outBss <- runConduit $ src .| builderToByteStringFlush .| CL.consume+ if bss == outBss then return () else error (show (bss, outBss))+ return $ bss == outBss+ it "large flush input" $ do+ let lbs = L.pack $ concat $ replicate 100000 [0..255]+ let chunks = map (Chunk . byteString) (L.toChunks lbs)+ let src = CL.sourceList chunks+ bss <- runConduit $ src .| builderToByteStringFlush .| CL.consume+ let unFlush (Chunk x) = [x]+ unFlush Flush = []+ L.fromChunks (concatMap unFlush bss) `shouldBe` lbs
test/Data/Conduit/ExtraSpec.hs view
@@ -13,18 +13,18 @@ spec :: Spec spec = describe "Data.Conduit.Extra" $ do it "basic test" $ do- let sink2 :: Sink a IO (Maybe a, Maybe a)+ let sink2 :: ConduitT a o IO (Maybe a, Maybe a) sink2 = do ma1 <- fuseLeftovers id (isolate 10) peek ma2 <- peek return (ma1, ma2) source = yield 1 >> yield 2- res <- source $$ sink2- res `shouldBe` (Just 1, Just 1)+ res <- runConduit $ source .| sink2+ res `shouldBe` (Just 1, Just (1 :: Int)) it "get leftovers" $ do- let sink2 :: Sink a IO ([a], [a], [a])+ let sink2 :: ConduitT a o IO ([a], [a], [a]) sink2 = do (x, y) <- fuseReturnLeftovers (isolate 2) peek3 z <- CL.consume@@ -35,12 +35,12 @@ mapM_ leftover $ reverse x return x - source = mapM_ yield [1..5]- res <- source $$ sink2+ source = mapM_ yield [1..5 :: Int]+ res <- runConduit $ source .| sink2 res `shouldBe` ([1..2], [1..2], [3..5]) it "multiple values" $ do- let sink2 :: Sink a IO ([a], Maybe a)+ let sink2 :: ConduitT a o IO ([a], Maybe a) sink2 = do ma1 <- fuseLeftovers id (isolate 10) peek3 ma2 <- peek@@ -52,16 +52,16 @@ return x source = mapM_ yield [1..5]- res <- source $$ sink2- res `shouldBe` ([1..3], Just 1)+ res <- runConduit $ source .| sink2+ res `shouldBe` ([1..3], Just (1 :: Int)) prop "more complex" $ \ss cnt -> do let ts = map T.pack ss src = mapM_ (yield . T.encodeUtf8) ts conduit = CL.map T.decodeUtf8- sink = CT.take cnt =$ consume+ sink = CT.take cnt .| consume undo = return . T.encodeUtf8 . T.concat- res <- src $$ do+ res <- runConduit $ src .| do x <- fuseLeftovers undo conduit sink y <- consume return (T.concat x, T.decodeUtf8 $ S.concat y)
test/Data/Conduit/FilesystemSpec.hs view
@@ -5,34 +5,34 @@ import qualified Data.Conduit.List as CL import Data.Conduit.Filesystem import Data.List (sort, isSuffixOf)-import Control.Monad.Trans.Resource (runResourceT)+import System.FilePath ((</>)) spec :: Spec spec = describe "Data.Conduit.Filesystem" $ do it "sourceDirectory" $ do- res <- runResourceT- $ sourceDirectory "test/filesystem"- $$ CL.filter (not . (".swp" `isSuffixOf`))- =$ CL.consume+ res <- runConduitRes+ $ sourceDirectory ("test" </> "filesystem")+ .| CL.filter (not . (".swp" `isSuffixOf`))+ .| CL.consume sort res `shouldBe`- [ "test/filesystem/bar.txt"- , "test/filesystem/baz.txt"- , "test/filesystem/bin"- , "test/filesystem/foo.txt"+ [ "test" </> "filesystem" </> "bar.txt"+ , "test" </> "filesystem" </> "baz.txt"+ , "test" </> "filesystem" </> "bin"+ , "test" </> "filesystem" </> "foo.txt" ] it "sourceDirectoryDeep" $ do- res1 <- runResourceT- $ sourceDirectoryDeep False "test/filesystem"- $$ CL.filter (not . (".swp" `isSuffixOf`))- =$ CL.consume- res2 <- runResourceT- $ sourceDirectoryDeep True "test/filesystem"- $$ CL.filter (not . (".swp" `isSuffixOf`))- =$ CL.consume+ res1 <- runConduitRes+ $ sourceDirectoryDeep False ("test" </> "filesystem")+ .| CL.filter (not . (".swp" `isSuffixOf`))+ .| CL.consume+ res2 <- runConduitRes+ $ sourceDirectoryDeep True ("test" </> "filesystem")+ .| CL.filter (not . (".swp" `isSuffixOf`))+ .| CL.consume sort res1 `shouldBe`- [ "test/filesystem/bar.txt"- , "test/filesystem/baz.txt"- , "test/filesystem/bin/bin.txt"- , "test/filesystem/foo.txt"+ [ "test" </> "filesystem" </> "bar.txt"+ , "test" </> "filesystem" </> "baz.txt"+ , "test" </> "filesystem" </> "bin" </> "bin.txt"+ , "test" </> "filesystem" </> "foo.txt" ] sort res1 `shouldBe` sort res2
test/Data/Conduit/LazySpec.hs view
@@ -6,7 +6,6 @@ import qualified Data.Conduit as C import qualified Data.Conduit.Binary as CB import Control.Monad.Trans.Resource-import Data.Monoid import qualified Data.IORef as I import Control.Monad (forever)
test/Data/Conduit/NetworkSpec.hs view
@@ -3,19 +3,46 @@ import Data.Conduit import Data.Conduit.Network-import Control.Concurrent (forkIO, threadDelay)+import Control.Concurrent (forkIO, threadDelay, putMVar, newEmptyMVar, takeMVar, killThread) import Control.Monad (replicateM_) import Test.Hspec spec :: Spec-spec = it "Data.Conduit.Network" $ do- _ <- forkIO $ runTCPServer (serverSettings 4009 "*4") echo- threadDelay 1000000- replicateM_ 10000- $ runTCPClient (clientSettings 4009 "127.0.0.1") doNothing+spec = describe "Data.Conduit.Network" $ do+ describe "run general server" $ do+ it "running tcp server" $ do+ _ <- forkIO $ runTCPServer (serverSettings 4009 "*4") echo+ threadDelay 1000000+ replicateM_ 100+ $ runTCPClient (clientSettings 4009 "127.0.0.1") doNothing+ describe "fork server" $ do+ it "can connect to server" $ do+ let set = serverSettings 4010 "*4"+ threadId <- forkTCPServer set echo+ replicateM_ 100+ $ runTCPClient (clientSettings 4010 "127.0.0.1") doNothing+ killThread threadId + it "fork server also executes custom afterBind" $ do+ assertMVar <- newEmptyMVar+ let set = serverSettings 4010 "*4"+ setWithAfterBind = setAfterBind (\_ -> putMVar assertMVar ()) set+ threadId <- forkTCPServer setWithAfterBind echo+ takeMVar assertMVar+ killThread threadId++ it "fork server really waits for server to be finalized before returning" $ do+ let set = serverSettings 4010 "*4"+ setWithAfterBind = setAfterBind (\_ -> threadDelay 1000000) set+ threadId <- forkTCPServer setWithAfterBind echo+ replicateM_ 100+ $ runTCPClient (clientSettings 4010 "127.0.0.1") doNothing+ killThread threadId+++ echo :: AppData -> IO ()-echo ad = appSource ad $$ appSink ad+echo ad = runConduit $ appSource ad .| appSink ad doNothing :: AppData -> IO () doNothing _ = return ()
+ test/Data/Conduit/Process/TypedSpec.hs view
@@ -0,0 +1,31 @@+module Data.Conduit.Process.TypedSpec (spec) where++import Test.Hspec+import Data.Conduit+import Data.Conduit.Process.Typed+import qualified Data.Conduit.List as CL+import qualified Data.ByteString as B++spec :: Spec+spec = do+ it "cat works" $ do+ let fp = "ChangeLog.md"+ pc = setStdout createSource $ proc "cat" [fp]+ bs <- B.readFile fp+ bss <- withProcess_ pc $ \p ->+ runConduit (getStdout p .| CL.consume) <* waitExitCode p+ B.concat bss `shouldBe` bs+ it "cat works with withLoggedProcess_" $ do+ let fp = "ChangeLog.md"+ pc = proc "cat" [fp]+ bs <- B.readFile fp+ bss <- withLoggedProcess_ pc $ \p ->+ runConduit (getStdout p .| CL.consume) <* waitExitCode p+ B.concat bss `shouldBe` bs+ it "failing process throws" $ do+ (withLoggedProcess_ (proc "cat" ["does not exist"]) $ \p -> do+ runConduit $ getStdout p .| CL.mapM_ (error "shouldn't have data"))+ `shouldThrow` anyException+ it "failing process throws" $ do+ (withProcess_ (proc "cat" ["does not exist"]) $ const $ return ())+ `shouldThrow` anyException
test/Data/Conduit/ProcessSpec.hs view
@@ -1,14 +1,17 @@ {-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-} module Data.Conduit.ProcessSpec (spec, main) where import Test.Hspec import Test.Hspec.QuickCheck (prop) import Data.Conduit+import qualified Data.Conduit.Combinators as CC import qualified Data.Conduit.List as CL import Data.Conduit.Process import Control.Concurrent.Async (concurrently) import qualified Data.ByteString.Lazy as L import qualified Data.ByteString as S+import qualified Data.ByteString.Char8 as S8 import System.Exit import Control.Concurrent (threadDelay) @@ -23,16 +26,16 @@ ((sink, closeStdin), source, Inherited, cph) <- streamingProcess (shell "cat") ((), bss) <- concurrently (do- mapM_ yield (L.toChunks lbs) $$ sink+ runConduit $ mapM_ yield (L.toChunks lbs) .| sink closeStdin)- (source $$ CL.consume)+ (runConduit $ source .| CL.consume) L.fromChunks bss `shouldBe` lbs ec <- waitForStreamingProcess cph ec `shouldBe` ExitSuccess it "closed stream" $ do (ClosedStream, source, Inherited, cph) <- streamingProcess (shell "cat")- bss <- source $$ CL.consume+ bss <- runConduit $ source .| CL.consume bss `shouldBe` [] ec <- waitForStreamingProcess cph@@ -45,6 +48,44 @@ `shouldReturn` (ExitFailure 11, ()) (sourceCmdWithConsumer "exit 12" CL.sinkNull) `shouldReturn` (ExitFailure 12, ())+ (sourceCmdWithStreams "exit 0" CL.sourceNull CL.sinkNull CL.sinkNull)+ `shouldReturn` (ExitSuccess, (), ())+ (sourceCmdWithStreams "exit 11" CL.sourceNull CL.sinkNull CL.sinkNull)+ `shouldReturn` (ExitFailure 11, (), ())+ (sourceCmdWithStreams "exit 12" CL.sourceNull CL.sinkNull CL.sinkNull)+ `shouldReturn` (ExitFailure 12, (), ())++ it "consumes stdout" $ do+ let mystr = "this is a test string" :: String+ sourceCmdWithStreams ("bash -c \"echo -n " ++ mystr ++ "\"")+ CL.sourceNull+ CL.consume -- stdout+ CL.consume -- stderr+ `shouldReturn` (ExitSuccess, [S8.pack mystr], [])++ it "consumes stderr" $ do+ let mystr = "this is a test string" :: String+ sourceCmdWithStreams ("bash -c \">&2 echo -n " ++ mystr ++ "\"")+ CL.sourceNull+ CL.consume -- stdout+ CL.consume -- stderr+ `shouldReturn` (ExitSuccess, [], [S8.pack mystr])++ it "feeds stdin" $ do+ let mystr = "this is a test string" :: S.ByteString+ sourceCmdWithStreams "cat"+ (yield mystr)+ CL.consume -- stdout+ CL.consume -- stderr+ `shouldReturn` (ExitSuccess, [mystr], [])+ it "gracefully handles closed stdin" $ do+ let blob = L.iterate (+1) 0+ blobHead = L.toStrict $ L.take 10000 blob+ sourceCmdWithStreams "head -c 10000"+ (CC.sourceLazy blob)+ (L.toStrict <$> CC.sinkLazy) -- stdout+ CL.consume -- stderr+ `shouldReturn` (ExitSuccess, blobHead, []) #endif it "blocking vs non-blocking" $ do (ClosedStream, ClosedStream, ClosedStream, cph) <- streamingProcess (shell "sleep 1")@@ -54,8 +95,17 @@ threadDelay 1500000 - mec2 <- getStreamingProcessExitCode cph- mec2 `shouldBe` Just ExitSuccess+ -- For slow systems where sleep may take longer than 1.5 seconds, do+ -- this in a loop.+ let loop 0 = error "Took too long for sleep to exit, your system is acting funny"+ loop i = do+ mec2 <- getStreamingProcessExitCode cph+ case mec2 of+ Nothing -> do+ threadDelay 500000+ loop (pred i)+ Just _ -> mec2 `shouldBe` Just ExitSuccess+ loop (5 :: Int) ec <- waitForStreamingProcess cph ec `shouldBe` ExitSuccess
test/Data/Conduit/TextSpec.hs view
@@ -1,69 +1,67 @@-{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts, OverloadedStrings #-} module Data.Conduit.TextSpec (spec) where +import Data.Conduit ((.|), runConduit, runConduitPure)+import Control.Exception (SomeException) import qualified Data.Conduit.Text as CT import qualified Data.Conduit as C-import qualified Data.Conduit.Lift as C+import Data.Conduit.Lift (runCatchC, catchCatchC)+import Data.Functor.Identity (runIdentity) import qualified Data.Conduit.List as CL import Test.Hspec import Test.Hspec.QuickCheck-import Data.Monoid-import Control.Monad.ST import qualified Data.Text as T import qualified Data.Text.Encoding as TE import qualified Data.Text.Encoding.Error as TEE import qualified Data.Text.Lazy.Encoding as TLE-import Data.Functor.Identity import Control.Arrow-import Control.Applicative-import Control.Monad.Trans.Resource import qualified Data.ByteString as S import qualified Data.Text.Lazy as TL import qualified Data.ByteString.Lazy as L-import Control.Monad.Trans.Resource (runExceptionT_)+import Control.Monad.Catch.Pure (runCatchT) spec :: Spec spec = describe "Data.Conduit.Text" $ do describe "text" $ do let go enc tenc tdec cenc = describe enc $ do- prop "single chunk" $ \chars -> runST $ runExceptionT_ $ do+ prop "single chunk" $ \chars -> do let tl = TL.pack chars lbs = tenc tl src = CL.sourceList $ L.toChunks lbs- ts <- src C.$= CT.decode cenc C.$$ CL.consume- return $ TL.fromChunks ts == tl- prop "many chunks" $ \chars -> runIdentity $ runExceptionT_ $ do+ ts <- runConduit $ src .| CT.decode cenc .| CL.consume+ TL.fromChunks ts `shouldBe` tl+ prop "many chunks" $ \chars -> do let tl = TL.pack chars lbs = tenc tl src = mconcat $ map (CL.sourceList . return . S.singleton) $ L.unpack lbs - ts <- src C.$= CT.decode cenc C.$$ CL.consume- return $ TL.fromChunks ts == tl+ ts <- runConduit $ src .| CT.decode cenc .| CL.consume+ TL.fromChunks ts `shouldBe` tl -- Check whether raw bytes are decoded correctly, in -- particular that Text decoding produces an error if -- and only if Conduit does.- prop "raw bytes" $ \bytes ->+ prop "raw bytes" $ \bytes -> do let lbs = L.pack bytes src = CL.sourceList $ L.toChunks lbs- etl = runException $ src C.$= CT.decode cenc C.$$ CL.consume tl' = tdec lbs- in case etl of+ etl = runConduit $ src .| CT.decode cenc .| CL.consume+ case etl of (Left _) -> (return $! TL.toStrict tl') `shouldThrow` anyException (Right tl) -> TL.fromChunks tl `shouldBe` tl'- prop "encoding" $ \chars -> runIdentity $ runExceptionT_ $ do+ prop "encoding" $ \chars -> do let tss = map T.pack chars lbs = tenc $ TL.fromChunks tss src = mconcat $ map (CL.sourceList . return) tss- bss <- src C.$= CT.encode cenc C.$$ CL.consume- return $ L.fromChunks bss == lbs- prop "valid then invalid" $ \x y chars -> runIdentity $ runExceptionT_ $ do+ bss <- runConduit $ src .| CT.encode cenc .| CL.consume+ L.fromChunks bss `shouldBe` lbs+ prop "valid then invalid" $ \x y chars -> do let tss = map T.pack ([x, y]:chars) ts = T.concat tss lbs = tenc (TL.fromChunks tss) `L.append` "\0\0\0\0\0\0\0" src = mapM_ C.yield $ L.toChunks lbs- Just x' <- src C.$$ CT.decode cenc C.=$ C.await- return $ x' `T.isPrefixOf` ts+ Just x' <- runConduit $ src .| CT.decode cenc .| C.await+ (x' `T.isPrefixOf` ts) `shouldBe` True go "utf8" TLE.encodeUtf8 TLE.decodeUtf8 CT.utf8 go "utf16_le" TLE.encodeUtf16LE TLE.decodeUtf16LE CT.utf16_le go "utf16_be" TLE.encodeUtf16BE TLE.decodeUtf16BE CT.utf16_be@@ -71,25 +69,25 @@ go "utf32_be" TLE.encodeUtf32BE TLE.decodeUtf32BE CT.utf32_be it "mixed utf16 and utf8" $ do let bs = "8\NUL:\NULu\NUL\215\216\217\218"- src = C.yield bs C.$= CT.decode CT.utf16_le- text <- src C.$$ C.await+ src = C.yield bs .| CT.decode CT.utf16_le+ text <- runConduit $ src .| C.await text `shouldBe` Just "8:u"- (src C.$$ CL.sinkNull) `shouldThrow` anyException+ (runConduit $ src .| CL.sinkNull) `shouldThrow` anyException it "invalid utf8" $ do let bs = S.pack [0..255]- src = C.yield bs C.$= CT.decode CT.utf8- text <- src C.$$ C.await+ src = C.yield bs .| CT.decode CT.utf8+ text <- runConduit $ src .| C.await text `shouldBe` Just (T.pack $ map toEnum [0..127])- (src C.$$ CL.sinkNull) `shouldThrow` anyException+ (runConduit $ src .| CL.sinkNull) `shouldThrow` anyException it "catch UTF8 exceptions" $ do let badBS = "this is good\128\128\0that was bad" grabExceptions inner = C.catchC- (inner C.=$= CL.map Right)+ (inner .| CL.map Right) (\e -> C.yield (Left (e :: CT.TextException))) - res <- C.yield badBS C.$$ (,)- <$> (grabExceptions (CT.decode CT.utf8) C.=$ CL.consume)+ res <- runConduit $ C.yield badBS .| (,)+ <$> (grabExceptions (CT.decode CT.utf8) .| CL.consume) <*> CL.consume first (map (either (Left . show) Right)) res `shouldBe`@@ -102,13 +100,13 @@ let badBS = "this is good\128\128\0that was bad" grabExceptions inner = do- res <- C.runCatchC $ inner C.=$= CL.map Right+ res <- runCatchC $ inner .| CL.map Right case res of Left e -> C.yield $ Left e Right () -> return () - let res = runIdentity $ C.yield badBS C.$$ (,)- <$> (grabExceptions (CT.decode CT.utf8) C.=$ CL.consume)+ let res = runConduitPure $ C.yield badBS .| (,)+ <$> (grabExceptions (CT.decode CT.utf8) .| CL.consume) <*> CL.consume first (map (either (Left . show) Right)) res `shouldBe`@@ -120,12 +118,12 @@ it "catch UTF8 exceptions, catchExceptionC" $ do let badBS = "this is good\128\128\0that was bad" - grabExceptions inner = C.catchCatchC- (inner C.=$= CL.map Right)+ grabExceptions inner = catchCatchC+ (inner .| CL.map Right) (\e -> C.yield $ Left e) - let res = runException_ $ C.yield badBS C.$$ (,)- <$> (grabExceptions (CT.decode CT.utf8) C.=$ CL.consume)+ let Right res = runIdentity $ runCatchT $ runConduit $ C.yield badBS .| (,)+ <$> (grabExceptions (CT.decode CT.utf8) .| CL.consume) <*> CL.consume first (map (either (Left . show) Right)) res `shouldBe`@@ -135,19 +133,20 @@ , ["\128\128\0that was bad"] ) it "catch UTF8 exceptions, catchExceptionC, decodeUtf8" $ do- let badBS = "this is good\128\128\0that was bad"+ let badBS = ["this is good", "\128\128\0that was bad"] - grabExceptions inner = C.catchCatchC- (inner C.=$= CL.map Right)+ grabExceptions inner = catchCatchC+ (inner .| CL.map Right) (\e -> C.yield $ Left e) - let res = runException_ $ C.yield badBS C.$$ (,)- <$> (grabExceptions CT.decodeUtf8 C.=$ CL.consume)+ let Right res = runIdentity $ runCatchT $ runConduit $+ mapM_ C.yield badBS .| (,)+ <$> (grabExceptions CT.decodeUtf8 .| CL.consume) <*> CL.consume - first (map (either (Left . show) Right)) res `shouldBe`+ first (map (either (Left . const ()) Right)) res `shouldBe` ( [ Right "this is good"- , Left $ show $ CT.NewDecodeException "UTF-8" 12 "\128\128\0t"+ , Left () ] , ["\128\128\0that was bad"] )@@ -155,78 +154,77 @@ let bss = [TE.encodeUtf8 $ T.pack good1, "\128\129\130", TE.encodeUtf8 $ T.pack good2] bs = S.concat bss expected = TE.decodeUtf8With TEE.lenientDecode bs- actual = runIdentity $ mapM_ C.yield bss C.$$ CT.decodeUtf8Lenient C.=$ CL.consume+ actual = runConduitPure $ mapM_ C.yield bss .| CT.decodeUtf8Lenient .| CL.consume T.concat actual `shouldBe` expected describe "text lines" $ do it "yields nothing given nothing" $- (CL.sourceList [] C.$= CT.lines C.$$ CL.consume) ==+ (runConduit $ CL.sourceList [] .| CT.lines .| CL.consume) == [[]] it "yields nothing given only empty text" $- (CL.sourceList [""] C.$= CT.lines C.$$ CL.consume) ==+ (runConduit $ CL.sourceList [""] .| CT.lines .| CL.consume) == [[]] it "works across split lines" $- (CL.sourceList ["abc", "d\nef"] C.$= CT.lines C.$$ CL.consume) ==+ (runConduit $ CL.sourceList ["abc", "d\nef"] .| CT.lines .| CL.consume) == [["abcd", "ef"]] it "works with multiple lines in an item" $- (CL.sourceList ["ab\ncd\ne"] C.$= CT.lines C.$$ CL.consume) ==+ (runConduit $ CL.sourceList ["ab\ncd\ne"] .| CT.lines .| CL.consume) == [["ab", "cd", "e"]] it "works with ending on a newline" $- (CL.sourceList ["ab\n"] C.$= CT.lines C.$$ CL.consume) ==+ (runConduit $ CL.sourceList ["ab\n"] .| CT.lines .| CL.consume) == [["ab"]] it "works with ending a middle item on a newline" $- (CL.sourceList ["ab\n", "cd\ne"] C.$= CT.lines C.$$ CL.consume) ==+ (runConduit $ CL.sourceList ["ab\n", "cd\ne"] .| CT.lines .| CL.consume) == [["ab", "cd", "e"]] it "works with empty text" $- (CL.sourceList ["ab", "", "cd"] C.$= CT.lines C.$$ CL.consume) ==+ (runConduit $ CL.sourceList ["ab", "", "cd"] .| CT.lines .| CL.consume) == [["abcd"]] it "works with empty lines" $- (CL.sourceList ["\n\n"] C.$= CT.lines C.$$ CL.consume) ==+ (runConduit $ CL.sourceList ["\n\n"] .| CT.lines .| CL.consume) == [["", ""]]- it "is not too eager" $ do- x <- CL.sourceList ["foobarbaz", error "ignore me"] C.$$ CT.decode CT.utf8 C.=$ CL.head- x `shouldBe` Just "foobarbaz" describe "text lines bounded" $ do it "yields nothing given nothing" $- (CL.sourceList [] C.$= CT.linesBounded 80 C.$$ CL.consume) ==+ (runConduit $ CL.sourceList [] .| CT.linesBounded 80 .| CL.consume) == [[]] it "yields nothing given only empty text" $- (CL.sourceList [""] C.$= CT.linesBounded 80 C.$$ CL.consume) ==+ (runConduit $ CL.sourceList [""] .| CT.linesBounded 80 .| CL.consume) == [[]] it "works across split lines" $- (CL.sourceList ["abc", "d\nef"] C.$= CT.linesBounded 80 C.$$ CL.consume) ==+ (runConduit $ CL.sourceList ["abc", "d\nef"] .| CT.linesBounded 80 .| CL.consume) == [["abcd", "ef"]] it "works with multiple lines in an item" $- (CL.sourceList ["ab\ncd\ne"] C.$= CT.linesBounded 80 C.$$ CL.consume) ==+ (runConduit $ CL.sourceList ["ab\ncd\ne"] .| CT.linesBounded 80 .| CL.consume) == [["ab", "cd", "e"]] it "works with ending on a newline" $- (CL.sourceList ["ab\n"] C.$= CT.linesBounded 80 C.$$ CL.consume) ==+ (runConduit $ CL.sourceList ["ab\n"] .| CT.linesBounded 80 .| CL.consume) `shouldBe` [["ab"]] it "works with ending a middle item on a newline" $- (CL.sourceList ["ab\n", "cd\ne"] C.$= CT.linesBounded 80 C.$$ CL.consume) ==+ (runConduit $ CL.sourceList ["ab\n", "cd\ne"] .| CT.linesBounded 80 .| CL.consume) `shouldBe` [["ab", "cd", "e"]] it "works with empty text" $- (CL.sourceList ["ab", "", "cd"] C.$= CT.linesBounded 80 C.$$ CL.consume) ==+ (runConduit $ CL.sourceList ["ab", "", "cd"] .| CT.linesBounded 80 .| CL.consume) `shouldBe` [["abcd"]] it "works with empty lines" $- (CL.sourceList ["\n\n"] C.$= CT.linesBounded 80 C.$$ CL.consume) ==+ (runConduit (CL.sourceList ["\n\n"] .| CT.linesBounded 80 .| CL.consume)) `shouldBe` [["", ""]]- it "is not too eager" $ do- x <- CL.sourceList ["foobarbaz", error "ignore me"] C.$$ CT.decode CT.utf8 C.=$ CL.head- x `shouldBe` Just "foobarbaz" it "throws an exception when lines are too long" $ do- x <- runExceptionT $ CL.sourceList ["hello\nworld"] C.$$ CT.linesBounded 4 C.=$ CL.consume+ let x :: Either SomeException [T.Text]+ x = runConduit $ CL.sourceList ["hello\nworld"] .| CT.linesBounded 4 .| CL.consume show x `shouldBe` show (Left $ CT.LengthExceeded 4 :: Either CT.TextException ()) it "works with infinite input" $ do- x <- runExceptionT $ CL.sourceList (cycle ["hello"]) C.$$ CT.linesBounded 256 C.=$ CL.consume+ let x :: Either SomeException [T.Text]+ x = runConduit $ CL.sourceList (cycle ["hello"]) .| CT.linesBounded 256 .| CL.consume show x `shouldBe` show (Left $ CT.LengthExceeded 256 :: Either CT.TextException ()) describe "text decode" $ do it' "doesn't throw runtime exceptions" $ do- let x = runIdentity $ runExceptionT $ C.yield "\x89\x243" C.$$ CT.decode CT.utf8 C.=$ CL.consume+ let x = runConduit $ C.yield "\x89\x243" .| CT.decode CT.utf8 .| CL.consume case x of Left _ -> return () Right t -> error $ "This should have failed: " ++ show t+ it "is not too eager" $ do+ x <- runConduit $ CL.sourceList ["foobarbaz", error "ignore me"] .| CT.decode CT.utf8 .| CL.head+ x `shouldBe` Just "foobarbaz" it' :: String -> IO () -> Spec it' = it
test/Data/Conduit/ZlibSpec.hs view
@@ -1,49 +1,92 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE OverloadedStrings #-} module Data.Conduit.ZlibSpec (spec) where import Test.Hspec import Test.Hspec.QuickCheck (prop) +import Data.Conduit ((.|), runConduit) import qualified Data.Conduit as C import qualified Data.Conduit.List as CL import qualified Data.Conduit.Zlib as CZ-import Control.Monad.ST (runST)-import Data.Monoid import qualified Data.ByteString as S import qualified Data.ByteString.Lazy as L+import Data.ByteString.Char8 () import Data.ByteString.Lazy.Char8 ()-import Control.Monad.Trans.Resource (runExceptionT_)-import Control.Monad.Trans.Class-import Control.Monad.Catch.Pure-import Control.Monad.Base--instance MonadBase base m => MonadBase base (CatchT m) where- liftBase = lift . liftBase+import Control.Monad (replicateM_) spec :: Spec spec = describe "Data.Conduit.Zlib" $ do- prop "idempotent" $ \bss' -> runST $ do+ prop "idempotent" $ \bss' -> do let bss = map S.pack bss' lbs = L.fromChunks bss src = mconcat $ map (CL.sourceList . return) bss- outBss <- runExceptionT_ $ src C.$= CZ.gzip C.$= CZ.ungzip C.$$ CL.consume- return $ lbs == L.fromChunks outBss- prop "flush" $ \bss' -> runST $ do+ outBss <- runConduit $ src .| CZ.gzip .| CZ.ungzip .| CL.consume+ L.fromChunks outBss `shouldBe` lbs+ prop "flush" $ \bss' -> do let bss = map S.pack $ filter (not . null) bss' bssC = concatMap (\bs -> [C.Chunk bs, C.Flush]) bss src = mconcat $ map (CL.sourceList . return) bssC- outBssC <- runExceptionT_- $ src C.$= CZ.compressFlush 5 (CZ.WindowBits 31)- C.$= CZ.decompressFlush (CZ.WindowBits 31)- C.$$ CL.consume- return $ bssC == outBssC+ outBssC <- runConduit+ $ src+ .| CZ.compressFlush 5 (CZ.WindowBits 31)+ .| CZ.decompressFlush (CZ.WindowBits 31)+ .| CL.consume+ outBssC `shouldBe` bssC it "compressFlush large data" $ do let content = L.pack $ map (fromIntegral . fromEnum) $ concat $ ["BEGIN"] ++ map show [1..100000 :: Int] ++ ["END"] src = CL.sourceList $ map C.Chunk $ L.toChunks content- bssC <- src C.$$ CZ.compressFlush 5 (CZ.WindowBits 31) C.=$ CL.consume+ bssC <- runConduit $ src .| CZ.compressFlush 5 (CZ.WindowBits 31) .| CL.consume let unChunk (C.Chunk x) = [x] unChunk C.Flush = []- bss <- CL.sourceList bssC C.$$ CL.concatMap unChunk C.=$ CZ.ungzip C.=$ CL.consume+ bss <- runConduit $ CL.sourceList bssC .| CL.concatMap unChunk .| CZ.ungzip .| CL.consume L.fromChunks bss `shouldBe` content++ it "uncompressed after compressed" $ do+ let c = "This data is stored compressed."+ u = "This data isn't."+ let src1 = do+ C.yield c .| CZ.gzip+ C.yield u+ encoded <- runConduit $ src1 .| CL.consume+ let src2 = mapM_ C.yield encoded+ (c', u') <- runConduit $ src2 .| do+ c' <- CZ.ungzip .| CL.consume+ u' <- CL.consume+ return (S.concat c', S.concat u')+ c' `shouldBe` c+ u' `shouldBe` u++ it "multiple compressed values" $ do+ let s1 = "hello"+ s2 = "world"+ src = do+ C.yield s1 .| CZ.gzip+ C.yield s2 .| CZ.gzip+ actual <- runConduit $ src .| CZ.multiple CZ.ungzip .| CL.consume+ S.concat actual `shouldBe` S.concat [s1, s2]++ it "single compressed, multiple uncompressed chunks" $ do+ let s1 = "hello"+ s2 = "there"+ s3 = "world"+ s1Z <- fmap S.concat $ runConduit $ C.yield s1 .| CZ.gzip .| CL.consume+ let src = do+ C.yield $ S.append s1Z s2+ C.yield s3+ actual <- runConduit $ src .| do+ x <- fmap S.concat $ CZ.ungzip .| CL.consume+ y <- CL.consume+ return (x, y)+ actual `shouldBe` (s1, [s2, s3])++ it "multiple, over 32k" $ do+ let str = "One line"+ cnt = 30000+ src = replicateM_ cnt $ C.yield str .| CZ.gzip+ actual <- fmap S.concat $ runConduit $ src .| CZ.multiple CZ.ungzip .| CL.consume+ let expected = S.concat (replicate cnt str)+ S.length actual `shouldBe` S.length expected+ actual `shouldBe` expected