conduit-extra 1.1.17 → 1.3.8
raw patch · 27 files changed
Files
- ChangeLog.md +76/−0
- Data/Conduit/Attoparsec.hs +32/−30
- Data/Conduit/Binary.hs +55/−217
- Data/Conduit/Blaze.hs +0/−103
- Data/Conduit/ByteString/Builder.hs +9/−122
- Data/Conduit/Filesystem.hs +5/−62
- Data/Conduit/Foldl.hs +2/−2
- Data/Conduit/Lazy.hs +30/−25
- Data/Conduit/Network.hs +28/−25
- Data/Conduit/Network/UDP.hs +6/−6
- Data/Conduit/Process.hs +85/−32
- Data/Conduit/Process/Typed.hs +107/−0
- Data/Conduit/Text.hs +42/−100
- Data/Conduit/Zlib.hs +22/−23
- bench/blaze.hs +19/−26
- conduit-extra.cabal +17/−20
- test/Data/Conduit/AttoparsecSpec.hs +49/−42
- test/Data/Conduit/BinarySpec.hs +51/−38
- test/Data/Conduit/ByteString/BuilderSpec.hs +12/−15
- 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 +1/−1
- test/Data/Conduit/Process/TypedSpec.hs +31/−0
- test/Data/Conduit/ProcessSpec.hs +12/−3
- test/Data/Conduit/TextSpec.hs +67/−66
- test/Data/Conduit/ZlibSpec.hs +23/−29
ChangeLog.md view
@@ -1,3 +1,79 @@+# 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.
Data/Conduit/Attoparsec.hs view
@@ -38,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@@ -53,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@@ -74,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@@ -87,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@@ -99,29 +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.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 -> Consumer a m (Either ParseError b)-sinkParserEither = (fmap.fmap) snd . sinkParserPos (Position 1 1)+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@@ -129,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 1+ conduit $ Position 1 1 0 where conduit !pos = await >>= maybe (return ()) go where@@ -143,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 () #-} @@ -156,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 1+ conduit $ Position 1 1 0 where conduit !pos = await >>= maybe (return ()) go where@@ -173,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 () #-} @@ -186,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 #-} @@ -198,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@@ -232,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,8 +1,11 @@ {-# LANGUAGE CPP, RankNTypes #-} {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE ScopedTypeVariables #-}--- | Functions for interacting with bytes.+-- | /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@@ -13,20 +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- , sinkFileCautious- , sinkTempFile- , sinkSystemTempFile- , 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@@ -50,7 +59,7 @@ , 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)@@ -58,9 +67,9 @@ 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)@@ -69,117 +78,19 @@ #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, castPtr) import Foreign.Storable (Storable, peek, sizeOf)-import GHC.ForeignPtr (mallocPlainForeignPtrBytes) import Control.Monad.Trans.Resource (MonadResource)-import Control.Monad.Catch (MonadThrow (..)) import Control.Exception (Exception) import Data.Typeable (Typeable) import Foreign.Ptr (Ptr) #ifndef ALLOW_UNALIGNED_ACCESS import Foreign.Marshal (alloca, copyBytes) #endif-import System.Directory (renameFile)-import System.FilePath (takeDirectory, takeFileName, (<.>))-import System.IO (hClose, openBinaryTempFile)-import Control.Exception (throwIO, catch)-import System.IO.Error (isDoesNotExistError) --- | 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 close 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 (liftIO . S.hPut 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- -- | Stream the contents of a file as binary data, starting from a certain -- offset and only consuming up to a certain number of bytes. --@@ -188,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@@ -202,7 +113,7 @@ => IO.Handle -> Maybe Integer -- ^ Offset -> Maybe Integer -- ^ Maximum count- -> Producer m S.ByteString+ -> ConduitT i S.ByteString m () sourceHandleRange handle offset count = sourceHandleRangeWithBuffer handle offset count defaultChunkSize @@ -216,7 +127,7 @@ -> Maybe Integer -- ^ Offset -> Maybe Integer -- ^ Maximum count -> Int -- ^ Buffer size- -> Producer m S.ByteString+ -> ConduitT i S.ByteString m () sourceHandleRangeWithBuffer handle offset count buffer = do case offset of Nothing -> return ()@@ -243,86 +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)---- | Cautious version of 'sinkFile'. The idea here is to stream the--- values to a temporary file in the same directory of the destination--- file, and only on successfully writing the entire file, moves it--- atomically to the destination path.------ In the event of an exception occurring, the temporary file will be--- deleted and no move will be made. If the application shuts down--- without running exception handling (such as machine failure or a--- SIGKILL), the temporary file will remain and the destination file--- will be untouched.------ @since 1.1.14-sinkFileCautious- :: MonadResource m- => FilePath- -> ConduitM S.ByteString o m ()-sinkFileCautious fp =- bracketP acquire cleanup inner- where- acquire = openBinaryTempFile (takeDirectory fp) (takeFileName fp <.> "tmp")- cleanup (tmpFP, h) = do- hClose h- removeFile tmpFP `Control.Exception.catch` \e ->- if isDoesNotExistError e- then return ()- else throwIO e- inner (tmpFP, h) = do- sinkHandle h- liftIO $ do- hClose h- renameFile tmpFP fp---- | Stream data into a temporary file in the given directory with the--- given filename pattern, and return the temporary filename. The--- temporary file will be automatically deleted when exiting the--- active 'ResourceT' block, if it still exists.------ @since 1.1.15-sinkTempFile :: MonadResource m- => FilePath -- ^ temp directory- -> String -- ^ filename pattern- -> ConduitM ByteString o m FilePath-sinkTempFile tmpdir pattern = do- (_releaseKey, (fp, h)) <- allocate- (IO.openBinaryTempFile tmpdir pattern)- (\(fp, h) -> IO.hClose h `finally` (removeFile fp `Control.Exception.catch` \e ->- if isDoesNotExistError e- then return ()- else throwIO e))- sinkHandle h- liftIO $ IO.hClose h- return fp---- | Same as 'sinkTempFile', but will use the default temp file--- directory for the system as the first argument.------ @since 1.1.15-sinkSystemTempFile- :: MonadResource m- => String -- ^ filename pattern- -> ConduitM ByteString o m FilePath-sinkSystemTempFile pattern = do- dir <- liftIO getTemporaryDirectory- sinkTempFile dir pattern- -- | 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@@ -333,7 +171,7 @@ -- 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 consumed by the inner@@ -343,7 +181,7 @@ -- Since 0.3.0 isolate :: Monad m => Int- -> Conduit S.ByteString m S.ByteString+ -> ConduitT S.ByteString S.ByteString m () isolate = loop where@@ -363,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@@ -376,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@@ -392,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@@ -407,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@@ -425,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@@ -444,7 +282,7 @@ -- (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 [] where@@ -464,7 +302,7 @@ -- | 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@@ -475,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@@ -483,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@@ -500,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 ()@@ -508,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 #-}@@ -518,7 +356,7 @@ -- | 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_ #-} @@ -527,7 +365,7 @@ -- all unused input as leftovers. -- -- @since 1.1.13-sinkStorable :: (Monad m, Storable a) => Consumer S.ByteString m (Maybe a)+sinkStorable :: (Monad m, Storable a) => ConduitT S.ByteString o m (Maybe a) sinkStorable = sinkStorableHelper Just (return Nothing) -- | Same as 'sinkStorable', but throws a 'SinkStorableInsufficientBytes'@@ -536,13 +374,13 @@ -- construct/deconstruct a @Maybe@ wrapper in the success case. -- -- @since 1.1.13-sinkStorableEx :: (MonadThrow m, Storable a) => Consumer S.ByteString m a+sinkStorableEx :: (MonadThrow m, Storable a) => ConduitT S.ByteString o m a sinkStorableEx = sinkStorableHelper id (throwM SinkStorableInsufficientBytes) -sinkStorableHelper :: forall m a b. (Monad m, Storable a)+sinkStorableHelper :: forall m a b o. (Monad m, Storable a) => (a -> b)- -> (Consumer S.ByteString m b)- -> Consumer S.ByteString m b+ -> (ConduitT S.ByteString o m b)+ -> ConduitT S.ByteString o m b sinkStorableHelper wrap failure = do start where@@ -561,13 +399,13 @@ -- 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+ let bs' = S.concat $ L.toChunks lbs+ case compare (S.length bs') size of LT -> do- leftover bs+ leftover bs' failure- EQ -> process bs- GT -> assert False (process bs)+ EQ -> process bs'+ GT -> assert False (process bs') EQ -> process bs GT -> do let (x, y) = S.splitAt size bs@@ -575,7 +413,7 @@ process x -- Given a bytestring of exactly the correct size, grab the value- process bs = return $! wrap $! inlinePerformIO $!+ process bs = return $! wrap $! accursedUnutterablePerformIO $! unsafeUseAsCString bs (safePeek undefined . castPtr) safePeek :: a -> Ptr a -> IO a
− Data/Conduit/Blaze.hs
@@ -1,103 +0,0 @@--- | 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@.------ Since 1.1.7.0, the functions here call their counterparts in--- "Data.Conduit.ByteString.Builder", which work with both--- 'Data.ByteString.Builder.Builder' and blaze-builder 0.3's--- 'Blaze.ByteString.Builder.Builder'.-module Data.Conduit.Blaze- (-- -- * Conduits from builders to bytestrings- builderToByteString- , unsafeBuilderToByteString- , builderToByteStringWith-- -- ** Flush- , builderToByteStringFlush- , builderToByteStringWithFlush-- -- * Buffers- , B.Buffer-- -- ** Status information- , B.freeSize- , B.sliceSize- , B.bufferSize-- -- ** Creation and modification- , B.allocBuffer- , B.reuseBuffer- , B.nextSlice-- -- ** Conversion to bytestings- , B.unsafeFreezeBuffer- , B.unsafeFreezeNonEmptyBuffer-- -- * Buffer allocation strategies- , B.BufferAllocStrategy- , B.allNewBuffersStrategy- , B.reuseBufferStrategy- ) where--import Data.Conduit--import qualified Data.ByteString as S--import Blaze.ByteString.Builder (Builder)-import Control.Monad.Primitive (PrimMonad)-import Control.Monad.Base (MonadBase)-import Data.Streaming.Blaze--import qualified Data.Conduit.ByteString.Builder as B---- | Incrementally execute builders and pass on the filled chunks as--- bytestrings.-builderToByteString :: (MonadBase base m, PrimMonad base) => Conduit Builder m S.ByteString-builderToByteString = B.builderToByteString-{-# INLINE builderToByteString #-}---- |------ Since 0.0.2-builderToByteStringFlush :: (MonadBase base m, PrimMonad base) => Conduit (Flush Builder) m (Flush S.ByteString)-builderToByteStringFlush = B.builderToByteStringFlush-{-# INLINE builderToByteStringFlush #-}---- | 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 = B.unsafeBuilderToByteString-{-# INLINE unsafeBuilderToByteString #-}----- | 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 = B.builderToByteStringWith-{-# INLINE builderToByteStringWith #-}---- |------ Since 0.0.2-builderToByteStringWithFlush- :: (MonadBase base m, PrimMonad base)- => BufferAllocStrategy- -> Conduit (Flush Builder) m (Flush S.ByteString)-builderToByteStringWithFlush = B.builderToByteStringWithFlush-{-# INLINE builderToByteStringWithFlush #-}
Data/Conduit/ByteString/Builder.hs view
@@ -17,131 +17,18 @@ ( -- * Conduits from builders to bytestrings- builderToByteString- , unsafeBuilderToByteString- , builderToByteStringWith+ CC.builderToByteString+ , CC.unsafeBuilderToByteString+ , CC.builderToByteStringWith -- ** Flush- , builderToByteStringFlush- , builderToByteStringWithFlush-- -- * Buffers- , Buffer-- -- ** Status information- , freeSize- , sliceSize- , bufferSize-- -- ** Creation and modification- , allocBuffer- , reuseBuffer- , nextSlice-- -- ** Conversion to bytestings- , unsafeFreezeBuffer- , unsafeFreezeNonEmptyBuffer+ , CC.builderToByteStringFlush+ , CC.builderToByteStringWithFlush -- * Buffer allocation strategies- , BufferAllocStrategy- , allNewBuffersStrategy- , reuseBufferStrategy+ , CC.BufferAllocStrategy+ , CC.allNewBuffersStrategy+ , CC.reuseBufferStrategy ) where -import Data.Conduit-import Control.Monad (unless, liftM)-import Control.Monad.Trans.Class (lift, MonadTrans)--import qualified Data.ByteString as S--import Control.Monad.Primitive (PrimMonad, unsafePrimToPrim)-import Control.Monad.Base (MonadBase, liftBase)-import Data.Streaming.ByteString.Builder.Class--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, StreamingBuilder b)- => Conduit b m S.ByteString-builderToByteString =- builderToByteStringWith defaultStrategy-{-# INLINE builderToByteString #-}---- |------ Since 0.0.2-builderToByteStringFlush :: (MonadBase base m, PrimMonad base, StreamingBuilder b)- => Conduit (Flush b) m (Flush S.ByteString)-builderToByteStringFlush =- builderToByteStringWithFlush defaultStrategy-{-# INLINE builderToByteStringFlush #-}---- | 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, StreamingBuilder b)- => IO Buffer -- action yielding the inital buffer.- -> Conduit b m S.ByteString-unsafeBuilderToByteString = builderToByteStringWith . reuseBufferStrategy-{-# INLINE unsafeBuilderToByteString #-}----- | 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, StreamingBuilder b)- => BufferAllocStrategy- -> Conduit b m S.ByteString-builderToByteStringWith =- helper (liftM (fmap Chunk) await) yield'- where- yield' Flush = return ()- yield' (Chunk bs) = yield bs-{-# INLINE builderToByteStringWith #-}---- |------ Since 0.0.2-builderToByteStringWithFlush- :: (MonadBase base m, PrimMonad base, StreamingBuilder b)- => BufferAllocStrategy- -> Conduit (Flush b) m (Flush S.ByteString)-builderToByteStringWithFlush = helper await yield-{-# INLINE builderToByteStringWithFlush #-}--helper :: (MonadBase base m, PrimMonad base, Monad (t m), MonadTrans t, StreamingBuilder b)- => t m (Maybe (Flush b))- -> (Flush S.ByteString -> t m ())- -> BufferAllocStrategy- -> t m ()-helper await' yield' strat = do- (recv, finish) <- lift $ unsafeLiftIO $ newBuilderRecv 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 -> builderFlush- 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-{-# INLINE helper #-}+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
@@ -13,7 +13,7 @@ -- <https://hackage.haskell.org/package/foldl foldl> package. -- -- @since 1.1.16-sinkFold :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Consumer a m b+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@@ -21,6 +21,6 @@ -- <https://hackage.haskell.org/package/foldl foldl> package. -- -- @since 1.1.16-sinkFoldM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> Consumer a m b+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,6 +1,7 @@ {-# 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@@ -13,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 )@@ -48,25 +50,26 @@ -- state has been closed. -- -- Since 0.3.0-lazyConsume :: (MonadBaseControl IO m, MonadActive m) => Source m a -> m [a]-lazyConsume =-#if MIN_VERSION_conduit(1, 2, 0)- go . flip unConduitM Done-#else- go . unConduitM-#endif- 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@@ -99,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)@@ -114,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
@@ -46,18 +46,18 @@ import qualified GHC.Conc as Conc (yield) import qualified Data.ByteString as S import Control.Monad.IO.Class (MonadIO (liftIO))-import Control.Monad (unless, void)-import Control.Monad.Trans.Control (MonadBaseControl, control, liftBaseWith)+import Control.Monad (unless) import Control.Monad.Trans.Class (lift) 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@@ -72,7 +72,7 @@ -- 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@@ -84,7 +84,7 @@ 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@@ -95,7 +95,7 @@ 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@@ -112,15 +112,16 @@ -- connections. Will return the thread id of the server -- -- Since 1.1.4-forkTCPServer :: MonadBaseControl IO m- => SN.ServerSettings- -> (SN.AppData -> m ())- -> m ThreadId+forkTCPServer+ :: MonadUnliftIO m+ => SN.ServerSettings+ -> (SN.AppData -> m ())+ -> m ThreadId forkTCPServer set f =- liftBaseWith $ \run -> do+ withRunInIO $ \run -> do isBound <- newEmptyMVar let setWithWaitForBind = addBoundSignal isBound set- threadId <- forkIO . void . run $ runGeneralTCPServer setWithWaitForBind f+ threadId <- forkIO . run $ runGeneralTCPServer setWithWaitForBind f takeMVar isBound return threadId @@ -129,7 +130,7 @@ -- | 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@@ -137,21 +138,23 @@ -- variables. -- -- Since 1.1.3-runGeneralTCPServer :: MonadBaseControl IO m- => SN.ServerSettings- -> (SN.AppData -> m ())- -> m a-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
@@ -27,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@@ -39,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.@@ -47,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.@@ -57,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.@@ -67,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/Process.hs view
@@ -6,6 +6,9 @@ -- | 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@@ -18,6 +21,9 @@ , sourceCmdWithStreams , sourceProcessWithStreams , withCheckedProcessCleanup+ -- * InputSource types+ , FlushInput(..)+ , BuilderInput(..) -- * Reexport , module Data.Streaming.Process ) where@@ -25,26 +31,51 @@ 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.Monad.Catch (MonadMask, onException, throwM, finally, bracket)+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,@@ -55,14 +86,16 @@ -- 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 -- ^ stdout+ -> 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)@@ -70,10 +103,12 @@ -- | Like @sourceProcessWithConsumer@ but providing the command to be run as -- a @String@. --+-- Requires the threaded runtime.+-- -- Since 1.1.2 sourceCmdWithConsumer :: MonadIO m => String -- ^command- -> Consumer ByteString m a -- ^stdout+ -> ConduitT ByteString Void m a -- ^stdout -> m (ExitCode, a) sourceCmdWithConsumer cmd = sourceProcessWithConsumer (shell cmd) @@ -93,63 +128,81 @@ -- 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 :: CreateProcess- -> Producer IO ByteString -- ^stdin- -> Consumer ByteString IO a -- ^stdout- -> Consumer ByteString IO b -- ^stderr- -> IO (ExitCode, a, b)-sourceProcessWithStreams cp producerStdin consumerStdout consumerStderr = do+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 ((producerStdin $$ sinkStdin) `finally` closeStdin)- <*> Concurrently (sourceStdout $$ consumerStdout)- <*> Concurrently (sourceStderr $$ consumerStderr))+ <$> 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 :: String -- ^command- -> Producer IO ByteString -- ^stdin- -> Consumer ByteString IO a -- ^stdout- -> Consumer ByteString IO b -- ^stderr- -> IO (ExitCode, a, b)+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- , MonadIO m- , MonadMask m+ , MonadUnliftIO m ) => CreateProcess -> (stdin -> stdout -> stderr -> m b) -> m b-withCheckedProcessCleanup cp f = bracket+withCheckedProcessCleanup cp f = withRunInIO $ \run -> bracket (streamingProcess cp) (\(_, _, _, sph) -> closeStreamingProcessHandle sph) $ \(x, y, z, sph) -> do- res <- f x y z `onException` liftIO (terminateStreamingProcess sph)+ res <- run (f x y z) `onException` terminateStreamingProcess sph ec <- waitForStreamingProcess sph if ec == ExitSuccess then return res- else throwM $ ProcessExitedUnsuccessfully cp ec+ else throwIO $ ProcessExitedUnsuccessfully cp ec -terminateStreamingProcess :: StreamingProcessHandle -> IO ()-terminateStreamingProcess = terminateProcess . streamingProcessHandleRaw+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,5 +1,7 @@ {-# 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 --@@ -32,9 +34,9 @@ , drop , foldLines , withLine- , Data.Conduit.Text.decodeUtf8- , decodeUtf8Lenient- , encodeUtf8+ , CC.decodeUtf8+ , CC.decodeUtf8Lenient+ , CC.encodeUtf8 , detectUtf ) where @@ -51,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 @@ -84,21 +87,23 @@ -- | 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:) @@ -111,7 +116,7 @@ -- 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@@ -123,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'@@ -136,19 +141,19 @@ -- 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+ -> ConduitT B.ByteString T.Text m () decodeNew onFailure _name = loop where@@ -172,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@@ -201,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@@ -211,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@@ -317,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@@ -333,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@@ -347,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@@ -364,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@@ -382,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@@ -400,71 +388,25 @@ -- -- 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.------ 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)- 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 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-- 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 #-}---- | 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 #-}- -- | 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.1.9-detectUtf :: MonadThrow m => Conduit B.ByteString m T.Text+detectUtf :: MonadThrow m => ConduitT B.ByteString T.Text m () detectUtf = go id where
Data/Conduit/Zlib.hs view
@@ -21,20 +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:@@ -42,9 +41,9 @@ -- > 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' leftover where@@ -53,12 +52,12 @@ -- | 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)+ -> 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 ())@@ -127,7 +126,7 @@ yield' (Chunk bs) pop -- An error occurred inside zlib, throw it- PRError e -> lift $ monadThrow e+ PRError e -> lift $ throwM e -- We've been asked to flush the stream Just Flush -> do -- Get any uncompressed data waiting for us@@ -142,10 +141,10 @@ -- 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@@ -154,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@@ -180,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@@ -192,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 @@ -201,7 +200,7 @@ 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@@ -221,8 +220,8 @@ -- -- @since 1.1.10 multiple :: Monad m- => Conduit ByteString m a- -> Conduit ByteString m a+ => ConduitT ByteString a m ()+ -> ConduitT ByteString a m () multiple inner = loop where
bench/blaze.hs view
@@ -1,20 +1,16 @@ {-# LANGUAGE OverloadedStrings #-} import Data.Conduit import qualified Data.Conduit.List as CL-import Data.Conduit.Blaze-import Criterion.Main-import Blaze.ByteString.Builder+import Data.Conduit.ByteString.Builder+import Gauge.Main import Data.Monoid-import qualified Data.ByteString.Builder as BS-import Data.Functor.Identity (runIdentity)-import Control.Monad.ST (runST)-import Data.ByteString.Lazy.Internal (defaultChunkSize)+import Data.ByteString.Builder count :: Int count = 100000 single :: Builder-single = copyByteString "Hello World!\n"+single = shortByteString "Hello World!\n" oneBuilderLeft :: Builder oneBuilderLeft =@@ -30,43 +26,40 @@ loop 0 b = b loop i b = loop (i - 1) (b <> single) -builderSource :: Monad m => Source m Builder+builderSource :: Monad m => ConduitT i Builder m () builderSource = CL.replicate count single -singleBS :: BS.Builder-singleBS = BS.shortByteString "Hello World!\n"--oneBSBuilderLeft :: BS.Builder+oneBSBuilderLeft :: Builder oneBSBuilderLeft = loop count mempty where loop 0 b = b- loop i b = loop (i - 1) (b <> singleBS)+ loop i b = loop (i - 1) (b <> single) -oneBSBuilderRight :: BS.Builder+oneBSBuilderRight :: Builder oneBSBuilderRight = loop count mempty where loop 0 b = b- loop i b = loop (i - 1) (b <> singleBS)+ loop i b = loop (i - 1) (b <> single) -builderBSSource :: Monad m => Source m BS.Builder-builderBSSource = CL.replicate count singleBS+builderBSSource :: Monad m => ConduitT i Builder m ()+builderBSSource = CL.replicate count single main :: IO () main = defaultMain- [ bench "conduit, strict, safe" $ whnfIO $- builderSource $$ builderToByteString =$ CL.sinkNull- , bench "conduit, strict, unsafe" $ whnfIO $- builderSource $$ unsafeBuilderToByteString (allocBuffer defaultChunkSize) =$ CL.sinkNull+ [ 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 $ runIdentity $ src $$ CL.fold (<>) mempty+ toLazyByteString $ runConduitPure $ src .| CL.fold (<>) mempty - , bench "one bs builder, left" $ nf BS.toLazyByteString oneBSBuilderLeft- , bench "one bs builder, right" $ nf BS.toLazyByteString oneBSBuilderRight+ , bench "one bs builder, left" $ nf toLazyByteString oneBSBuilderLeft+ , bench "one bs builder, right" $ nf toLazyByteString oneBSBuilderRight , bench "conduit BS, lazy" $ flip nf builderBSSource $ \src ->- BS.toLazyByteString $ runIdentity $ src $$ CL.fold (<>) mempty+ toLazyByteString $ runConduitPure $ src .| CL.fold (<>) mempty ]
conduit-extra.cabal view
@@ -1,5 +1,6 @@+Cabal-version: >=1.10 Name: conduit-extra-Version: 1.1.17+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,7 +10,6 @@ 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@@ -19,40 +19,34 @@ 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 if arch(x86_64) || arch(i386) -- These architectures are able to perform unaligned memory accesses cpp-options: -DALLOW_UNALIGNED_ACCESS - Build-depends: base >= 4.5 && < 5- , conduit >= 1.1 && < 1.3+ Build-depends: base >= 4.12 && < 5+ , conduit >= 1.3 && < 1.4 - -- No version bounds necessary, since they're inherited- -- from conduit.- , bytestring- , exceptions- , monad-control+ , bytestring >= 0.10.2 , text , transformers- , transformers-base , async , attoparsec >= 0.10- , blaze-builder >= 0.3 , directory , filepath , network >= 2.3@@ -60,12 +54,15 @@ , process , resourcet >= 1.1 , stm- , streaming-commons >= 0.1.16+ , 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@@ -77,8 +74,6 @@ , async , attoparsec- , blaze-builder- , bytestring-builder , bytestring , exceptions , process@@ -90,6 +85,8 @@ , transformers , transformers-base , directory+ , filepath+ build-tool-depends: hspec-discover:hspec-discover ghc-options: -Wall if os(windows) cpp-options: -DWINDOWS@@ -101,19 +98,19 @@ 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- , blaze-builder , conduit , conduit-extra- , criterion+ , gauge , bytestring- , bytestring-builder , transformers main-is: blaze.hs ghc-options: -Wall -O2 -rtsopts
test/Data/Conduit/AttoparsecSpec.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE BangPatterns #-} {-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TupleSections #-}@@ -8,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@@ -22,148 +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 sink' = sinkParserEither parser- ea <- runExceptionT $ CL.sourceList input $$ sink+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol- ea' <- CL.sourceList input $$ sink'+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink' case ea' of Left pe ->- errorPosition pe `shouldBe` Position badLine badCol+ 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 sink' = sinkParserEither parser- ea <- runExceptionT $ CL.sourceList input $$ sink+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol- ea' <- CL.sourceList input $$ sink'+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink' case ea' of Left pe ->- errorPosition pe `shouldBe` Position badLine badCol+ 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 sink' = sinkParserEither parser- ea <- runExceptionT $ CL.sourceList input $$ sink+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol- ea' <- CL.sourceList input $$ sink'+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink' case ea' of Left pe ->- errorPosition pe `shouldBe` Position badLine badCol+ 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 sink' = sinkParserEither parser- ea <- runExceptionT $ CL.sourceList input $$ sink+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol- ea' <- CL.sourceList input $$ sink'+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink' case ea' of Left pe ->- errorPosition pe `shouldBe` Position badLine badCol+ 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 sink' = sinkParserEither parser- ea <- runExceptionT $ CL.sourceList input $$ sink+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol- ea' <- CL.sourceList input $$ sink'+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink' case ea' of Left pe ->- errorPosition pe `shouldBe` Position badLine badCol+ 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 sink' = sinkParserEither parser- ea <- runExceptionT $ CL.sourceList input $$ sink+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol- ea' <- CL.sourceList input $$ sink'+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink' case ea' of Left pe ->- errorPosition pe `shouldBe` Position badLine badCol+ 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 <- runExceptionT $ CL.sourceList input $$ sink+ ea = runConduit $ CL.sourceList input .| sink case ea of Left e -> case fromException e of Just pe -> do- errorPosition pe `shouldBe` Position badLine badCol- ea' <- CL.sourceList input $$ sink'+ errorPosition pe `shouldBe` Position badLine badCol badOff+ ea' <- runConduit $ CL.sourceList input .| sink' case ea' of Left pe ->- errorPosition pe `shouldBe` Position badLine badCol+ 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 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 <- yield "hihihi\nhihi"- $$ conduitParser (Data.Attoparsec.Text.string "\n" <|> Data.Attoparsec.Text.string "hi")- =$ CL.consume- let f (a, b, c, d, e) = (PositionRange (Position a b) (Position c d), e)+ 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, 1, 3, "hi")- , (1, 3, 1, 5, "hi")- , (1, 5, 1, 7, "hi")+ [ (1, 1, 0, 1, 3, 2, "hi")+ , (1, 3, 2, 1, 5, 4, "hi")+ , (1, 5, 4, 1, 7, 6, "hi") - , (1, 7, 2, 1, "\n")+ , (1, 7, 6, 2, 1, 7, "\n") - , (2, 1, 2, 3, "hi")- , (2, 3, 2, 5, "hi")+ , (2, 1, 7, 2, 3, 9, "hi")+ , (2, 3, 9, 2, 5, 11, "hi") ]
test/Data/Conduit/BinarySpec.hs view
@@ -3,6 +3,7 @@ {-# 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@@ -26,7 +27,8 @@ import Foreign.Ptr (alignPtr, minusPtr) import System.Directory (doesFileExist) import System.IO.Unsafe (unsafePerformIO)-import Control.Applicative ((<$>), (<*>))+import Control.Applicative ((<$>))+import qualified Data.ByteString.Builder as BB spec :: Spec spec = describe "Data.Conduit.Binary" $ do@@ -34,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"@@ -59,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 @@ -81,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@@ -105,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)@@ -145,14 +157,14 @@ 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@@ -160,16 +172,16 @@ describe "sinkFileCautious" $ do it' "success" $ do- runResourceT $ CB.sourceFile "conduit-extra.cabal" C.$$ CB.sinkFileCautious "tmp"+ 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- runResourceT+ runConduitRes ( (CB.sourceFile "conduit-extra.cabal" >> error "FIXME")- C.$$ CB.sinkFileCautious "tmp")+ .| CB.sinkFileCautious "tmp") `shouldThrow` anyException bs2 <- S.readFile "tmp" bs2 `shouldBe` bs1@@ -177,10 +189,11 @@ it "sinkSystemTempFile" $ do let bs = "Hello World!" fp <- runResourceT $ do- fp <- C.yield bs C.$$ CB.sinkSystemTempFile "temp-file-test"- actual <- liftIO $ S.readFile fp- liftIO $ actual `shouldBe` bs- return fp+ 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 @@ -189,7 +202,7 @@ 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@@ -201,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)@@ -212,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@@ -222,9 +235,9 @@ 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@@ -242,7 +255,7 @@ loop y sink :: [SomeStorable]- -> C.Sink S.ByteString IO ()+ -> ConduitT S.ByteString Void IO () sink [] = do mw <- CB.head case mw of@@ -254,7 +267,7 @@ checkOne :: (Storable a, Eq a, Show a) => a- -> C.Sink S.ByteString IO ()+ -> ConduitT S.ByteString Void IO () checkOne expected = do mactual <- if func@@ -266,7 +279,7 @@ Just actual -> return actual liftIO $ actual `shouldBe` expected - src C.$$ sink stores0 :: IO ()+ runConduit $ src .| sink stores0 :: IO () mapM_ test' [1, 5, 10, 100] test "sink Maybe" True@@ -274,7 +287,7 @@ it' "insufficient bytes are leftovers, one chunk" $ do let src = C.yield $ S.singleton 1- src C.$$ do+ runConduit $ src .| do mactual <- CB.sinkStorable liftIO $ mactual `shouldBe` (Nothing :: Maybe Int) lbs <- CB.sinkLbs@@ -284,7 +297,7 @@ let src = do C.yield $ S.singleton 1 C.yield $ S.singleton 2- src C.$$ do+ runConduit $ src .| do mactual <- CB.sinkStorable liftIO $ mactual `shouldBe` (Nothing :: Maybe Int) lbs <- CB.sinkLbs
test/Data/Conduit/ByteString/BuilderSpec.hs view
@@ -4,17 +4,14 @@ import Test.Hspec import Test.Hspec.QuickCheck (prop) -import qualified Data.Conduit as C+import Data.Conduit ((.|), runConduit, Flush (..)) import qualified Data.Conduit.List as CL-import Data.ByteString.Char8 () import Data.Conduit.ByteString.Builder (builderToByteString, builderToByteStringFlush) import Control.Monad.ST (runST)-import Data.Monoid 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-import Data.ByteString.Lazy.Char8 () spec :: Spec spec =@@ -24,41 +21,41 @@ let builders = map byteString bss let lbs = toLazyByteString $ mconcat builders let src = mconcat $ map (CL.sourceList . return) builders- outBss <- src C.$= builderToByteString C.$$ CL.consume+ 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 <- src C.$= builderToByteString C.$$ CL.consume+ 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 <- src C.$= builderToByteString C.$$ CL.consume+ 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 C.$= CL.map ((`mappend` flush) . byteString)- out <- src C.$= builderToByteString C.$$ CL.consume+ 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 -> [C.Chunk $ S.pack bs, C.Flush]) $ filter (not . null) bss'+ 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 <- src C.$= builderToByteStringFlush C.$$ CL.consume+ 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 (C.Chunk . byteString) (L.toChunks lbs)+ let chunks = map (Chunk . byteString) (L.toChunks lbs) let src = CL.sourceList chunks- bss <- src C.$$ builderToByteStringFlush C.=$ CL.consume- let unFlush (C.Chunk x) = [x]- unFlush C.Flush = []+ 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
@@ -42,7 +42,7 @@ 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
@@ -5,6 +5,7 @@ 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)@@ -25,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@@ -77,6 +78,14 @@ 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")
test/Data/Conduit/TextSpec.hs view
@@ -1,69 +1,67 @@ {-# 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,74 +154,76 @@ 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) == [["", ""]] 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 "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 <- CL.sourceList ["foobarbaz", error "ignore me"] C.$$ CT.decode CT.utf8 C.=$ CL.head+ x <- runConduit $ CL.sourceList ["foobarbaz", error "ignore me"] .| CT.decode CT.utf8 .| CL.head x `shouldBe` Just "foobarbaz" it' :: String -> IO () -> Spec
test/Data/Conduit/ZlibSpec.hs view
@@ -7,59 +7,53 @@ 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 import Control.Monad (replicateM_) -instance MonadBase base m => MonadBase base (CatchT m) where- liftBase = lift . liftBase- 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+ 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 <- src C.$= CZ.compressFlush 5 (CZ.WindowBits 31)- C.$= CZ.decompressFlush (CZ.WindowBits 31)- C.$$ CL.consume+ 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 C.$= CZ.gzip+ C.yield c .| CZ.gzip C.yield u- encoded <- src1 C.$$ CL.consume+ encoded <- runConduit $ src1 .| CL.consume let src2 = mapM_ C.yield encoded- (c', u') <- src2 C.$$ do- c' <- CZ.ungzip C.=$ CL.consume+ (c', u') <- runConduit $ src2 .| do+ c' <- CZ.ungzip .| CL.consume u' <- CL.consume return (S.concat c', S.concat u') c' `shouldBe` c@@ -69,21 +63,21 @@ let s1 = "hello" s2 = "world" src = do- C.yield s1 C.$= CZ.gzip- C.yield s2 C.$= CZ.gzip- actual <- src C.$$ CZ.multiple CZ.ungzip C.=$ CL.consume+ 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 $ C.yield s1 C.$= CZ.gzip C.$$ CL.consume+ 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 <- src C.$$ do- x <- fmap S.concat $ CZ.ungzip C.=$ CL.consume+ actual <- runConduit $ src .| do+ x <- fmap S.concat $ CZ.ungzip .| CL.consume y <- CL.consume return (x, y) actual `shouldBe` (s1, [s2, s3])@@ -91,8 +85,8 @@ it "multiple, over 32k" $ do let str = "One line" cnt = 30000- src = replicateM_ cnt $ C.yield str C.$= CZ.gzip- actual <- fmap S.concat $ src C.$$ CZ.multiple CZ.ungzip C.=$ CL.consume+ 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