streaming-eversion (empty) → 0.1.0.0
raw patch · 9 files changed
+849/−0 lines, 9 filesdep +basedep +bifunctorsdep +comonadsetup-changed
Dependencies added: base, bifunctors, comonad, doctest, foldl, free, pipes, pipes-text, profunctors, streaming, streaming-eversion, tasty, tasty-hunit, tasty-quickcheck, transformers
Files
- CHANGELOG +0/−0
- LICENSE +28/−0
- README.md +50/−0
- Setup.hs +2/−0
- src/Streaming/Eversion.hs +413/−0
- src/Streaming/Eversion/Pipes.hs +123/−0
- streaming-eversion.cabal +67/−0
- tests/doctests.hs +10/−0
- tests/tests.hs +156/−0
+ CHANGELOG view
+ LICENSE view
@@ -0,0 +1,28 @@+Copyright (c) 2015, Daniel Díaz Carrete+All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++* Redistributions of source code must retain the above copyright notice, this+ list of conditions and the following disclaimer.++* Redistributions in binary form must reproduce the above copyright notice,+ this list of conditions and the following disclaimer in the documentation+ and/or other materials provided with the distribution.++* Neither the name of foldl-transduce nor the names of its+ contributors may be used to endorse or promote products derived from+ this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.+
+ README.md view
@@ -0,0 +1,50 @@+## What's in this library?++Functions that turn pull-based stream operations from the pipes/streaming+ecosystem into push-based, iteratee-like stream operations. ++Inspired by the blog post [Programmatic translation to iteratees from pull-based code](http://pchiusano.blogspot.com.es/2011/12/programmatic-translation-to-iteratees.html).++## Could you go into more detail?++There are three streaming libraries that often go together:+[pipes](http://hackage.haskell.org/package/pipes),+[streaming](http://hackage.haskell.org/package/streaming), and+[foldl](http://hackage.haskell.org/package/foldl).++Of these, the first two are pull-based: you take some (possibly effectful)+source of values and keep extracting stuff until the source is exhausted and/or+you have obtained all the info you need.++Meanwhile, foldl is push-based: foldl folds are not directly aware of any+source, they are like little state machines that keep running as long as+someone feeds them input. ++Usually, defining stream transformations in pull-based mode is easier and feels+more natural. The pipes ecosystem already provides a lot of them:+[parsers](http://hackage.haskell.org/package/pipes-parse),+[decoders](http://hackage.haskell.org/package/pipes-text),+[splitters](http://hackage.haskell.org/package/pipes-group)...++However, push-based mode also has advantages. Push-based abstractions are not+tied to a particular type of source because data is fed externally. And foldl+folds have very useful Applicative and Comonad instances. ++Also, sometimes, a library will only offer a push-based interface. ++Wouldn't it be nice if you could adapt already existing pull-based operations+to work on push-based consumers? For example, using a decoding function from+[Pipes.Text.Encoding](http://hackage.haskell.org/package/pipes-text-0.0.2.4/docs/Pipes-Text-Encoding.html#g:6)+to preprocess the inputs of a+[Fold](http://hackage.haskell.org/package/foldl-1.2.1/docs/Control-Foldl-Text.html).++This library provides that.++## Why so many newtypes?++To avoid having to enable [-XImpredicativeTypes](https://downloads.haskell.org/~ghc/latest/docs/html/users_guide/glasgow_exts.html#impredicative-polymorphism).++## Is it fast?++I haven't benchmarked or optimized it. It is likely to be slow.+
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ src/Streaming/Eversion.hs view
@@ -0,0 +1,413 @@+{-# LANGUAGE RankNTypes #-}+++{-| The pull-to-push transformations in this module require functions that are + polymorphic over a monad transformer. + + Because of this, some of the type signatures look scary, but actually many+ (suitably polymorphic) operations on 'Stream's will unify with them.+ + To get "interruptible" operations that can exit early with an error, put a+ 'ExceptT' transformer just below the polymorphic monad transformer. See+ 'foldE'.+ + Inspired by http://pchiusano.blogspot.com.es/2011/12/programmatic-translation-to-iteratees.html+-}++module Streaming.Eversion (+ -- * Evertible Stream folds+ Evertible+ , evertible+ , evert+ , EvertibleM+ , evertibleM+ , evertM+ , EvertibleMIO+ , evertibleMIO+ , evertMIO+ -- * Transvertible Stream transformations+ , Transvertible+ , transvertible+ , transvert+ , TransvertibleM+ , transvertibleM+ , transvertM+ , TransvertibleMIO+ , transvertibleMIO+ , transvertMIO+ -- * Auxiliary functions+ , foldE+ ) where++import Data.Bifunctor+import Data.Profunctor++import Control.Foldl (Fold(..),FoldM(..))+import qualified Control.Foldl as Foldl+import Streaming (Stream,Of(..))+import Streaming.Prelude (yield,next)+import qualified Streaming.Prelude as S++import Control.Monad.IO.Class+import Control.Monad.Trans.Class+import Control.Monad.Free+import qualified Control.Monad.Trans.Free as TF+import Control.Monad.Trans.Except+import Control.Comonad++{- $setup+>>> import Data.Functor.Identity+>>> import Control.Monad.Trans.Except+>>> import Control.Monad.Trans.Identity+>>> import Control.Foldl (Fold(..),FoldM(..))+>>> import qualified Control.Foldl as L+>>> import Streaming (Stream,Of(..))+>>> import Streaming.Prelude (yield,next)+>>> import qualified Streaming.Prelude as S+-}++-----------------------------------------------------------------------------------------++data Feed a = Input a | EOF++-- What type could go here for efficiency?+type Iteratee a = Free ((->) a) ++evertedStream :: forall a. Stream (Of a) (Iteratee (Feed a)) ()+evertedStream = do+ r <- lift (liftF id)+ case r of+ Input a -> do+ yield a+ evertedStream+ EOF -> return ()++type IterateeT a m = TF.FreeT ((->) a) m ++evertedStreamM :: forall a m. Monad m => Stream (Of a) (IterateeT (Feed a) m) ()+evertedStreamM = do+ r <- lift (TF.liftF id)+ case r of+ Input a -> do+ yield a+ evertedStreamM+ EOF -> return ()++-----------------------------------------------------------------------------------------++-- | A stream-consuming function that can be turned into a pure, push-based fold. +newtype Evertible a x = + Evertible (forall m r. Monad m => Stream (Of a) m r -> m (Of x r)) ++instance Functor (Evertible a) where+ fmap f (Evertible somefold) = Evertible (fmap (first f) . somefold) ++instance Profunctor Evertible where+ lmap f (Evertible somefold) = Evertible (somefold . S.map f)+ rmap = fmap++stoppedBeforeEOF :: String+stoppedBeforeEOF = "Stopped before receiving EOF."++continuedAfterEOF :: String+continuedAfterEOF = "Continued after receiving EOF."++evertible :: (forall m r. Monad m => Stream (Of a) m r -> m (Of x r)) -> Evertible a x+evertible = Evertible++evert :: Evertible a x -> Fold a x+evert (Evertible consumer) = Fold step begin done+ where+ begin = consumer evertedStream+ step s a = case s of+ Pure _ -> error stoppedBeforeEOF+ Free f -> f (Input a)+ done s = case s of+ Pure _ -> error stoppedBeforeEOF+ Free f -> case f EOF of+ Pure (a :> ()) -> a+ Free _ -> error continuedAfterEOF+++{- | Like 'Evertible', but gives the stream-consuming function access to a base monad.+ +>>> :{+ let f stream = fmap ((:>) ()) (lift (putStrLn "x") >> S.effects stream)+ in L.foldM (evertM (evertibleM f)) ["a","b","c"]+ :}+x++ Note however that control operations can't be lifted through the transformer.+-}+newtype EvertibleM m a x = + EvertibleM (forall t r. (MonadTrans t, Monad (t m)) => Stream (Of a) (t m) r -> t m (Of x r)) ++instance Functor (EvertibleM m a) where+ fmap f (EvertibleM somefold) = EvertibleM (fmap (first f) . somefold) ++instance Profunctor (EvertibleM m) where+ lmap f (EvertibleM somefold) = EvertibleM (somefold . S.map f)+ rmap = fmap++evertibleM ::(forall t r . (MonadTrans t, Monad (t m)) => Stream (Of a) (t m) r -> t m (Of x r)) -- ^+ -> EvertibleM m a x+evertibleM = EvertibleM ++evertM :: Monad m => EvertibleM m a x -> FoldM m a x+evertM (EvertibleM consumer) = FoldM step begin done+ where+ begin = return (consumer evertedStreamM)+ step (TF.FreeT ms) i = do+ s <- ms+ case s of+ TF.Pure _ -> error stoppedBeforeEOF+ TF.Free f -> return (f (Input i))+ done (TF.FreeT ms) = do+ s <- ms+ case s of + TF.Pure _ -> error stoppedBeforeEOF+ TF.Free f -> do+ let TF.FreeT ms' = f EOF+ s' <- ms'+ case s' of+ TF.Pure (a :> ()) -> return a+ TF.Free _ -> error continuedAfterEOF++{-| Like 'EvertibleM', but gives the stream-consuming function the ability to use 'liftIO'.+ +>>> L.foldM (evertMIO (evertibleMIO (\stream -> fmap ((:>) ()) (S.print stream)))) ["a","b","c"]+"a"+"b"+"c"++-}+newtype EvertibleMIO m a x = + EvertibleMIO (forall t r. (MonadTrans t, MonadIO (t m)) => Stream (Of a) (t m) r -> t m (Of x r)) ++instance Functor (EvertibleMIO m a) where+ fmap f (EvertibleMIO somefold) = EvertibleMIO (fmap (first f) . somefold) ++instance Profunctor (EvertibleMIO m) where+ lmap f (EvertibleMIO somefold) = EvertibleMIO (somefold . S.map f)+ rmap = fmap++evertibleMIO ::(forall t r . (MonadTrans t, MonadIO (t m)) => Stream (Of a) (t m) r -> t m (Of x r)) -- ^+ -> EvertibleMIO m a x+evertibleMIO = EvertibleMIO ++evertMIO :: MonadIO m => EvertibleMIO m a x -> FoldM m a x +evertMIO (EvertibleMIO consumer) = FoldM step begin done+ where+ begin = return (consumer evertedStreamM)+ step (TF.FreeT ms) i = do+ s <- ms+ case s of+ TF.Pure _ -> error stoppedBeforeEOF+ TF.Free f -> return (f (Input i))+ done (TF.FreeT ms) = do+ s <- ms+ case s of + TF.Pure _ -> error stoppedBeforeEOF+ TF.Free f -> do+ let TF.FreeT ms' = f EOF+ s' <- ms'+ case s' of+ TF.Pure (a :> ()) -> return a+ TF.Free _ -> error continuedAfterEOF++-- | A stream-transforming function that can be turned into fold-transforming function.+newtype Transvertible a b = + Transvertible (forall m r. Monad m => Stream (Of a) m r -> Stream (Of b) m r)++instance Functor (Transvertible a) where+ fmap f (Transvertible transducer) = Transvertible (S.map f . transducer) ++instance Profunctor Transvertible where+ lmap f (Transvertible somefold) = Transvertible (somefold . S.map f)+ rmap = fmap++data Pair a b = Pair !a !b++data StreamState a b = Pristine (Stream (Of b) (Iteratee (Feed a)) ())+ | Waiting (Feed a -> Iteratee (Feed a) (Either () (b, Stream (Of b) (Iteratee (Feed a)) ())))+++transvertible :: (forall m r. Monad m => Stream (Of a) m r -> Stream (Of b) m r) -- ^+ -> Transvertible a b+transvertible = Transvertible++transvert :: Transvertible b a + -> (forall x. Fold a x -> Fold b x)+transvert (Transvertible transducer) somefold = Fold step begin done+ where+ begin = Pair somefold (Pristine (transducer evertedStream))+ step (Pair innerfold (Pristine pristine)) i = step (advance innerfold pristine) i+ step (Pair innerfold (Waiting waiting)) i = + case waiting (Input i) of+ Pure (Left ()) -> error stoppedBeforeEOF+ Pure (Right (a, stream)) -> advance (Foldl.fold (duplicate innerfold) [a]) stream+ Free f -> Pair innerfold (Waiting f)+ advance innerfold stream = + case next stream of+ Pure (Left ()) -> error stoppedBeforeEOF+ Pure (Right (a,future)) -> advance (Foldl.fold (duplicate innerfold) [a]) future+ Free f -> Pair innerfold (Waiting f)+ done (Pair innerfold (Pristine pristine)) = done (advance innerfold pristine) + done (Pair innerfold (Waiting waiting)) =+ case waiting EOF of+ Pure (Left ()) -> extract innerfold+ Pure (Right (a, stream)) -> extract (advancefinal (Foldl.fold (duplicate innerfold) [a]) stream)+ Free _ -> error continuedAfterEOF+ advancefinal innerfold stream = + case next stream of+ Pure (Left ()) -> innerfold + Pure (Right (a,future)) -> advancefinal (Foldl.fold (duplicate innerfold) [a]) future+ Free _ -> error continuedAfterEOF++data StreamStateM m a b = PristineM (Stream (Of b) (IterateeT (Feed a) m) ())+ | WaitingM (Feed a -> IterateeT (Feed a) m (Either () (b, Stream (Of b) (IterateeT (Feed a) m) ())))++-- | Like 'Transvertible', but gives the stream-transforming function access to a base monad.+-- +-- Note however that control operations can't be lifted through the transformer.+--+newtype TransvertibleM m a b = + TransvertibleM (forall t r. (MonadTrans t, Monad (t m)) => Stream (Of a) (t m) r -> Stream (Of b) (t m) r)++transvertibleM :: (forall t r. (MonadTrans t, Monad (t m)) => Stream (Of a) (t m) r -> Stream (Of b) (t m) r) -- ^+ -> TransvertibleM m a b +transvertibleM = TransvertibleM++instance Functor (TransvertibleM m a) where+ fmap f (TransvertibleM transducer) = TransvertibleM (S.map f . transducer) ++instance Profunctor (TransvertibleM m) where+ lmap f (TransvertibleM somefold) = TransvertibleM (somefold . S.map f)+ rmap = fmap++transvertM :: Monad m + => TransvertibleM m b a + -> (forall x . FoldM m a x -> FoldM m b x)+transvertM (TransvertibleM transducer) somefold = FoldM step begin done+ where+ begin = return (Pair somefold (PristineM (transducer evertedStreamM)))+ step (Pair innerfold (PristineM pristine)) i = do+ s <- advance innerfold pristine + step s i+ step (Pair innerfold (WaitingM waiting)) i = do + s <- TF.runFreeT (waiting (Input i))+ case s of+ TF.Pure (Left ()) -> error stoppedBeforeEOF+ TF.Pure (Right (a, nexx)) -> do+ step1 <- Foldl.foldM (Foldl.duplicateM innerfold) [a]+ advance step1 nexx + TF.Free f -> return (Pair innerfold (WaitingM f))+ advance innerfold stream = do + r <- TF.runFreeT (next stream) + case r of+ TF.Pure (Left ()) -> error stoppedBeforeEOF+ TF.Pure (Right (a,future)) -> do+ step1 <- Foldl.foldM (Foldl.duplicateM innerfold) [a]+ advance step1 future+ TF.Free f -> return (Pair innerfold (WaitingM f))+ done (Pair innerfold (PristineM pristine)) = do+ s <- advance innerfold pristine + done s+ done (Pair innerfold (WaitingM waiting)) = do+ s <- TF.runFreeT (waiting EOF)+ case s of+ TF.Pure (Left ()) -> do+ Foldl.foldM innerfold []+ TF.Pure (Right (a,future)) -> do+ step1 <- Foldl.foldM (Foldl.duplicateM innerfold) [a]+ r <- advancefinal step1 future+ Foldl.foldM r []+ TF.Free _ -> error continuedAfterEOF+ advancefinal innerfold stream = do+ r <- TF.runFreeT (next stream) + case r of+ TF.Pure (Right (a,future)) -> do+ step1 <- Foldl.foldM (Foldl.duplicateM innerfold) [a]+ advancefinal step1 future+ TF.Pure (Left ()) -> return innerfold+ TF.Free _ -> error continuedAfterEOF+++-- | Like 'TransvertibleM', but gives the stream-consuming function the ability to use 'liftIO'.+-- +newtype TransvertibleMIO m a b = + TransvertibleMIO (forall t r. (MonadTrans t, MonadIO (t m)) => Stream (Of a) (t m) r -> Stream (Of b) (t m) r)++instance Functor (TransvertibleMIO m a) where+ fmap f (TransvertibleMIO transducer) = TransvertibleMIO (S.map f . transducer) ++instance Profunctor (TransvertibleMIO m) where+ lmap f (TransvertibleMIO somefold) = TransvertibleMIO (somefold . S.map f)+ rmap = fmap++transvertibleMIO :: (forall t r. (MonadTrans t, MonadIO (t m)) => Stream (Of a) (t m) r -> Stream (Of b) (t m) r) -- ^+ -> TransvertibleMIO m a b +transvertibleMIO = TransvertibleMIO++transvertMIO :: (MonadIO m) + => TransvertibleMIO m b a + -> (forall x . FoldM m a x -> FoldM m b x)++transvertMIO (TransvertibleMIO transducer) somefold = FoldM step begin done+ where+ begin = return (Pair somefold (PristineM (transducer evertedStreamM)))+ step (Pair innerfold (PristineM pristine)) i = do+ s <- advance innerfold pristine + step s i+ step (Pair innerfold (WaitingM waiting)) i = do + s <- TF.runFreeT (waiting (Input i))+ case s of+ TF.Pure (Left ()) -> error stoppedBeforeEOF+ TF.Pure (Right (a, nexx)) -> do+ step1 <- Foldl.foldM (Foldl.duplicateM innerfold) [a]+ advance step1 nexx + TF.Free f -> return (Pair innerfold (WaitingM f))+ advance innerfold stream = do + r <- TF.runFreeT (next stream) + case r of+ TF.Pure (Left ()) -> error stoppedBeforeEOF+ TF.Pure (Right (a,future)) -> do+ step1 <- Foldl.foldM (Foldl.duplicateM innerfold) [a]+ advance step1 future+ TF.Free f -> return (Pair innerfold (WaitingM f))+ done (Pair innerfold (PristineM pristine)) = do+ s <- advance innerfold pristine + done s+ done (Pair innerfold (WaitingM waiting)) = do+ s <- TF.runFreeT (waiting EOF)+ case s of+ TF.Pure (Left ()) -> do+ Foldl.foldM innerfold []+ TF.Pure (Right (a,future)) -> do+ step1 <- Foldl.foldM (Foldl.duplicateM innerfold) [a]+ r <- advancefinal step1 future+ Foldl.foldM r []+ TF.Free _ -> error continuedAfterEOF+ advancefinal innerfold stream = do+ r <- TF.runFreeT (next stream) + case r of+ TF.Pure (Right (a,future)) -> do+ step1 <- Foldl.foldM (Foldl.duplicateM innerfold) [a]+ advancefinal step1 future+ TF.Pure (Left ()) -> return innerfold+ TF.Free _ -> error continuedAfterEOF++{-| If your stream-folding computation can fail early returning a 'Left',+ compose it with this function before passing it to 'evertibleM'. ++ The result will be an 'EvertibleM' that works on 'ExceptT'.++>>> runExceptT $ L.foldM (evertM (evertibleM (foldE . (\_ -> return (Left ()))))) [1..10]+Left ()++-} +foldE :: (MonadTrans t, Monad m, Monad (t (ExceptT e m))) + => t (ExceptT e m) (Either e r) -- ^+ -> t (ExceptT e m) r+foldE action = action >>= lift . ExceptT . return+
+ src/Streaming/Eversion/Pipes.hs view
@@ -0,0 +1,123 @@+{-# LANGUAGE RankNTypes #-}++-- | Like "Streaming.Eversion", but for Producer folds and transformations.+-- ++module Streaming.Eversion.Pipes (+ -- * Evertible Producer folds+ pipeEvertible+ , evert+ , pipeEvertibleM+ , evertM+ , pipeEvertibleMIO+ , evertMIO+ -- * Transvertible Producer transformations+ , pipeTransvertible+ , transvert+ , pipeTransvertibleM+ , transvertM+ , pipeTransvertibleMIO+ , transvertMIO+ -- * Auxiliary functions+ , pipeLeftoversE+ , pipeTransE+ ) where++import Control.Monad.IO.Class+import Control.Monad.Trans.Class+import Control.Monad.Trans.Except++import Streaming(Of(..))+import qualified Streaming.Prelude+import Streaming.Eversion+import Pipes+import Pipes.Prelude++{- $setup+>>> :set -XOverloadedStrings+>>> import Data.Functor.Identity+>>> import Control.Monad.Trans.Except+>>> import Control.Monad.Trans.Identity+>>> import Control.Foldl (Fold(..),FoldM(..))+>>> import qualified Control.Foldl as L+>>> import Streaming (Stream,Of(..))+>>> import Streaming.Prelude (yield,next)+>>> import qualified Streaming.Prelude as S+>>> import Pipes+>>> import qualified Pipes.Prelude as P+>>> import qualified Pipes.Text as T+>>> import qualified Pipes.Text.Encoding as TE+-}++-----------------------------------------------------------------------------------------+++pipeEvertible :: (forall m r. Monad m => Producer a m r -> m (x,r)) -- ^+ -> Evertible a x+pipeEvertible f = evertible (\stream -> fmap (\(x,r) -> x :> r) (f (Pipes.Prelude.unfoldr Streaming.Prelude.next stream)))++pipeEvertibleM :: (forall t r. (MonadTrans t, Monad (t m)) => Producer a (t m) r -> t m (x,r)) -- ^+ -> EvertibleM m a x+pipeEvertibleM f = evertibleM (\stream -> fmap (\(x,r) -> x :> r) (f (Pipes.Prelude.unfoldr Streaming.Prelude.next stream)))++pipeEvertibleMIO :: (forall t r. (MonadTrans t, MonadIO (t m)) => Producer a (t m) r -> t m (x,r)) -- ^+ -> EvertibleMIO m a x+pipeEvertibleMIO f = evertibleMIO (\stream -> fmap (\(x,r) -> x :> r) (f (Pipes.Prelude.unfoldr Streaming.Prelude.next stream)))++pipeTransvertible :: (forall m r. Monad m => Producer a m r -> Producer b m r) -- ^+ -> Transvertible a b+pipeTransvertible pt = transvertible (\stream -> Streaming.Prelude.unfoldr Pipes.next (pt (Pipes.Prelude.unfoldr Streaming.Prelude.next stream)))++pipeTransvertibleM :: (forall t r. (MonadTrans t, Monad (t m)) => Producer a (t m) r -> Producer b (t m) r) -- ^+ -> TransvertibleM m a b+pipeTransvertibleM pt = transvertibleM (\stream -> Streaming.Prelude.unfoldr Pipes.next (pt (Pipes.Prelude.unfoldr Streaming.Prelude.next stream)))++-- -- | Ignore the somewhat baroque type and just remember that you can plug any of the "non-lens decoding functions" from "Pipes.Text.Encoding" here.+-- --+-- -- The result is a 'TransvertibleM' that works in 'ExceptT'. If any undecodable bytes are found, the computation halts with the undecodable bytes as the error.+-- pipeDecoderTransvertibleE :: Monad m => (forall t r .(MonadTrans t, Monad (t (ExceptT bytes m))) => (Producer bytes (t (ExceptT bytes m)) r -> Producer text (t (ExceptT bytes m)) (Producer bytes (t (ExceptT bytes m)) r))) -- ^+-- -> TransvertibleM (ExceptT bytes m) bytes text+-- pipeDecoderTransvertibleE decoder = pipeTransvertibleM (pipeLeftoversE . decoder)++pipeTransvertibleMIO :: (forall t r. (MonadTrans t, MonadIO (t m)) => Producer a (t m) r -> Producer b (t m) r) -- ^+ -> TransvertibleMIO m a b+pipeTransvertibleMIO pt = transvertibleMIO (\stream -> Streaming.Prelude.unfoldr Pipes.next (pt (Pipes.Prelude.unfoldr Streaming.Prelude.next stream)))++{-| Allows you to plug any of the "non-lens decoding functions" from "Pipes.Text.Encoding" into 'pipeTransvertibleM'. Just + compose the decoder with this function before passing it to 'pipeTransvertibleM'.++ The result will be a 'TransvertibleM' that works in 'ExceptT'. ++>>> :{ + let adapted = transvertM (pipeTransvertibleM (pipeLeftoversE . TE.decodeUtf8)) (L.generalize L.mconcat) + in runExceptT $ L.foldM adapted ["decode","this"]+ :}+Right "decodethis"++ If any undecodable bytes are found, the computation halts with the undecoded bytes as the error.++>>> :{ + let adapted = transvertM (pipeTransvertibleM (pipeLeftoversE . TE.decodeUtf8)) (L.generalize L.mconcat) + in runExceptT $ L.foldM adapted ["invalid \xc3\x28","sequence"]+ :}+Left "\195("++-}+pipeLeftoversE :: (MonadTrans t, Monad m, Monad (t (ExceptT bytes m))) => Producer text (t (ExceptT bytes m)) (Producer bytes (t (ExceptT bytes m)) r) -- ^+ -> Producer text (t (ExceptT bytes m)) r+pipeLeftoversE decodedProducer = decodedProducer >>= \leftoversProducer -> do+ leftovers <- lift (next leftoversProducer)+ case leftovers of + Left r -> return r+ Right (firstleftover,_) -> lift (lift (throwE firstleftover))++{-| If your producer-transforming computation can fail early returning a 'Left',+ compose it with this function before passing it to 'transvertibleM'. ++ The result will be an 'TransvertibleM' that works on 'ExceptT'.+-}+pipeTransE :: (MonadTrans t, Monad m, Monad (t (ExceptT e m))) + => Producer a (t (ExceptT e m)) (Either e r) -- ^+ -> Producer a (t (ExceptT e m)) r+pipeTransE producer = producer >>= lift . lift . ExceptT . return+
+ streaming-eversion.cabal view
@@ -0,0 +1,67 @@+Name: streaming-eversion+Version: 0.1.0.0+Cabal-Version: >=1.8.0.2+Build-Type: Simple+License: BSD3+License-File: LICENSE+Copyright: 2016 Daniel Diaz+Author: Daniel Diaz+Maintainer: diaz_carrete@yahoo.com+Bug-Reports: https://github.com/danidiaz/streaming-eversion/issues+Synopsis: Translate pull-based stream folds into push-based iteratees.+Description: Translate pull-based folds from the "streaming" package into+ push-based folds from the "foldl" package. +Category: Control++Extra-Source-Files:+ README.md+ CHANGELOG++Source-Repository head+ Type: git+ Location: git@github.com:danidiaz/streaming-eversion.git++Library+ HS-Source-Dirs: src+ Build-Depends:+ base >= 4 && < 5 ,+ transformers >= 0.4.0.0 ,+ comonad >= 4 ,+ bifunctors >= 4 ,+ profunctors >= 5 ,+ free >= 4 , + foldl >= 1.1.5 ,+ pipes >= 4.2.0 ,+ streaming >= 0.1.4.2 + Exposed-Modules:+ Streaming.Eversion+ Streaming.Eversion.Pipes+ GHC-Options: -O2 -Wall++test-suite doctests+ type: exitcode-stdio-1.0+ ghc-options: -Wall -threaded+ hs-source-dirs: tests+ main-is: doctests.hs+ build-depends:+ base >= 4.4 && < 5 ,+ doctest >= 0.10.1 ,+ foldl >= 1.1.5 ,+ pipes >= 4.1.9 ,+ pipes-text >= 0.0.2.2 ,+ streaming >= 0.1.4.2 ++test-suite tests+ type: exitcode-stdio-1.0+ ghc-options: -Wall -threaded+ hs-source-dirs: tests+ main-is: tests.hs+ build-depends:+ base >= 4.4 && < 5 ,+ tasty >= 0.10.1.1 ,+ tasty-hunit >= 0.9.2 ,+ tasty-quickcheck >= 0.8.3.2 , + streaming ,+ foldl ,+ streaming-eversion+
+ tests/doctests.hs view
@@ -0,0 +1,10 @@+module Main where++import Test.DocTest++main :: IO ()+main = doctest + [+ "src/Streaming/Eversion.hs",+ "src/Streaming/Eversion/Pipes.hs"+ ]
+ tests/tests.hs view
@@ -0,0 +1,156 @@+module Main where++import Data.Functor.Identity+import Data.IORef+import Test.Tasty+import Test.Tasty.HUnit+-- import Test.Tasty.QuickCheck++import qualified Control.Foldl as Foldl+import Streaming+import qualified Streaming.Prelude as S+import Streaming.Eversion++main :: IO ()+main = defaultMain tests++tests :: TestTree+tests = testGroup "tests" + [ + testGroup "evert"+ [+ testCaseEq+ "empty"+ ([]::[Integer])+ (Foldl.fold (evert (evertible S.toList)) [])+ , testCaseEq+ "toList"+ [1..10::Integer]+ (Foldl.fold (evert (evertible S.toList)) [1..10])+ ]+ , testGroup "evertM"+ [+ testCaseEq+ "empty"+ ([]::[Integer])+ (runIdentity (Foldl.foldM (evertM (evertibleM S.toList)) []))+ , testCaseEq+ "toList"+ [1..10::Integer]+ (runIdentity (Foldl.foldM (evertM (evertibleM S.toList)) [1..10]))+ , testCaseEqIO+ "ref"+ (True,[1..10::Integer])+ (do ref <- newIORef False + res <- Foldl.foldM (evertM (evertibleM (\s -> S.toList s <* lift (writeIORef ref True)))) [1..10]+ refval <- readIORef ref+ return (refval,res))+ ]+ , testGroup "evertMIO"+ [+ testCaseEqIO+ "empty"+ ([]::[Integer])+ (Foldl.foldM (evertMIO (evertibleMIO S.toList)) [])+ , testCaseEqIO+ "toList"+ [1..10::Integer]+ (Foldl.foldM (evertMIO (evertibleMIO S.toList)) [1..10])+ , testCaseEqIO+ "ref"+ (True,[1..10::Integer])+ (do ref <- newIORef False + res <- Foldl.foldM (evertMIO (evertibleMIO (\s -> S.toList s <* liftIO (writeIORef ref True)))) [1..10]+ refval <- readIORef ref+ return (refval,res))+ ]+ , testGroup "transduce"+ [+ testCaseEq+ "empty"+ ([]::[Integer])+ (Foldl.fold (transvert (transvertible id) Foldl.list) [])+ , testCaseEq+ "notempty"+ ([1..5]::[Integer])+ (Foldl.fold (transvert (transvertible id) Foldl.list) [1..5])+ , testCaseEq+ "surroundempty"+ ([1,2,3,4]::[Integer])+ (Foldl.fold (transvert (transvertible (\s -> S.yield 1 *> S.yield 2 *> s <* S.yield 3 <* S.yield 4)) Foldl.list) [])+ , testCaseEq+ "surround"+ ([1,2,3,4,5,6]::[Integer])+ (Foldl.fold (transvert (transvertible (\s -> S.yield 1 *> S.yield 2 *> s <* S.yield 5 <* S.yield 6)) Foldl.list) [3,4])+ , testCaseEq+ "group"+ ([[1,1],[2,2,2],[3,3,3]]::[[Integer]])+ (Foldl.fold (transvert (transvertible (mapped S.toList . S.group)) Foldl.list) [1,1,2,2,2,3,3,3])+ ]+ , testGroup "transduceM"+ [+ testCaseEq + "empty"+ ([]::[Integer])+ (runIdentity (Foldl.foldM (transvertM (transvertibleM id) (Foldl.generalize Foldl.list)) []))+ , testCaseEq+ "notempty"+ ([1..5]::[Integer])+ (runIdentity (Foldl.foldM (transvertM (transvertibleM id) (Foldl.generalize Foldl.list)) [1..5]))+ , testCaseEq+ "surroundempty"+ ([1,2,3,4]::[Integer])+ (runIdentity (Foldl.foldM (transvertM (transvertibleM (\s -> S.yield 1 *> S.yield 2 *> s <* S.yield 3 <* S.yield 4)) (Foldl.generalize Foldl.list)) []))+ , testCaseEq+ "surround"+ ([1,2,3,4,5,6]::[Integer])+ (runIdentity (Foldl.foldM (transvertM (transvertibleM (\s -> S.yield 1 *> S.yield 2 *> s <* S.yield 5 <* S.yield 6)) (Foldl.generalize Foldl.list)) [3,4]))+ , testCaseEq+ "group"+ ([[1,1],[2,2,2],[3,3,3]]::[[Integer]])+ (runIdentity (Foldl.foldM (transvertM (transvertibleM (mapped S.toList . S.group)) (Foldl.generalize Foldl.list)) [1,1,2,2,2,3,3,3]))+ , testCaseEqIO+ "ref"+ (True,[1,2,3,4,5,6]::[Integer])+ (do ref <- newIORef False + res <- Foldl.foldM (transvertM (transvertibleM (\s -> S.yield 1 *> S.yield 2 *> (lift (lift (writeIORef ref True))) *> s <* S.yield 5 <* S.yield 6)) (Foldl.generalize Foldl.list)) [3,4]+ refval <- readIORef ref+ return (refval,res))+ ]+ , testGroup "transduceMIO"+ [+ testCaseEqIO+ "empty"+ ([]::[Integer])+ (Foldl.foldM (transvertMIO (transvertibleMIO id) (Foldl.generalize Foldl.list)) [])+ , testCaseEqIO+ "notempty"+ ([1..5]::[Integer])+ (Foldl.foldM (transvertMIO (transvertibleMIO id) (Foldl.generalize Foldl.list)) [1..5])+ , testCaseEqIO+ "surroundempty"+ ([1,2,3,4]::[Integer])+ (Foldl.foldM (transvertMIO (transvertibleMIO (\s -> S.yield 1 *> S.yield 2 *> s <* S.yield 3 <* S.yield 4)) (Foldl.generalize Foldl.list)) [])+ , testCaseEqIO+ "surround"+ ([1,2,3,4,5,6]::[Integer])+ (Foldl.foldM (transvertMIO (transvertibleMIO (\s -> S.yield 1 *> S.yield 2 *> s <* S.yield 5 <* S.yield 6)) (Foldl.generalize Foldl.list)) [3,4])+ , testCaseEqIO+ "group"+ ([[1,1],[2,2,2],[3,3,3]]::[[Integer]])+ (Foldl.foldM (transvertMIO (transvertibleMIO (mapped S.toList . S.group)) (Foldl.generalize Foldl.list)) [1,1,2,2,2,3,3,3])+ , testCaseEqIO+ "ref"+ (True,[1,2,3,4,5,6]::[Integer])+ (do ref <- newIORef False + res <- Foldl.foldM (transvertMIO (transvertibleMIO (\s -> S.yield 1 *> S.yield 2 *> (liftIO (writeIORef ref True)) *> s <* S.yield 5 <* S.yield 6)) (Foldl.generalize Foldl.list)) [3,4]+ refval <- readIORef ref+ return (refval,res))+ ]+ ]+ where+ testCaseEq :: (Eq a, Show a) => TestName -> a -> a -> TestTree+ testCaseEq name a1 a2 = testCase name (assertEqual "" a1 a2)+ testCaseEqIO :: (Eq a, Show a) => TestName -> a -> IO a -> TestTree+ testCaseEqIO name a1 action = testCase name (action >>= assertEqual "" a1)+