pipes-transduce 0.4 → 0.4.1

raw patch · 6 files changed

+707/−567 lines, 6 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG +9/−0
• pipes-transduce.cabal +2/−1
• src/Pipes/Transduce.hs +18/−542
• src/Pipes/Transduce/Internal.hs +545/−0
• src/Pipes/Transduce/Text.hs +132/−23
• tests/doctests.hs +1/−1

@@ -1,3 +1,12 @@+0.4.1.0+=======++- The transducer1/Transducer thing is kind of akward. De-emphasized in the+  docs. Use the newly added functions instead.+- Added asUtf8, asUtf8x, bothAsUtf8x, asFoldedLines, combinedLines,+  combinedLinesPrefixing.+- New Pipes.Transduce.Internal module.+ 0.4.0.0 ======= 

@@ -1,5 +1,5 @@ Name: pipes-transduce-Version: 0.4+Version: 0.4.1 Cabal-Version: >=1.8.0.2 Build-Type: Simple License: BSD3@@ -44,6 +44,7 @@         Pipes.Transduce         Pipes.Transduce.ByteString         Pipes.Transduce.Text+        Pipes.Transduce.Internal     GHC-Options: -O2 -Wall  test-suite doctests

@@ -64,56 +64,40 @@     ,   separated     ,   combined         -- * Utilities+    ,   intoList     ,   trip     ,   tripx+        -- * Re-exports+    ,   runExceptT+    ,   throwE+    ,   next     ) where --import Data.Bifunctor-import Data.Monoid hiding (First)-import Data.Void-import Data.Foldable-import Control.Applicative-import Control.Applicative.Lift-import Control.Monad-import Control.Monad.Trans.Except-import Control.Monad.Trans.Free hiding (Pure)-import qualified Control.Foldl as Foldl-import Control.Concurrent (newMVar,withMVar)-import Control.Concurrent.Conceit-import Control.Exception-import Pipes -import Pipes.Lift (distribute) -import Pipes.Prelude (drain)-import qualified Pipes.Prelude as Pipes-import qualified Pipes.Group as Pipes-import qualified Pipes.Parse-import Pipes.Concurrent-import Pipes.Safe (SafeT, runSafeT)--import Lens.Micro+import           Control.Exception+import qualified Control.Foldl as L+import           Control.Monad.Trans.Except(runExceptT,throwE)+import           Pipes(next)+import           Pipes.Transduce.Internal  {- $setup >>> :set -XOverloadedStrings >>> import qualified Data.Text as T  >>> import qualified Data.Text.Lazy as TL ->>> import Control.Applicative->>> import Control.Monad+>>> import           Control.Applicative+>>> import           Control.Monad >>> import qualified Control.Foldl as L->>> import Pipes.Transduce +>>> import           Pipes.Transduce >>> import qualified Pipes.Transduce as PT->>> import Pipes.Transduce.Text->>> import qualified Pipes.Transduce.Text as PTT -}  {-|      Fail if the 'Producer' produces anything at all. The error value is what came     out of the 'Producer'. ->>> fold1Fallibly trip (mapM_ yield ['z']) +>>> PT.fold1Fallibly trip (mapM_ yield ['z'])  Left 'z' ->>> fold1Fallibly trip (mapM_ yield []) +>>> PT.fold1Fallibly trip (mapM_ yield [])  Right ((),()) -} trip :: Fold1 b b ()@@ -130,7 +114,7 @@     This 'Transducer may throw 'AssertionFailed'.     __/BEWARE!/__  ->>> fold1Fallibly tripx (mapM_ yield ['z']) +>>> PT.fold1Fallibly PT.tripx (mapM_ yield ['z'])  *** Exception: tripx -} tripx :: Fold1 b e ()@@ -140,6 +124,8 @@         Left r -> return (Right ((),r))         Right _ -> throwIO (AssertionFailed "tripx") +intoList :: Fold1 b e [b]+intoList = withFold L.list  {- $foldl  @@ -162,514 +148,4 @@     The most general way of constructing 'Fold1' values is from an arbitrary     function that consumes a 'Producer'. -}--{-| -    A computation in 'IO' that completely drains a 'Producer' of @b@ values,-    returning a value of type @a@, except when it fails early with an error of-    type @e@.--}-newtype Fold1 b e a = Fold1 { runFold1 :: Lift (Fold1_ b e) a } deriving (Functor)--data Fold1_ b e a = -         TrueFold (Foldl.FoldM (ExceptT e IO) b a)-       | ExhaustiveCont (forall r. Producer b IO r -> IO (Either e (a,r)))-       | NonexhaustiveCont (Producer b IO () -> IO (Either e a))-       deriving (Functor)--{-| -    'pure' creates a 'Fold1' that does nothing besides draining the-    'Producer'. --    '<*>' feeds both folds with the data of the same 'Producer'. If any of-    them fails the combination fails.--}-instance Applicative (Fold1 b e) where-    pure a = Fold1 (pure a)-    Fold1 fa <*> Fold1 a = Fold1 (fa <*> a)--instance Applicative (Fold1_ b e) where-    pure a = ExhaustiveCont (\producer -> do-        r <- runEffect (producer >-> Pipes.drain)-        pure (Right (a,r)))--    TrueFold f1 <*> TrueFold f2 = TrueFold (f1 <*> f2)-    s1 <*> s2 = bifurcate (nonexhaustiveCont s1) (nonexhaustiveCont s2)  -        where -        bifurcate fs as = ExhaustiveCont (\producer -> do-            (outbox1,inbox1,seal1) <- spawn' (bounded 1)-            (outbox2,inbox2,seal2) <- spawn' (bounded 1)-            runConceit $-                (\f x r -> (f x,r))-                <$>-                Conceit (fs (fromInput inbox1) `finally` atomically seal1)-                <*>-                Conceit (as (fromInput inbox2) `finally` atomically seal2)-                <*>-                (_Conceit $-                    (runEffect (producer >-> Pipes.tee (toOutput outbox1 *> Pipes.drain) -                                         >->           (toOutput outbox2 *> Pipes.drain)))-                    `finally` atomically seal1 -                    `finally` atomically seal2))--instance Bifunctor (Fold1_ b) where-  bimap f g s = case s of-      TrueFold (Foldl.FoldM step start done) -> TrueFold (Foldl.FoldM -          (\previous input -> withExceptT f (step previous input))-          (withExceptT f start)-          (\final -> withExceptT f (fmap g (done final))))-      ExhaustiveCont u -> ExhaustiveCont (fmap (liftM  (bimap f (bimap g id))) u)-      NonexhaustiveCont h -> NonexhaustiveCont (fmap (liftM  (bimap f g)) h)--{-| -    'first' is useful to massage errors.--}-instance Bifunctor (Fold1 b) where-  bimap f g (Fold1 s) = Fold1 (case s of-      Pure a -> Pure (g a)-      Other o -> Other (bimap f g o))--instance (Monoid a) => Monoid (Fold1 b e a) where-   mempty = pure mempty-   mappend s1 s2 = (<>) <$> s1 <*> s2--nonexhaustiveCont :: Fold1_ b e a -> Producer b IO () -> IO (Either e a)-nonexhaustiveCont (TrueFold e) = \producer -> runExceptT (Foldl.impurely Pipes.foldM e (hoist lift producer))-nonexhaustiveCont (ExhaustiveCont e) = \producer -> liftM (fmap fst) (e producer)-nonexhaustiveCont (NonexhaustiveCont u) = u--exhaustiveCont :: Fold1_ b e a -> Producer b IO r -> IO (Either e (a,r))-exhaustiveCont s = case s of -    TrueFold e -> \producer -> -        runExceptT (Foldl.impurely Pipes.foldM' e (hoist lift producer))-    ExhaustiveCont e -> e-    NonexhaustiveCont activity -> \producer -> do -        (outbox,inbox,seal) <- spawn' (bounded 1)-        runConceit $ -            (,) -            <$>-            Conceit (activity (fromInput inbox) `finally` atomically seal)-            <*>-            (_Conceit $-                (runEffect (producer >-> (toOutput outbox *> Pipes.drain)) -                `finally` atomically seal))---withFallibleCont -    :: (Producer b IO () -> IO (Either e a)) -- ^-    -> Fold1 b e a -withFallibleCont f = Fold1 (Other (NonexhaustiveCont f))--withFallibleCont'  -    :: (forall r. Producer b IO r -> IO (Either e (a,r))) -- ^-    -> Fold1 b e a -withFallibleCont' f = Fold1 (Other (ExhaustiveCont f))--withCont -    :: (Producer b IO () -> IO a) -- ^-    -> Fold1 b e a -- ^-withCont aFold = withFallibleCont $ fmap (fmap pure) $ aFold--withCont' -    :: (forall r. Producer b IO r -> IO (a,r)) -- ^-    -> Fold1 b e a -- ^-withCont' aFold = withFallibleCont' $ fmap (fmap pure) aFold--withFold :: Foldl.Fold b a -> Fold1 b e a -withFold aFold = Fold1 (Other (TrueFold (Foldl.generalize aFold)))--withFoldIO :: Foldl.FoldM IO b a -> Fold1 b e a -withFoldIO aFold = Fold1 (Other (TrueFold (hoistFold lift aFold)))--hoistFold :: Monad m => (forall a. m a -> n a) -> Foldl.FoldM m i r -> Foldl.FoldM n i r -hoistFold g (Foldl.FoldM step begin done) = Foldl.FoldM (\s i -> g (step s i)) (g begin) (g . done)--withFallibleFold :: Foldl.FoldM (ExceptT e IO) b a -> Fold1 b e a -withFallibleFold aFold = Fold1 (Other (TrueFold aFold))----withFoldM ---    :: MonadIO m ---    => (forall r. m (a,r) -> IO (Either e (c,r))) ---    -> Foldl.FoldM m b a ---    -> Fold1 b e c ---withFoldM whittle aFoldM = withFallibleCont' $ \producer -> ---    whittle $ Foldl.impurely Pipes.Prelude.foldM' aFoldM (hoist liftIO producer)--withConsumer :: Consumer b IO () -> Fold1 b e ()-withConsumer consumer = withCont $ \producer -> runEffect $ producer >-> consumer --{-| Builds a 'Fold1' out of a 'Consumer' that never stops by itself.---}-withConsumer' :: Consumer b IO Void -> Fold1 b e ()-withConsumer' consumer = withCont' $ \producer -> fmap ((,) ()) $ runEffect $ producer >-> fmap absurd consumer --withConsumerM :: MonadIO m -              => (m () -> IO (Either e a))  -- ^-              -> Consumer b m () -              -> Fold1 b e a-withConsumerM whittle consumer = withFallibleCont $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> consumer --withConsumerM' :: MonadIO m -               => (forall r. m r -> IO (Either e (a,r))) -- ^-               -> Consumer b m Void-               -> Fold1 b e a-withConsumerM' whittle consumer = withFallibleCont' $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> fmap absurd consumer --withSafeConsumer -    :: Consumer b (SafeT IO) Void -- ^-    -> Fold1 b e ()-withSafeConsumer = withConsumerM' (fmap (\r -> Right ((),r)) . runSafeT)--withFallibleConsumer -    :: Consumer b (ExceptT e IO) Void -- ^-    -> Fold1 b e ()-withFallibleConsumer = withConsumerM' (fmap (fmap (\r -> ((), r))) . runExceptT)---withParser -    :: Pipes.Parse.Parser b IO (Either e a) -- ^-    -> Fold1 b e a -withParser parser = withFallibleCont' $ \producer -> drainage $ Pipes.Parse.runStateT parser producer-  where-    drainage m = do -        (a,leftovers) <- m-        r <- runEffect (leftovers >-> Pipes.Prelude.drain)-        case a of-            Left e -> return (Left e)-            Right a' -> return (Right (a',r)) --withParserM :: MonadIO m -            => (forall r. m (a,r) -> IO (Either e (c,r))) -- ^-            -> Pipes.Parse.Parser b m a -> Fold1 b e c -withParserM f parser = withFallibleCont' $ \producer -> f $ drainage $ (Pipes.Parse.runStateT parser) (hoist liftIO producer)-  where-    drainage m = do -        (a,leftovers) <- m-        r <- runEffect (leftovers >-> Pipes.Prelude.drain)-        return (a,r)----------------------------------------------------------------------------------{-| -    Run a 'Fold1'.--}-fold1Fallibly :: Fold1 b e a -> Producer b IO r -> IO (Either e (a,r))-fold1Fallibly (Fold1 (unLift -> s)) = exhaustiveCont s--{-| -    Run a 'Fold1' that never returns an error value (but which may still throw exceptions!)--}-fold1 :: Fold1 b Void a -> Producer b IO r -> IO (a,r)-fold1 (Fold1 (unLift -> s)) = liftM (either absurd id) . exhaustiveCont s--{-| A transformation that takes the inputs of a 'Fold1' from type @a@ to type @b@.		--    Optionally, the transformation may delimit groups of elements in the-    stream. In that case the phantom type @x@ will be 'Delimited'. Otherwise, it will be-    'Continuous'.--}-data Transducer x b e a = -      M (b -> a)-    | F (b -> [a])-    | P (forall r. Producer b IO r -> Producer a IO r)-    | PE (forall r. Producer b IO r -> Producer a IO (Either e r))-    | S (forall r. Producer b IO r -> FreeT (Producer a IO) IO r)-    | SE (forall r. Producer b IO r -> FreeT (Producer a IO) IO (Either e r))--instance Functor (Transducer x b e) where-  fmap = second--instance Bifunctor (Transducer x b) where-  bimap f g s = case s of-      M x -> M (g . x)-      F x -> F (fmap g . x)-      P x -> P (\producer -> for (x producer) (Pipes.yield . g))-      PE x -> PE (\producer -> liftM (first f) (for (x producer) (Pipes.yield . g)))-      S x -> S (\producer -> transFreeT (\p -> for p (Pipes.yield . g)) (x producer))-      SE x -> SE (\producer -> liftM (first f) (transFreeT (\p -> (for p (Pipes.yield . g))) (x producer)))--mapper -    :: (a -> b) -- ^-    -> Transducer Continuous a e b-mapper = M--fallibleMapper -    :: (a -> Either e b) -- ^-    -> Transducer Continuous a e b  -- ^-fallibleMapper fallible = PE (\producer -> (runExceptT . distribute) (for (hoist lift producer) (\a -> do-    case fallible a of-        Left e -> lift (throwE e)-        Right b -> Pipes.yield b)))--mapperFoldable -    :: Foldable f -    => (a -> f b) -- ^-    -> Transducer Continuous a e b -- ^-mapperFoldable f = F (Data.Foldable.toList . f)--mapperEnumerable -    :: Enumerable f -    => (a -> f IO b) -- ^-    -> Transducer Continuous a e b  -- ^-mapperEnumerable enumerable = P (\producer -> for producer (enumerate . toListT . enumerable))--transducer -    :: (forall r. Producer b IO r -> Producer a IO r)  -- ^-    -> Transducer Continuous b e a -- ^-transducer = P--fallibleTransducer -    :: (forall r. Producer b IO r -> Producer a IO (Either e r))  -- ^-    -> Transducer Continuous b e a  -- ^-fallibleTransducer = PE--{-| Plug splitting functions from @pipes-group@ here.		---}-delimit -    :: (forall r. Producer a IO r -> FreeT (Producer a' IO) IO r) -- ^-    -> Transducer Continuous b e a -- ^-    -> Transducer Delimited b e a' -- ^-delimit f t = case t of-    M func -> S (\producer -> f (producer >-> Pipes.map func))-    F func -> S (\producer -> f (producer >-> Pipes.mapFoldable func))-    P g -> S (f . g)-    PE g -> SE (f . g)-    S g -> S (f . Pipes.concats . g)-    SE g -> SE (f . Pipes.concats . g)--{-| Apply a 'Transducer' to a 'Fold1'.		---}-transduce1 :: Transducer Continuous b e a -> Fold1 a e r -> Fold1 b e r-transduce1 (M _) (Fold1 (Pure x)) = -    Fold1 (Pure x)-transduce1 (M f) (Fold1 (Other s)) = (Fold1 (Other (case s of-    TrueFold x -> TrueFold (Foldl.premapM f x)-    ExhaustiveCont x -> ExhaustiveCont (\producer -> x (producer >-> Pipes.map f))-    NonexhaustiveCont x -> NonexhaustiveCont (\producer -> x (producer >-> Pipes.map f)))))-transduce1 (F _) (Fold1 (Pure x)) = -    Fold1 (Pure x)-transduce1 (F f) (Fold1 (Other s)) = (Fold1 (Other (case s of-    TrueFold x -> TrueFold (Foldl.handlesM (folding f) x)-    ExhaustiveCont x -> ExhaustiveCont (\producer -> x (producer >-> Pipes.mapFoldable f))-    NonexhaustiveCont x -> NonexhaustiveCont (\producer -> x (producer >-> Pipes.mapFoldable f)))))-transduce1 (P f) (Fold1 (unLift -> s)) = case s of-    NonexhaustiveCont x -> Fold1 (Other (NonexhaustiveCont (x . f)))-    _ -> Fold1 (Other (ExhaustiveCont (exhaustiveCont s . f)))-transduce1 (PE f) (Fold1 (exhaustiveCont . unLift -> s)) = do-    Fold1 (Other (ExhaustiveCont (\producer -> do-        (outbox,inbox,seal) <- spawn' (bounded 1)-        runConceit $ -            (\(r,()) r' -> (r,r'))-            <$>-            Conceit (s (fromInput inbox) `finally` atomically seal)-            <*>-            (Conceit $-                (runEffect (f producer >-> (toOutput outbox *> Pipes.drain)) -                `finally` atomically seal)))))-transduce1 (S f) somefold = transduce1 (P (Pipes.concats . f)) somefold-transduce1 (SE f) somefold = transduce1 (PE (Pipes.concats . f)) somefold--{-| Tweak each of the groups delimited by a 'Transducer'.		---}-groups -    :: (forall r. Producer b IO r -> Producer b' IO r) -- ^-    -> Transducer Delimited a e b  -- ^-    -> Transducer Delimited a e b' -- ^-groups f t = case t of-    M func -> P (f . (\producer -> producer >-> Pipes.map func))-    F func -> P (f . (\producer -> producer >-> Pipes.mapFoldable func))-    P g -> P (f . g)-    PE g -> PE (f . g)-    S g -> S (Pipes.maps f . g)-    SE g -> SE (Pipes.maps f . g)--folds -    :: Fold1 b Void b' -- ^-    -> Transducer Delimited a e b -    -> Transducer Continuous a e b'-folds somefold t = case t of-    M func -> folds somefold (P (\producer -> producer >-> Pipes.map func))-    F func -> folds somefold (P (\producer -> producer >-> Pipes.mapFoldable func))-    P g -> folds somefold (S (liftF . g))-    PE g -> folds somefold (SE (liftF . g))-    S g -> P (Pipes.concats . transFreeT ((\action -> lift action >>= (\(b',r) -> Pipes.yield b' >> return r)) . fold1 somefold) . g)-    SE g -> PE (Pipes.concats . transFreeT ((\action -> lift action >>= (\(b',r) -> Pipes.yield b' >> return r)) . fold1 somefold) . g)--data Delimited--data Continuous--concats -    :: Transducer Delimited a e b   -- ^-    -> Transducer Continuous a e b-concats t =  case t of-    M func -> M func-    F func -> F func-    P g -> P g-    PE g -> PE g-    S g -> P (Pipes.concats . g)-    SE g -> PE (Pipes.concats . g)--intercalates -    :: Producer b IO ()  -- ^-    -> Transducer Delimited a e b -    -> Transducer Continuous a e b-intercalates p t =  case t of-    M func -> M func-    F func -> F func-    P g -> P g-    PE g -> PE g-    S g -> P (Pipes.intercalates p . g)-    SE g -> PE (Pipes.intercalates p . g)----{-| -    A computation in 'IO' that completely drains two 'Producer's of @b@ values-    in a concurrent way, returning a value of type @a@, except when it fails early-    with an error of type @e@.--}-newtype Fold2 b1 b2 e a = Fold2 (Lift (Fold2_ b1 b2 e) a) deriving (Functor)--data Fold2_ b1 b2 e a = -      First (Fold1_ b1 e a)-    | Second (Fold1_ b2 e a)-    | Both (forall r1 r2. Producer b1 IO r1 -> Producer b2 IO r2 -> IO (Either e (a,r1,r2))) deriving (Functor)--fold2Fallibly_ :: Fold2_ b1 b2 e a -> Producer b1 IO r1 -> Producer b2 IO r2 -> IO (Either e (a,r1,r2))-fold2Fallibly_ theFold producer1 producer2 = case theFold of-        Both f -> f producer1 producer2-        First f -> runConceit $-            (\(r1,x1) (_,x2) -> (r1,x1,x2))-            <$>-            Conceit (exhaustiveCont f producer1)-            <*>-            Conceit (fold1Fallibly (pure ()) producer2)-        Second f -> runConceit $-            (\(_,x1) (r2,x2) -> (r2,x1,x2))-            <$>-            Conceit (fold1Fallibly (pure ()) producer1)-            <*>-            Conceit (exhaustiveCont f producer2)--instance Bifunctor (Fold2 b1 b2) where-    bimap f g (Fold2 x) = Fold2 (case x of-        Pure a -> Pure (g a)-        Other o -> Other (bimap f g o))--instance Bifunctor (Fold2_ b1 b2) where-    bimap f g (First s) = First (bimap f g s) -    bimap f g (Second s) = Second (bimap f g s) -    bimap f g (Both s) = Both (fmap (fmap (fmap (bimap f (\(x1,x2,x3) -> (g x1,x2,x3))))) s) --instance Applicative (Fold2 b1 b2 e) where-    pure a = Fold2 (pure a)-    Fold2 fa <*> Fold2 a = Fold2 (fa <*> a)--instance Applicative (Fold2_ b1 b2 e) where-    pure a = fmap (const a) (separated_ (pure ()) (pure ()))--    Both fs <*> Both as = Both (\producer1 producer2 -> do-        (outbox1a,inbox1a,seal1a) <- spawn' (bounded 1)-        (outbox2a,inbox2a,seal2a) <- spawn' (bounded 1)-        (outbox1b,inbox1b,seal1b) <- spawn' (bounded 1)-        (outbox2b,inbox2b,seal2b) <- spawn' (bounded 1)-        runConceit $-            (\(f,(),()) (x,(),()) r1 r2 -> (f x,r1,r2))-            <$>-            Conceit (fs (fromInput inbox1a) (fromInput inbox1b) `finally` atomically seal1a `finally` atomically seal1b)-            <*>-            Conceit (as (fromInput inbox2a) (fromInput inbox2b) `finally` atomically seal2a `finally` atomically seal2b)-            <*>-            (_Conceit $-                (runEffect (producer1 >-> Pipes.tee (toOutput outbox1a *> Pipes.drain) -                                      >->           (toOutput outbox2a *> Pipes.drain)))-                `finally` atomically seal1a -                `finally` atomically seal2a)-            <*>-            (_Conceit $-                (runEffect (producer2 >-> Pipes.tee (toOutput outbox1b *> Pipes.drain) -                                      >->           (toOutput outbox2b *> Pipes.drain)))-                `finally` atomically seal1b -                `finally` atomically seal2b))-    First fs <*> First as = First (fs <*> as)-    Second fs <*> Second as = Second (fs <*> as)-    First fs <*> Second as = uncurry ($) <$> separated_ fs as-    Second fs <*> First as = uncurry (flip ($)) <$> separated_ as fs-    First fs <*> Both as =  (\(f,()) x -> f x) <$> separated_ fs (pure ()) <*> Both as -    Both fs <*> First as =  (\f (x,()) -> f x) <$> Both fs <*> separated_ as (pure ())-    Second fs <*> Both as = (\((),f) x -> f x) <$> separated_ (pure ()) fs <*> Both as -    Both fs <*> Second as = (\f ((),x) -> f x) <$> Both fs <*> separated_ (pure ()) as --instance (Monoid a) => Monoid (Fold2 b1 b2 e a) where-   mempty = pure mempty-   mappend s1 s2 = (<>) <$> s1 <*> s2--{-| -    Run a 'Fold2'.--}-fold2Fallibly :: Fold2 b1 b2 e a -> Producer b1 IO r1 -> Producer b2 IO r2 -> IO (Either e (a,r1,r2))-fold2Fallibly (Fold2 (fold2Fallibly_ . unLift -> s)) = s ---{-| -    Run a 'Fold2' that never returns an error value (but which may still throw exceptions!)--}-fold2 :: Fold2 b1 b2 Void a -> Producer b1 IO r1 -> Producer b2 IO r2 -> IO (a,r1,r2)-fold2 s producer1 producer2 = liftM (either absurd id) (fold2Fallibly s producer1 producer2) --liftFirst :: Fold1 b1 e r1 -> Fold2 b1 b2 e r1-liftFirst (unLift . runFold1 -> f1) = Fold2 (Other (First f1))--liftSecond :: Fold1 b2 e r1 -> Fold2 b1 b2 e r1-liftSecond (unLift . runFold1 -> f1) = Fold2 (Other (Second f1))--separated_ :: Fold1_ b1 e r1 -> Fold1_ b2 e r2 -> Fold2_ b1 b2 e (r1,r2)-separated_ f1 f2 = Both (\producer1 producer2 ->-    runConceit $-        (\(r1,x1) (r2,x2) -> ((r1,r2),x1,x2))-        <$>-        Conceit (exhaustiveCont f1 producer1)-        <*>-        Conceit (exhaustiveCont f2 producer2))--{-|-    Consume the producers concurrently, each one independently of the other. --}-separated :: Fold1 b1 e r1 -> Fold1 b2 e r2 -> Fold2 b1 b2 e (r1,r2)-separated f1 f2 = Fold2 (Other (separated_ (unLift . runFold1 $ f1) (unLift . runFold1 $ f2)))--{-|-    Consume the producers concurrently, delimiting groups in each producer,-    and writing the groups into a common 'Fold1'. --    Possible use: find lines in two text producers and combine the lines in a-    single stream, preserving the integrity of each individual line.--}-combined :: Transducer Delimited b1 e x -> Transducer Delimited b2 e x -> Fold1 x e a -> Fold2 b1 b2 e a-combined t1 t2 f = Fold2 (Other (Both (\producer1 producer2 -> do-   (outbox, inbox, seal) <- spawn' (bounded 1)-   lock <- newMVar outbox-   runConceit $ -       (\(((),r1),((),r2)) (a,()) -> (a,r1,r2))-       <$>-       Conceit -           ((runConceit $-               (,)-               <$>-               Conceit (fold1Fallibly (transduce1 (folds (withCont' (iterTLines lock)) t1) (pure ())) producer1)-               <*>-               Conceit (fold1Fallibly (transduce1 (folds (withCont' (iterTLines lock)) t2) (pure ())) producer2)-           ) `finally` atomically seal)-       <*>-       Conceit (fold1Fallibly f (fromInput inbox) `finally` atomically seal))))-  where-    -- iterTLines mvar = iterT $ \textProducer -> do-    iterTLines mvar = \textProducer -> fmap (\x -> ((),x)) $ do-        -- the P.drain bit was difficult to figure out!!!-        withMVar mvar $ \output -> do-            runEffect $ textProducer >-> (toOutput output >> Pipes.drain) 

@@ -0,0 +1,545 @@+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE ViewPatterns #-}+{-# LANGUAGE EmptyDataDecls #-}++module Pipes.Transduce.Internal where++import Data.Bifunctor+import Data.Monoid hiding (First)+import Data.Void+import Data.Foldable+import Control.Applicative+import Control.Applicative.Lift+import Control.Monad+import Control.Monad.Trans.Except+import Control.Monad.Trans.Free hiding (Pure)+import qualified Control.Foldl as Foldl+import Control.Concurrent (newMVar,withMVar)+import Control.Concurrent.Conceit+import Control.Exception+import Pipes +import Pipes.Lift (distribute) +import Pipes.Prelude (drain)+import qualified Pipes.Prelude as Pipes+import qualified Pipes.Group as Pipes+import qualified Pipes.Parse+import Pipes.Concurrent+import Pipes.Safe (SafeT, runSafeT)++import Lens.Micro++{-| +    A computation in 'IO' that completely drains a 'Producer' of @b@ values,+    returning a value of type @a@, except when it fails early with an error of+    type @e@.+-}+newtype Fold1 b e a = Fold1 { runFold1 :: Lift (Fold1_ b e) a } deriving (Functor)++data Fold1_ b e a = +         TrueFold (Foldl.FoldM (ExceptT e IO) b a)+       | ExhaustiveCont (forall r. Producer b IO r -> IO (Either e (a,r)))+       | NonexhaustiveCont (Producer b IO () -> IO (Either e a))+       deriving (Functor)++{-| +    'pure' creates a 'Fold1' that does nothing besides draining the+    'Producer'. ++    '<*>' feeds both folds with the data of the same 'Producer'. If any of+    them fails the combination fails.+-}+instance Applicative (Fold1 b e) where+    pure a = Fold1 (pure a)+    Fold1 fa <*> Fold1 a = Fold1 (fa <*> a)++instance Applicative (Fold1_ b e) where+    pure a = ExhaustiveCont (\producer -> do+        r <- runEffect (producer >-> Pipes.drain)+        pure (Right (a,r)))++    TrueFold f1 <*> TrueFold f2 = TrueFold (f1 <*> f2)+    s1 <*> s2 = bifurcate (nonexhaustiveCont s1) (nonexhaustiveCont s2)  +        where +        bifurcate fs as = ExhaustiveCont (\producer -> do+            (outbox1,inbox1,seal1) <- spawn' (bounded 1)+            (outbox2,inbox2,seal2) <- spawn' (bounded 1)+            runConceit $+                (\f x r -> (f x,r))+                <$>+                Conceit (fs (fromInput inbox1) `finally` atomically seal1)+                <*>+                Conceit (as (fromInput inbox2) `finally` atomically seal2)+                <*>+                (_Conceit $+                    (runEffect (producer >-> Pipes.tee (toOutput outbox1 *> Pipes.drain) +                                         >->           (toOutput outbox2 *> Pipes.drain)))+                    `finally` atomically seal1 +                    `finally` atomically seal2))++instance Bifunctor (Fold1_ b) where+  bimap f g s = case s of+      TrueFold (Foldl.FoldM step start done) -> TrueFold (Foldl.FoldM +          (\previous input -> withExceptT f (step previous input))+          (withExceptT f start)+          (\final -> withExceptT f (fmap g (done final))))+      ExhaustiveCont u -> ExhaustiveCont (fmap (liftM  (bimap f (bimap g id))) u)+      NonexhaustiveCont h -> NonexhaustiveCont (fmap (liftM  (bimap f g)) h)++{-| +    'first' is useful to massage errors.+-}+instance Bifunctor (Fold1 b) where+  bimap f g (Fold1 s) = Fold1 (case s of+      Pure a -> Pure (g a)+      Other o -> Other (bimap f g o))++instance (Monoid a) => Monoid (Fold1 b e a) where+   mempty = pure mempty+   mappend s1 s2 = (<>) <$> s1 <*> s2++nonexhaustiveCont :: Fold1_ b e a -> Producer b IO () -> IO (Either e a)+nonexhaustiveCont (TrueFold e) = \producer -> runExceptT (Foldl.impurely Pipes.foldM e (hoist lift producer))+nonexhaustiveCont (ExhaustiveCont e) = \producer -> liftM (fmap fst) (e producer)+nonexhaustiveCont (NonexhaustiveCont u) = u++exhaustiveCont :: Fold1_ b e a -> Producer b IO r -> IO (Either e (a,r))+exhaustiveCont s = case s of +    TrueFold e -> \producer -> +        runExceptT (Foldl.impurely Pipes.foldM' e (hoist lift producer))+    ExhaustiveCont e -> e+    NonexhaustiveCont activity -> \producer -> do +        (outbox,inbox,seal) <- spawn' (bounded 1)+        runConceit $ +            (,) +            <$>+            Conceit (activity (fromInput inbox) `finally` atomically seal)+            <*>+            (_Conceit $+                (runEffect (producer >-> (toOutput outbox *> Pipes.drain)) +                `finally` atomically seal))+++withFallibleCont +    :: (Producer b IO () -> IO (Either e a)) -- ^+    -> Fold1 b e a +withFallibleCont f = Fold1 (Other (NonexhaustiveCont f))++withFallibleCont'  +    :: (forall r. Producer b IO r -> IO (Either e (a,r))) -- ^+    -> Fold1 b e a +withFallibleCont' f = Fold1 (Other (ExhaustiveCont f))++withCont +    :: (Producer b IO () -> IO a) -- ^+    -> Fold1 b e a -- ^+withCont aFold = withFallibleCont $ fmap (fmap pure) $ aFold++withCont' +    :: (forall r. Producer b IO r -> IO (a,r)) -- ^+    -> Fold1 b e a -- ^+withCont' aFold = withFallibleCont' $ fmap (fmap pure) aFold++withFold :: Foldl.Fold b a -> Fold1 b e a +withFold aFold = Fold1 (Other (TrueFold (Foldl.generalize aFold)))++withFoldIO :: Foldl.FoldM IO b a -> Fold1 b e a +withFoldIO aFold = Fold1 (Other (TrueFold (hoistFold lift aFold)))++hoistFold :: Monad m => (forall a. m a -> n a) -> Foldl.FoldM m i r -> Foldl.FoldM n i r +hoistFold g (Foldl.FoldM step begin done) = Foldl.FoldM (\s i -> g (step s i)) (g begin) (g . done)++withFallibleFold :: Foldl.FoldM (ExceptT e IO) b a -> Fold1 b e a +withFallibleFold aFold = Fold1 (Other (TrueFold aFold))++--withFoldM +--    :: MonadIO m +--    => (forall r. m (a,r) -> IO (Either e (c,r))) +--    -> Foldl.FoldM m b a +--    -> Fold1 b e c +--withFoldM whittle aFoldM = withFallibleCont' $ \producer -> +--    whittle $ Foldl.impurely Pipes.Prelude.foldM' aFoldM (hoist liftIO producer)++withConsumer :: Consumer b IO () -> Fold1 b e ()+withConsumer consumer = withCont $ \producer -> runEffect $ producer >-> consumer ++{-| Builds a 'Fold1' out of a 'Consumer' that never stops by itself.++-}+withConsumer' :: Consumer b IO Void -> Fold1 b e ()+withConsumer' consumer = withCont' $ \producer -> fmap ((,) ()) $ runEffect $ producer >-> fmap absurd consumer ++withConsumerM :: MonadIO m +              => (m () -> IO (Either e a))  -- ^+              -> Consumer b m () +              -> Fold1 b e a+withConsumerM whittle consumer = withFallibleCont $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> consumer ++withConsumerM' :: MonadIO m +               => (forall r. m r -> IO (Either e (a,r))) -- ^+               -> Consumer b m Void+               -> Fold1 b e a+withConsumerM' whittle consumer = withFallibleCont' $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> fmap absurd consumer ++withSafeConsumer +    :: Consumer b (SafeT IO) Void -- ^+    -> Fold1 b e ()+withSafeConsumer = withConsumerM' (fmap (\r -> Right ((),r)) . runSafeT)++withFallibleConsumer +    :: Consumer b (ExceptT e IO) Void -- ^+    -> Fold1 b e ()+withFallibleConsumer = withConsumerM' (fmap (fmap (\r -> ((), r))) . runExceptT)+++withParser +    :: Pipes.Parse.Parser b IO (Either e a) -- ^+    -> Fold1 b e a +withParser parser = withFallibleCont' $ \producer -> drainage $ Pipes.Parse.runStateT parser producer+  where+    drainage m = do +        (a,leftovers) <- m+        r <- runEffect (leftovers >-> Pipes.Prelude.drain)+        case a of+            Left e -> return (Left e)+            Right a' -> return (Right (a',r)) ++withParserM :: MonadIO m +            => (forall r. m (a,r) -> IO (Either e (c,r))) -- ^+            -> Pipes.Parse.Parser b m a -> Fold1 b e c +withParserM f parser = withFallibleCont' $ \producer -> f $ drainage $ (Pipes.Parse.runStateT parser) (hoist liftIO producer)+  where+    drainage m = do +        (a,leftovers) <- m+        r <- runEffect (leftovers >-> Pipes.Prelude.drain)+        return (a,r)++------------------------------------------------------------------------------++{-| +    Run a 'Fold1'.+-}+fold1Fallibly :: Fold1 b e a -> Producer b IO r -> IO (Either e (a,r))+fold1Fallibly (Fold1 (unLift -> s)) = exhaustiveCont s++{-| +    Run a 'Fold1' that never returns an error value (but which may still throw exceptions!)+-}+fold1 :: Fold1 b Void a -> Producer b IO r -> IO (a,r)+fold1 (Fold1 (unLift -> s)) = liftM (either absurd id) . exhaustiveCont s++{-| A transformation that takes the inputs of a 'Fold1' from type @a@ to type @b@.		++    Optionally, the transformation may delimit groups of elements in the+    stream. In that case the phantom type @x@ will be 'Delimited'. Otherwise, it will be+    'Continuous'.+-}+data Transducer x b e a = +      M (b -> a)+    | F (b -> [a])+    | P (forall r. Producer b IO r -> Producer a IO r)+    | PE (forall r. Producer b IO r -> Producer a IO (Either e r))+    | S (forall r. Producer b IO r -> FreeT (Producer a IO) IO r)+    | SE (forall r. Producer b IO r -> FreeT (Producer a IO) IO (Either e r))++instance Functor (Transducer x b e) where+  fmap = second++instance Bifunctor (Transducer x b) where+  bimap f g s = case s of+      M x -> M (g . x)+      F x -> F (fmap g . x)+      P x -> P (\producer -> for (x producer) (Pipes.yield . g))+      PE x -> PE (\producer -> liftM (first f) (for (x producer) (Pipes.yield . g)))+      S x -> S (\producer -> transFreeT (\p -> for p (Pipes.yield . g)) (x producer))+      SE x -> SE (\producer -> liftM (first f) (transFreeT (\p -> (for p (Pipes.yield . g))) (x producer)))++mapper +    :: (a -> b) -- ^+    -> Transducer Continuous a e b+mapper = M++fallibleMapper +    :: (a -> Either e b) -- ^+    -> Transducer Continuous a e b  -- ^+fallibleMapper fallible = PE (\producer -> (runExceptT . distribute) (for (hoist lift producer) (\a -> do+    case fallible a of+        Left e -> lift (throwE e)+        Right b -> Pipes.yield b)))++mapperFoldable +    :: Foldable f +    => (a -> f b) -- ^+    -> Transducer Continuous a e b -- ^+mapperFoldable f = F (Data.Foldable.toList . f)++mapperEnumerable +    :: Enumerable f +    => (a -> f IO b) -- ^+    -> Transducer Continuous a e b  -- ^+mapperEnumerable enumerable = P (\producer -> for producer (enumerate . toListT . enumerable))++transducer +    :: (forall r. Producer b IO r -> Producer a IO r)  -- ^+    -> Transducer Continuous b e a -- ^+transducer = P++fallibleTransducer +    :: (forall r. Producer b IO r -> Producer a IO (Either e r))  -- ^+    -> Transducer Continuous b e a  -- ^+fallibleTransducer = PE++{-| Plug splitting functions from @pipes-group@ here.		++-}+delimit +    :: (forall r. Producer a IO r -> FreeT (Producer a' IO) IO r) -- ^+    -> Transducer Continuous b e a -- ^+    -> Transducer Delimited b e a' -- ^+delimit f t = case t of+    M func -> S (\producer -> f (producer >-> Pipes.map func))+    F func -> S (\producer -> f (producer >-> Pipes.mapFoldable func))+    P g -> S (f . g)+    PE g -> SE (f . g)+    S g -> S (f . Pipes.concats . g)+    SE g -> SE (f . Pipes.concats . g)++{-| Apply a 'Transducer' to a 'Fold1'.		++-}+transduce1 :: Transducer Continuous b e a -> Fold1 a e r -> Fold1 b e r+transduce1 (M _) (Fold1 (Pure x)) = +    Fold1 (Pure x)+transduce1 (M f) (Fold1 (Other s)) = (Fold1 (Other (case s of+    TrueFold x -> TrueFold (Foldl.premapM f x)+    ExhaustiveCont x -> ExhaustiveCont (\producer -> x (producer >-> Pipes.map f))+    NonexhaustiveCont x -> NonexhaustiveCont (\producer -> x (producer >-> Pipes.map f)))))+transduce1 (F _) (Fold1 (Pure x)) = +    Fold1 (Pure x)+transduce1 (F f) (Fold1 (Other s)) = (Fold1 (Other (case s of+    TrueFold x -> TrueFold (Foldl.handlesM (folding f) x)+    ExhaustiveCont x -> ExhaustiveCont (\producer -> x (producer >-> Pipes.mapFoldable f))+    NonexhaustiveCont x -> NonexhaustiveCont (\producer -> x (producer >-> Pipes.mapFoldable f)))))+transduce1 (P f) (Fold1 (unLift -> s)) = case s of+    NonexhaustiveCont x -> Fold1 (Other (NonexhaustiveCont (x . f)))+    _ -> Fold1 (Other (ExhaustiveCont (exhaustiveCont s . f)))+transduce1 (PE f) (Fold1 (exhaustiveCont . unLift -> s)) = do+    Fold1 (Other (ExhaustiveCont (\producer -> do+        (outbox,inbox,seal) <- spawn' (bounded 1)+        runConceit $ +            (\(r,()) r' -> (r,r'))+            <$>+            Conceit (s (fromInput inbox) `finally` atomically seal)+            <*>+            (Conceit $+                (runEffect (f producer >-> (toOutput outbox *> Pipes.drain)) +                `finally` atomically seal)))))+transduce1 (S f) somefold = transduce1 (P (Pipes.concats . f)) somefold+transduce1 (SE f) somefold = transduce1 (PE (Pipes.concats . f)) somefold++{-| Tweak each of the groups delimited by a 'Transducer'.		++-}+groups +    :: (forall r. Producer b IO r -> Producer b' IO r) -- ^+    -> Transducer Delimited a e b  -- ^+    -> Transducer Delimited a e b' -- ^+groups f t = case t of+    M func -> P (f . (\producer -> producer >-> Pipes.map func))+    F func -> P (f . (\producer -> producer >-> Pipes.mapFoldable func))+    P g -> P (f . g)+    PE g -> PE (f . g)+    S g -> S (Pipes.maps f . g)+    SE g -> SE (Pipes.maps f . g)++folds +    :: Fold1 b Void b' -- ^+    -> Transducer Delimited a e b +    -> Transducer Continuous a e b'+folds somefold t = case t of+    M func -> folds somefold (P (\producer -> producer >-> Pipes.map func))+    F func -> folds somefold (P (\producer -> producer >-> Pipes.mapFoldable func))+    P g -> folds somefold (S (liftF . g))+    PE g -> folds somefold (SE (liftF . g))+    S g -> P (Pipes.concats . transFreeT ((\action -> lift action >>= (\(b',r) -> Pipes.yield b' >> return r)) . fold1 somefold) . g)+    SE g -> PE (Pipes.concats . transFreeT ((\action -> lift action >>= (\(b',r) -> Pipes.yield b' >> return r)) . fold1 somefold) . g)++data Delimited++data Continuous++concats +    :: Transducer Delimited a e b   -- ^+    -> Transducer Continuous a e b+concats t =  case t of+    M func -> M func+    F func -> F func+    P g -> P g+    PE g -> PE g+    S g -> P (Pipes.concats . g)+    SE g -> PE (Pipes.concats . g)++intercalates +    :: Producer b IO ()  -- ^+    -> Transducer Delimited a e b +    -> Transducer Continuous a e b+intercalates p t =  case t of+    M func -> M func+    F func -> F func+    P g -> P g+    PE g -> PE g+    S g -> P (Pipes.intercalates p . g)+    SE g -> PE (Pipes.intercalates p . g)++++{-| +    A computation in 'IO' that completely drains two 'Producer's of @b@ values+    in a concurrent way, returning a value of type @a@, except when it fails early+    with an error of type @e@.+-}+newtype Fold2 b1 b2 e a = Fold2 (Lift (Fold2_ b1 b2 e) a) deriving (Functor)++data Fold2_ b1 b2 e a = +      First (Fold1_ b1 e a)+    | Second (Fold1_ b2 e a)+    | Both (forall r1 r2. Producer b1 IO r1 -> Producer b2 IO r2 -> IO (Either e (a,r1,r2))) deriving (Functor)++fold2Fallibly_ :: Fold2_ b1 b2 e a -> Producer b1 IO r1 -> Producer b2 IO r2 -> IO (Either e (a,r1,r2))+fold2Fallibly_ theFold producer1 producer2 = case theFold of+        Both f -> f producer1 producer2+        First f -> runConceit $+            (\(r1,x1) (_,x2) -> (r1,x1,x2))+            <$>+            Conceit (exhaustiveCont f producer1)+            <*>+            Conceit (fold1Fallibly (pure ()) producer2)+        Second f -> runConceit $+            (\(_,x1) (r2,x2) -> (r2,x1,x2))+            <$>+            Conceit (fold1Fallibly (pure ()) producer1)+            <*>+            Conceit (exhaustiveCont f producer2)++instance Bifunctor (Fold2 b1 b2) where+    bimap f g (Fold2 x) = Fold2 (case x of+        Pure a -> Pure (g a)+        Other o -> Other (bimap f g o))++instance Bifunctor (Fold2_ b1 b2) where+    bimap f g (First s) = First (bimap f g s) +    bimap f g (Second s) = Second (bimap f g s) +    bimap f g (Both s) = Both (fmap (fmap (fmap (bimap f (\(x1,x2,x3) -> (g x1,x2,x3))))) s) ++instance Applicative (Fold2 b1 b2 e) where+    pure a = Fold2 (pure a)+    Fold2 fa <*> Fold2 a = Fold2 (fa <*> a)++instance Applicative (Fold2_ b1 b2 e) where+    pure a = fmap (const a) (separated_ (pure ()) (pure ()))++    Both fs <*> Both as = Both (\producer1 producer2 -> do+        (outbox1a,inbox1a,seal1a) <- spawn' (bounded 1)+        (outbox2a,inbox2a,seal2a) <- spawn' (bounded 1)+        (outbox1b,inbox1b,seal1b) <- spawn' (bounded 1)+        (outbox2b,inbox2b,seal2b) <- spawn' (bounded 1)+        runConceit $+            (\(f,(),()) (x,(),()) r1 r2 -> (f x,r1,r2))+            <$>+            Conceit (fs (fromInput inbox1a) (fromInput inbox1b) `finally` atomically seal1a `finally` atomically seal1b)+            <*>+            Conceit (as (fromInput inbox2a) (fromInput inbox2b) `finally` atomically seal2a `finally` atomically seal2b)+            <*>+            (_Conceit $+                (runEffect (producer1 >-> Pipes.tee (toOutput outbox1a *> Pipes.drain) +                                      >->           (toOutput outbox2a *> Pipes.drain)))+                `finally` atomically seal1a +                `finally` atomically seal2a)+            <*>+            (_Conceit $+                (runEffect (producer2 >-> Pipes.tee (toOutput outbox1b *> Pipes.drain) +                                      >->           (toOutput outbox2b *> Pipes.drain)))+                `finally` atomically seal1b +                `finally` atomically seal2b))+    First fs <*> First as = First (fs <*> as)+    Second fs <*> Second as = Second (fs <*> as)+    First fs <*> Second as = uncurry ($) <$> separated_ fs as+    Second fs <*> First as = uncurry (flip ($)) <$> separated_ as fs+    First fs <*> Both as =  (\(f,()) x -> f x) <$> separated_ fs (pure ()) <*> Both as +    Both fs <*> First as =  (\f (x,()) -> f x) <$> Both fs <*> separated_ as (pure ())+    Second fs <*> Both as = (\((),f) x -> f x) <$> separated_ (pure ()) fs <*> Both as +    Both fs <*> Second as = (\f ((),x) -> f x) <$> Both fs <*> separated_ (pure ()) as ++instance (Monoid a) => Monoid (Fold2 b1 b2 e a) where+   mempty = pure mempty+   mappend s1 s2 = (<>) <$> s1 <*> s2++{-| +    Run a 'Fold2'.+-}+fold2Fallibly :: Fold2 b1 b2 e a -> Producer b1 IO r1 -> Producer b2 IO r2 -> IO (Either e (a,r1,r2))+fold2Fallibly (Fold2 (fold2Fallibly_ . unLift -> s)) = s +++{-| +    Run a 'Fold2' that never returns an error value (but which may still throw exceptions!)+-}+fold2 :: Fold2 b1 b2 Void a -> Producer b1 IO r1 -> Producer b2 IO r2 -> IO (a,r1,r2)+fold2 s producer1 producer2 = liftM (either absurd id) (fold2Fallibly s producer1 producer2) ++liftFirst :: Fold1 b1 e r1 -> Fold2 b1 b2 e r1+liftFirst (unLift . runFold1 -> f1) = Fold2 (Other (First f1))++liftSecond :: Fold1 b2 e r1 -> Fold2 b1 b2 e r1+liftSecond (unLift . runFold1 -> f1) = Fold2 (Other (Second f1))++separated_ :: Fold1_ b1 e r1 -> Fold1_ b2 e r2 -> Fold2_ b1 b2 e (r1,r2)+separated_ f1 f2 = Both (\producer1 producer2 ->+    runConceit $+        (\(r1,x1) (r2,x2) -> ((r1,r2),x1,x2))+        <$>+        Conceit (exhaustiveCont f1 producer1)+        <*>+        Conceit (exhaustiveCont f2 producer2))++{-|+    Consume the producers concurrently, each one independently of the other. +-}+separated :: Fold1 b1 e r1 -> Fold1 b2 e r2 -> Fold2 b1 b2 e (r1,r2)+separated f1 f2 = Fold2 (Other (separated_ (unLift . runFold1 $ f1) (unLift . runFold1 $ f2)))++{-|+    Consume the producers concurrently, delimiting groups in each producer,+    and writing the groups into a common 'Fold1'. ++    Possible use: find lines in two text producers and combine the lines in a+    single stream, preserving the integrity of each individual line.+-}+combined :: Transducer Delimited b1 e x -> Transducer Delimited b2 e x -> Fold1 x e a -> Fold2 b1 b2 e a+combined t1 t2 f = Fold2 (Other (Both (\producer1 producer2 -> do+   (outbox, inbox, seal) <- spawn' (bounded 1)+   lock <- newMVar outbox+   runConceit $ +       (\(((),r1),((),r2)) (a,()) -> (a,r1,r2))+       <$>+       Conceit +           ((runConceit $+               (,)+               <$>+               Conceit (fold1Fallibly (transduce1 (folds (withCont' (iterTLines lock)) t1) (pure ())) producer1)+               <*>+               Conceit (fold1Fallibly (transduce1 (folds (withCont' (iterTLines lock)) t2) (pure ())) producer2)+           ) `finally` atomically seal)+       <*>+       Conceit (fold1Fallibly f (fromInput inbox) `finally` atomically seal))))+  where+    -- iterTLines mvar = iterT $ \textProducer -> do+    iterTLines mvar = \textProducer -> fmap (\x -> ((),x)) $ do+        -- the P.drain bit was difficult to figure out!!!+        withMVar mvar $ \output -> do+            runEffect $ textProducer >-> (toOutput output >> Pipes.drain)+

@@ -1,22 +1,31 @@ {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ViewPatterns #-}  module Pipes.Transduce.Text (-        -- * Collecting input+        -- * Text folds         intoLazyText -        -- * Splitting-    ,   lines-    ,   lines_-        -- * Grouping-    ,   foldedLines+    ,   asUtf8+    ,   asUtf8x+    ,   bothAsUtf8x+    ,   Line+    ,   asFoldedLines     ,   eachLine-        -- * Decoding+    ,   combinedLines+    ,   combinedLinesPrefixing+        -- * Text transducers+        -- ** Decoding     ,   decoder     ,   decoderx     ,   utf8     ,   utf8x+        -- ** Splitting+    ,   lines+    ,   lines_+    ,   foldedLines     ) where  import Prelude hiding (lines)+import Data.Bifunctor import Data.ByteString import qualified Data.Text  import qualified Data.Text.Lazy@@ -25,6 +34,7 @@ import qualified Control.Foldl as Foldl import Control.Exception import Control.Applicative+import Control.Applicative.Lift import Control.Monad import Control.Monad.Trans.Class import Control.Monad.Trans.Except@@ -32,7 +42,7 @@ import qualified Pipes.Text import Pipes.Text.Encoding (decodeUtf8)  -import Pipes.Transduce+import           Pipes.Transduce.Internal   {- $setup >>> :set -XOverloadedStrings@@ -41,44 +51,87 @@ >>> import Control.Applicative >>> import Control.Monad >>> import qualified Control.Foldl as L->>> import Pipes.Transduce ->>> import qualified Pipes.Transduce as PT+>>> import Pipes.Transduce.Internal >>> import Pipes.Transduce.Text->>> import qualified Pipes.Transduce.Text as PTT -}  {-| +    Whole lines are represented as lazy 'Data.Text.Lazy.Text' values.+-}+type Line = Data.Text.Lazy.Text ++{-|      Split the stream into lines, collect them into lazy 'Text' values, and pass     them downstream.  ->>> PT.fold1 (transduce1 foldedLines (withFold L.list)) (mapM_ yield ["aa","aa\nbb","bb"]) +>>> fold1 (transduce1 foldedLines (withFold L.list)) (mapM_ yield ["aa","aa\nbb","bb"])  (["aaaa","bbbb"],())  -} foldedLines -    :: Transducer Continuous Text e Data.Text.Lazy.Text +    :: Transducer Continuous Text e Line foldedLines = -    Pipes.Transduce.folds -    (fmap Data.Text.Lazy.fromChunks (Pipes.Transduce.withFold Foldl.list)) -    (lines_ (Pipes.Transduce.mapper id))+    folds +    (fmap Data.Text.Lazy.fromChunks (withFold Foldl.list)) +    (lines_ (mapper id)) +{-|  +Transforms a 'Fold1' that accepts whole lines into a 'Fold1' that accepts and+undivided text stream.++>>> fold1 (asFoldedLines (withFold L.list)) (mapM_ yield ["aa","aa\nbb","bb"]) +(["aaaa","bbbb"],())++-}+asFoldedLines :: Fold1 Line e r -> Fold1 Text e r+asFoldedLines = transduce1 foldedLines + {-|      Split the stream into lines, collect them into lazy 'Text' values, and     apply an effectul function to each line. ->>> PT.fold1Fallibly (eachLine $ \l -> pure $ if TL.head l == 'b' then (Left l) else (Right ())) (mapM_ yield ["aa","\nbb"]) +>>> fold1Fallibly (eachLine $ \l -> pure $ if TL.head l == 'b' then (Left l) else (Right ())) (mapM_ yield ["aa","\nbb"])  Left "bb"  -}-eachLine :: (Data.Text.Lazy.Text -> IO (Either e ())) -> Fold1 Data.Text.Text e ()+eachLine :: (Line -> IO (Either e ())) -> Fold1 Data.Text.Text e () eachLine action = transduce1 foldedLines (withFallibleConsumer (forever (do     await >>= lift . ExceptT . action)))  {-| +    Process two streams of text, combined as a single text stream.++    The streams are combined line by line, but the resulting stream is undivided.++>>> fold2 (combinedLines intoLazyText) (mapM_ yield ["aa"]) (mapM_ yield ["aa"])+("aa\naa\n",(),())++-}+combinedLines :: Fold1 Text e r -> Fold2 Text Text e r+combinedLines = +    combined (Pipes.Transduce.Text.lines (transducer id)) +             (Pipes.Transduce.Text.lines (transducer id))++{-| +    Like 'combinedLines', but adding different prefixes to lines from stdout+    and stderr.++>>> fold2 (combinedLinesPrefixing "-" "-" intoLazyText) (mapM_ yield ["aa"]) (mapM_ yield ["aa"])+("-aa\n-aa\n",(),())++-}+combinedLinesPrefixing :: Text -> Text -> Fold1 Text e r -> Fold2 Text Text e r+combinedLinesPrefixing outprefix errprefix = +    let tag prefix = groups (\producer -> Pipes.yield prefix *> producer)+    in+    combined (tag outprefix (Pipes.Transduce.Text.lines (transducer id))) +             (tag errprefix (Pipes.Transduce.Text.lines (transducer id)))++{-|      Split into lines, eliding newlines. ->>> PT.fold1 (transduce1 (concats . groups (\p -> yield "x" *> p) . lines_ $ utf8x) intoLazyText) (mapM_ yield ["aa\n","bb"]) +>>> fold1 (transduce1 (concats . groups (\p -> yield "x" *> p) . lines_ $ utf8x) intoLazyText) (mapM_ yield ["aa\n","bb"])  ("xaaxbb",())  -}@@ -90,7 +143,7 @@ {-|      Split into lines, preserving newlines. ->>> PT.fold1 (transduce1 (concats . groups (\p -> yield "x" *> p) . lines $ utf8x) intoLazyText) (mapM_ yield ["aa\n","bb"]) +>>> fold1 (transduce1 (concats . groups (\p -> yield "x" *> p) . lines $ utf8x) intoLazyText) (mapM_ yield ["aa\n","bb"])  ("xaa\nxbb\n",())  -}@@ -127,10 +180,19 @@         Left r -> return r         Right b -> throwIO (DecodeError "transducer decoding error" (Just (Data.ByteString.head (fst b))))))  +decoderx'+    :: (forall r. Producer ByteString IO r -> Producer Text IO (Producer ByteString IO r))+    -> Producer ByteString IO x -> Producer Text IO x +decoderx' f = (\producer -> f producer >>= \producer' -> lift (do+    n <- next producer'+    case n of+        Left r -> return r+        Right b -> throwIO (DecodeError "transducer decoding error" (Just (Data.ByteString.head (fst b)))))) + {-|      The first undecodable bytes will be the error value. ->>> PT.fold1Fallibly (transduce1 utf8 intoLazyText) (mapM_ yield ["aa"]) +>>> fold1Fallibly (transduce1 utf8 intoLazyText) (mapM_ yield ["aa"])  Right ("aa",())  -}@@ -139,7 +201,7 @@  {-|  ->>> PT.fold1  (transduce1 utf8x intoLazyText) (mapM_ yield ["aa"]) +>>> fold1  (transduce1 utf8x intoLazyText) (mapM_ yield ["aa"])  ("aa",())      __/BEWARE!/__ @@ -150,9 +212,56 @@ utf8x = decoderx decodeUtf8  {-| ++Turns a fold that accepts `Text` into a fold that accepts UTF8-encoded+`ByteString`.++It also takes a function that maps undecoded leftovers to a more general error+type. ++>>> fold1Fallibly (asUtf8 id intoLazyText) (mapM_ yield ["aa"]) +Right ("aa",())+ + -}+asUtf8 :: (ByteString -> e) -> Fold1 Text e r -> Fold1 ByteString e r+asUtf8 erradapt = transduce1 (first erradapt utf8)++{-| ++Like 'asUtf8', but throws exceptions in case of decoding errors.++>>> fold1  (asUtf8x intoLazyText) (mapM_ yield ["aa"]) +("aa",())+ +    __/BEWARE!/__ +    This 'Transducer' may throw 'DecodeError'.+    __/BEWARE!/__ + -}+asUtf8x :: Fold1 Text e r -> Fold1 ByteString e r+asUtf8x = transduce1 utf8x++{-| +>>> fold2 (bothAsUtf8x (combinedLines intoLazyText)) (mapM_ yield ["aa"]) (mapM_ yield ["aa"])+("aa\naa\n",(),())++    __/BEWARE!/__ +    This 'Transducer' may throw 'DecodeError'.+    __/BEWARE!/__ + -}+bothAsUtf8x :: Fold2 Text Text e r -> Fold2 ByteString ByteString e r+bothAsUtf8x (Fold2 (unLift -> f)) = case f of+    First f1  -> Fold2 (Other (First $ unwrap (trans f1)))+    Second f1 -> Fold2 (Other (Second $ unwrap (trans f1)))+    Both both -> Fold2 (Other (Both $ \f1 f2 -> both (dec f1) (dec f2)))+    where +    dec = decoderx' decodeUtf8+    trans = transduce1 (transducer dec) . Fold1 . Other+    unwrap (Fold1 z) = unLift z++{-|      Collect strict 'Text's into a lazy 'Text'. ->>> PT.fold1  intoLazyText (mapM_ yield ["aa","bb","cc"]) +>>> fold1  intoLazyText (mapM_ yield ["aa","bb","cc"])  ("aabbcc",())  -}

@@ -5,7 +5,7 @@ main :: IO () main = doctest      [-        "src/Pipes/Transduce.hs"+        "src/Pipes/Transduce/Internal.hs"     ,   "src/Pipes/Transduce/ByteString.hs"     ,   "src/Pipes/Transduce/Text.hs"     ]