diff --git a/CHANGELOG b/CHANGELOG --- a/CHANGELOG +++ b/CHANGELOG @@ -1,3 +1,6 @@ +0.2.3.0 +- More efficient Applicative for Fold2 + 0.2.0.0 - Added folds of two Producers. diff --git a/pipes-transduce.cabal b/pipes-transduce.cabal --- a/pipes-transduce.cabal +++ b/pipes-transduce.cabal @@ -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 diff --git a/src/Pipes/Transduce.hs b/src/Pipes/Transduce.hs --- a/src/Pipes/Transduce.hs +++ b/src/Pipes/Transduce.hs @@ -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) + diff --git a/src/Pipes/Transduce/ByteString.hs b/src/Pipes/Transduce/ByteString.hs --- a/src/Pipes/Transduce/ByteString.hs +++ b/src/Pipes/Transduce/ByteString.hs @@ -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 diff --git a/src/Pipes/Transduce/Internal.hs b/src/Pipes/Transduce/Internal.hs deleted file mode 100644 --- a/src/Pipes/Transduce/Internal.hs +++ /dev/null @@ -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) diff --git a/src/Pipes/Transduce/Text.hs b/src/Pipes/Transduce/Text.hs --- a/src/Pipes/Transduce/Text.hs +++ b/src/Pipes/Transduce/Text.hs @@ -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",()) -}