conduit-combinators 1.0.0 → 1.0.1

raw patch · 7 files changed

+201/−66 lines, 7 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog.md +4/−0
• Data/Conduit/Combinators.hs +67/−18
• Data/Conduit/Combinators/Stream.hs +36/−0
• Data/Conduit/Combinators/Unqualified.hs +22/−27
• conduit-combinators.cabal +1/−1
• test/Spec.hs +46/−18
• test/StreamSpec.hs +25/−2

@@ -1,3 +1,7 @@+# 1.0.1++* mapAccumWhile, mapAccumWhileM, mapAccumS+ # 1.0.0  * Drop system-filepath/system-fileio

@@ -141,6 +141,7 @@     , mapWhile     , conduitVector     , scanl+    , mapAccumWhile     , concatMapAccum     , intersperse     , slidingWindow@@ -162,6 +163,7 @@     , filterME     , iterM     , scanlM+    , mapAccumWhileM     , concatMapAccumM        -- ** Textual@@ -179,6 +181,7 @@        -- * Special     , vectorBuilder+    , mapAccumS     , peekForever     ) where @@ -213,12 +216,10 @@ import qualified Data.Vector.Generic         as V import qualified Data.Vector.Generic.Mutable as VM import           Data.Void                   (absurd)-import qualified System.FilePath             as F-import           System.FilePath             ((</>)) import           Prelude                     (Bool (..), Eq (..), Int,-                                              Maybe (..), Monad (..), Num (..),-                                              Ord (..), fromIntegral, maybe,-                                              ($), Functor (..), Enum, seq, Show, Char, (||),+                                              Maybe (..), Either (..), Monad (..), Num (..),+                                              Ord (..), fromIntegral, maybe, either,+                                              ($), Functor (..), Enum, seq, Show, Char,                                               mod, otherwise, Either (..),                                               ($!), succ, FilePath) import Data.Word (Word8)@@ -233,14 +234,9 @@ import Data.Conduit.Combinators.Internal import Data.Conduit.Combinators.Stream import Data.Conduit.Internal.Fusion-import qualified System.PosixCompat.Files as PosixC import           Data.Primitive.MutVar       (MutVar, newMutVar, readMutVar,                                               writeMutVar) -#ifndef WINDOWS-import qualified System.Posix.Directory as Dir-#endif- -- Defines INLINE_RULE0, INLINE_RULE, STREAMING0, and STREAMING. #include "fusion-macros.h" @@ -540,11 +536,11 @@         then return ()         else await >>= maybe (return ()) (go i) -    go i seq = do+    go i sq = do         unless (onull y) $ leftover y         loop i'       where-        (x, y) = Seq.splitAt i seq+        (x, y) = Seq.splitAt i sq         i' = i - fromIntegral (olength x) {-# INLINEABLE dropE #-} @@ -572,10 +568,10 @@   where     loop = await >>= maybe (return ()) go -    go seq =+    go sq =         if onull x then loop else leftover x       where-        x = Seq.dropWhile f seq+        x = Seq.dropWhile f sq {-# INLINE dropWhileE #-}  -- | Monoidally combine all values in the stream.@@ -802,7 +798,11 @@ elemE :: (Monad m, Seq.EqSequence seq)       => Element seq       -> Consumer seq m Bool+#if MIN_VERSION_mono_traversable(0,8,0)+INLINE_RULE(elemE, f, any (oelem f))+#else INLINE_RULE(elemE, f, any (Seq.elem f))+#endif  -- | Are no values in the stream equal to the given value? --@@ -824,7 +824,11 @@ notElemE :: (Monad m, Seq.EqSequence seq)          => Element seq          -> Consumer seq m Bool+#if MIN_VERSION_mono_traversable(0,8,0)+INLINE_RULE(notElemE, x, all (onotElem x))+#else INLINE_RULE(notElemE, x, all (Seq.notElem x))+#endif  -- | Consume all incoming strict chunks into a lazy sequence. -- Note that the entirety of the sequence will be resident at memory.@@ -1349,12 +1353,12 @@         then return ()         else await >>= maybe (return ()) (go i) -    go i seq = do+    go i sq = do         unless (onull x) $ yield x         unless (onull y) $ leftover y         loop i'       where-        (x, y) = Seq.splitAt i seq+        (x, y) = Seq.splitAt i sq         i' = i - fromIntegral (olength x) {-# INLINEABLE takeE #-} @@ -1388,13 +1392,13 @@   where     loop = await >>= maybe (return ()) go -    go seq = do+    go sq = do         unless (onull x) $ yield x         if onull y             then loop             else leftover y       where-        (x, y) = Seq.span f seq+        (x, y) = Seq.span f sq {-# INLINE takeWhileE #-}  -- | Consume precisely the given number of values and feed them downstream.@@ -1509,6 +1513,23 @@             loop seed' STREAMING(scanl, scanlC, scanlS, f x) +-- | 'mapWhile' with a break condition dependent on an accumulator.+-- Equivalently, 'CL.mapAccum' as long as the result is @Right@. Instead of+-- producing a leftover, the breaking input determines the resulting+-- accumulator via @Left@.+--+-- Subject to fusion+mapAccumWhile, mapAccumWhileC :: Monad m =>+    (a -> s -> Either s (s, b)) -> s -> ConduitM a b m s+mapAccumWhileC f =+    loop+  where+    loop s = await >>= maybe (return s) go+      where+        go a = either return (\(s', b) -> yield b >> loop s') $ f a s+{-# INLINE mapAccumWhileC #-}+STREAMING(mapAccumWhile, mapAccumWhileC, mapAccumWhileS, f s)+ -- | 'concatMap' with an accumulator. -- -- Subject to fusion@@ -1751,6 +1772,20 @@             loop seed' STREAMING(scanlM, scanlMC, scanlMS, f x) +-- | Monadic `mapAccumWhile`.+--+-- Subject to fusion+mapAccumWhileM, mapAccumWhileMC :: Monad m =>+    (a -> s -> m (Either s (s, b))) -> s -> ConduitM a b m s+mapAccumWhileMC f =+    loop+  where+    loop s = await >>= maybe (return s) go+      where+        go a = lift (f a s) >>= either return (\(s', b) -> yield b >> loop s')+{-# INLINE mapAccumWhileMC #-}+STREAMING(mapAccumWhileM, mapAccumWhileMC, mapAccumWhileMS, f s)+ -- | 'concatMapM' with an accumulator. -- -- Subject to fusion@@ -1991,6 +2026,20 @@             writeMutVar ref $! S 0 mv' front'         else writeMutVar ref $! S idx' mv front {-# INLINE addE #-}++-- | Consume a source in a way piecewise defined by a controlling stream. The+-- latter will be evaluated until it terminates.+--+-- >>> let f a s = liftM (:s) $ mapC (*a) =$ CL.take a+-- >>> reverse $ runIdentity $ yieldMany [0..3] $$ mapAccumS f [] (yieldMany [1..])+-- [[],[1],[4,6],[12,15,18]] :: [[Int]]+mapAccumS :: Monad m => (a -> s -> Sink b m s) -> s -> Source m b -> Sink a m s+mapAccumS f s xs = do+    (zs, u) <- loop (newResumableSource xs, s)+    lift (closeResumableSource zs) >> return u+    where loop r@(ys, t) = await >>= maybe (return r) go+              where go a = lift (ys $$++ f a t) >>= loop+{-# INLINE mapAccumS #-}  -- | Run a consuming conduit repeatedly, only stopping when there is no more -- data available from upstream.

