packages feed

conduit-extra 1.0.0.1 → 1.1.0

raw patch · 26 files changed

+2659/−4 lines, 26 filesdep +attoparsecdep +blaze-builderdep +bytestringdep ~basedep ~conduitbinary-added

Dependencies added: attoparsec, blaze-builder, bytestring, conduit-extra, directory, exceptions, filepath, hspec, monad-control, network, primitive, resourcet, streaming-commons, text, transformers, transformers-base

Dependency ranges changed: base, conduit

Files

+ Data/Conduit/Attoparsec.hs view
@@ -0,0 +1,234 @@+{-# LANGUAGE BangPatterns       #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE FlexibleContexts   #-}+{-# LANGUAGE RankNTypes         #-}++-- |+-- Copyright: 2011 Michael Snoyman, 2010 John Millikin+-- License: MIT+--+-- Consume attoparsec parsers via conduit.+--+-- This code was taken from attoparsec-enumerator and adapted for conduits.+module Data.Conduit.Attoparsec+    ( -- * Sink+      sinkParser+      -- * Conduit+    , conduitParser+    , conduitParserEither++      -- * Types+    , ParseError (..)+    , Position (..)+    , PositionRange (..)+      -- * Classes+    , AttoparsecInput+    ) where++import           Control.Exception          (Exception)+import           Control.Monad              (unless)+import qualified Data.ByteString            as B+import qualified Data.Text                  as T+import qualified Data.Text.Internal         as TI+import           Data.Typeable              (Typeable)+import           Prelude                    hiding (lines)++import qualified Data.Attoparsec.ByteString+import qualified Data.Attoparsec.Text+import qualified Data.Attoparsec.Types      as A+import           Data.Conduit+import Control.Monad.Trans.Resource (MonadThrow, monadThrow)++-- | The context and message from a 'A.Fail' value.+data ParseError = ParseError+    { errorContexts :: [String]+    , errorMessage  :: String+    , errorPosition :: Position+    } | DivergentParser+    deriving (Show, Typeable)++instance Exception ParseError++data Position = Position+    { posLine :: {-# UNPACK #-} !Int+    , posCol  :: {-# UNPACK #-} !Int+    }+    deriving (Eq, Ord)++instance Show Position where+    show (Position l c) = show l ++ ':' : show c++data PositionRange = PositionRange+    { posRangeStart :: {-# UNPACK #-} !Position+    , posRangeEnd   :: {-# UNPACK #-} !Position+    }+    deriving (Eq, Ord)++instance Show PositionRange where+    show (PositionRange s e) = show s ++ '-' : show e++-- | A class of types which may be consumed by an Attoparsec parser.+class AttoparsecInput a where+    parseA :: A.Parser a b -> a -> A.IResult a b+    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+    -- the second input removed. Assumes the second string is shorter+    -- than the first.+    stripFromEnd :: a -> a -> a++instance AttoparsecInput B.ByteString where+    parseA = Data.Attoparsec.ByteString.parse+    feedA = Data.Attoparsec.ByteString.feed+    empty = B.empty+    isNull = B.null+    notEmpty = filter (not . B.null)+    getLinesCols = B.foldl' f (Position 0 0)+      where+        f (Position l c) ch | ch == 10 = Position (l + 1) 0+                            | otherwise = Position l (c + 1)+    stripFromEnd b1 b2 = B.take (B.length b1 - B.length b2) b1++instance AttoparsecInput T.Text where+    parseA = Data.Attoparsec.Text.parse+    feedA = Data.Attoparsec.Text.feed+    empty = T.empty+    isNull = T.null+    notEmpty = filter (not . T.null)+    getLinesCols = T.foldl' f (Position 0 0)+      where+        f (Position l c) ch | ch == '\n' = Position (l + 1) 0+                            | otherwise = Position l (c + 1)+    stripFromEnd (TI.Text arr1 off1 len1) (TI.Text _ _ len2) =+        TI.textP 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'.+--+-- Since 0.5.0+sinkParser :: (AttoparsecInput a, MonadThrow m) => A.Parser a b -> Consumer a m b+sinkParser = fmap snd . sinkParserPosErr (Position 1 1)++++-- | Consume a stream of parsed tokens, returning both the token and+-- the position it appears at. This function will raise a 'ParseError'+-- on bad input.+--+-- Since 0.5.0+conduitParser :: (AttoparsecInput a, MonadThrow m) => A.Parser a b -> Conduit a m (PositionRange, b)+conduitParser parser =+    conduit $ Position 1 0+       where+         conduit !pos = await >>= maybe (return ()) go+             where+               go x = do+                   leftover x+                   (!pos', !res) <- sinkParserPosErr pos parser+                   yield (PositionRange pos pos', res)+                   conduit pos'+{-# SPECIALIZE conduitParser+                   :: MonadThrow m+                   => A.Parser T.Text b+                   -> Conduit T.Text m (PositionRange, b) #-}+{-# SPECIALIZE conduitParser+                   :: MonadThrow m+                   => A.Parser B.ByteString b+                   -> Conduit B.ByteString m (PositionRange, b) #-}++++-- | Same as 'conduitParser', but we return an 'Either' type instead+-- of raising an exception.+conduitParserEither+    :: (Monad m, AttoparsecInput a)+    => A.Parser a b+    -> Conduit a m (Either ParseError (PositionRange, b))+conduitParserEither parser =+    conduit $ Position 1 0+  where+    conduit !pos = await >>= maybe (return ()) go+      where+        go x = do+          leftover x+          eres <- sinkParserPos pos parser+          case eres of+            Left e -> yield $ Left e+            Right (!pos', !res) -> do+              yield $! Right (PositionRange pos pos', res)+              conduit pos'+{-# SPECIALIZE conduitParserEither+                   :: Monad m+                   => A.Parser T.Text b+                   -> Conduit T.Text m (Either ParseError (PositionRange, b)) #-}+{-# SPECIALIZE conduitParserEither+                   :: Monad m+                   => A.Parser B.ByteString b+                   -> Conduit B.ByteString m (Either ParseError (PositionRange, b)) #-}+++++sinkParserPosErr+    :: (AttoparsecInput a, MonadThrow m)+    => Position+    -> A.Parser a b+    -> Consumer a m (Position, b)+sinkParserPosErr pos0 p = sinkParserPos pos0 p >>= f+    where+      f (Left e) = monadThrow e+      f (Right a) = return a+{-# INLINE sinkParserPosErr #-}+++sinkParserPos+    :: (AttoparsecInput a, Monad m)+    => Position+    -> A.Parser a b+    -> Consumer a m (Either ParseError (Position, b))+sinkParserPos pos0 p = sink empty pos0 (parseA p)+  where+    sink prev pos parser = await >>= maybe close push+      where+        push c+            | isNull c  = sink prev pos parser+            | otherwise = go False c $ parser c++        close = go True prev (feedA (parser empty) empty)++        go end c (A.Done lo x) = do+            let pos'+                    | end       = pos+                    | otherwise = addLinesCols prev pos+                y = stripFromEnd c lo+                pos'' = addLinesCols y pos'+            unless (isNull lo) $ leftover lo+            pos'' `seq` return $! Right (pos'', x)+        go end c (A.Fail rest contexts msg) =+            let x = stripFromEnd c rest+                pos'+                    | end       = pos+                    | otherwise = addLinesCols prev pos+                pos'' = addLinesCols x pos'+             in pos'' `seq` return $! Left (ParseError contexts msg pos'')+        go end c (A.Partial parser')+            | end       = return $! Left DivergentParser+            | otherwise =+                pos' `seq` sink c pos' parser'+              where+                pos' = addLinesCols prev pos++    addLinesCols :: AttoparsecInput a => a -> Position -> Position+    addLinesCols x (Position lines cols) =+        lines' `seq` cols' `seq` Position lines' cols'+      where+        Position dlines dcols = getLinesCols x+        lines' = lines + dlines+        cols' = (if dlines > 0 then 1 else cols) + dcols+{-# INLINE sinkParserPos #-}
+ Data/Conduit/Binary.hs view
@@ -0,0 +1,419 @@+{-# LANGUAGE CPP, RankNTypes #-}+-- | Functions for interacting with bytes.+module Data.Conduit.Binary+    ( -- * Files and @Handle@s++      -- | Note that most of these functions live in the @MonadResource@ monad+      -- to ensure resource finalization even in the presence of exceptions. In+      -- order to run such code, you will need to use @runResourceT@.++      -- ** Sources+      sourceFile+    , sourceHandle+    , sourceHandleUnsafe+    , sourceIOHandle+    , sourceFileRange+    , sourceHandleRange+      -- ** Sinks+    , sinkFile+    , sinkHandle+    , sinkIOHandle+      -- ** Conduits+    , conduitFile+    , conduitHandle+      -- * Utilities+      -- ** Sources+    , sourceLbs+      -- ** Sinks+    , head+    , dropWhile+    , take+    , drop+    , sinkCacheLength+    , sinkLbs+    , mapM_+      -- ** Conduits+    , isolate+    , takeWhile+    , Data.Conduit.Binary.lines+    ) where++import qualified Data.Streaming.FileRead as FR+import Prelude hiding (head, take, drop, takeWhile, dropWhile, mapM_)+import qualified Data.ByteString as S+import qualified Data.ByteString.Lazy as L+import Data.Conduit+import Data.Conduit.List (sourceList, consume)+import Control.Exception (assert, finally)+import Control.Monad (unless, when)+import Control.Monad.IO.Class (liftIO, MonadIO)+import Control.Monad.Trans.Resource (allocate, release)+import Control.Monad.Trans.Class (lift)+import qualified System.IO as IO+import Data.Word (Word8, Word64)+import Control.Applicative ((<$>))+import System.Directory (getTemporaryDirectory, removeFile)+import Data.ByteString.Lazy.Internal (defaultChunkSize)+import Data.ByteString.Internal (ByteString (PS), inlinePerformIO)+import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)+import Foreign.ForeignPtr (touchForeignPtr)+import Foreign.Ptr (plusPtr)+import Foreign.Storable (peek)+import GHC.ForeignPtr           (mallocPlainForeignPtrBytes)+import Control.Monad.Trans.Resource (MonadResource)++-- | Stream the contents of a file as binary data.+--+-- Since 0.3.0+sourceFile :: MonadResource m+           => FilePath+           -> Producer m S.ByteString+sourceFile fp =+    bracketP+        (FR.openFile fp)+         FR.closeFile+         loop+  where+    loop h = do+        bs <- liftIO $ FR.readChunk h+        unless (S.null bs) $ do+            yield bs+            loop h++-- | Stream the contents of a 'IO.Handle' as binary data. Note that this+-- function will /not/ automatically close the @Handle@ when processing+-- completes, since it did not acquire the @Handle@ in the first place.+--+-- Since 0.3.0+sourceHandle :: MonadIO m+             => IO.Handle+             -> Producer m S.ByteString+sourceHandle h =+    loop+  where+    loop = do+        bs <- liftIO (S.hGetSome h defaultChunkSize)+        if S.null bs+            then return ()+            else yield bs >> loop++-- | Same as @sourceHandle@, but instead of allocating a new buffer for each+-- incoming chunk of data, reuses the same buffer. Therefore, the @ByteString@s+-- yielded by this function are not referentially transparent between two+-- different @yield@s.+--+-- This function will be slightly more efficient than @sourceHandle@ by+-- avoiding allocations and reducing garbage collections, but should only be+-- used if you can guarantee that you do not reuse a @ByteString@ (or any slice+-- thereof) between two calls to @await@.+--+-- Since 1.0.12+sourceHandleUnsafe :: MonadIO m => IO.Handle -> Source m ByteString+sourceHandleUnsafe handle = do+    fptr <- liftIO $ mallocPlainForeignPtrBytes defaultChunkSize+    let ptr = unsafeForeignPtrToPtr fptr+        loop = do+            count <- liftIO $ IO.hGetBuf handle ptr defaultChunkSize+            when (count > 0) $ do+                yield (PS fptr 0 count)+                loop++    loop++    liftIO $ touchForeignPtr fptr++-- | An alternative to 'sourceHandle'.+-- Instead of taking a pre-opened 'IO.Handle', it takes an action that opens+-- a 'IO.Handle' (in read mode), so that it can open it only when needed+-- and closed it as soon as possible.+--+-- Since 0.3.0+sourceIOHandle :: MonadResource m+               => IO IO.Handle+               -> Producer m S.ByteString+sourceIOHandle alloc = bracketP alloc IO.hClose sourceHandle++-- | Stream all incoming data to the given 'IO.Handle'. Note that this function+-- will /not/ automatically close the @Handle@ when processing completes.+--+-- Since 0.3.0+sinkHandle :: MonadIO m+           => IO.Handle+           -> Consumer S.ByteString m ()+sinkHandle h = awaitForever $ 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.+--+-- Since 0.3.0+sourceFileRange :: MonadResource m+                => FilePath+                -> Maybe Integer -- ^ Offset+                -> Maybe Integer -- ^ Maximum count+                -> Producer m S.ByteString+sourceFileRange fp offset count = bracketP+    (IO.openBinaryFile fp IO.ReadMode)+    IO.hClose+    (\h -> sourceHandleRange h offset count)++-- | Stream the contents of a handle as binary data, starting from a certain+-- offset and only consuming up to a certain number of bytes.+--+-- Since 1.0.8+sourceHandleRange :: MonadIO m+                  => IO.Handle+                  -> Maybe Integer -- ^ Offset+                  -> Maybe Integer -- ^ Maximum count+                  -> Producer m S.ByteString+sourceHandleRange handle offset count = do+    case offset of+        Nothing -> return ()+        Just off -> liftIO $ IO.hSeek handle IO.AbsoluteSeek off+    case count of+        Nothing -> pullUnlimited+        Just c -> pullLimited (fromInteger c)+  where+    pullUnlimited = do+        bs <- liftIO $ S.hGetSome handle 4096+        if S.null bs+            then return ()+            else do+                yield bs+                pullUnlimited++    pullLimited c = do+        bs <- liftIO $ S.hGetSome handle (min c 4096)+        let c' = c - S.length bs+        assert (c' >= 0) $+            if S.null bs+                then return ()+                else do+                    yield bs+                    pullLimited c'++-- | Stream all incoming data to the given file.+--+-- Since 0.3.0+sinkFile :: MonadResource m+         => FilePath+         -> Consumer S.ByteString m ()+sinkFile fp = sinkIOHandle (IO.openBinaryFile fp IO.WriteMode)++-- | Stream the contents of the input to a file, and also send it along the+-- pipeline. Similar in concept to the Unix command @tee@.+--+-- Since 0.3.0+conduitFile :: MonadResource m+            => FilePath+            -> Conduit S.ByteString m S.ByteString+conduitFile fp = bracketP+    (IO.openBinaryFile fp IO.WriteMode)+    IO.hClose+    conduitHandle++-- | Stream the contents of the input to a @Handle@, and also send it along the+-- pipeline. Similar in concept to the Unix command @tee@. Like @sourceHandle@,+-- 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 h = awaitForever $ \bs -> liftIO (S.hPut h bs) >> yield bs++-- | Ensure that only up to the given number of bytes are consume by the inner+-- sink. Note that this does /not/ ensure that all of those bytes are in fact+-- consumed.+--+-- Since 0.3.0+isolate :: Monad m+        => Int+        -> Conduit S.ByteString m S.ByteString+isolate =+    loop+  where+    loop 0 = return ()+    loop count = do+        mbs <- await+        case mbs of+            Nothing -> return ()+            Just bs -> do+                let (a, b) = S.splitAt count bs+                case count - S.length a of+                    0 -> do+                        unless (S.null b) $ leftover b+                        yield a+                    count' -> assert (S.null b) $ yield a >> loop count'++-- | Return the next byte from the stream, if available.+--+-- Since 0.3.0+head :: Monad m => Consumer S.ByteString m (Maybe Word8)+head = do+    mbs <- await+    case mbs of+        Nothing -> return Nothing+        Just bs ->+            case S.uncons bs of+                Nothing -> head+                Just (w, bs') -> leftover bs' >> return (Just w)++-- | Return all bytes while the predicate returns @True@.+--+-- Since 0.3.0+takeWhile :: Monad m => (Word8 -> Bool) -> Conduit S.ByteString m S.ByteString+takeWhile p =+    loop+  where+    loop = await >>= maybe (return ()) go++    go bs+        | S.null x = next+        | otherwise = yield x >> next+      where+        next = if S.null y then loop else leftover y+        (x, y) = S.span p bs++-- | Ignore all bytes while the predicate returns @True@.+--+-- Since 0.3.0+dropWhile :: Monad m => (Word8 -> Bool) -> Consumer S.ByteString m ()+dropWhile p =+    loop+  where+    loop = do+        mbs <- await+        case S.dropWhile p <$> mbs of+            Nothing -> return ()+            Just bs+                | S.null bs -> loop+                | otherwise -> leftover bs++-- | Take the given number of bytes, if available.+--+-- Since 0.3.0+take :: Monad m => Int -> Consumer S.ByteString m L.ByteString+take  0 = return L.empty+take n0 = go n0 id+  where+    go n front =+        await >>= maybe (return $ L.fromChunks $ front []) go'+      where+        go' bs =+            case S.length bs `compare` n of+                LT -> go (n - S.length bs) (front . (bs:))+                EQ -> return $ L.fromChunks $ front [bs]+                GT ->+                    let (x, y) = S.splitAt n bs+                     in assert (not $ S.null y) $ leftover y >> return (L.fromChunks $ front [x])++-- | Drop up to the given number of bytes.+--+-- Since 0.5.0+drop :: Monad m => Int -> Consumer S.ByteString m ()+drop  0 = return ()+drop n0 = go n0+  where+    go n =+        await >>= maybe (return ()) go'+      where+        go' bs =+            case S.length bs `compare` n of+                LT -> go (n - S.length bs)+                EQ -> return ()+                GT ->+                    let y = S.drop n bs+                     in assert (not $ S.null y) $ leftover y >> return ()++-- | Split the input bytes into lines. In other words, split on the LF byte+-- (10), and strip it from the output.+--+-- Since 0.3.0+lines :: Monad m => Conduit S.ByteString m S.ByteString+lines =+    loop id+  where+    loop front = await >>= maybe (finish front) (go front)++    finish front =+        let final = front S.empty+         in unless (S.null final) (yield final)++    go sofar more =+        case S.uncons second of+            Just (_, second') -> yield (sofar first) >> go id second'+            Nothing ->+                let rest = sofar more+                 in loop $ S.append rest+      where+        (first, second) = S.breakByte 10 more++-- | Stream the chunks from a lazy bytestring.+--+-- Since 0.5.0+sourceLbs :: Monad m => L.ByteString -> Producer m S.ByteString+sourceLbs = sourceList . L.toChunks++-- | Stream the input data into a temp file and count the number of bytes+-- present. When complete, return a new @Source@ reading from the temp file+-- together with the length of the input in bytes.+--+-- All resources will be cleaned up automatically.+--+-- Since 1.0.5+sinkCacheLength :: (MonadResource m1, MonadResource m2)+                => Sink S.ByteString m1 (Word64, Source m2 S.ByteString)+sinkCacheLength = do+    tmpdir <- liftIO getTemporaryDirectory+    (releaseKey, (fp, h)) <- allocate+        (IO.openBinaryTempFile tmpdir "conduit.cache")+        (\(fp, h) -> IO.hClose h `finally` removeFile fp)+    len <- sinkHandleLen h+    liftIO $ IO.hClose h+    return (len, sourceFile fp >> release releaseKey)+  where+    sinkHandleLen :: MonadResource m => IO.Handle -> Sink S.ByteString m Word64+    sinkHandleLen h =+        loop 0+      where+        loop x =+            await >>= maybe (return x) go+          where+            go bs = do+                liftIO $ S.hPut h bs+                loop $ x + fromIntegral (S.length bs)++-- | Consume a stream of input into a lazy bytestring. Note that no lazy I\/O+-- 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 = fmap L.fromChunks consume++mapM_BS :: Monad m => (Word8 -> m ()) -> S.ByteString -> m ()+mapM_BS f (PS fptr offset len) = do+    let start = unsafeForeignPtrToPtr fptr `plusPtr` offset+        end = start `plusPtr` len+        loop ptr+            | ptr >= end = inlinePerformIO (touchForeignPtr fptr) `seq` return ()+            | otherwise = do+                f (inlinePerformIO (peek ptr))+                loop (ptr `plusPtr` 1)+    loop start+{-# INLINE mapM_BS #-}++-- | Perform a computation on each @Word8@ in a stream.+--+-- Since 1.0.10+mapM_ :: Monad m => (Word8 -> m ()) -> Consumer S.ByteString m ()+mapM_ f = awaitForever (lift . mapM_BS f)+{-# INLINE mapM_ #-}
+ Data/Conduit/Blaze.hs view
@@ -0,0 +1,155 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE RankNTypes #-}+-- | Convert a stream of blaze-builder @Builder@s into a stream of @ByteString@s.+--+-- Adapted from blaze-builder-enumerator, written by myself and Simon Meier.+--+-- Note that the functions here can work in any monad built on top of @IO@ or+-- @ST@.+module Data.Conduit.Blaze+    (++  -- * Conduits from builders to bytestrings+    builderToByteString+  , unsafeBuilderToByteString+  , builderToByteStringWith++  -- ** Flush+  , builderToByteStringFlush+  , builderToByteStringWithFlush++  -- * Buffers+  , Buffer++  -- ** Status information+  , freeSize+  , sliceSize+  , bufferSize++  -- ** Creation and modification+  , allocBuffer+  , reuseBuffer+  , nextSlice++  -- ** Conversion to bytestings+  , unsafeFreezeBuffer+  , unsafeFreezeNonEmptyBuffer++  -- * Buffer allocation strategies+  , BufferAllocStrategy+  , allNewBuffersStrategy+  , reuseBufferStrategy+    ) where++import Data.Conduit+import Control.Monad (unless, liftM)+import Control.Monad.Trans.Class (lift, MonadTrans)++import qualified Data.ByteString                   as S++import Blaze.ByteString.Builder.Internal+import Blaze.ByteString.Builder.Internal.Types+import Blaze.ByteString.Builder.Internal.Buffer+import Control.Monad.Primitive (PrimMonad, unsafePrimToPrim)+import Control.Monad.Base (MonadBase, liftBase)++unsafeLiftIO :: (MonadBase base m, PrimMonad base) => IO a -> m a+unsafeLiftIO = liftBase . unsafePrimToPrim++-- | Incrementally execute builders and pass on the filled chunks as+-- bytestrings.+builderToByteString :: (MonadBase base m, PrimMonad base) => Conduit Builder m S.ByteString+builderToByteString =+  builderToByteStringWith (allNewBuffersStrategy defaultBufferSize)++-- |+--+-- Since 0.0.2+builderToByteStringFlush :: (MonadBase base m, PrimMonad base) => Conduit (Flush Builder) m (Flush S.ByteString)+builderToByteStringFlush =+  builderToByteStringWithFlush (allNewBuffersStrategy defaultBufferSize)++-- | Incrementally execute builders on the given buffer and pass on the filled+-- chunks as bytestrings. Note that, if the given buffer is too small for the+-- execution of a build step, a larger one will be allocated.+--+-- WARNING: This conduit yields bytestrings that are NOT+-- referentially transparent. Their content will be overwritten as soon+-- as control is returned from the inner sink!+unsafeBuilderToByteString :: (MonadBase base m, PrimMonad base)+                          => IO Buffer  -- action yielding the inital buffer.+                          -> Conduit Builder m S.ByteString+unsafeBuilderToByteString = builderToByteStringWith . reuseBufferStrategy+++-- | A conduit that incrementally executes builders and passes on the+-- filled chunks as bytestrings to an inner sink.+--+-- INV: All bytestrings passed to the inner sink are non-empty.+builderToByteStringWith :: (MonadBase base m, PrimMonad base)+                        => BufferAllocStrategy+                        -> Conduit Builder m S.ByteString+builderToByteStringWith =+    helper (liftM (fmap Chunk) await) yield'+  where+    yield' Flush = return ()+    yield' (Chunk bs) = yield bs++-- |+--+-- Since 0.0.2+builderToByteStringWithFlush+    :: (MonadBase base m, PrimMonad base)+    => BufferAllocStrategy+    -> Conduit (Flush Builder) m (Flush S.ByteString)+builderToByteStringWithFlush = helper await yield++helper :: (MonadBase base m, PrimMonad base, Monad (t m), MonadTrans t)+       => t m (Maybe (Flush Builder))+       -> (Flush S.ByteString -> t m ())+       -> BufferAllocStrategy+       -> t m ()+helper await' yield' (ioBufInit, nextBuf) =+    loop ioBufInit+  where+    loop ioBuf = do+        await' >>= maybe (close ioBuf) (cont' ioBuf)++    cont' ioBuf Flush = push ioBuf flush $ \ioBuf' -> yield' Flush >> loop ioBuf'+    cont' ioBuf (Chunk builder) = push ioBuf builder loop++    close ioBuf = do+        buf <- lift $ unsafeLiftIO $ ioBuf+        maybe (return ()) (yield' . Chunk) (unsafeFreezeNonEmptyBuffer buf)++    push ioBuf0 x continue = do+        go (unBuilder x (buildStep finalStep)) ioBuf0+      where+        finalStep !(BufRange pf _) = return $ Done pf ()++        go bStep ioBuf = do+            !buf   <- lift $ unsafeLiftIO $ ioBuf+            signal <- lift $ unsafeLiftIO $ execBuildStep bStep buf+            case signal of+                Done op' _ -> continue $ return $ updateEndOfSlice buf op'+                BufferFull minSize op' bStep' -> do+                    let buf' = updateEndOfSlice buf op'+                        {-# INLINE cont #-}+                        cont = do+                            -- sequencing the computation of the next buffer+                            -- construction here ensures that the reference to the+                            -- foreign pointer `fp` is lost as soon as possible.+                            ioBuf' <- lift $ unsafeLiftIO $ nextBuf minSize buf'+                            go bStep' ioBuf'+                    case unsafeFreezeNonEmptyBuffer buf' of+                        Nothing -> return ()+                        Just bs -> yield' (Chunk bs)+                    cont+                InsertByteString op' bs bStep' -> do+                    let buf' = updateEndOfSlice buf op'+                    case unsafeFreezeNonEmptyBuffer buf' of+                        Nothing -> return ()+                        Just bs' -> yield' $ Chunk bs'+                    unless (S.null bs) $ yield' $ Chunk bs+                    lift (unsafeLiftIO $ nextBuf 1 buf') >>= go bStep'
+ Data/Conduit/Filesystem.hs view
@@ -0,0 +1,66 @@+{-# LANGUAGE RankNTypes #-}+module Data.Conduit.Filesystem+    ( sourceDirectory+    , 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 ()
+ Data/Conduit/Lazy.hs view
@@ -0,0 +1,109 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE FlexibleContexts #-}+-- | Use lazy I\/O for consuming the contents of a source. Warning: All normal+-- warnings of lazy I\/O apply. In particular, if you are using this with a+-- @ResourceT@ transformer, you must force the list to be evaluated before+-- exiting the @ResourceT@.+module Data.Conduit.Lazy+    ( lazyConsume+    , MonadActive (..)+    ) where++import Data.Conduit+import Data.Conduit.Internal (Pipe (..), unConduitM)+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.Trans.Identity ( IdentityT)+import Control.Monad.Trans.List     ( ListT    )+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  )+import Control.Monad.Trans.RWS      ( RWST     )++import qualified Control.Monad.Trans.RWS.Strict    as Strict ( RWST   )+import qualified Control.Monad.Trans.State.Strict  as Strict ( StateT )+import qualified Control.Monad.Trans.Writer.Strict as Strict ( WriterT )++import Data.Monoid (Monoid)+import Control.Monad.ST (ST)+import qualified Control.Monad.ST.Lazy as Lazy+import Data.Functor.Identity (Identity)+import Control.Monad.Trans.Resource.Internal (ResourceT (ResourceT), ReleaseMap (ReleaseMapClosed))+import qualified Data.IORef as I++-- | Use lazy I\/O to consume all elements from a @Source@.+--+-- This function relies on 'monadActive' to determine if the underlying monadic+-- state has been closed.+--+-- Since 0.3.0+lazyConsume :: (MonadBaseControl IO m, MonadActive m) => Source m a -> m [a]+lazyConsume =+    go . unConduitM+  where+    go (Done _) = return []+    go (HaveOutput src _ x) = do+        xs <- liftBaseOp_ unsafeInterleaveIO $ go src+        return $ x : xs+    go (PipeM msrc) = liftBaseOp_ unsafeInterleaveIO $ do+        a <- monadActive+        if a+            then msrc >>= go+            else return []+    go (NeedInput _ c) = go (c ())+    go (Leftover p _) = go p++-- | 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+-- cases as lazy I\/O, where an unevaluated thunk may still refer to a+-- closed @ResourceT@.+--+-- Since 0.3.0+class Monad m => MonadActive m where+    monadActive :: m Bool++instance (MonadIO m, MonadActive m) => MonadActive (ResourceT m) where+    monadActive = ResourceT $ \rmMap -> do+        rm <- liftIO $ I.readIORef rmMap+        case rm of+            ReleaseMapClosed -> return False+            _ -> monadActive -- recurse++instance MonadActive Identity where+    monadActive = return True++instance MonadActive IO where+    monadActive = return True++instance MonadActive (ST s) where+    monadActive = return True++instance MonadActive (Lazy.ST s) where+    monadActive = return True++#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)+GO(ListT)+GO(MaybeT)+GOX(Error e, ErrorT e)+GO(ReaderT r)+GO(StateT s)+GOX(Monoid w, WriterT w)+GOX(Monoid w, RWST r w s)+GOX(Monoid w, Strict.RWST r w s)+GO(Strict.StateT s)+GOX(Monoid w, Strict.WriterT w)+#undef GO+#undef GOX++instance MonadActive m => MonadActive (Pipe l i o u m) where+    monadActive = lift monadActive+instance MonadActive m => MonadActive (ConduitM i o m) where+    monadActive = lift monadActive
+ Data/Conduit/Network.hs view
@@ -0,0 +1,95 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE CPP #-}+module Data.Conduit.Network+    ( -- * Basic utilities+      sourceSocket+    , sinkSocket+      -- * Simple TCP server/client interface.+    , SN.AppData+    , appSource+    , appSink+    , SN.appSockAddr+    , SN.appLocalAddr+      -- ** Server+    , SN.ServerSettings+    , serverSettings+    , SN.runTCPServer+    , SN.runTCPServerWithHandle+      -- ** Client+    , SN.ClientSettings+    , clientSettings+    , SN.runTCPClient+      -- ** Getters+    , SN.getPort+    , SN.getHost+    , SN.getAfterBind+    , SN.getNeedLocalAddr+      -- ** Setters+    , SN.setPort+    , SN.setHost+    , SN.setAfterBind+    , SN.setNeedLocalAddr+      -- * Types+    , SN.HostPreference+    ) where++import Prelude hiding (catch)+import Data.Conduit+import qualified Network.Socket as NS+import Network.Socket (Socket)+import Network.Socket.ByteString (sendAll, recv)+import Data.ByteString (ByteString)+import qualified Data.ByteString as S+import qualified Data.ByteString.Char8 as S8+import Control.Monad.IO.Class (MonadIO (liftIO))+import Control.Exception (throwIO, SomeException, try, finally, bracket, IOException, catch)+import Control.Monad (forever, unless)+import Control.Monad.Trans.Control (MonadBaseControl, control)+import Control.Monad.Trans.Class (lift)+import Control.Concurrent (forkIO, threadDelay, newEmptyMVar, putMVar, takeMVar)+import qualified Data.Streaming.Network as SN++-- | 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 socket =+    loop+  where+    loop = do+        bs <- lift $ liftIO $ SN.safeRecv socket 4096+        if S.null bs+            then return ()+            else yield bs >> loop++-- | Stream data to the socket.+--+-- This function does /not/ automatically close the socket.+--+-- Since 0.0.0+sinkSocket :: MonadIO m => Socket -> Consumer ByteString m ()+sinkSocket socket =+    loop+  where+    loop = await >>= maybe (return ()) (\bs -> lift (liftIO $ sendAll socket bs) >> loop)++serverSettings = SN.serverSettingsTCP+clientSettings = SN.clientSettingsTCP++appSource :: (SN.HasReadWrite ad, MonadIO m) => ad -> Producer m ByteString+appSource ad =+    loop+  where+    read' = SN.appRead ad+    loop = do+        bs <- liftIO read'+        unless (S.null bs) $ do+            yield bs+            loop++appSink :: (SN.HasReadWrite ad, MonadIO m) => ad -> Consumer ByteString m ()+appSink ad = awaitForever $ liftIO . SN.appWrite ad
+ Data/Conduit/Network/UDP.hs view
@@ -0,0 +1,83 @@+{-# LANGUAGE RankNTypes #-}+module Data.Conduit.Network.UDP+    ( -- * UDP message representation+      SN.Message (..)+      -- * Basic utilities+    , sourceSocket+    , sinkSocket+    , sinkAllSocket+    , sinkToSocket+    , sinkAllToSocket+      -- * Helper Utilities+    , SN.HostPreference+    ) where++import Data.Conduit+import Network.Socket (AddrInfo, SockAddr, Socket)+import qualified Network.Socket as NS+import Network.Socket.ByteString (recvFrom, send, sendAll, sendTo, sendAllTo)+import Data.ByteString (ByteString)+import Control.Monad.IO.Class (MonadIO (liftIO))+import Control.Monad (void)+import Control.Monad.Trans.Class (lift)+import qualified Data.Streaming.Network as SN++-- | Stream messages from the socket.+--+-- The given @len@ defines the maximum packet size. Every produced item+-- 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 socket len = loop+  where+    loop = do+        (bs, addr) <- lift $ liftIO $ recvFrom socket len+        yield (SN.Message bs addr) >> loop++-- | Stream messages to the connected socket.+--+-- 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 = sinkSocketHelper (\sock bs -> void $ send sock bs)++-- | Stream messages to the connected socket.+--+-- 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 = sinkSocketHelper sendAll++-- | Stream messages to the socket.+--+-- Every handled item contains the message payload and the destination+-- address. The payload is sent using @sendTo@, so some of it might be+-- lost.+--+-- This function does /not/ automatically close the socket.+sinkToSocket :: MonadIO m => Socket -> Consumer SN.Message m ()+sinkToSocket = sinkSocketHelper (\sock (SN.Message bs addr) -> void $ sendTo sock bs addr)++-- | Stream messages to the socket.+--+-- Every handled item contains the message payload and the destination+-- address. The payload is sent using @sendAllTo@, so it might end up in+-- multiple packets.+--+-- This function does /not/ automatically close the socket.+sinkAllToSocket :: MonadIO m => Socket -> Consumer SN.Message m ()+sinkAllToSocket = sinkSocketHelper (\sock (SN.Message bs addr) -> sendAllTo sock bs addr)++-- Internal+sinkSocketHelper :: MonadIO m => (Socket -> a -> IO ())+                              -> Socket+                              -> Consumer a m ()+sinkSocketHelper act socket = loop+  where+    loop = await >>= maybe+                        (return ())+                        (\a -> lift (liftIO $ act socket a) >> loop)+{-# INLINE sinkSocketHelper #-}
+ Data/Conduit/Network/Unix.hs view
@@ -0,0 +1,45 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+module Data.Conduit.Network.Unix+    ( -- * Basic utilities+      sourceSocket+    , sinkSocket+      -- * Simple server/client interface+    , SN.AppDataUnix+    , appSource+    , appSink+      -- ** Server+    , SN.ServerSettingsUnix+    , serverSettings+    , SN.runUnixServer+      -- ** Client+    , SN.ClientSettingsUnix+    , clientSettings+    , SN.runUnixClient+      -- ** Getters+    , SN.getPath+    , SN.getAfterBind+      -- ** Setters+    , SN.setPath+    , SN.setAfterBind+    ) where++import Data.Conduit+import Network.Socket (Socket)+import qualified Network.Socket as NS+import Data.Conduit.Network (appSource, appSink, sourceSocket, sinkSocket)+import qualified Data.Streaming.Network as SN+import Control.Monad.IO.Class (MonadIO (liftIO))+import Control.Exception (throwIO, SomeException, try, finally, bracket,+                          bracketOnError, catch)+import Control.Monad (forever, void)+import Control.Monad.Trans.Control (control)+import Control.Concurrent (forkIO)+import System.Directory (removeFile)+import System.IO.Error (isDoesNotExistError)+import Control.Monad.Trans.Resource (MonadBaseControl)++clientSettings = SN.clientSettingsUnix+serverSettings = SN.serverSettingsUnix
+ Data/Conduit/Text.hs view
@@ -0,0 +1,440 @@+{-# LANGUAGE DeriveDataTypeable, RankNTypes #-}+-- |+-- Copyright: 2011 Michael Snoyman, 2010-2011 John Millikin+-- License: MIT+--+-- Handle streams of text.+--+-- Parts of this code were taken from enumerator and adapted for conduits.+module Data.Conduit.Text+    (++    -- * Text codecs+      Codec+    , encode+    , decode+    , utf8+    , utf16_le+    , utf16_be+    , utf32_le+    , utf32_be+    , ascii+    , iso8859_1+    , lines+    , linesBounded+    , TextException (..)+    , takeWhile+    , dropWhile+    , take+    , drop+    , foldLines+    , withLine+    , Data.Conduit.Text.decodeUtf8+    , encodeUtf8+    ) where++import qualified Prelude+import           Prelude hiding (head, drop, takeWhile, lines, zip, zip3, zipWith, zipWith3, take, dropWhile)++import qualified Control.Exception as Exc+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as B8+import           Data.Char (ord)+import qualified Data.Text as T+import qualified Data.Text.Encoding as TE+import           Data.Word (Word8)+import           Data.Typeable (Typeable)++import Data.Conduit+import qualified Data.Conduit.List as CL+import Control.Monad.Trans.Class (lift)+import Control.Monad.Trans.Resource (MonadThrow, monadThrow)+import Control.Monad (unless,when)+import Data.Streaming.Text++-- | A specific character encoding.+--+-- Since 0.3.0+data Codec = Codec+    { codecName :: T.Text+    , codecEncode+        :: T.Text+        -> (B.ByteString, Maybe (TextException, T.Text))+    , codecDecode+        :: B.ByteString+        -> (T.Text, Either+            (TextException, B.ByteString)+            B.ByteString)+    }+    | NewCodec T.Text (T.Text -> B.ByteString) (B.ByteString -> DecodeResult)++instance Show Codec where+    showsPrec d c = showParen (d > 10) $+        showString "Codec " . shows (codecName c)+++++-- | Emit each line separately+--+-- Since 0.4.1+lines :: Monad m => Conduit T.Text m T.Text+lines =+    loop id+  where+    loop front = await >>= maybe (finish front) (go front)++    finish front =+        let final = front T.empty+         in unless (T.null final) (yield final)++    go sofar more =+        case T.uncons second of+            Just (_, second') -> yield (sofar first') >> go id second'+            Nothing ->+                let rest = sofar more+                 in loop $ T.append rest+      where+        (first', second) = T.break (== '\n') more++++-- | Variant of the lines function with an integer parameter.+-- The text length of any emitted line+-- never exceeds the value of the paramater. Whenever+-- this is about to happen a LengthExceeded exception+-- is thrown. This function should be used instead+-- of the lines function whenever we are dealing with+-- user input (e.g. a file upload) because we can't be sure that+-- user input won't have extraordinarily large lines which would+-- require large amounts of memory if consumed.+linesBounded :: MonadThrow m => Int -> Conduit T.Text m T.Text+linesBounded maxLineLen =+    loop 0 id+  where+    loop len front = await >>= maybe (finish front) (go len front)++    finish front =+        let final = front T.empty+         in unless (T.null final) (yield final)+    go len sofar more =+        case T.uncons second of+            Just (_, second') -> do+                let toYield = sofar first'+                    len' = len + T.length first'+                when (len' > maxLineLen)+                    (lift $ monadThrow (LengthExceeded maxLineLen))+                yield toYield+                go 0 id second'+            Nothing -> do+                let len' = len + T.length more+                when (len' > maxLineLen) $+                    (lift $ monadThrow (LengthExceeded maxLineLen))+                let rest = sofar more+                loop len' $ T.append rest+      where+        (first', second) = T.break (== '\n') more++++-- | Convert text into bytes, using the provided codec. If the codec is+-- 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 (NewCodec _ enc _) = CL.map enc+encode codec = CL.mapM $ \t -> do+    let (bs, mexc) = codecEncode codec t+    maybe (return bs) (monadThrow . fst) mexc+++-- | 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 (NewCodec name _ start) =+    loop 0 start+  where+    loop consumed dec =+        await >>= maybe finish go+      where+        finish =+            case dec B.empty of+                DecodeResultSuccess _ _ -> return ()+                DecodeResultFailure t rest -> onFailure B.empty t rest+        {-# INLINE finish #-}++        go bs | B.null bs = loop consumed dec+        go bs =+            case dec bs of+                DecodeResultSuccess t dec' -> do+                    let consumed' = consumed + B.length bs+                        next = do+                            unless (T.null t) (yield t)+                            loop consumed' dec'+                     in consumed' `seq` next+                DecodeResultFailure t rest -> onFailure bs t rest++        onFailure bs t rest = do+            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)+        {-# INLINE onFailure #-}+decode codec =+    loop id+  where+    loop front = await >>= maybe (finish front) (go front)++    finish front =+        case B.uncons $ front B.empty of+            Nothing -> return ()+            Just (w, _) -> lift $ monadThrow $ DecodeException codec w++    go front bs' =+        case extra of+            Left (exc, _) -> lift $ monadThrow exc+            Right bs'' -> yield text >> loop (B.append bs'')+      where+        (text, extra) = codecDecode codec bs+        bs = front bs'++-- |+-- Since 0.3.0+data TextException = DecodeException Codec Word8+                   | EncodeException Codec Char+                   | LengthExceeded Int+                   | TextException Exc.SomeException+                   | NewDecodeException !T.Text !Int !B.ByteString+    deriving Typeable+instance Show TextException where+    show (DecodeException codec w) = concat+        [ "Error decoding legacy Data.Conduit.Text codec "+        , show codec+        , " when parsing byte: "+        , show w+        ]+    show (EncodeException codec c) = concat+        [ "Error encoding legacy Data.Conduit.Text codec "+        , show codec+        , " when parsing char: "+        , show c+        ]+    show (LengthExceeded i) = "Data.Conduit.Text.linesBounded: line too long: " ++ show i+    show (TextException se) = "Data.Conduit.Text.TextException: " ++ show se+    show (NewDecodeException codec consumed next) = concat+        [ "Data.Conduit.Text.decode: Error decoding stream of "+        , T.unpack codec+        , " bytes. Error encountered in stream at offset "+        , show consumed+        , ". Encountered at byte sequence "+        , show next+        ]+instance Exc.Exception TextException++-- |+-- Since 0.3.0+utf8 :: Codec+utf8 = NewCodec (T.pack "UTF-8") TE.encodeUtf8 Data.Streaming.Text.decodeUtf8++-- |+-- Since 0.3.0+utf16_le :: Codec+utf16_le = NewCodec (T.pack "UTF-16-LE") TE.encodeUtf16LE decodeUtf16LE++-- |+-- Since 0.3.0+utf16_be :: Codec+utf16_be = NewCodec (T.pack "UTF-16-BE") TE.encodeUtf16BE decodeUtf16BE++-- |+-- Since 0.3.0+utf32_le :: Codec+utf32_le = NewCodec (T.pack "UTF-32-LE") TE.encodeUtf32LE decodeUtf32LE++-- |+-- Since 0.3.0+utf32_be :: Codec+utf32_be = NewCodec (T.pack "UTF-32-BE") TE.encodeUtf32BE decodeUtf32BE++-- |+-- Since 0.3.0+ascii :: Codec+ascii = Codec name enc dec where+    name = T.pack "ASCII"+    enc text = (bytes, extra) where+        (safe, unsafe) = T.span (\c -> ord c <= 0x7F) text+        bytes = B8.pack (T.unpack safe)+        extra = if T.null unsafe+            then Nothing+            else Just (EncodeException ascii (T.head unsafe), unsafe)++    dec bytes = (text, extra) where+        (safe, unsafe) = B.span (<= 0x7F) bytes+        text = T.pack (B8.unpack safe)+        extra = if B.null unsafe+            then Right B.empty+            else Left (DecodeException ascii (B.head unsafe), unsafe)++-- |+-- Since 0.3.0+iso8859_1 :: Codec+iso8859_1 = Codec name enc dec where+    name = T.pack "ISO-8859-1"+    enc text = (bytes, extra) where+        (safe, unsafe) = T.span (\c -> ord c <= 0xFF) text+        bytes = B8.pack (T.unpack safe)+        extra = if T.null unsafe+            then Nothing+            else Just (EncodeException iso8859_1 (T.head unsafe), unsafe)++    dec bytes = (T.pack (B8.unpack bytes), Right B.empty)++-- |+--+-- Since 1.0.8+takeWhile :: Monad m+          => (Char -> Bool)+          -> Conduit T.Text m T.Text+takeWhile p =+    loop+  where+    loop = await >>= maybe (return ()) go+    go t =+        case T.span p t of+            (x, y)+                | T.null y -> yield x >> loop+                | otherwise -> yield x >> leftover y++-- |+--+-- Since 1.0.8+dropWhile :: Monad m+          => (Char -> Bool)+          -> Consumer T.Text m ()+dropWhile p =+    loop+  where+    loop = await >>= maybe (return ()) go+    go t+        | T.null x = loop+        | otherwise = leftover x+      where+        x = T.dropWhile p t++-- |+--+-- Since 1.0.8+take :: Monad m => Int -> Conduit T.Text m T.Text+take =+    loop+  where+    loop i = await >>= maybe (return ()) (go i)+    go i t+        | diff == 0 = yield t+        | diff < 0 =+            let (x, y) = T.splitAt i t+             in yield x >> leftover y+        | otherwise = yield t >> loop diff+      where+        diff = i - T.length t++-- |+--+-- Since 1.0.8+drop :: Monad m => Int -> Consumer T.Text m ()+drop =+    loop+  where+    loop i = await >>= maybe (return ()) (go i)+    go i t+        | diff == 0 = return ()+        | diff < 0 = leftover $ T.drop i t+        | otherwise = loop diff+      where+        diff = i - T.length t++-- |+--+-- Since 1.0.8+foldLines :: Monad m+          => (a -> ConduitM T.Text o m a)+          -> a+          -> ConduitM 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+            CL.sinkNull+            return a+        drop 1+        start a++-- |+--+-- Since 1.0.8+withLine :: Monad m+         => Sink T.Text m a+         -> Consumer T.Text 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+                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 #-}
+ Data/Conduit/Zlib.hs view
@@ -0,0 +1,158 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE RankNTypes #-}+-- | Streaming compression and decompression using conduits.+--+-- Parts of this code were taken from zlib-enum and adapted for conduits.+module Data.Conduit.Zlib (+    -- * Conduits+    compress, decompress, gzip, ungzip,+    -- * Flushing+    compressFlush, decompressFlush,+    -- * Re-exported from zlib-bindings+    WindowBits (..), defaultWindowBits+) where++import Data.Streaming.Zlib+import Data.Conduit+import Data.ByteString (ByteString)+import qualified Data.ByteString as S+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)++-- | Gzip compression with default parameters.+gzip :: (MonadThrow m, MonadBase base m, PrimMonad base) => Conduit ByteString m ByteString+gzip = compress 1 (WindowBits 31)++-- | Gzip decompression with default parameters.+ungzip :: (MonadBase base m, PrimMonad base, MonadThrow m) => Conduit ByteString m ByteString+ungzip = decompress (WindowBits 31)++unsafeLiftIO :: (MonadBase base m, PrimMonad base, MonadThrow m) => IO a -> m a+unsafeLiftIO = liftBase . unsafePrimToPrim++-- |+-- Decompress (inflate) a stream of 'ByteString's. For example:+--+-- >    sourceFile "test.z" $= decompress defaultWindowBits $$ sinkFile "test"++decompress+    :: (MonadBase base m, PrimMonad base, MonadThrow m)+    => WindowBits -- ^ Zlib parameter (see the zlib-bindings package as well as the zlib C library)+    -> Conduit ByteString m ByteString+decompress =+    helperDecompress (liftM (fmap Chunk) await) yield'+  where+    yield' Flush = return ()+    yield' (Chunk bs) = yield bs++-- | Same as 'decompress', but allows you to explicitly flush the stream.+decompressFlush+    :: (MonadBase base m, PrimMonad base, MonadThrow m)+    => WindowBits -- ^ Zlib parameter (see the zlib-bindings package as well as the zlib C library)+    -> Conduit (Flush ByteString) m (Flush ByteString)+decompressFlush = helperDecompress await yield++helperDecompress :: (Monad (t m), MonadBase base m, PrimMonad base, MonadThrow m, MonadTrans t)+                 => t m (Maybe (Flush ByteString))+                 -> (Flush ByteString -> t m ())+                 -> WindowBits+                 -> t m ()+helperDecompress await' yield' config =+    await' >>= maybe (return ()) start+  where+    start input = do+        inf <- lift $ unsafeLiftIO $ initInflate config+        push inf input++    continue inf = await' >>= maybe (close inf) (push inf)++    goPopper popper = do+        mbs <- lift $ unsafeLiftIO popper+        case mbs of+            PRDone -> return ()+            PRNext bs -> yield' (Chunk bs) >> goPopper popper+            PRError e -> lift $ monadThrow e++    push inf (Chunk x) = do+        popper <- lift $ unsafeLiftIO $ feedInflate inf x+        goPopper popper+        continue inf++    push inf Flush = do+        chunk <- lift $ unsafeLiftIO $ flushInflate inf+        unless (S.null chunk) $ yield' $ Chunk chunk+        yield' Flush+        continue inf++    close inf = do+        chunk <- lift $ unsafeLiftIO $ finishInflate inf+        unless (S.null chunk) $ yield' $ Chunk chunk++-- |+-- Compress (deflate) a stream of 'ByteString's. The 'WindowBits' also control+-- the format (zlib vs. gzip).++compress+    :: (MonadBase base m, PrimMonad base, MonadThrow m)+    => Int         -- ^ Compression level+    -> WindowBits  -- ^ Zlib parameter (see the zlib-bindings package as well as the zlib C library)+    -> Conduit ByteString m ByteString+compress =+    helperCompress (liftM (fmap Chunk) await) yield'+  where+    yield' Flush = return ()+    yield' (Chunk bs) = yield bs++-- | Same as 'compress', but allows you to explicitly flush the stream.+compressFlush+    :: (MonadBase base m, PrimMonad base, 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)+compressFlush = helperCompress await yield++helperCompress :: (Monad (t m), MonadBase base m, PrimMonad base, MonadThrow m, MonadTrans t)+               => t m (Maybe (Flush ByteString))+               -> (Flush ByteString -> t m ())+               -> Int+               -> WindowBits+               -> t m ()+helperCompress await' yield' level config =+    await' >>= maybe (return ()) start+  where+    start input = do+        def <- lift $ unsafeLiftIO $ initDeflate level config+        push def input++    continue def = await' >>= maybe (close def) (push def)++    goPopper popper = do+        mbs <- lift $ unsafeLiftIO popper+        case mbs of+            PRDone -> return ()+            PRNext bs -> yield' (Chunk bs) >> goPopper popper+            PRError e -> lift $ monadThrow e++    push def (Chunk x) = do+        popper <- lift $ unsafeLiftIO $ feedDeflate def x+        goPopper popper+        continue def++    push def Flush = do+        mchunk <- lift $ unsafeLiftIO $ flushDeflate def+        case mchunk of+            PRDone -> return ()+            PRNext x -> yield' $ Chunk x+            PRError e -> lift $ monadThrow e+        yield' Flush+        continue def++    close def = do+        mchunk <- lift $ unsafeLiftIO $ finishDeflate def+        case mchunk of+            PRDone -> return ()+            PRNext chunk -> yield' (Chunk chunk) >> close def+            PRError e -> lift $ monadThrow e
conduit-extra.cabal view
@@ -1,7 +1,8 @@ Name:                conduit-extra-Version:             1.0.0.1-Synopsis:            Temporary placeholder package.-Description:         This package will soon contain much more functionality. This is a placeholder until the next release is ready.+Version:             1.1.0+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. License:             MIT License-file:        LICENSE Author:              Michael Snoyman@@ -10,10 +11,75 @@ Build-type:          Simple Cabal-version:       >=1.8 Homepage:            http://github.com/snoyberg/conduit+extra-source-files:+    test/random+    test/filesystem/*.txt+    test/filesystem/bin/*.txt  Library+  Exposed-modules:     Data.Conduit.Attoparsec+                       Data.Conduit.Binary+                       Data.Conduit.Blaze+                       Data.Conduit.Filesystem+                       Data.Conduit.Lazy+                       Data.Conduit.Network+                       Data.Conduit.Network.UDP+                       Data.Conduit.Text+                       Data.Conduit.Zlib+  if !os(windows)+      Exposed-modules: Data.Conduit.Network.Unix+   Build-depends:       base                     >= 4            && < 5-                     , conduit < 1.1+                     , conduit                  >= 1.1          && < 1.2++                       -- No version bounds necessary, since they're inherited+                       -- from conduit.+                     , bytestring+                     , monad-control+                     , text+                     , transformers+                     , transformers-base++                     , attoparsec               >= 0.10         && < 0.12+                     , blaze-builder            >= 0.3          && < 0.4+                     , directory+                     , filepath+                     , network                  >= 2.3          && < 2.5+                     , primitive                >= 0.5          && < 0.6+                     , resourcet                >= 1.1          && < 1.2+                     , streaming-commons        >= 0.1          && < 0.2++  ghc-options:     -Wall++test-suite test+    hs-source-dirs: test+    main-is: Spec.hs+    type: exitcode-stdio-1.0+    ghc-options:   -threaded+    cpp-options:   -DTEST+    build-depends:   conduit+                   , conduit-extra+                   , base+                   , hspec >= 1.3++                   , attoparsec+                   , blaze-builder+                   , bytestring+                   , exceptions+                   , resourcet+                   , text+                   , transformers+                   , transformers-base+    ghc-options:     -Wall+    other-modules:   Data.Conduit.AttoparsecSpec+                     Data.Conduit.BinarySpec+                     Data.Conduit.BlazeSpec+                     Data.Conduit.ExtraSpec+                     Data.Conduit.FilesystemSpec+                     Data.Conduit.LazySpec+                     Data.Conduit.NetworkSpec+                     Data.Conduit.TextSpec+                     Data.Conduit.ZlibSpec  source-repository head   type:     git
+ test/Data/Conduit/AttoparsecSpec.hs view
@@ -0,0 +1,108 @@+{-# LANGUAGE CPP               #-}+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -fno-warn-incomplete-patterns #-}+module Data.Conduit.AttoparsecSpec (spec) where+import           Control.Exception                (fromException)+import           Test.Hspec++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+import           Data.Conduit.Attoparsec+import qualified Data.Conduit.List                as CL++spec :: Spec+spec = describe "Data.Conduit.AttoparsecSpec" $ do+    describe "error position" $ do+        it "works for text" $ do+            let input = ["aaa\na", "aaa\n\n", "aaa", "aab\n\naaaa"]+                badLine = 4+                badCol = 6+                parser = Data.Attoparsec.Text.endOfInput <|> (Data.Attoparsec.Text.notChar 'b' >> parser)+                sink = sinkParser parser+            ea <- runExceptionT $ CL.sourceList input $$ sink+            case ea of+                Left e ->+                    case fromException e of+                        Just pe -> do+                            errorPosition pe `shouldBe` Position badLine badCol+        it "works for bytestring" $ do+            let input = ["aaa\na", "aaa\n\n", "aaa", "aab\n\naaaa"]+                badLine = 4+                badCol = 6+                parser = Data.Attoparsec.ByteString.Char8.endOfInput <|> (Data.Attoparsec.ByteString.Char8.notChar 'b' >> parser)+                sink = sinkParser parser+            ea <- runExceptionT $ CL.sourceList input $$ sink+            case ea of+                Left e ->+                    case fromException e of+                        Just pe -> do+                            errorPosition pe `shouldBe` Position badLine badCol+        it "works in last chunk" $ do+            let input = ["aaa\na", "aaa\n\n", "aaa", "aab\n\naaaa"]+                badLine = 6+                badCol = 5+                parser = Data.Attoparsec.Text.char 'c' <|> (Data.Attoparsec.Text.anyChar >> parser)+                sink = sinkParser parser+            ea <- runExceptionT $ CL.sourceList input $$ sink+            case ea of+                Left e ->+                    case fromException e of+                        Just pe -> do+                            errorPosition pe `shouldBe` Position badLine badCol+        it "works in last chunk" $ do+            let input = ["aaa\na", "aaa\n\n", "aaa", "aa\n\naaaab"]+                badLine = 6+                badCol = 6+                parser = Data.Attoparsec.Text.string "bc" <|> (Data.Attoparsec.Text.anyChar >> parser)+                sink = sinkParser parser+            ea <- runExceptionT $ CL.sourceList input $$ sink+            case ea of+                Left e ->+                    case fromException e of+                        Just pe -> do+                            errorPosition pe `shouldBe` Position badLine badCol+        it "works after new line in text" $ do+            let input = ["aaa\n", "aaa\n\n", "aaa", "aa\nb\naaaa"]+                badLine = 5+                badCol = 1+                parser = Data.Attoparsec.Text.endOfInput <|> (Data.Attoparsec.Text.notChar 'b' >> parser)+                sink = sinkParser parser+            ea <- runExceptionT $ CL.sourceList input $$ sink+            case ea of+                Left e ->+                    case fromException e of+                        Just pe -> do+                            errorPosition pe `shouldBe` Position badLine badCol+        it "works after new line in bytestring" $ do+            let input = ["aaa\n", "aaa\n\n", "aaa", "aa\nb\naaaa"]+                badLine = 5+                badCol = 1+                parser = Data.Attoparsec.ByteString.Char8.endOfInput <|> (Data.Attoparsec.ByteString.Char8.notChar 'b' >> parser)+                sink = sinkParser parser+            ea <- runExceptionT $ CL.sourceList input $$ sink+            case ea of+                Left e ->+                    case fromException e of+                        Just pe -> do+                            errorPosition pe `shouldBe` Position badLine badCol++    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+            let chk a = case a of+                          Left{} -> False+                          Right (_, xs) -> xs == "aaa"+                chkp 1 = (PositionRange (Position 1 0) (Position 2 1))+                chkp l = (PositionRange (Position l 1) (Position (l+1) 1))+            forM_ ea $ \ a -> a `shouldSatisfy` chk :: Expectation+            forM_ (zip ea [1..]) $ \ (Right (pos, _), l) -> pos `shouldBe` chkp l+            length ea `shouldBe` 4++
+ test/Data/Conduit/BinarySpec.hs view
@@ -0,0 +1,193 @@+{-# LANGUAGE OverloadedStrings #-}+module Data.Conduit.BinarySpec (spec) where++import qualified Data.Conduit.Binary as CB+import qualified Data.Conduit as C+import qualified Data.Conduit.List as CL+import Control.Monad.Trans.Resource+import Control.Monad.IO.Class+import Control.Exception (IOException)+import qualified Data.ByteString.Lazy as L+import Test.Hspec+import Test.Hspec.QuickCheck+import qualified Data.IORef as I+import Control.Monad.Trans.Writer.Strict+import qualified Data.ByteString as S+import qualified Data.ByteString.Char8 as S8+import Data.Functor.Identity++spec :: Spec+spec = describe "Data.Conduit.Binary" $ do++    describe "file access" $ do+        it "read" $ do+            bs <- S.readFile "conduit-extra.cabal"+            bss <- runResourceT $ CB.sourceFile "conduit-extra.cabal" C.$$ 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+            S.concat bss `shouldBe` "234"++        it "write" $ do+            runResourceT $ CB.sourceFile "conduit-extra.cabal" C.$$ CB.sinkFile "tmp"+            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"+            bs1 <- S.readFile "conduit-extra.cabal"+            bs2 <- S.readFile "tmp"+            bs3 <- S.readFile "tmp2"+            bs2 `shouldBe` bs1+            bs3 `shouldBe` bs1+    describe "binary isolate" $ do+        it "works" $ do+            bss <- runResourceT $ CL.sourceList (replicate 1000 "X")+                           C.$= CB.isolate 6+                           C.$$ 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+            lbs <- L.readFile "test/random"+            L.fromChunks x `shouldBe` lbs++    describe "binary head" $ do+        let go lbs = do+                x <- CB.head+                case (x, L.uncons lbs) of+                    (Nothing, Nothing) -> return True+                    (Just y, Just (z, lbs'))+                        | y == z -> go lbs'+                    _ -> return False++        prop "works" $ \bss' ->+            let bss = map S.pack bss'+             in runIdentity $+                CL.sourceList bss C.$$ 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+                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+                    y <- CL.consume+                    return (S.concat x, S.concat y)+                let expected = S.span (>= 5) $ S.concat bss+                if result == expected+                    then return True+                    else error $ show (S.concat bss, result, expected)++    describe "binary dropWhile" $ do+        prop "works" $ \bss' ->+            let bss = map S.pack bss'+             in runIdentity $ do+                bss2 <- CL.sourceList bss C.$$ 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+          a <- CB.take n+          b <- CL.consume+          return (a, b)++      -- Taking nothing should result in an empty Bytestring+      it "nothing" $ do+        (a, b) <- runResourceT $ go 0 ["abc", "defg"]+        a              `shouldBe` L.empty+        L.fromChunks b `shouldBe` "abcdefg"++      it "normal" $ do+        (a, b) <- runResourceT $ go 4 ["abc", "defg"]+        a              `shouldBe` "abcd"+        L.fromChunks b `shouldBe` "efg"++      -- Taking exactly the data that is available should result in no+      -- leftover.+      it "all" $ do+        (a, b) <- runResourceT $ go 7 ["abc", "defg"]+        a `shouldBe` "abcdefg"+        b `shouldBe` []++      -- Take as much as possible.+      it "more" $ do+        (a, b) <- runResourceT $ go 10 ["abc", "defg"]+        a `shouldBe` "abcdefg"+        b `shouldBe` []++    describe "binary" $ do+        prop "lines" $ \bss' -> runIdentity $ do+            let bss = map S.pack bss'+                bs = S.concat bss+                src = CL.sourceList bss+            res <- src C.$$ CB.lines C.=$ 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+            liftIO $ do+                fromIntegral len `shouldBe` L.length lbs+                lbs' `shouldBe` lbs+                fromIntegral len `shouldBe` L.length lbs'++    describe "Data.Conduit.Binary.mapM_" $ do+        prop "telling works" $ \bytes ->+            let lbs = L.pack bytes+                src = CB.sourceLbs lbs+                sink = CB.mapM_ (tell . return . S.singleton)+                bss = execWriter $ src C.$$ sink+             in L.fromChunks bss == lbs++    describe "exception handling" $ do+        it "catchC" $ do+            ref <- I.newIORef 0+            let src = do+                    C.catchC (CB.sourceFile "some-file-that-does-not-exist") onErr+                    C.handleC onErr $ CB.sourceFile "conduit-extra.cabal"+                onErr :: MonadIO m => IOException -> m ()+                onErr _ = liftIO $ I.modifyIORef ref (+ 1)+            contents <- L.readFile "conduit-extra.cabal"+            res <- runResourceT $ src C.$$ CB.sinkLbs+            res `shouldBe` contents+            errCount <- I.readIORef ref+            errCount `shouldBe` (1 :: Int)+        it "tryC" $ do+            ref <- I.newIORef undefined+            let src = do+                    res1 <- C.tryC $ CB.sourceFile "some-file-that-does-not-exist"+                    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 `shouldBe` contents+            exc <- I.readIORef ref+            case exc :: (Either IOException (), Either IOException ()) of+                (Left _, Right ()) ->+                    return ()+                _ -> error $ show exc++    describe "normalFuseLeft" $ do+        it "does not double close conduit" $ do+            x <- runResourceT $ do+                let src = CL.sourceList ["foobarbazbin"]+                src C.$= CB.isolate 10 C.$$ CL.head+            x `shouldBe` Just "foobarbazb"++it' :: String -> IO () -> Spec+it' = it
+ test/Data/Conduit/BlazeSpec.hs view
@@ -0,0 +1,63 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE CPP #-}+module Data.Conduit.BlazeSpec (spec) where++import Test.Hspec+import Test.Hspec.QuickCheck (prop)++import qualified Data.Conduit as C+import qualified Data.Conduit.List as CL+import Data.ByteString.Char8 ()+import Data.Conduit.Blaze (builderToByteString, builderToByteStringFlush)+import Control.Monad.ST (runST)+import Data.Monoid+import qualified Data.ByteString as S+import Blaze.ByteString.Builder (fromByteString, toLazyByteString, insertLazyByteString, flush)+import qualified Data.ByteString.Lazy as L+import Data.ByteString.Lazy.Char8 ()++spec :: Spec+spec =+    describe "Data.Conduit.Blaze" $ do+        prop "idempotent to toLazyByteString" $ \bss' -> runST $ do+            let bss = map S.pack bss'+            let builders = map fromByteString bss+            let lbs = toLazyByteString $ mconcat builders+            let src = mconcat $ map (CL.sourceList . return) builders+            outBss <- src C.$= builderToByteString C.$$ CL.consume+            return $ lbs == L.fromChunks outBss++        it "works for large input" $ do+            let builders = replicate 10000 (fromByteString "hello world!")+            let lbs = toLazyByteString $ mconcat builders+            let src = mconcat $ map (CL.sourceList . return) builders+            outBss <- src C.$= builderToByteString C.$$ CL.consume+            lbs `shouldBe` L.fromChunks outBss++        it "works for lazy bytestring insertion" $ do+            let builders = replicate 10000 (insertLazyByteString "hello world!")+            let lbs = toLazyByteString $ mconcat builders+            let src = mconcat $ map (CL.sourceList . return) builders+            outBss <- src C.$= builderToByteString C.$$ CL.consume+            lbs `shouldBe` L.fromChunks outBss++        it "flush shouldn't bring in empty strings." $ do+            let dat = ["hello", "world"]+                src = CL.sourceList dat C.$= CL.map ((`mappend` flush) . fromByteString)+            out <- src C.$= builderToByteString C.$$ CL.consume+            dat `shouldBe` out++        prop "flushing" $ \bss' -> runST $ do+            let bss = concatMap (\bs -> [C.Chunk $ S.pack bs, C.Flush]) $ filter (not . null) bss'+            let src = CL.sourceList $ map (fmap fromByteString) bss+            outBss <- src C.$= builderToByteStringFlush C.$$ CL.consume+            if bss == outBss then return () else error (show (bss, outBss))+            return $ bss == outBss+        it "large flush input" $ do+            let lbs = L.pack $ concat $ replicate 100000 [0..255]+                chunks = map (C.Chunk . fromByteString) (L.toChunks lbs)+                src = CL.sourceList chunks+            bss <- src C.$$ builderToByteStringFlush C.=$ CL.consume+            let unFlush (C.Chunk x) = [x]+                unFlush C.Flush = []+            L.fromChunks (concatMap unFlush bss) `shouldBe` lbs
+ test/Data/Conduit/ExtraSpec.hs view
@@ -0,0 +1,71 @@+module Data.Conduit.ExtraSpec where++import Data.Conduit+import Test.Hspec+import Test.Hspec.QuickCheck+import Data.Conduit.List (isolate, peek, consume)+import qualified Data.Conduit.List as CL+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import qualified Data.ByteString as S+import qualified Data.Conduit.Text as CT++spec :: Spec+spec = describe "Data.Conduit.Extra" $ do+    it "basic test" $ do+        let sink2 :: Sink a 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)++    it "get leftovers" $ do+        let sink2 :: Sink a IO ([a], [a], [a])+            sink2 = do+                (x, y) <- fuseReturnLeftovers (isolate 2) peek3+                z <- CL.consume+                return (x, y, z)++            peek3 = do+                x <- CL.take 3+                mapM_ leftover $ reverse x+                return x++            source = mapM_ yield [1..5]+        res <- source $$ sink2+        res `shouldBe` ([1..2], [1..2], [3..5])++    it "multiple values" $ do+        let sink2 :: Sink a IO ([a], Maybe a)+            sink2 = do+                ma1 <- fuseLeftovers id (isolate 10) peek3+                ma2 <- peek+                return (ma1, ma2)++            peek3 = do+                x <- CL.take 3+                mapM_ leftover $ reverse x+                return x++            source = mapM_ yield [1..5]+        res <- source $$ sink2+        res `shouldBe` ([1..3], Just 1)++    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+            undo = return . T.encodeUtf8 . T.concat+        res <- src $$ do+            x <- fuseLeftovers undo conduit sink+            y <- consume+            return (T.concat x, T.decodeUtf8 $ S.concat y)+        res `shouldBe` T.splitAt cnt (T.concat ts)++main :: IO ()+main = hspec spec
+ test/Data/Conduit/FilesystemSpec.hs view
@@ -0,0 +1,38 @@+module Data.Conduit.FilesystemSpec (spec) where++import Test.Hspec+import Data.Conduit+import qualified Data.Conduit.List as CL+import Data.Conduit.Filesystem+import Data.List (sort, isSuffixOf)+import Control.Monad.Trans.Resource (runResourceT)++spec :: Spec+spec = describe "Data.Conduit.Filesystem" $ do+    it "sourceDirectory" $ do+        res <- runResourceT+             $ 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"+            ]+    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+        sort res1 `shouldBe`+            [ "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
@@ -0,0 +1,44 @@+module Data.Conduit.LazySpec (spec) where++import qualified Data.Conduit.Lazy as CLazy+import Test.Hspec+import Control.Monad.IO.Class+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)++spec :: Spec+spec = describe "Data.Conduit.Lazy" $ do++    describe "lazy" $ do+        it' "works inside a ResourceT" $ runResourceT $ do+            counter <- liftIO $ I.newIORef 0+            let incr i = do+                    istate <- liftIO $ I.newIORef $ Just (i :: Int)+                    let loop = do+                            res <- liftIO $ I.atomicModifyIORef istate ((,) Nothing)+                            case res of+                                Nothing -> return ()+                                Just x -> do+                                    count <- liftIO $ I.atomicModifyIORef counter+                                        (\j -> (j + 1, j + 1))+                                    liftIO $ count `shouldBe` i+                                    C.yield x+                                    loop+                    loop+            nums <- CLazy.lazyConsume $ mconcat $ map incr [1..10]+            liftIO $ nums `shouldBe` [1..10]++        it' "returns nothing outside ResourceT" $ do+            bss <- runResourceT $ CLazy.lazyConsume $ CB.sourceFile "test/main.hs"+            bss `shouldBe` []++        it' "works with pure sources" $ do+            nums <- CLazy.lazyConsume $ forever $ C.yield 1+            take 100 nums `shouldBe` replicate 100 (1 :: Int)++it' :: String -> IO () -> Spec+it' = it
+ test/Data/Conduit/NetworkSpec.hs view
@@ -0,0 +1,21 @@+{-# LANGUAGE OverloadedStrings #-}+module Data.Conduit.NetworkSpec (spec) where++import Data.Conduit+import Data.Conduit.Network+import Control.Concurrent (forkIO, threadDelay)+import Control.Monad (replicateM_)+import Test.Hspec++spec :: Spec+spec = it "Data.Conduit.Network" $ do+    _ <- forkIO $ runTCPServer (serverSettings 4009 "*4") echo+    threadDelay 1000000+    replicateM_ 10000+        $ runTCPClient (clientSettings 4009 "127.0.0.1") doNothing++echo :: AppData -> IO ()+echo ad = appSource ad $$ appSink ad++doNothing :: AppData -> IO ()+doNothing _ = return ()
+ test/Data/Conduit/TextSpec.hs view
@@ -0,0 +1,197 @@+{-# LANGUAGE OverloadedStrings #-}+module Data.Conduit.TextSpec (spec) where++import qualified Data.Conduit.Text as CT+import qualified Data.Conduit as C+import qualified Data.Conduit.Lift as C+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.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_)++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+                    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+                    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++                -- 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 ->+                    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+                          (Left _) -> (return $! TL.toStrict tl') `shouldThrow` anyException+                          (Right tl) -> TL.fromChunks tl `shouldBe` tl'+                prop "encoding" $ \chars -> runIdentity $ runExceptionT_ $ 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+                    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+        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+        go "utf32_le" TLE.encodeUtf32LE TLE.decodeUtf32LE CT.utf32_le+        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+            text `shouldBe` Just "8:u"+            (src C.$$ 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+            text `shouldBe` Just (T.pack $ map toEnum [0..127])+            (src C.$$ 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)+                    (\e -> C.yield (Left (e :: CT.TextException)))++            res <- C.yield badBS C.$$ (,)+                <$> (grabExceptions (CT.decode CT.utf8) C.=$ CL.consume)+                <*> CL.consume++            first (map (either (Left . show) Right)) res `shouldBe`+                ( [ Right "this is good"+                  , Left $ show $ CT.NewDecodeException "UTF-8" 12 "\128\128\0t"+                  ]+                , ["\128\128\0that was bad"]+                )+        it "catch UTF8 exceptions, pure" $ do+            let badBS = "this is good\128\128\0that was bad"++                grabExceptions inner = do+                    res <- C.runCatchC $ inner C.=$= 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)+                        <*> CL.consume++            first (map (either (Left . show) Right)) res `shouldBe`+                ( [ Right "this is good"+                  , Left $ show $ CT.NewDecodeException "UTF-8" 12 "\128\128\0t"+                  ]+                , ["\128\128\0that was bad"]+                )+        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)+                    (\e -> C.yield $ Left e)++            let res = runException_ $ C.yield badBS C.$$ (,)+                        <$> (grabExceptions (CT.decode CT.utf8) C.=$ CL.consume)+                        <*> CL.consume++            first (map (either (Left . show) Right)) res `shouldBe`+                ( [ Right "this is good"+                  , Left $ show $ CT.NewDecodeException "UTF-8" 12 "\128\128\0t"+                  ]+                , ["\128\128\0that was bad"]+                )+        it "catch UTF8 exceptions, catchExceptionC, decodeUtf8" $ do+            let badBS = "this is good\128\128\0that was bad"++                grabExceptions inner = C.catchCatchC+                    (inner C.=$= CL.map Right)+                    (\e -> C.yield $ Left e)++            let res = runException_ $ C.yield badBS C.$$ (,)+                        <$> (grabExceptions CT.decodeUtf8 C.=$ CL.consume)+                        <*> CL.consume++            first (map (either (Left . show) Right)) res `shouldBe`+                ( [ Right "this is good"+                  , Left $ show $ CT.NewDecodeException "UTF-8" 12 "\128\128\0t"+                  ]+                , ["\128\128\0that was bad"]+                )++    describe "text lines" $ do+        it "works across split lines" $+            (CL.sourceList [T.pack "abc", T.pack "d\nef"] C.$= CT.lines C.$$ CL.consume) ==+                [[T.pack "abcd", T.pack "ef"]]+        it "works with multiple lines in an item" $+            (CL.sourceList [T.pack "ab\ncd\ne"] C.$= CT.lines C.$$ CL.consume) ==+                [[T.pack "ab", T.pack "cd", T.pack "e"]]+        it "works with ending on a newline" $+            (CL.sourceList [T.pack "ab\n"] C.$= CT.lines C.$$ CL.consume) ==+                [[T.pack "ab"]]+        it "works with ending a middle item on a newline" $+            (CL.sourceList [T.pack "ab\n", T.pack "cd\ne"] C.$= CT.lines C.$$ CL.consume) ==+                [[T.pack "ab", T.pack "cd", T.pack "e"]]+        it "is not too eager" $ do+            x <- CL.sourceList ["foobarbaz", error "ignore me"] C.$$ CT.decode CT.utf8 C.=$ CL.head+            x `shouldBe` Just "foobarbaz"++    describe "text lines bounded" $ do+        it "works across split lines" $+            (CL.sourceList [T.pack "abc", T.pack "d\nef"] C.$= CT.linesBounded 80 C.$$ CL.consume) ==+                [[T.pack "abcd", T.pack "ef"]]+        it "works with multiple lines in an item" $+            (CL.sourceList [T.pack "ab\ncd\ne"] C.$= CT.linesBounded 80 C.$$ CL.consume) ==+                [[T.pack "ab", T.pack "cd", T.pack "e"]]+        it "works with ending on a newline" $+            (CL.sourceList [T.pack "ab\n"] C.$= CT.linesBounded 80 C.$$ CL.consume) ==+                [[T.pack "ab"]]+        it "works with ending a middle item on a newline" $+            (CL.sourceList [T.pack "ab\n", T.pack "cd\ne"] C.$= CT.linesBounded 80 C.$$ CL.consume) ==+                [[T.pack "ab", T.pack "cd", T.pack "e"]]+        it "is not too eager" $ do+            x <- CL.sourceList ["foobarbaz", error "ignore me"] C.$$ CT.decode CT.utf8 C.=$ CL.head+            x `shouldBe` Just "foobarbaz"+        it "throws an exception when lines are too long" $ do+            x <- runExceptionT $ CL.sourceList ["hello\nworld"] C.$$ CT.linesBounded 4 C.=$ CL.consume+            show x `shouldBe` show (Left $ CT.LengthExceeded 4 :: 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+            case x of+                Left _ -> return ()+                Right t -> error $ "This should have failed: " ++ show t++it' :: String -> IO () -> Spec+it' = it
+ test/Data/Conduit/ZlibSpec.hs view
@@ -0,0 +1,49 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE UndecidableInstances #-}+module Data.Conduit.ZlibSpec (spec) where++import Test.Hspec+import Test.Hspec.QuickCheck (prop)++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.Lazy.Char8 ()+import Control.Monad.Trans.Resource (runExceptionT_)+import Control.Monad.Trans.Class+import Control.Monad.Catch.Pure+import Control.Monad.Base++instance MonadBase base m => MonadBase base (CatchT m) where+    liftBase = lift . liftBase++spec :: Spec+spec = describe "Data.Conduit.Zlib" $ do+        prop "idempotent" $ \bss' -> runST $ do+            let bss = map S.pack bss'+                lbs = L.fromChunks bss+                src = mconcat $ map (CL.sourceList . return) bss+            outBss <- runExceptionT_ $ src C.$= CZ.gzip C.$= CZ.ungzip C.$$ CL.consume+            return $ lbs == L.fromChunks outBss+        prop "flush" $ \bss' -> runST $ do+            let bss = map S.pack $ filter (not . null) bss'+                bssC = concatMap (\bs -> [C.Chunk bs, C.Flush]) bss+                src = mconcat $ map (CL.sourceList . return) bssC+            outBssC <- runExceptionT_+                     $ src C.$= CZ.compressFlush 5 (CZ.WindowBits 31)+                           C.$= CZ.decompressFlush (CZ.WindowBits 31)+                           C.$$ CL.consume+            return $ bssC == outBssC+        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+            let unChunk (C.Chunk x) = [x]+                unChunk C.Flush = []+            bss <- CL.sourceList bssC C.$$ CL.concatMap unChunk C.=$ CZ.ungzip C.=$ CL.consume+            L.fromChunks bss `shouldBe` content
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}
+ test/filesystem/bar.txt view
+ test/filesystem/baz.txt view
+ test/filesystem/bin/bin.txt view
+ test/filesystem/foo.txt view
+ test/random view

binary file changed (absent → 1024 bytes)