conduit 1.2.5.1 → 1.2.6

raw patch · 6 files changed

+161/−6 lines, 6 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Data/Conduit.hs +2/−0
• Data/Conduit/Internal/Conduit.hs +43/−2
• Data/Conduit/Internal/Pipe.hs +2/−2
• changelog.md +5/−0
• conduit.cabal +2/−2
• test/main.hs +107/−0

@@ -52,7 +52,9 @@     , mapOutput     , mapOutputMaybe     , mapInput+    , mergeSource     , passthroughSink+    , sourceToList        -- * Connect-and-resume     , ResumableSource

@@ -74,7 +74,9 @@     , zipSources     , zipSourcesApp     , zipConduitApp+    , mergeSource     , passthroughSink+    , sourceToList     , fuseBoth     , fuseBothMaybe     , fuseUpstream@@ -123,13 +125,13 @@     fmap f (ConduitM c) = ConduitM $ \rest -> c (rest . f)  instance Applicative (ConduitM i o m) where-    pure = return+    pure x = ConduitM ($ x)     {-# INLINE pure #-}     (<*>) = ap     {-# INLINE (<*>) #-}  instance Monad (ConduitM i o m) where-    return x = ConduitM ($ x)+    return = pure     ConduitM f >>= g = ConduitM $ \h -> f $ \a -> unConduitM (g a) h  instance MonadThrow m => MonadThrow (ConduitM i o m) where@@ -664,6 +666,25 @@ newResumableConduit :: Monad m => Conduit i m o -> ResumableConduit i m o newResumableConduit (ConduitM c) = ResumableConduit (c Done) (return ()) ++-- | Merge a @Source@ into a @Conduit@.+-- The new conduit will stop processing once either source or upstream have been exhausted.+mergeSource+  :: Monad m+  => Source m i+  -> Conduit a m (i, a)+mergeSource = loop . newResumableSource+  where+    loop :: Monad m => ResumableSource m i -> Conduit a m (i, a)+    loop src0 = await >>= maybe (lift $ closeResumableSource src0) go+      where+        go a = do+          (src1, mi) <- lift $ src0 $$++ await+          case mi of+            Nothing -> lift $ closeResumableSource src1+            Just i  -> yield (i, a) >> loop src1++ -- | Turn a @Sink@ into a @Conduit@ in the following way: -- -- * All input passed to the @Sink@ is yielded downstream.@@ -699,6 +720,26 @@                 CI.yield x                 go [] (next x)     in go [] (sink0 Done)++-- | Convert a @Source@ into a list. The basic functionality can be explained as:+--+-- > sourceToList src = src $$ Data.Conduit.List.consume+--+-- However, @sourceToList@ is able to produce its results lazily, which cannot+-- be done when running a conduit pipeline in general. Unlike the+-- @Data.Conduit.Lazy@ module, this function performs no unsafe I\/O+-- operations, and therefore can only be as lazily as the underlying monad.+--+-- Since 1.2.6+sourceToList :: Monad m => Source m a -> m [a]+sourceToList =+    go . flip unConduitM Done+  where+    go (Done _) = return []+    go (HaveOutput src _ x) = liftM (x:) (go src)+    go (PipeM msrc) = msrc >>= go+    go (NeedInput _ c) = go (c ())+    go (Leftover p _) = go p  -- Define fixity of all our operators infixr 0 $$

@@ -108,13 +108,13 @@     {-# INLINE fmap #-}  instance Monad m => Applicative (Pipe l i o u m) where-    pure = return+    pure = Done     {-# INLINE pure #-}     (<*>) = ap     {-# INLINE (<*>) #-}  instance Monad m => Monad (Pipe l i o u m) where-    return = Done+    return = pure     {-# INLINE return #-}      HaveOutput p c o >>= fp = HaveOutput (p >>= fp)            c          o

@@ -1,3 +1,8 @@+## 1.2.6++* `sourceToList`+* Canonicalise Monad instances [#237](https://github.com/snoyberg/conduit/pull/237)+ ## 1.2.5  * mapAccum and mapAccumM should be strict in their state [#218](https://github.com/snoyberg/conduit/issues/218)

@@ -1,5 +1,5 @@ Name:                conduit-Version:             1.2.5.1+Version:             1.2.6 Synopsis:            Streaming data processing library. description:     Hackage documentation generation is not reliable. For up to date documentation, please see: <http://www.stackage.org/package/conduit>.@@ -27,7 +27,7 @@                        Data.Conduit.Internal.List.Stream   other-modules:       Data.Conduit.Internal.Pipe                        Data.Conduit.Internal.Conduit-  Build-depends:       base                     >= 4.3          && < 5+  Build-depends:       base                     >= 4.5          && < 5                      , resourcet                >= 1.1          && < 1.2                      , exceptions               >= 0.6                      , lifted-base              >= 0.1

@@ -12,6 +12,7 @@ import qualified Data.Conduit.Internal as CI import qualified Data.Conduit.List as CL import Data.Typeable (Typeable)+import Control.Exception (throw) import Control.Monad.Trans.Resource as C (runResourceT) import Data.Maybe   (fromMaybe,catMaybes,fromJust) import qualified Data.List as DL@@ -730,6 +731,103 @@             res <- C.yield 10 C.$$ C.awaitForever (C.toProducer . src) C.=$ (C.toConsumer sink >>= C.yield) C.=$ C.await             res `shouldBe` Just (sum [1..10]) +    describe "mergeSource" $ do+        it "works" $ do+            let src :: C.Source IO String+                src = CL.sourceList ["A", "B", "C"]+                withIndex :: C.Conduit String IO (Integer, String)+                withIndex = CI.mergeSource (CL.sourceList [1..])+            output <- src C.$= withIndex C.$$ CL.consume+            output `shouldBe` [(1, "A"), (2, "B"), (3, "C")]+        it "does stop processing when the source exhausted" $ do+            let src :: C.Source IO Integer+                src = CL.sourceList [1..]+                withShortAlphaIndex :: C.Conduit Integer IO (String, Integer)+                withShortAlphaIndex = CI.mergeSource (CL.sourceList ["A", "B", "C"])+            output <- src C.$= withShortAlphaIndex C.$$ CL.consume+            output `shouldBe` [("A", 1), ("B", 2), ("C", 3)]++        let modFlag ref cur next = do+                prev <- I.atomicModifyIORef ref $ (,) next+                prev `shouldBe` cur+            flagShouldBe ref expect = do+                cur <- I.readIORef ref+                cur `shouldBe` expect+        it "properly run the finalizer - When the main Conduit is fully consumed" $ do+            called <- I.newIORef ("RawC" :: String)+            let src :: MonadIO m => C.Source m String+                src = CL.sourceList ["A", "B", "C"]+                withIndex :: MonadIO m => C.Conduit String m (Integer, String)+                withIndex = C.addCleanup (\f -> liftIO $ modFlag called "AllocC-3" ("FinalC:" ++ show f)) . CI.mergeSource $ do+                    liftIO $ modFlag called "RawC" "AllocC-1"+                    C.yield 1+                    liftIO $ modFlag called "AllocC-1" "AllocC-2"+                    C.yield 2+                    liftIO $ modFlag called "AllocC-2" "AllocC-3"+                    C.yield 3+                    liftIO $ modFlag called "AllocC-3" "AllocC-4"+                    C.yield 4+            output <- src C.$= withIndex C.$$ CL.consume+            output `shouldBe` [(1, "A"), (2, "B"), (3, "C")]+            called `flagShouldBe` "FinalC:True"+        it "properly run the finalizer - When the branch Source is fully consumed" $ do+            called <- I.newIORef ("RawS" :: String)+            let src :: MonadIO m => C.Source m Integer+                src = CL.sourceList [1..]+                withIndex :: MonadIO m => C.Conduit Integer m (String, Integer)+                withIndex = C.addCleanup (\f -> liftIO $ modFlag called "AllocS-C" ("FinalS:" ++ show f)) . CI.mergeSource $ do+                    liftIO $ modFlag called "RawS" "AllocS-A"+                    C.yield "A"+                    liftIO $ modFlag called "AllocS-A" "AllocS-B"+                    C.yield "B"+                    liftIO $ modFlag called "AllocS-B" "AllocS-C"+                    C.yield "C"+            output <- src C.$= withIndex C.$$ CL.consume+            output `shouldBe` [("A", 1), ("B", 2), ("C", 3)]+            called `flagShouldBe` "FinalS:True"+        it "properly DO NOT run the finalizer - When nothing consumed" $ do+            called <- I.newIORef ("Raw0" :: String)+            let src :: MonadIO m => C.Source m String+                src = CL.sourceList ["A", "B", "C"]+                withIndex :: MonadIO m => C.Conduit String m (Integer, String)+                withIndex = C.addCleanup (\f -> liftIO $ modFlag called "WONT CALLED" ("Final0:" ++ show f)) . CI.mergeSource $ do+                    liftIO $ modFlag called "Raw0" "Alloc0-1"+                    C.yield 1+            output <- src C.$= withIndex C.$$ return ()+            output `shouldBe` ()+            called `flagShouldBe` "Raw0"+        it "properly run the finalizer - When only one item consumed" $ do+            called <- I.newIORef ("Raw1" :: String)+            let src :: MonadIO m => C.Source m Integer+                src = CL.sourceList [1..]+                withIndex :: MonadIO m => C.Conduit Integer m (String, Integer)+                withIndex = C.addCleanup (\f -> liftIO $ modFlag called "Alloc1-A" ("Final1:" ++ show f)) . CI.mergeSource $ do+                    liftIO $ modFlag called "Raw1" "Alloc1-A"+                    C.yield "A"+                    liftIO $ modFlag called "Alloc1-A" "Alloc1-B"+                    C.yield "B"+                    liftIO $ modFlag called "Alloc1-B" "Alloc1-C"+                    C.yield "C"+            output <- src C.$= withIndex C.$= CL.isolate 1 C.$$ CL.consume+            output `shouldBe` [("A", 1)]+            called `flagShouldBe` "Final1:False"++        it "handles finalizers" $ do+            ref <- I.newIORef (0 :: Int)+            let src1 = C.addCleanup+                    (const $ I.modifyIORef ref (+1))+                    (mapM_ C.yield [1 :: Int ..])+                src2 = mapM_ C.yield ("hi" :: String)+            res1 <- src1 C.$$ C.mergeSource src2 C.=$ CL.consume+            res1 `shouldBe` [('h', 1), ('i', 2)]+            i1 <- I.readIORef ref+            i1 `shouldBe` 1++            res2 <- src2 C.$$ C.mergeSource src1 C.=$ CL.consume+            res2 `shouldBe` [(1, 'h'), (2, 'i')]+            i2 <- I.readIORef ref+            i2 `shouldBe` 2+     describe "passthroughSink" $ do         it "works" $ do             ref <- I.newIORef (-1)@@ -965,6 +1063,15 @@                     Catch.catch src (\DummyError -> C.yield (3 :: Int))             res <- src' C.$$ CL.consume             res `shouldBe` [1, 3]++    describe "sourceToList" $ do+        it "works lazily in Identity" $ do+            let src = C.yield 1 >> C.yield 2 >> throw DummyError+            let res = runIdentity $ C.sourceToList src+            take 2 res `shouldBe` [1, 2 :: Int]+        it "is not lazy in IO" $ do+            let src = C.yield 1 >> C.yield (2 :: Int) >> throw DummyError+            C.sourceToList src `shouldThrow` (==DummyError)      ZipConduit.spec     Stream.spec