@@ -28,6 +28,8 @@   , concatS   , scanlS   , scanlMS+  , mapAccumWhileS+  , mapAccumWhileMS   , intersperseS   , slidingWindowS   , filterMS@@ -49,7 +51,9 @@ import           Data.IOData import           Data.Maybe (isNothing, isJust) import           Data.MonoTraversable+#if ! MIN_VERSION_base(4,8,0) import           Data.Monoid (Monoid (..))+#endif import qualified Data.NonNull as NonNull import qualified Data.Sequences as Seq import           Data.Sequences.Lazy@@ -304,6 +308,38 @@                 !seed' <- f seed x                 return $ Emit (ScanContinues seed' s') seed {-# INLINE scanlMS #-}++mapAccumWhileS :: Monad m =>+    (a -> s -> Either s (s, b)) -> s -> StreamConduitM a b m s+mapAccumWhileS f initial (Stream step ms0) =+    Stream step' (liftM (initial, ) ms0)+  where+    step' (accum, s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop accum+            Skip s' -> Skip (accum, s')+            Emit s' x -> case f x accum of+                Right (accum', r) -> Emit (accum', s') r+                Left   accum'     -> Stop accum'+{-# INLINE mapAccumWhileS #-}++mapAccumWhileMS :: Monad m =>+    (a -> s -> m (Either s (s, b))) -> s -> StreamConduitM a b m s+mapAccumWhileMS f initial (Stream step ms0) =+    Stream step' (liftM (initial, ) ms0)+  where+    step' (accum, s) = do+        res <- step s+        case res of+            Stop () -> return $ Stop accum+            Skip s' -> return $ Skip (accum, s')+            Emit s' x -> do+                lr <- f x accum+                case lr of+                    Right (accum', r) -> return $ Emit (accum', s') r+                    Left   accum'     -> return $ Stop accum'+{-# INLINE mapAccumWhileMS #-}  data IntersperseState a s     = IFirstValue s

