pipes-transduce 0.2.0.0 → 0.2.3.0
raw patch · 6 files changed
+543/−558 lines, 6 filesdep +kan-extensions
Dependencies added: kan-extensions
Files
- CHANGELOG +3/−0
- pipes-transduce.cabal +3/−3
- src/Pipes/Transduce.hs +533/−23
- src/Pipes/Transduce/ByteString.hs +1/−23
- src/Pipes/Transduce/Internal.hs +0/−489
- src/Pipes/Transduce/Text.hs +3/−20
CHANGELOG view
@@ -1,3 +1,6 @@+0.2.3.0+- More efficient Applicative for Fold2+ 0.2.0.0 - Added folds of two Producers.
pipes-transduce.cabal view
@@ -1,5 +1,5 @@ Name: pipes-transduce-Version: 0.2.0.0+Version: 0.2.3.0 Cabal-Version: >=1.8.0.2 Build-Type: Simple License: BSD3@@ -35,7 +35,7 @@ comonad == 4.* , free == 4.* , pipes == 4.* ,- pipes-concurrency >= 2.0.2 && < 3 ,+ pipes-concurrency >= 2.0.2 && < 3 , pipes-group >= 1.0.1 , pipes-parse , pipes-safe ,@@ -44,12 +44,12 @@ monoid-subclasses == 0.4.* , void >= 0.6 && < 1.0 , conceit >= 0.3.2.0 && < 0.4.0.0,+ kan-extensions >= 4.2, lens-family-core >= 1.1 && < 2 Exposed-Modules: Pipes.Transduce Pipes.Transduce.ByteString Pipes.Transduce.Text- Pipes.Transduce.Internal GHC-Options: -O2 -Wall test-suite doctests
src/Pipes/Transduce.hs view
@@ -1,21 +1,24 @@-module Pipes.Transduce (+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE ViewPatterns #-}+{-# LANGUAGE EmptyDataDecls #-}++module Pipes.Transduce ( -- * Producer folds Fold1- , foldFallibly1- , Pipes.Transduce.Internal.fold1- -- * Multiple Producer folds- , Fold2- , foldFallibly2- , fold2- , separated- , combined- -- * Building folds- -- ** From foldl folds+ , fold1+ , fold1Fallibly+ -- ** Building producer folds+ -- *** From foldl folds -- $foldl , withFold , withFoldIO , withFallibleFold- -- ** From consumers+ -- *** From consumers -- $consumers , withConsumer , withConsumer' @@ -23,43 +26,51 @@ , withConsumerM' , withSafeConsumer , withFallibleConsumer - -- ** From parsers+ -- *** From parsers -- $parsers , withParser , withParserM - -- ** From continuations+ -- *** From continuations -- $continuations , withCont , withCont' , withFallibleCont , withFallibleCont' - -- * Transducers+ -- * Fold transducers , Transducer , Delimited , Continuous , transduce1- -- * Building transducers+ -- ** Building fold transducers , mapper , fallibleMapper , mapperFoldable , mapperEnumerable , transducer , fallibleTransducer- -- * Transducer group operations+ -- ** Transducer group operations , delimit , groups , folds , concats , intercalates+ -- * Multiple producer folds+ , Fold2+ , fold2+ , fold2Fallibly+ -- ** Building multiple producer folds+ , liftFirst+ , liftSecond+ , separated+ , combined -- * Utilities , trip , tripx ) where -import Pipes.Transduce.Internal import Data.Bifunctor-import Data.Monoid+import Data.Monoid hiding (First) import Data.Void import Data.Foldable import Control.Applicative@@ -68,12 +79,20 @@ 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+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.Family (folding) + {- $setup >>> :set -XOverloadedStrings >>> import qualified Data.Text as T @@ -91,10 +110,10 @@ Fail if the 'Producer' produces anything at all. The error value is what came out of the 'Producer'. ->>> PT.foldFallibly1 trip (mapM_ yield ['z']) +>>> fold1Fallibly trip (mapM_ yield ['z']) Left 'z' ->>> PT.foldFallibly1 trip (mapM_ yield []) +>>> fold1Fallibly trip (mapM_ yield []) Right ((),()) -} trip :: Fold1 b b ()@@ -111,7 +130,7 @@ This 'Transducer may throw 'AssertionFailed'. __/BEWARE!/__ ->>> PT.foldFallibly1 tripx (mapM_ yield ['z']) +>>> fold1Fallibly tripx (mapM_ yield ['z']) *** Exception: tripx -} tripx :: Fold1 b e ()@@ -143,3 +162,494 @@ 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)++++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))++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)))++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 (fold1Fallibly (transduce1 (folds (withCont' (iterTLines lock)) t1) (pure ())) producer1 `finally` atomically seal)+ <*>+ 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)+
src/Pipes/Transduce/ByteString.hs view
@@ -14,33 +14,11 @@ import Prelude hiding (lines) import Data.Bifunctor-import Data.Monoid import Data.Void-import Data.Foldable import Data.ByteString import qualified Data.ByteString.Lazy-import Data.Text hiding (lines)-import Data.Text.Encoding.Error (UnicodeException(..))-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.Exception-import Pipes -import qualified Pipes.Text-import Pipes.Text.Encoding (decodeUtf8) -import Pipes.Lift (distribute) ---import Pipes.Prelude import qualified Pipes.ByteString (hGetSome)-import qualified Pipes.Prelude as Pipes-import qualified Pipes.Group as Pipes-import qualified Pipes.Parse-import qualified Pipes.Text-import qualified Data.Text.Lazy-import Pipes.Concurrent-import Lens.Family (view) import System.IO import Data.ByteString.Lazy.Internal (defaultChunkSize) @@ -69,7 +47,7 @@ -> Handle -> IO (Either e r) drainHandleFallibly somefold (ChunkSize csize) handle =- fmap (bimap id fst) (Pipes.Transduce.foldFallibly1 somefold (Pipes.ByteString.hGetSome csize handle))+ fmap (bimap id fst) (Pipes.Transduce.fold1Fallibly somefold (Pipes.ByteString.hGetSome csize handle)) drainHandle :: Fold1 ByteString Void r
− src/Pipes/Transduce/Internal.hs
@@ -1,489 +0,0 @@-{-# 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-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-import Control.Concurrent.Conceit-import Control.Exception-import Pipes -import Pipes.Lift (distribute) -import Pipes.Prelude-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.Family (folding)--{-| - 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 (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'.--}-foldFallibly1 :: Fold1 b e a -> Producer b IO r -> IO (Either e (a,r))-foldFallibly1 (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--fallibleM - :: (a -> Either e b) -- ^- -> Transducer Continuous a e b -- ^-fallibleM 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)))--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.Prelude.map func))- F func -> S (\producer -> f (producer >-> 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.Prelude.map f))- NonexhaustiveCont x -> NonexhaustiveCont (\producer -> x (producer >-> Pipes.Prelude.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.Prelude.mapFoldable f))- NonexhaustiveCont x -> NonexhaustiveCont (\producer -> x (producer >-> Pipes.Prelude.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.Prelude.map func))- F func -> P (f . (\producer -> producer >-> 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.Prelude.map func))- F func -> folds somefold (P (\producer -> producer >-> 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)) . Pipes.Transduce.Internal.fold1 somefold) . g)- SE g -> PE (Pipes.concats . transFreeT ((\action -> lift action >>= (\(b',r) -> Pipes.yield b' >> return r)) . Pipes.Transduce.Internal.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)----newtype Fold2 b1 b2 e a = Fold2 (forall r1 r2. Producer b1 IO r1 -> Producer b2 IO r2 -> IO (Either e (a,r1,r2))) deriving (Functor)--instance Bifunctor (Fold2 b1 b2) where- bimap f g (Fold2 s) = Fold2 (fmap (fmap (fmap (bimap f (\(x1,x2,x3) -> (g x1,x2,x3))))) s) --instance Applicative (Fold2 b1 b2 e) where- pure a = Fold2 (\producer1 producer2 -> do- (r1,r2) <- _runConceit $- (,)- <$>- _Conceit (runEffect (producer1 >-> Pipes.drain))- <*>- _Conceit (runEffect (producer2 >-> Pipes.drain))- pure (Right (a,r1,r2)))-- Fold2 fs <*> Fold2 as = Fold2 (\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))--instance (Monoid a) => Monoid (Fold2 b1 b2 e a) where- mempty = pure mempty- mappend s1 s2 = (<>) <$> s1 <*> s2--{-| - Run a 'Fold2'.--}-foldFallibly2 :: Fold2 b1 b2 e a -> Producer b1 IO r1 -> Producer b2 IO r2 -> IO (Either e (a,r1,r2))-foldFallibly2 (Fold2 s) producer1 producer2 = s producer1 producer2---{-| - 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 (Fold2 s) producer1 producer2 = liftM (either absurd id) (s producer1 producer2) --separated :: Fold1 b1 e r1 -> Fold1 b2 e r2 -> Fold2 b1 b2 e (r1,r2)-separated f1 f2 = Fold2 (\producer1 producer2 ->- runConceit $- (\(r1,x1) (r2,x2) -> ((r1,r2),x1,x2))- <$>- Conceit (foldFallibly1 f1 producer1)- <*>- Conceit (foldFallibly1 f2 producer2))--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 (\producer1 producer2 -> do- (outbox, inbox, seal) <- spawn' (bounded 1)- lock <- newMVar outbox- runConceit $ - (\((),r1) ((),r2) (a,()) -> (a,r1,r2))- <$>- Conceit (foldFallibly1 (transduce1 (folds (withCont' (iterTLines lock)) t1) (pure ())) producer1 `finally` atomically seal)- <*>- Conceit (foldFallibly1 (transduce1 (folds (withCont' (iterTLines lock)) t2) (pure ())) producer2 `finally` atomically seal)- <*>- Conceit (foldFallibly1 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)
src/Pipes/Transduce/Text.hs view
@@ -16,36 +16,19 @@ ) where import Prelude hiding (lines)-import Data.Bifunctor-import Data.Monoid-import Data.Void-import Data.Foldable import Data.ByteString import qualified Data.Text import Data.Text hiding (lines) import Data.Text.Encoding.Error (UnicodeException(..))-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.Exception import Pipes import qualified Pipes.Text import Pipes.Text.Encoding (decodeUtf8) -import Pipes.Lift (distribute) ---import Pipes.Prelude-import qualified Pipes.Prelude as Pipes-import qualified Pipes.Group as Pipes-import qualified Pipes.Parse-import qualified Pipes.Text import qualified Data.Text.Lazy-import Pipes.Concurrent import Lens.Family (view) import Pipes.Transduce-import Pipes.Transduce.Internal {- $setup >>> :set -XOverloadedStrings@@ -104,7 +87,7 @@ decoder :: (forall r. Producer ByteString IO r -> Producer Text IO (Producer ByteString IO r)) -> Transducer Continuous ByteString ByteString Text -- ^-decoder f = PE (\producer -> f producer >>= \producer' -> lift (do+decoder f = fallibleTransducer (\producer -> f producer >>= \producer' -> lift (do n <- next producer' case n of Left r -> return (Right r)@@ -119,7 +102,7 @@ decoderx :: (forall r. Producer ByteString IO r -> Producer Text IO (Producer ByteString IO r)) -> Transducer Continuous ByteString e Text -- ^-decoderx f = P (\producer -> f producer >>= \producer' -> lift (do+decoderx f = transducer (\producer -> f producer >>= \producer' -> lift (do n <- next producer' case n of Left r -> return r@@ -128,7 +111,7 @@ {-| The first undecodable bytes will be the error value. ->>> PT.foldFallibly1 (transduce1 utf8 intoLazyText) (mapM_ yield ["aa"]) +>>> PT.fold1Fallibly (transduce1 utf8 intoLazyText) (mapM_ yield ["aa"]) Right ("aa",()) -}