diff --git a/Data/Conduit/Attoparsec.hs b/Data/Conduit/Attoparsec.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Attoparsec.hs @@ -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 #-} diff --git a/Data/Conduit/Binary.hs b/Data/Conduit/Binary.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Binary.hs @@ -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_ #-} diff --git a/Data/Conduit/Blaze.hs b/Data/Conduit/Blaze.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Blaze.hs @@ -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' diff --git a/Data/Conduit/Filesystem.hs b/Data/Conduit/Filesystem.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Filesystem.hs @@ -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 () diff --git a/Data/Conduit/Lazy.hs b/Data/Conduit/Lazy.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Lazy.hs @@ -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 diff --git a/Data/Conduit/Network.hs b/Data/Conduit/Network.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Network.hs @@ -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 diff --git a/Data/Conduit/Network/UDP.hs b/Data/Conduit/Network/UDP.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Network/UDP.hs @@ -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 #-} diff --git a/Data/Conduit/Network/Unix.hs b/Data/Conduit/Network/Unix.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Network/Unix.hs @@ -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 diff --git a/Data/Conduit/Text.hs b/Data/Conduit/Text.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Text.hs @@ -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 #-} diff --git a/Data/Conduit/Zlib.hs b/Data/Conduit/Zlib.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Zlib.hs @@ -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 diff --git a/conduit-extra.cabal b/conduit-extra.cabal --- a/conduit-extra.cabal +++ b/conduit-extra.cabal @@ -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 diff --git a/test/Data/Conduit/AttoparsecSpec.hs b/test/Data/Conduit/AttoparsecSpec.hs new file mode 100644 --- /dev/null +++ b/test/Data/Conduit/AttoparsecSpec.hs @@ -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 + + diff --git a/test/Data/Conduit/BinarySpec.hs b/test/Data/Conduit/BinarySpec.hs new file mode 100644 --- /dev/null +++ b/test/Data/Conduit/BinarySpec.hs @@ -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 diff --git a/test/Data/Conduit/BlazeSpec.hs b/test/Data/Conduit/BlazeSpec.hs new file mode 100644 --- /dev/null +++ b/test/Data/Conduit/BlazeSpec.hs @@ -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 diff --git a/test/Data/Conduit/ExtraSpec.hs b/test/Data/Conduit/ExtraSpec.hs new file mode 100644 --- /dev/null +++ b/test/Data/Conduit/ExtraSpec.hs @@ -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 diff --git a/test/Data/Conduit/FilesystemSpec.hs b/test/Data/Conduit/FilesystemSpec.hs new file mode 100644 --- /dev/null +++ b/test/Data/Conduit/FilesystemSpec.hs @@ -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 diff --git a/test/Data/Conduit/LazySpec.hs b/test/Data/Conduit/LazySpec.hs new file mode 100644 --- /dev/null +++ b/test/Data/Conduit/LazySpec.hs @@ -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 diff --git a/test/Data/Conduit/NetworkSpec.hs b/test/Data/Conduit/NetworkSpec.hs new file mode 100644 --- /dev/null +++ b/test/Data/Conduit/NetworkSpec.hs @@ -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 () diff --git a/test/Data/Conduit/TextSpec.hs b/test/Data/Conduit/TextSpec.hs new file mode 100644 --- /dev/null +++ b/test/Data/Conduit/TextSpec.hs @@ -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 diff --git a/test/Data/Conduit/ZlibSpec.hs b/test/Data/Conduit/ZlibSpec.hs new file mode 100644 --- /dev/null +++ b/test/Data/Conduit/ZlibSpec.hs @@ -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 diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,1 @@ +{-# OPTIONS_GHC -F -pgmF hspec-discover #-} diff --git a/test/filesystem/bar.txt b/test/filesystem/bar.txt new file mode 100644 --- /dev/null +++ b/test/filesystem/bar.txt diff --git a/test/filesystem/baz.txt b/test/filesystem/baz.txt new file mode 100644 --- /dev/null +++ b/test/filesystem/baz.txt diff --git a/test/filesystem/bin/bin.txt b/test/filesystem/bin/bin.txt new file mode 100644 --- /dev/null +++ b/test/filesystem/bin/bin.txt diff --git a/test/filesystem/foo.txt b/test/filesystem/foo.txt new file mode 100644 --- /dev/null +++ b/test/filesystem/foo.txt diff --git a/test/random b/test/random new file mode 100644 Binary files /dev/null and b/test/random differ