diff --git a/CHANGELOG b/CHANGELOG new file mode 100644 --- /dev/null +++ b/CHANGELOG diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. + diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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. + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/src/Streaming/Eversion.hs b/src/Streaming/Eversion.hs new file mode 100644 --- /dev/null +++ b/src/Streaming/Eversion.hs @@ -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 + diff --git a/src/Streaming/Eversion/Pipes.hs b/src/Streaming/Eversion/Pipes.hs new file mode 100644 --- /dev/null +++ b/src/Streaming/Eversion/Pipes.hs @@ -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 + diff --git a/streaming-eversion.cabal b/streaming-eversion.cabal new file mode 100644 --- /dev/null +++ b/streaming-eversion.cabal @@ -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 + diff --git a/tests/doctests.hs b/tests/doctests.hs new file mode 100644 --- /dev/null +++ b/tests/doctests.hs @@ -0,0 +1,10 @@ +module Main where + +import Test.DocTest + +main :: IO () +main = doctest + [ + "src/Streaming/Eversion.hs", + "src/Streaming/Eversion/Pipes.hs" + ] diff --git a/tests/tests.hs b/tests/tests.hs new file mode 100644 --- /dev/null +++ b/tests/tests.hs @@ -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) +