@@ -7,7 +7,6 @@ {-# LANGUAGE MultiParamTypeClasses     #-} {-# LANGUAGE NoImplicitPrelude         #-} {-# LANGUAGE NoMonomorphismRestriction #-}-{-# LANGUAGE BangPatterns #-} module Data.Conduit.Combinators.Unqualified     ( -- ** Producers       -- *** Pure@@ -128,6 +127,7 @@     , mapWhileC     , conduitVector     , scanlC+    , mapAccumWhileC     , concatMapAccumC     , intersperseC     , slidingWindowC@@ -149,6 +149,7 @@     , filterMCE     , iterMC     , scanlMC+    , mapAccumWhileMC     , concatMapAccumMC        -- *** Textual@@ -164,6 +165,7 @@        -- ** Special     , vectorBuilderC+    , CC.mapAccumS     , CC.peekForever     ) where @@ -175,21 +177,14 @@ import Data.Builder import qualified Data.NonNull as NonNull import qualified Data.Traversable-import qualified Data.ByteString as S-import qualified Data.ByteString.Base16 as B16-import qualified Data.ByteString.Base64 as B64-import qualified Data.ByteString.Base64.URL as B64U+#if ! MIN_VERSION_base(4,8,0) import           Control.Applicative         ((<$>))-import           Control.Exception           (assert)-import           Control.Category            (Category (..))-import           Control.Monad               (unless, when, (>=>), liftM, forever)-import           Control.Monad.Base          (MonadBase (liftBase))+#endif+import           Control.Monad.Base          (MonadBase (..)) import           Control.Monad.IO.Class      (MonadIO (..)) import           Control.Monad.Primitive     (PrimMonad, PrimState)-import           Control.Monad.Trans.Class   (lift) import           Control.Monad.Trans.Resource (MonadResource, MonadThrow) import           Data.Conduit-import           Data.Conduit.Internal       (ConduitM (..), Pipe (..)) import qualified Data.Conduit.List           as CL import           Data.IOData import           Data.Monoid                 (Monoid (..))@@ -197,33 +192,19 @@ import qualified Data.Sequences              as Seq import           Data.Sequences.Lazy import qualified Data.Vector.Generic         as V-import qualified Data.Vector.Generic.Mutable as VM-import           Data.Void                   (absurd)-import qualified System.FilePath             as F-import           System.FilePath             ((</>)) import           Prelude                     (Bool (..), Eq (..), Int,                                               Maybe (..), Monad (..), Num (..),-                                              Ord (..), fromIntegral, maybe,-                                              ($), Functor (..), Enum, seq, Show, Char, (||),-                                              mod, otherwise, Either (..),-                                              ($!), succ, FilePath)+                                              Ord (..), Functor (..), Either (..),+                                              Enum, Show, Char, FilePath) import Data.Word (Word8) import qualified Prelude import           System.IO                   (Handle) import qualified System.IO                   as SIO import qualified Data.Textual.Encoding as DTE-import qualified Data.Conduit.Text as CT import Data.ByteString (ByteString) import Data.Text (Text) import qualified System.Random.MWC as MWC-import Data.Conduit.Combinators.Internal-import qualified System.PosixCompat.Files as PosixC-import           Data.Primitive.MutVar       (MutVar, newMutVar, readMutVar,-                                              writeMutVar) -#ifndef WINDOWS-import qualified System.Posix.Directory as Dir-#endif  -- END IMPORTS @@ -1147,6 +1128,15 @@ scanlC = CC.scanl {-# INLINE scanlC #-} +-- | 'mapWhileC' with a break condition dependent on an accumulator.+-- Equivalently, 'CL.mapAccum' as long as the result is @Right@. Instead of+-- producing a leftover, the breaking input determines the resulting+-- accumulator via @Left@.+mapAccumWhileC :: Monad m =>+    (a -> s -> Either s (s, b)) -> s -> ConduitM a b m s+mapAccumWhileC = CC.mapAccumWhile+{-# INLINE mapAccumWhileC #-}+ -- | 'concatMap' with an accumulator. -- -- Since 1.0.0@@ -1293,6 +1283,11 @@ scanlMC :: Monad m => (a -> b -> m a) -> a -> Conduit b m a scanlMC = CC.scanlM {-# INLINE scanlMC #-}++-- | Monadic `mapAccumWhileC`.+mapAccumWhileMC :: Monad m => (a -> s -> m (Either s (s, b))) -> s -> ConduitM a b m s+mapAccumWhileMC = CC.mapAccumWhileM+{-# INLINE mapAccumWhileMC #-}  -- | 'concatMapM' with an accumulator. --

@@ -1,5 +1,5 @@ name:                conduit-combinators-version:             1.0.0+version:             1.0.1 synopsis:            Commonly used conduit functions, for both chunked and unchunked data description:         Provides a replacement for Data.Conduit.List, as well as a convenient Conduit module. homepage:            https://github.com/fpco/conduit-combinators

@@ -8,7 +8,8 @@ import Data.Maybe (listToMaybe) import Data.Conduit.Combinators.Internal import Data.Conduit.Combinators (slidingWindow)-import Data.List (intersperse, sort, find)+import Data.List (intersperse, sort, find, mapAccumL)+import Safe (tailSafe) import System.FilePath (takeExtension) import Test.Hspec import Test.Hspec.QuickCheck@@ -18,11 +19,11 @@ import qualified Data.Vector as V import qualified Data.Vector.Unboxed as VU import qualified Data.Vector.Storable as VS+import Control.Monad (liftM)+import Control.Monad.ST (runST) import Control.Monad.Trans.Writer import qualified System.IO as IO-#if MIN_VERSION_base(4,8,0)-import Control.Applicative ((<$>))-#else+#if ! MIN_VERSION_base(4,8,0) import Data.Monoid (Monoid (..)) import Control.Applicative ((<$>), (<*>)) #endif@@ -42,7 +43,6 @@ import qualified Data.ByteString.Base64.Lazy as B64L import qualified Data.ByteString.Base64.URL.Lazy as B64LU import qualified Data.ByteString.Base64.URL as B64U-import Control.Monad.ST (runST) import qualified StreamSpec  main :: IO ()@@ -436,16 +436,14 @@         runIdentity (yield input $$ filterCE evenInt =$ foldC)         `shouldBe` filter evenInt input     prop "mapWhile" $ \input (min 20 -> highest) ->-        let f i =-                if i < highest-                    then Just (i + 2 :: Int)-                    else Nothing+        let f i | i < highest = Just (i + 2 :: Int)+                | otherwise   = Nothing             res = runIdentity $ yieldMany input $$ do                 x <- (mapWhileC f >>= \() -> mempty) =$ sinkList                 y <- sinkList                 return (x, y)-            expected = (map (+ 2) $ takeWhile (< highest) input, dropWhile (< highest) input)-         in res `shouldBe` expected+            (taken, dropped) = span (< highest) input+         in res `shouldBe` (map (+ 2) taken, dropped)     prop "conduitVector" $ \(take 200 -> input) size' -> do         let size = min 30 $ succ $ abs size'         res <- yieldMany input $$ conduitVector size =$ sinkList@@ -456,10 +454,20 @@         let f a b = a + b :: Int             res = runIdentity $ yieldMany input $$ scanlC f seed =$ sinkList          in res `shouldBe` scanl f seed input-    it "concatMapAccum" $+    prop "mapAccumWhile" $ \input (min 20 -> highest) ->+        let f i accum | i < highest = Right (i + accum, 2 * i :: Int)+                      | otherwise   = Left accum+            res = runIdentity $ yieldMany input $$ do+                (s, x) <- fuseBoth (mapAccumWhileC f 0) sinkList+                y <- sinkList+                return (s, x, y)+            (taken, dropped) = span (< highest) input+         in res `shouldBe` (sum taken, map (* 2) taken, tailSafe dropped)+    prop "concatMapAccum" $ \(input :: [Int]) ->         let f a accum = (a + accum, [a, accum])-            res = runIdentity $ yieldMany [1..3] $$ concatMapAccumC f 0 =$ sinkList-         in res `shouldBe` [1, 0, 2, 1, 3, 3]+            res = runIdentity $ yieldMany input $$ concatMapAccumC f 0 =$ sinkList+            expected = concat $ snd $ mapAccumL (flip f) 0 input+         in res `shouldBe` expected     prop "intersperse" $ \xs x ->         runIdentity (yieldMany xs $$ intersperseC x =$ sinkList)         `shouldBe` intersperse (x :: Int) xs@@ -520,10 +528,20 @@             fm a b = return $ a + b             res = runIdentity $ yieldMany input $$ scanlMC fm seed =$ sinkList          in res `shouldBe` scanl f seed input-    it "concatMapAccumM" $-        let f a accum = return (a + accum, [a, accum])-            res = runIdentity $ yieldMany [1..3] $$ concatMapAccumMC f 0 =$ sinkList-         in res `shouldBe` [1, 0, 2, 1, 3, 3]+    prop "mapAccumWhileM" $ \input (min 20 -> highest) ->+        let f i accum | i < highest = Right (i + accum, 2 * i :: Int)+                      | otherwise   = Left accum+            res = runIdentity $ yieldMany input $$ do+                (s, x) <- fuseBoth (mapAccumWhileMC ((return.).f) 0) sinkList+                y <- sinkList+                return (s, x, y)+            (taken, dropped) = span (< highest) input+         in res `shouldBe` (sum taken, map (* 2) taken, tailSafe dropped)+    prop "concatMapAccumM" $ \(input :: [Int]) ->+        let f a accum = (a + accum, [a, accum])+            res = runIdentity $ yieldMany input $$ concatMapAccumMC ((return.).f) 0 =$ sinkList+            expected = concat $ snd $ mapAccumL (flip f) 0 input+         in res `shouldBe` expected     prop "encode UTF8" $ \(map T.pack -> inputs) -> do         let expected = encodeUtf8 $ fromChunks inputs         actual <- yieldMany inputs@@ -631,6 +649,16 @@                   where                     (y, z) = splitAt size x         res `shouldBe` expected+    prop "mapAccumS" $ \input ->+        let ints  = [1..]+            f a s = liftM (:s) $ mapC (* a) =$ takeC a =$ sinkList+            res   = reverse $ runIdentity $ yieldMany input+                           $$ mapAccumS f [] (yieldMany ints)+            expected = loop input ints+                where  loop []     _  = []+                       loop (a:as) xs = let (y, ys) = Prelude.splitAt a xs+                                        in  map (* a) y : loop as ys+        in  res `shouldBe` expected     prop "peekForever" $ \(strs' :: [String]) -> do         let strs = filter (not . null) strs'         res1 <- yieldMany strs $$ linesUnboundedC =$ sinkList

@@ -2,11 +2,13 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ViewPatterns #-}+{-# LANGUAGE TupleSections #-} {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE CPP #-} {-# OPTIONS_GHC -fno-warn-orphans #-} module StreamSpec where +import           Control.Arrow (first) import           Control.Applicative import qualified Control.Monad import           Control.Monad (liftM)@@ -29,8 +31,8 @@ import           Data.Vector (Vector) import qualified Prelude import           Prelude-    ((.), ($), (=<<), return, id, Maybe(..), Monad, Bool(..), Int,-     Eq, Show, String, Functor, fst, snd)+    ((.), ($), (>>=), (=<<), return, id, Maybe(..), Either(..), Monad,+     Bool(..), Int, Eq, Show, String, Functor, fst, snd, either) import qualified Safe import           System.Directory (removeFile) import qualified System.IO as IO@@ -163,6 +165,16 @@             \(getBlind -> (f :: Int -> Int -> M Int), initial) ->                 scanlMS f initial `checkStreamConduitM`                 scanlML f initial+        qit "mapAccumWhileS" $+            \(getBlind -> ( f :: Int -> [Int] -> Either [Int] ([Int], Int))+                          , initial :: [Int]) ->+                mapAccumWhileS f initial `checkStreamConduitResult`+                mapAccumWhileL f initial+        qit "mapAccumWhileMS" $+            \(getBlind -> ( f :: Int -> [Int] -> M (Either [Int] ([Int], Int)))+                          , initial :: [Int]) ->+                mapAccumWhileMS f initial `checkStreamConduitResultM`+                mapAccumWhileML f initial         qit "intersperse" $             \(sep :: Int) ->                 intersperse sep `checkConduit`@@ -256,6 +268,17 @@     go l (r:rs) = do         l' <- f l r         liftM (l:) (go l' rs)++mapAccumWhileL :: (a -> s -> Either s (s, b)) -> s -> [a] -> ([b], s)+mapAccumWhileL f = (runIdentity.) . mapAccumWhileML ((return.) . f)++mapAccumWhileML :: Monad m =>+    (a -> s -> m (Either s (s, b))) -> s -> [a] -> m ([b], s)+mapAccumWhileML f = go+    where go s []     = return ([], s)+          go s (a:as) = f a s >>= either+              (return . ([], ))+              (\(s', b) -> liftM (first (b:)) $ go s' as)  --FIXME: the following code is directly copied from the conduit test --suite.  How to share this code??