streaming 0.1.4.2 → 0.1.4.3
raw patch · 9 files changed
+4259/−4202 lines, 9 filessetup-changed
Files
- README.md +60/−43
- Setup.hs +1/−0
- Streaming.hs +0/−176
- Streaming/Internal.hs +0/−1145
- Streaming/Prelude.hs +0/−2787
- src/Streaming.hs +176/−0
- src/Streaming/Internal.hs +1158/−0
- src/Streaming/Prelude.hs +2805/−0
- streaming.cabal +59/−51
README.md view
@@ -256,7 +256,7 @@ Rather than wrapping each step in a monadic 'layer', such a layer is put alongside separate 'pure' constructors for a functor 'layer' and a final return value. The maneuver is very friendly to the compiler, but requires a bit of subtlety to protect a sound monad instance. Just such an optimization is adopted internally by the `pipes` library. As in `pipes`, the constructors are here left in an `Internal` module; the main `Streaming` module exporting the type itself and various operations and instances. -I ran a simple [benchmark](https://gist.github.com/michaelt/7f89dc8b366b30bb6acc) (adjusting a [script](https://github.com/jwiegley/streaming-tests) of John Weigly) using a very simple composition of functions:+I ran a simple [benchmark](https://gist.github.com/michaelt/ee3710c5bab9b7d0892bd552e0eedfd9) (adjusting a [script](https://github.com/jwiegley/streaming-tests) of John Weigly) using a very simple composition of functions: toList . filter (\x -> x `mod` 2 == 0) @@ -266,59 +266,76 @@ . filter even . each -as it interpreted by various libraries - `streaming`, `conduit`, (Weigley's) `simple-conduit`, `io-streams` and `machines`.+as it interpreted by various libraries - `streaming`, `conduit`, `io-streams` and `machines`. -The the results were fairly pleasing:+The results were fairly pleasing: - benchmarking basic/stream- time 85.45 ms (81.63 ms .. 89.32 ms)- 0.994 R² (0.982 R² .. 0.999 R²)- mean 86.53 ms (84.16 ms .. 90.51 ms)- std dev 4.987 ms (2.301 ms .. 7.906 ms)- variance introduced by outliers: 18% (moderately inflated)+ benchmarking sum/streaming+ time 8.996 ms (8.910 ms .. 9.068 ms)+ 0.999 R² (0.998 R² .. 1.000 R²)+ mean 9.060 ms (9.004 ms .. 9.122 ms)+ std dev 164.6 μs (123.9 μs .. 251.9 μs) - benchmarking basic/conduit- time 101.3 ms (88.77 ms .. 111.3 ms)- 0.976 R² (0.911 R² .. 0.996 R²)- mean 95.56 ms (84.90 ms .. 103.6 ms)- std dev 13.76 ms (8.210 ms .. 21.79 ms)- variance introduced by outliers: 43% (moderately inflated)+ benchmarking sum/conduit+ time 15.77 ms (15.66 ms .. 15.89 ms)+ 0.999 R² (0.998 R² .. 1.000 R²)+ mean 15.78 ms (15.70 ms .. 15.89 ms)+ std dev 245.3 μs (176.5 μs .. 379.7 μs) - benchmarking basic/simple-conduit- time 199.2 ms (174.1 ms .. 215.6 ms)- 0.993 R² (0.978 R² .. 1.000 R²)- mean 198.4 ms (190.0 ms .. 202.2 ms)- std dev 7.091 ms (1.565 ms .. 10.000 ms)- variance introduced by outliers: 14% (moderately inflated)+ benchmarking sum/pipes+ time 57.94 ms (57.68 ms .. 58.27 ms)+ 1.000 R² (1.000 R² .. 1.000 R²)+ mean 58.10 ms (57.92 ms .. 58.27 ms)+ std dev 324.2 μs (214.1 μs .. 468.8 μs) - benchmarking basic/pipes- time 211.7 ms (180.8 ms .. 232.7 ms)- 0.991 R² (0.974 R² .. 1.000 R²)- mean 207.7 ms (199.1 ms .. 218.7 ms)- std dev 12.34 ms (5.989 ms .. 17.67 ms)- variance introduced by outliers: 15% (moderately inflated)+ benchmarking sum/iostreams+ time 61.96 ms (61.36 ms .. 62.53 ms)+ 1.000 R² (0.999 R² .. 1.000 R²)+ mean 61.80 ms (61.54 ms .. 62.08 ms)+ std dev 543.7 μs (375.1 μs .. 715.7 μs) - benchmarking basic/data-list- time 202.7 ms (186.5 ms .. 225.5 ms)- 0.990 R² (0.970 R² .. 1.000 R²)- mean 199.3 ms (188.4 ms .. 207.4 ms)- std dev 11.67 ms (6.966 ms .. 15.11 ms)- variance introduced by outliers: 15% (moderately inflated)+ benchmarking sum/machine+ time 260.4 ms (257.2 ms .. 263.6 ms)+ 1.000 R² (0.999 R² .. 1.000 R²)+ mean 259.7 ms (258.4 ms .. 260.6 ms)+ std dev 1.284 ms (565.9 μs .. 1.690 ms)+ variance introduced by outliers: 16% (moderately inflated) + benchmarking basic/streaming+ time 74.86 ms (70.07 ms .. 78.78 ms)+ 0.994 R² (0.987 R² .. 0.999 R²)+ mean 78.25 ms (75.55 ms .. 84.10 ms)+ std dev 6.301 ms (1.995 ms .. 10.17 ms)+ variance introduced by outliers: 19% (moderately inflated)++ benchmarking basic/conduit+ time 90.06 ms (66.61 ms .. 114.4 ms)+ 0.876 R² (0.658 R² .. 0.977 R²)+ mean 98.63 ms (85.28 ms .. 116.5 ms)+ std dev 23.06 ms (10.61 ms .. 30.72 ms)+ variance introduced by outliers: 65% (severely inflated)++ benchmarking basic/pipes+ time 180.9 ms (158.7 ms .. 201.3 ms)+ 0.989 R² (0.971 R² .. 1.000 R²)+ mean 190.5 ms (183.0 ms .. 197.8 ms)+ std dev 10.16 ms (4.910 ms .. 14.86 ms)+ variance introduced by outliers: 14% (moderately inflated)+ benchmarking basic/iostreams- time 265.7 ms (247.2 ms .. 284.8 ms)- 0.997 R² (0.990 R² .. 1.000 R²)- mean 265.6 ms (261.9 ms .. 272.8 ms)- std dev 7.094 ms (146.8 μs .. 8.387 ms)+ time 269.7 ms (243.8 ms .. 303.9 ms)+ 0.995 R² (0.985 R² .. 1.000 R²)+ mean 264.2 ms (254.0 ms .. 272.0 ms)+ std dev 10.87 ms (5.762 ms .. 15.06 ms) variance introduced by outliers: 16% (moderately inflated) - benchmarking basic/machines- time 1.123 s (NaN s .. 1.206 s)- 0.999 R² (0.999 R² .. 1.000 R²)- mean 1.134 s (1.114 s .. 1.145 s)- std dev 17.29 ms (0.0 s .. 19.07 ms)+ benchmarking basic/machine+ time 397.7 ms (324.4 ms .. 504.8 ms)+ 0.992 R² (0.977 R² .. 1.000 R²)+ mean 407.7 ms (391.1 ms .. 420.3 ms)+ std dev 19.40 ms (0.0 s .. 21.88 ms) variance introduced by outliers: 19% (moderately inflated) -+ This sequence of pre-packaged combinators is, I think, as friendly as it could possibly be to the more recent conduit fusion framework. That framework of course doesn't apply to user-defined operations; there we should expect times like those shown for pipes. Since the combinators from `streaming` are defined with naive recursion, more or less as the user might, we have reason to think this result is characteristic, but much more benchmarking is needed before anything can be said with certainty.
Setup.hs view
@@ -1,2 +1,3 @@+ import Distribution.Simple main = defaultMain
− Streaming.hs
@@ -1,176 +0,0 @@-{-#LANGUAGE RankNTypes, CPP, Trustworthy #-}-module Streaming - (- -- * An iterable streaming monad transformer- -- $stream- Stream, - -- * Constructing a 'Stream' on a given functor- yields,- effect,- wrap,- replicates,- repeats,- repeatsM,- unfold,- never,- untilJust,- streamBuild,- delays,- - -- * Transforming streams- maps,- mapsM,- mapped, - distribute,- groups,- - -- * Inspecting a stream- inspect,- - -- * Splitting and joining 'Stream's - splitsAt,- takes,- chunksOf,- concats,- intercalates,- cutoff, - -- period,- -- periods,- - - -- * Zipping, unzipping, separating and unseparating streams- zipsWith,- zips,- unzips,- interleaves,- separate,- unseparate,- decompose,--- -- * Eliminating a 'Stream'- mapsM_,- run,- streamFold, - iterTM,- iterT,- destroy,-- -- * Base functor for streams of individual items- Of (..),- lazily,- strictly,- - -- * ResourceT help- - bracketStream,- - -- * re-exports- MFunctor(..),- MMonad(..),- MonadTrans(..),- MonadIO(..),- Compose(..),- Sum(..),- Identity(..),- Alternative((<|>)),- MonadThrow(..),- MonadResource(..),- MonadBase(..),- ResourceT(..),- runResourceT,-#if MIN_VERSION_base(4,8,0)- Bifunctor(..),-#endif- - join,- liftM,- liftM2,- liftA2,- liftA3,- void,- (<>)- )- where-import Streaming.Internal -import Streaming.Prelude -import Control.Monad.Morph-import Control.Monad-import Data.Monoid ((<>))-import Control.Applicative-import Control.Monad.Trans-import Data.Functor.Compose -import Data.Functor.Sum-import Data.Functor.Identity--import Control.Monad.Base-import Control.Monad.Trans.Resource-#if MIN_VERSION_base(4,8,0)-import Data.Bifunctor-#endif--{- $stream-- The 'Stream' data type can be used to represent any effectful- succession of steps arising in some monad. - The form of the steps is specified by the first (\"functor\") - parameter in @Stream f m r@. The monad of the underlying effects - is expressed by the second parameter. -- This module exports combinators that pertain to that general case.- Some of these are quite abstract and pervade any use of the library, - e.g. --> maps :: (forall x . f x -> g x) -> Stream f m r -> Stream g m r -> mapped :: (forall x . f x -> m (g x)) -> Stream f m r -> Stream g m r -> hoist :: (forall x . m x -> n x) -> Stream f m r -> Stream f n r -- from the MFunctor instance-> concats :: Stream (Stream f m) m r -> Stream f m r -- (assuming here and thoughout that @m@ or @n@ satisfies a @Monad@ constraint, and- @f@ or @g@ a @Functor@ constraint.)-- Others are surprisingly determinate in content:--> chunksOf :: Int -> Stream f m r -> Stream (Stream f m) m r-> splitsAt :: Int -> Stream f m r -> Stream f m (Stream f m r)-> zipsWith :: (forall x y. f x -> g y -> h (x, y)) -> Stream f m r -> Stream g m r -> Stream h m r-> intercalates :: Stream f m () -> Stream (Stream f m) m r -> Stream f m r-> unzips :: Stream (Compose f g) m r -> Stream f (Stream g m) r -> separate :: Stream (Sum f g) m r -> Stream f (Stream g) m r -- cp. partitionEithers-> unseparate :: Stream f (Stream g) m r -> Stream (Sum f g) m r-> groups :: Stream (Sum f g) m r -> Stream (Sum (Stream f m) (Stream g m)) m r-- One way to see that /any/ streaming library needs some such general type is- that it is required to represent the segmentation of a stream, and to- express the equivalents of @Prelude/Data.List@ combinators that involve- 'lists of lists' and the like. See for example this - <http://www.haskellforall.com/2013/09/perfect-streaming-using-pipes-bytestring.html post> - on the correct expression of a streaming \'lines\' function. -- The module @Streaming.Prelude@ exports combinators relating to --> Stream (Of a) m r-- where @Of a r = !a :> r@ is a left-strict pair.--- This expresses the concept of a 'Producer' or 'Source' or 'Generator' and- easily inter-operates with types with such names in e.g. 'conduit', - 'iostreams' and 'pipes'.--}--{-| Map a stream to its church encoding; compare @Data.List.foldr@- This is the @safe_destroy@ exported by the @Internal@ module.-- Typical @FreeT@ operators can be defined in terms of @destroy@- e.g.--> iterT :: (Functor f, Monad m) => (f (m a) -> m a) -> Stream f m a -> m a-> iterT out stream = destroy stream out join return-> iterTM :: (Functor f, Monad m, MonadTrans t, Monad (t m)) => (f (t m a) -> t m a) -> Stream f m a -> t m a-> iterTM out stream = destroy stream out (join . lift) return-> concats :: (Monad m, MonadTrans t, Monad (t m)) => Stream (t m) m a -> t m a-> concats stream = destroy stream join (join . lift) return--}--
− Streaming/Internal.hs
@@ -1,1145 +0,0 @@-{-# LANGUAGE RankNTypes, StandaloneDeriving,DeriveDataTypeable, BangPatterns #-}-{-# LANGUAGE UndecidableInstances, CPP, FlexibleInstances, MultiParamTypeClasses #-} -{-#LANGUAGE Trustworthy #-}-module Streaming.Internal (- -- * The free monad transformer- -- $stream- Stream (..)- - -- * Introducing a stream- , unfold - , replicates- , repeats- , repeatsM- , effect- , wrap- , yields- , streamBuild - , cycles- , delays- , never- , untilJust- - -- * Eliminating a stream- , intercalates - , concats - , iterT - , iterTM - , destroy - , streamFold- - -- * Inspecting a stream wrap by wrap- , inspect - - -- * Transforming streams- , maps - , mapsM - , decompose- , mapsM_- , run- , distribute- , groups--- , groupInL- - -- * Splitting streams- , chunksOf - , splitsAt- , takes- , cutoff- -- , period- -- , periods- - -- * Zipping and unzipping streams- , zipsWith- , zips- , unzips- , interleaves- , separate- , unseparate-- - -- * Assorted Data.Functor.x help- - , switch- - -- * ResourceT help- - , bracketStream- - -- * For use in implementation- , unexposed- , hoistExposed- , mapsExposed- , mapsMExposed- , destroyExposed- - ) where--import Control.Monad-import Control.Monad.Trans-import Control.Monad.Trans.Class-import Control.Monad.Reader.Class-import Control.Monad.Writer.Class-import Control.Monad.State.Class-import Control.Monad.Error.Class-import Control.Monad.Cont.Class-import Control.Applicative-import Data.Foldable ( Foldable(..) )-import Data.Traversable-import Control.Monad.Morph-import Data.Monoid (Monoid (..), (<>))-import Data.Functor.Identity-import Data.Data ( Data, Typeable )-import Prelude hiding (splitAt)-import Data.Functor.Compose-import Data.Functor.Sum-import Control.Concurrent (threadDelay)-import Control.Monad.Base-import Control.Monad.Trans.Resource-import Control.Monad.Catch (MonadCatch (..))-import Control.Monad.Trans.Control---{- $stream-- The 'Stream' data type is equivalent to @FreeT@ and can represent any effectful- succession of steps, where the form of the steps or 'commands' is - specified by the first (functor) parameter. --> data Stream f m r = Step !(f (Stream f m r)) | Effect (m (Stream f m r)) | Return r-- The /producer/ concept uses the simple functor @ (a,_) @ \- or the stricter - @ Of a _ @. Then the news at each step or layer is just: an individual item of type @a@. - Since @Stream (Of a) m r@ is equivalent to @Pipe.Producer a m r@, much of- the @pipes@ @Prelude@ can easily be mirrored in a @streaming@ @Prelude@. Similarly, - a simple @Consumer a m r@ or @Parser a m r@ concept arises when the base functor is- @ (a -> _) @ . @Stream ((->) input) m result@ consumes @input@ until it returns a - @result@.-- To avoid breaking reasoning principles, the constructors - should not be used directly. A pattern-match should go by way of 'inspect' \- \- or, in the producer case, 'Streaming.Prelude.next'- The constructors are exported by the 'Internal' module.--}-data Stream f m r = Step !(f (Stream f m r))- | Effect (m (Stream f m r))- | Return r-#if __GLASGOW_HASKELL__ >= 710- deriving (Typeable)-#endif-deriving instance (Show r, Show (m (Stream f m r))- , Show (f (Stream f m r))) => Show (Stream f m r)-deriving instance (Eq r, Eq (m (Stream f m r))- , Eq (f (Stream f m r))) => Eq (Stream f m r)-#if __GLASGOW_HASKELL__ >= 710-deriving instance (Typeable f, Typeable m, Data r, Data (m (Stream f m r))- , Data (f (Stream f m r))) => Data (Stream f m r)-#endif-instance (Functor f, Monad m) => Functor (Stream f m) where- fmap f = loop where- loop stream = case stream of- Return r -> Return (f r)- Effect m -> Effect (do {stream' <- m; return (loop stream')})- Step f -> Step (fmap loop f)- {-# INLINABLE fmap #-}- a <$ stream0 = loop stream0 where- loop stream = case stream of- Return r -> Return a- Effect m -> Effect (do {stream' <- m; return (loop stream')})- Step f -> Step (fmap loop f)- {-# INLINABLE (<$) #-} - -instance (Functor f, Monad m) => Monad (Stream f m) where- return = Return- {-# INLINE return #-}- stream1 >> stream2 = loop stream1 where- loop stream = case stream of- Return _ -> stream2- Effect m -> Effect (liftM loop m)- Step f -> Step (fmap loop f) - {-# INLINABLE (>>) #-}- (>>=) = _bind- {-#INLINE (>>=) #-}- - -- stream >>= f = - -- loop stream where- -- loop stream0 = case stream0 of- -- Step fstr -> Step (fmap loop fstr)- -- Effect m -> Effect (liftM loop m)- -- Return r -> f r- -- {-# INLINABLE (>>=) #-} -- fail = lift . fail- {-#INLINE fail #-}- --_bind- :: (Functor f, Monad m)- => Stream f m r- -> (r -> Stream f m s)- -> Stream f m s-_bind p0 f = go p0 where- go p = case p of- Step fstr -> Step (fmap go fstr)- Effect m -> Effect (m >>= \s -> return (go s))- Return r -> f r-{-#INLINABLE[0] _bind #-}- -{-# RULES- "_bind (Step fstr) f" forall fstr f .- _bind (Step fstr) f = Step (fmap (\p -> _bind p f) fstr);- "_bind (Effect m) f" forall m f .- _bind (Effect m) f = Effect (m >>= \p -> return (_bind p f));- "_bind (Return r) f" forall r f .- _bind (Return r) f = f r;- #-}- -instance (Functor f, Monad m) => Applicative (Stream f m) where- pure = Return- {-# INLINE pure #-}- streamf <*> streamx = do {f <- streamf; x <- streamx; return (f x)} - {-# INLINE (<*>) #-} --{- | The 'Alternative' instance glues streams together stepwise. --> empty = never-> (<|>) = zipsWith (liftA2 (,))-- See also 'never', 'untilJust' and 'delays'--}-instance (Applicative f, Monad m) => Alternative (Stream f m) where- empty = never- {-#INLINE empty #-}- - str <|> str' = zipsWith (liftA2 (,)) str str'- {-#INLINE (<|>) #-}--instance (Applicative f, Monad m) => MonadPlus (Stream f m) where- mzero = empty- mplus = (<|>)- -instance Functor f => MonadTrans (Stream f) where- lift = Effect . liftM Return- {-# INLINE lift #-}--instance Functor f => MFunctor (Stream f) where- hoist trans = loop where- loop stream = case stream of - Return r -> Return r- Effect m -> Effect (trans (liftM loop m))- Step f -> Step (fmap loop f)- {-# INLINABLE hoist #-} --instance Functor f => MMonad (Stream f) where- embed phi = loop where- loop stream = case stream of- Return r -> Return r- Effect m -> phi m >>= loop- Step f -> Step (fmap loop f)- {-# INLINABLE embed #-} --instance (MonadIO m, Functor f) => MonadIO (Stream f m) where- liftIO = Effect . liftM Return . liftIO- {-# INLINE liftIO #-}- -instance (MonadBase b m, Functor f) => MonadBase b (Stream f m) where- liftBase = effect . fmap return . liftBase- {-#INLINE liftBase #-}- -instance (MonadThrow m, Functor f) => MonadThrow (Stream f m) where- throwM = lift . throwM - {-#INLINE throwM #-}--instance (MonadCatch m, Functor f) => MonadCatch (Stream f m) where- catch str f = go str- where- go p = case p of- Step f -> Step (fmap go f)- Return r -> Return r- Effect m -> Effect (catch (do- p' <- m- return (go p')) - (\e -> return (f e)) )- {-#INLINABLE catch #-}- -instance (MonadResource m, Functor f) => MonadResource (Stream f m) where- liftResourceT = lift . liftResourceT- {-#INLINE liftResourceT #-}---instance (Functor f, MonadReader r m) => MonadReader r (Stream f m) where- ask = lift ask- {-# INLINE ask #-}- local f = hoist (local f)- {-# INLINE local #-}---- instance (Functor f, MonadWriter w m) => MonadWriter w (Stream f m) where--- tell = lift . tell--- {-# INLINE tell #-}--- -- listen (FreeT m) = FreeT $ liftM concat' $ listen (fmap listen `liftM` m)--- where--- concat' (Pure x, w) = Pure (x, w)--- concat' (Free y, w) = Free $ fmap (second (w <>)) <$> y--- pass m = FreeT . pass' . runFreeT . hoist clean $ listen m--- where--- clean = pass . liftM (\x -> (x, const mempty))--- pass' = join . liftM g--- g (Pure ((x, f), w)) = tell (f w) >> return (Pure x)--- g (Free f) = return . Free . fmap (FreeT . pass' . runFreeT) $ f--- #if MIN_VERSION_mtl(2,1,1)--- writer w = lift (writer w)--- {-# INLINE writer #-}--- #endif----instance (Functor f, MonadState s m) => MonadState s (Stream f m) where- get = lift get- {-# INLINE get #-}- put = lift . put- {-# INLINE put #-}-#if MIN_VERSION_mtl(2,1,1)- state f = lift (state f)- {-# INLINE state #-}-#endif--instance (Functor f, MonadError e m) => MonadError e (Stream f m) where- throwError = lift . throwError- {-# INLINE throwError #-}- str `catchError` f = loop str where- loop str = case str of- Return r -> Return r- Effect m -> Effect $ liftM loop m `catchError` (return . f)- Step f -> Step (fmap loop f)- {-# INLINABLE catchError #-}- -bracketStream :: (Functor f, MonadResource m) =>- IO a -> (a -> IO ()) -> (a -> Stream f m b) -> Stream f m b-bracketStream alloc free inside = do- (key, seed) <- lift (allocate alloc free)- clean key (inside seed)- where- clean key = loop where- loop str = case str of - Return r -> Effect (release key >> return (Return r))- Effect m -> Effect (liftM loop m)- Step f -> Step (fmap loop f)-{-#INLINABLE bracketStream #-}---{-| Map a stream directly to its church encoding; compare @Data.List.foldr@--}-destroy- :: (Functor f, Monad m) =>- Stream f m r -> (f b -> b) -> (m b -> b) -> (r -> b) -> b-destroy stream0 construct effect done = loop stream0 where- loop stream = case stream of- Return r -> done r- Effect m -> effect (liftM loop m)- Step fs -> construct (fmap loop fs)-{-# INLINABLE destroy #-}---{-| 'streamFold' reorders the arguments of 'destroy' to be more akin- to @foldr@ It is more convenient to query in ghci to figure out- what kind of \'algebra\' you need to write. -->>> :t streamFold return join -(Monad m, Functor f) => - (f (m a) -> m a) -> Stream f m a -> m a -- iterT-->>> :t streamFold return (join . lift)-(Monad m, Monad (t m), Functor f, MonadTrans t) =>- (f (t m a) -> t m a) -> Stream f m a -> t m a -- iterTM-->>> :t streamFold return effect -(Monad m, Functor f, Functor g) =>- (f (Stream g m r) -> Stream g m r) -> Stream f m r -> Stream g m r-->>> :t \f -> streamFold return effect (wrap . f)-(Monad m, Functor f, Functor g) =>- (f (Stream g m a) -> g (Stream g m a))- -> Stream f m a -> Stream g m a -- maps-->>> :t \f -> streamFold return effect (effect . liftM wrap . f)-(Monad m, Functor f, Functor g) =>- (f (Stream g m a) -> m (g (Stream g m a)))- -> Stream f m a -> Stream g m a -- mapped---}-streamFold- :: (Functor f, Monad m) =>- (r -> b) -> (m b -> b) -> (f b -> b) -> Stream f m r -> b-streamFold done effect construct stream = destroy stream construct effect done-{-#INLINE streamFold #-}--{- | Reflect a church-encoded stream; cp. @GHC.Exts.build@--> streamFold return_ effect_ step_ (streamBuild psi) = psi return_ effect_ step_--}-streamBuild- :: (forall b . (r -> b) -> (m b -> b) -> (f b -> b) -> b) -> Stream f m r-streamBuild = \phi -> phi Return Effect Step-{-# INLINE streamBuild #-}---{-| Inspect the first stage of a freely layered sequence. - Compare @Pipes.next@ and the replica @Streaming.Prelude.next@. - This is the 'uncons' for the general 'unfold'.--> unfold inspect = id-> Streaming.Prelude.unfoldr StreamingPrelude.next = id--}-inspect :: (Functor f, Monad m) =>- Stream f m r -> m (Either r (f (Stream f m r)))-inspect = loop where- loop stream = case stream of- Return r -> return (Left r)- Effect m -> m >>= loop- Step fs -> return (Right fs)-{-# INLINABLE inspect #-}- -{-| Build a @Stream@ by unfolding steps starting from a seed. See also- the specialized 'Streaming.Prelude.unfoldr' in the prelude.--> unfold inspect = id -- modulo the quotient we work with-> unfold Pipes.next :: Monad m => Producer a m r -> Stream ((,) a) m r-> unfold (curry (:>) . Pipes.next) :: Monad m => Producer a m r -> Stream (Of a) m r---}-unfold :: (Monad m, Functor f) - => (s -> m (Either r (f s))) - -> s -> Stream f m r-unfold step = loop where- loop s0 = Effect $ do - e <- step s0- case e of- Left r -> return (Return r)- Right fs -> return (Step (fmap loop fs))-{-# INLINABLE unfold #-}---{- | Map layers of one functor to another with a transformation. Compare- hoist, which has a similar effect on the 'monadic' parameter. --> maps id = id-> maps f . maps g = maps (f . g)---}-maps :: (Monad m, Functor f) - => (forall x . f x -> g x) -> Stream f m r -> Stream g m r-maps phi = loop where- loop stream = case stream of - Return r -> Return r- Effect m -> Effect (liftM loop m)- Step f -> Step (phi (fmap loop f))-{-# INLINABLE maps #-}---{- | Map layers of one functor to another with a transformation involving the base monad- @maps@ is more fundamental than @mapsM@, which is best understood as a convenience- for effecting this frequent composition:--> mapsM phi = decompose . maps (Compose . phi)-- The streaming prelude exports the same function under the better name @mapped@,- which overlaps with the lens libraries. - --}-mapsM :: (Monad m, Functor f) => (forall x . f x -> m (g x)) -> Stream f m r -> Stream g m r-mapsM phi = loop where- loop stream = case stream of - Return r -> Return r- Effect m -> Effect (liftM loop m)- Step f -> Effect (liftM Step (phi (fmap loop f)))-{-# INLINABLE mapsM #-}---{-| Rearrange a succession of layers of the form @Compose m (f x)@. -- we could as well define @decompose@ by @mapsM@:--> decompose = mapped getCompose-- but @mapped@ is best understood as:--> mapped phi = decompose . maps (Compose . phi)-- since @maps@ and @hoist@ are the really fundamental operations that preserve the- shape of the stream:--> maps :: (Monad m, Functor f) => (forall x. f x -> g x) -> Stream f m r -> Stream g m r-> hoist :: (Monad m, Functor f) => (forall a. m a -> n a) -> Stream f m r -> Stream f n r---}-decompose :: (Monad m, Functor f) => Stream (Compose m f) m r -> Stream f m r-decompose = loop where- loop stream = case stream of - Return r -> Return r - Effect m -> Effect (liftM loop m)- Step (Compose mstr) -> Effect $ do- str <- mstr- return (Step (fmap loop str))---{-| Run the effects in a stream that merely layers effects.--}-run :: Monad m => Stream m m r -> m r-run = loop where- loop stream = case stream of- Return r -> return r- Effect m -> m >>= loop- Step mrest -> mrest >>= loop-{-# INLINABLE run #-}---{-| Map each layer to an effect, and run them all.--}-mapsM_ :: (Functor f, Monad m) => (forall x . f x -> m x) -> Stream f m r -> m r-mapsM_ f = run . maps f -{-# INLINE mapsM_ #-}---{-| Interpolate a layer at each segment. This specializes to e.g.--> intercalates :: (Monad m, Functor f) => Stream f m () -> Stream (Stream f m) m r -> Stream f m r--}-intercalates :: (Monad m, Monad (t m), MonadTrans t) =>- t m x -> Stream (t m) m r -> t m r-intercalates sep = go0- where- go0 f = case f of - Return r -> return r - Effect m -> lift m >>= go0 - Step fstr -> do- f' <- fstr- go1 f'- go1 f = case f of - Return r -> return r - Effect m -> lift m >>= go1- Step fstr -> do- sep- f' <- fstr- go1 f'-{-# INLINABLE intercalates #-}--{-| Specialized fold following the usage of @Control.Monad.Trans.Free@--> iterTM alg = streamFold return (join . lift)--}-iterTM ::- (Functor f, Monad m, MonadTrans t,- Monad (t m)) =>- (f (t m a) -> t m a) -> Stream f m a -> t m a-iterTM out stream = destroyExposed stream out (join . lift) return-{-# INLINE iterTM #-}--{-| Specialized fold following the usage of @Control.Monad.Trans.Free@--> iterT alg = streamFold return join alg --}-iterT ::- (Functor f, Monad m) => (f (m a) -> m a) -> Stream f m a -> m a-iterT out stream = destroyExposed stream out join return-{-# INLINE iterT #-}--{-| Dissolves the segmentation into layers of @Stream f m@ layers.---}-concats :: (Monad m, Functor f) => Stream (Stream f m) m r -> Stream f m r-concats = loop where- loop stream = case stream of- Return r -> return r- Effect m -> join $ lift (liftM loop m)- Step fs -> join (fmap loop fs)-{-# INLINE concats #-}--{-| Split a succession of layers after some number, returning a streaming or- effectful pair.-->>> rest <- S.print $ S.splitAt 1 $ each [1..3]-1->>> S.print rest-2-3--> splitAt 0 = return-> splitAt n >=> splitAt m = splitAt (m+n)-- Thus, e.g. -->>> rest <- S.print $ splitsAt 2 >=> splitsAt 2 $ each [1..5]-1-2-3-4->>> S.print rest-5---}-splitsAt :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Stream f m r)-splitsAt = loop where- loop !n stream - | n <= 0 = Return stream- | otherwise = case stream of- Return r -> Return (Return r)- Effect m -> Effect (liftM (loop n) m)- Step fs -> case n of - 0 -> Return (Step fs)- _ -> Step (fmap (loop (n-1)) fs)-{-# INLINABLE splitsAt #-} --{- Functor-general take. -- @takes 3@ can take three individual values-->>> S.print $ takes 3 $ each [1..]-1-2-3--- or three sub-streams-->>> S.print $ mapped S.toList $ takes 3 $ chunksOf 2 $ each [1..]-[1,2]-[3,4]-[5,6]-- Or, using 'Data.ByteString.Streaming.Char' (here called @Q@),- three byte streams.-->>> Q.stdout $ Q.unlines $ takes 3 $ Q.lines $ Q.chunk "a\nb\nc\nd\ne\nf"-a-b-c---}-takes :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m ()-takes n = void . splitsAt n-{-# INLINE takes #-} --{-| Break a stream into substreams each with n functorial layers. -->>> S.print $ mapped S.sum $ chunksOf 2 $ each [1,1,1,1,1]-2-2-1--}-chunksOf :: (Monad m, Functor f) => Int -> Stream f m r -> Stream (Stream f m) m r-chunksOf n0 = loop where- loop stream = case stream of- Return r -> Return r- Effect m -> Effect (liftM loop m)- Step fs -> Step (Step (fmap (fmap loop . splitsAt (n0-1)) fs))-{-# INLINABLE chunksOf #-} --{- | Make it possible to \'run\' the underlying transformed monad. --}-distribute :: (Monad m, Functor f, MonadTrans t, MFunctor t, Monad (t (Stream f m)))- => Stream f (t m) r -> t (Stream f m) r-distribute = loop where- loop stream = case stream of - Return r -> lift (Return r)- Effect tmstr -> hoist lift tmstr >>= loop- Step fstr -> join (lift (Step (fmap (Return . loop) fstr)))-{-#INLINABLE distribute #-}- --- | Repeat a functorial layer (a \"command\" or \"instruction\") forever.-repeats :: (Monad m, Functor f) => f () -> Stream f m r -repeats f = loop where- loop = Effect (return (Step (fmap (\_ -> loop) f)))---- | Repeat an effect containing a functorial layer, command or instruction forever.-repeatsM :: (Monad m, Functor f) => m (f ()) -> Stream f m r -repeatsM mf = loop where- loop = Effect $ do- f <- mf- return $ Step $ fmap (\_ -> loop) f--{- | Repeat a functorial layer, command or instruction a fixed number of times.--> replicates n = takes n . repeats --}-replicates :: (Monad m, Functor f) => Int -> f () -> Stream f m ()-replicates n f = splitsAt n (repeats f) >> return ()--{-| Construct an infinite stream by cycling a finite one--> cycles = forever-->>> --}- -cycles :: (Monad m, Functor f) => Stream f m () -> Stream f m r-cycles = forever----hoistExposed trans = loop where- loop stream = case stream of - Return r -> Return r- Effect m -> Effect (trans (liftM loop m))- Step f -> Step (fmap loop f)--mapsExposed :: (Monad m, Functor f) - => (forall x . f x -> g x) -> Stream f m r -> Stream g m r-mapsExposed phi = loop where- loop stream = case stream of - Return r -> Return r- Effect m -> Effect (liftM loop m)- Step f -> Step (phi (fmap loop f))-{-# INLINABLE mapsExposed #-}--mapsMExposed phi = loop where- loop stream = case stream of - Return r -> Return r- Effect m -> Effect (liftM loop m)- Step f -> Effect (liftM Step (phi (fmap loop f)))-{-# INLINABLE mapsMExposed #-}---- Map a stream directly to its church encoding; compare @Data.List.foldr@--- It permits distinctions that should be hidden, as can be seen from--- e.g.------ isPure stream = destroy (const True) (const False) (const True)------ and similar nonsense. The crucial--- constraint is that the @m x -> x@ argument is an /Eilenberg-Moore algebra/.--- See Atkey "Reasoning about Stream Processing with Effects"---destroyExposed stream0 construct effect done = loop stream0 where- loop stream = case stream of- Return r -> done r- Effect m -> effect (liftM loop m)- Step fs -> construct (fmap loop fs)-{-# INLINABLE destroyExposed #-}---{-| This is akin to the @observe@ of @Pipes.Internal@ . It reeffects the layering- in instances of @Stream f m r@ so that it replicates that of - @FreeT@. ---}-unexposed :: (Functor f, Monad m) => Stream f m r -> Stream f m r-unexposed = Effect . loop where- loop stream = case stream of - Return r -> return (Return r)- Effect m -> m >>= loop- Step f -> return (Step (fmap (Effect . loop) f))-{-# INLINABLE unexposed #-} ---{- Wrap a new layer of a stream. So, e.g.--> S.cons :: Monad m => a -> Stream (Of a) m r -> Stream (Of a) m r-> S.cons a str = wrap (a :> str)-- and, recursively:--> S.each :: (Monad m, Foldable t) => t a -> Stream (Of a) m ()-> S.each = foldr (\a b -> wrap (a :> b)) (return ())-- The two operations --> wrap :: (Monad m, Functor f ) => f (Stream f m r) -> Stream f m r-> effect :: (Monad m, Functor f ) => m (Stream f m r) -> Stream f m r-- are fundamental. We can define the parallel operations @yields@ and @lift@ in - terms of them--> yields :: (Monad m, Functor f ) => f r -> Stream f m r-> yields = wrap . fmap return-> lift :: (Monad m, Functor f ) => m r -> Stream f m r-> lift = effect . fmap return---}-wrap :: (Monad m, Functor f ) => f (Stream f m r) -> Stream f m r-wrap = Step-{-#INLINE wrap #-}---{- | Wrap an effect that returns a stream--> effect = join . lift ---}-effect :: (Monad m, Functor f ) => m (Stream f m r) -> Stream f m r-effect = Effect-{-#INLINE effect #-}---{-| @yields@ is like @lift@ for items in the streamed functor. - It makes a singleton or one-layer succession. --> lift :: (Monad m, Functor f) => m r -> Stream f m r-> yields :: (Monad m, Functor f) => f r -> Stream f m r-- Viewed in another light, it is like a functor-general version of @yield@:--> S.yield a = yields (a :> ())---}--yields :: (Monad m, Functor f) => f r -> Stream f m r-yields fr = Step (fmap Return fr)-{-#INLINE yields #-}---zipsWith :: (Monad m, Functor h) - => (forall x y . f x -> g y -> h (x,y)) - -> Stream f m r -> Stream g m r -> Stream h m r-zipsWith phi = curry loop where- loop (s1, s2) = Effect (go s1 s2)- go (Return r) p = return (Return r)- go q (Return s) = return (Return s)- go (Effect m) p = m >>= (\s -> go s p)- go q (Effect m) = m >>= (\s -> go q s)- go (Step f) (Step g) = return (Step (fmap loop (phi f g)))-{-# INLINABLE zipsWith #-} - -zips :: (Monad m, Functor f, Functor g) - => Stream f m r -> Stream g m r -> Stream (Compose f g) m r -zips = zipsWith go where- go fx gy = Compose (fmap (\x -> fmap (\y -> (x,y)) gy) fx)-{-# INLINE zips #-} ----{-| Interleave functor layers, with the effects of the first preceding- the effects of the second.--> interleaves = zipsWith (liftA2 (,))-->>> let paste = \a b -> interleaves (Q.lines a) (maps (Q.cons' '\t') (Q.lines b))->>> Q.stdout $ Q.unlines $ paste "hello\nworld\n" "goodbye\nworld\n"-hello goodbye-world world---}- -interleaves- :: (Monad m, Applicative h) =>- Stream h m r -> Stream h m r -> Stream h m r-interleaves = zipsWith (liftA2 (,))-{-# INLINE interleaves #-} ---{-| Swap the order of functors in a sum of functors. --->>> S.toListM' $ S.print $ separate $ maps S.switch $ maps (S.distinguish (=='a')) $ S.each "banana"-'a'-'a'-'a'-"bnn" :> ()->>> S.toListM' $ S.print $ separate $ maps (S.distinguish (=='a')) $ S.each "banana"-'b'-'n'-'n'-"aaa" :> ()--}-switch :: Sum f g r -> Sum g f r-switch s = case s of InL a -> InR a; InR a -> InL a-{-#INLINE switch #-}--- -{-| Given a stream on a sum of functors, make it a stream on the left functor,- with the streaming on the other functor as the governing monad. This is- useful for acting on one or the other functor with a fold. It generalizes- 'Data.Either.partitionEithers' massively, but actually streams properly.-->>> let odd_even = S.maps (S.distinguish even) $ S.each [1..10::Int]->>> :t separate odd_even-separate odd_even- :: Monad m => Stream (Of Int) (Stream (Of Int) m) ()-- Now, for example, it is convenient to fold on the left and right values separately:-->>> S.toList $ S.toList $ separate odd_even-[2,4,6,8,10] :> ([1,3,5,7,9] :> ())--- Or we can write them to separate files or whatever:-->>> runResourceT $ S.writeFile "even.txt" . S.show $ S.writeFile "odd.txt" . S.show $ S.separate odd_even ->>> :! cat even.txt -2-4-6-8-10->>> :! cat odd.txt-1-3-5-7-9-- Of course, in the special case of @Stream (Of a) m r@, we can achieve the above - effects more simply by using 'Streaming.Prelude.copy'-->>> S.toList . S.filter even $ S.toList . S.filter odd $ S.copy $ each [1..10::Int]-[2,4,6,8,10] :> ([1,3,5,7,9] :> ())--- But 'separate' and 'unseparate' are functor-general. ---}--separate :: (Monad m, Functor f, Functor g) => Stream (Sum f g) m r -> Stream f (Stream g m) r-separate str = destroyExposed - str - (\x -> case x of InL fss -> wrap fss; InR gss -> effect (yields gss))- (effect . lift) - return -{-#INLINABLE separate #-}--unseparate :: (Monad m, Functor f, Functor g) => Stream f (Stream g m) r -> Stream (Sum f g) m r-unseparate str = destroyExposed - str - (wrap . InL) - (join . maps InR) - return -{-#INLINABLE unseparate #-}---unzips :: (Monad m, Functor f, Functor g) => - Stream (Compose f g) m r -> Stream f (Stream g m) r -unzips str = destroyExposed- str - (\(Compose fgstr) -> Step (fmap (Effect . yields) fgstr))- (Effect . lift) - return -{-#INLINABLE unzips #-}--{-| Group layers in an alternating stream into adjoining sub-streams- of one type or another. --}-groups :: (Monad m, Functor f, Functor g) - => Stream (Sum f g) m r - -> Stream (Sum (Stream f m) (Stream g m)) m r-groups = loop - where- loop str = do- e <- lift $ inspect str- case e of- Left r -> return r- Right ostr -> case ostr of- InR gstr -> wrap $ InR (fmap loop (cleanR (wrap (InR gstr))))- InL fstr -> wrap $ InL (fmap loop (cleanL (wrap (InL fstr))))-- cleanL :: (Monad m, Functor f, Functor g) =>- Stream (Sum f g) m r -> Stream f m (Stream (Sum f g) m r)- cleanL = loop where- loop s = do- e <- lift $ inspect s- case e of- Left r -> return (return r)- Right (InL fstr) -> wrap (fmap loop fstr)- Right (InR gstr) -> return (wrap (InR gstr))-- cleanR :: (Monad m, Functor f, Functor g) =>- Stream (Sum f g) m r -> Stream g m (Stream (Sum f g) m r)--- cleanR = fmap (maps switch) . cleanL . maps switch- cleanR = loop where- loop s = do- e <- lift $ inspect s- case e of- Left r -> return (return r)- Right (InL fstr) -> return (wrap (InL fstr))- Right (InR gstr) -> wrap (fmap loop gstr)-{-#INLINABLE groups #-}- --- groupInL :: (Monad m, Functor f, Functor g)--- => Stream (Sum f g) m r--- -> Stream (Sum (Stream f m) g) m r--- groupInL = loop--- where--- loop str = do--- e <- lift $ inspect str--- case e of--- Left r -> return r--- Right ostr -> case ostr of--- InR gstr -> wrap $ InR (fmap loop gstr)--- InL fstr -> wrap $ InL (fmap loop (cleanL (wrap (InL fstr))))--- cleanL :: (Monad m, Functor f, Functor g) =>--- Stream (Sum f g) m r -> Stream f m (Stream (Sum f g) m r)--- cleanL = loop where--- loop s = dos--- e <- lift $ inspect s--- case e of--- Left r -> return (return r)--- Right (InL fstr) -> wrap (fmap loop fstr)--- Right (InR gstr) -> return (wrap (InR gstr))--{- | 'never' interleaves the pure applicative action with the return of the monad forever. - It is the 'empty' of the 'Alternative' instance, thus--> never <|> a = a-> a <|> never = a-- and so on. If w is a monoid then @never :: Stream (Of w) m r@ is- the infinite sequence of 'mempty', and- @str1 \<|\> str2@ appends the elements monoidally until one of streams ends.- Thus we have, e.g.-->>> S.stdoutLn $ S.take 2 $ S.stdinLn <|> S.repeat " " <|> S.stdinLn <|> S.repeat " " <|> S.stdinLn -1<Enter> -2<Enter>-3<Enter>-1 2 3-4<Enter>-5<Enter>-6<Enter>-4 5 6-- This is equivalent to -->>> S.stdoutLn $ S.take 2 $ foldr (<|>) never [S.stdinLn, S.repeat " ", S.stdinLn, S.repeat " ", S.stdinLn ]-- Where 'f' is a monad, @(\<|\>)@ sequences the conjoined streams stepwise. See the- definition of @paste@ <https://gist.github.com/michaelt/6c6843e6dd8030e95d58 here>,- where the separate steps are bytestreams corresponding to the lines of a file. -- Given, say,--> data Branch r = Branch r r deriving Functor -- add obvious applicative instance-- then @never :: Stream Branch Identity r@ is the pure infinite binary tree with- (inaccessible) @r@s in its leaves. Given two binary trees, @tree1 \<|\> tree2@- intersects them, preserving the leaves that came first, - so @tree1 \<|\> never = tree1@-- @Stream Identity m r@ is an action in @m@ that is indefinitely delayed. Such an - action can be constructed with e.g. 'untilJust'.--> untilJust :: (Monad m, Applicative f) => m (Maybe r) -> Stream f m r-- Given two such items, @\<|\>@ instance races them. - It is thus the iterative monad transformer specially defined in - <https://hackage.haskell.org/package/free-4.12.1/docs/Control-Monad-Trans-Iter.html Control.Monad.Trans.Iter>-- So, for example, we might write-->>> let justFour str = if length str == 4 then Just str else Nothing->>> let four = untilJust (liftM justFour getLine) ->>> run four-one<Enter>-two<Enter>-three<Enter>-four<Enter>-"four"--- The 'Alternative' instance in - <https://hackage.haskell.org/package/free-4.12.1/docs/Control-Monad-Trans-Free.html Control.Monad.Trans.Free> - is avowedly wrong, though no explanation is given for this.----}-never :: (Monad m, Applicative f) => Stream f m r-never = let loop = Effect $ return $ Step $ pure loop in loop-{-#INLINABLE never #-} ---delays :: (MonadIO m, Applicative f) => Double -> Stream f m r-delays seconds = loop where- loop = Effect $ liftIO (threadDelay delay) >> return (Step (pure loop))- delay = fromInteger (truncate (1000000 * seconds)) -{-#INLINABLE delays #-}---- {-| Permit streamed actions to proceed unless the clock has run out.------ -}--- period :: (MonadIO m, Functor f) => Double -> Stream f m r -> Stream f m (Stream f m r)--- period seconds str = do--- utc <- liftIO getCurrentTime--- let loop s = do--- utc' <- liftIO getCurrentTime--- if diffUTCTime utc' utc > (cutoff / 1000000000)--- then return s--- else case s of--- Return r -> Return (Return r)--- Effect m -> Effect (liftM loop m)--- Step f -> Step (fmap loop f)--- loop str--- where--- cutoff = fromInteger (truncate (1000000000 * seconds))--- {-#INLINABLE period #-}--------- {-| Divide a succession of phases according to a specified time interval. If time runs out--- while an action is proceeding, it is allowed to run to completion. The clock is only then--- restarted.--- -}--- periods :: (MonadIO m, Functor f) => Double -> Stream f m r -> Stream (Stream f m) m r--- periods seconds s = do--- utc <- liftIO getCurrentTime--- loop (addUTCTime cutoff utc) s------ where--- cutoff = fromInteger (truncate (1000000000 * seconds)) / 1000000000--- loop final stream = do--- utc <- liftIO getCurrentTime--- if utc > final--- then loop (addUTCTime cutoff utc) stream--- else case stream of--- Return r -> Return r--- Effect m -> Effect $ liftM (loop final) m--- Step fstr -> Step $ fmap (periods seconds) (cutoff_ final (Step fstr))------ -- do--- -- let sloop s = do--- -- utc' <- liftIO getCurrentTime--- -- if final < utc'--- -- then return s--- -- else case s of--- -- Return r -> Return (Return r)--- -- Effect m -> Effect (liftM sloop m)--- -- Step f -> Step (fmap sloop f)--- -- Step (Step (fmap (fmap (periods seconds) . sloop) fstr))--- -- str <- m--- -- utc' <- liftIO getCurrentTime--- -- if diffUTCTime utc' utc > (cutoff / 1000000000)--- -- then return (loop utc' str)--- -- else return (loop utc str)--- -- Step fs -> do--- -- let check str = do--- -- utc' <- liftIO getCurrentTime--- -- loop utc' str--- ----- {-# INLINABLE periods #-}------ cutoff_ final str = do--- let loop s = do--- utc' <- liftIO getCurrentTime--- if utc' > final--- then Return s--- else case s of--- Return r -> Return (Return r)--- Effect m -> Effect (liftM loop m)--- Step f -> Step (fmap loop f)--- loop str--{- | Repeat a ---}--untilJust :: (Monad m, Applicative f) => m (Maybe r) -> Stream f m r-untilJust act = loop where- loop = Effect $ do- m <- act- case m of - Nothing -> return $ Step $ pure loop- Just a -> return $ Return a- - -cutoff :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Maybe r)-cutoff = loop where- loop 0 str = return Nothing- loop n str = do - e <- lift $ inspect str- case e of- Left r -> return (Just r)- Right (frest) -> Step $ fmap (loop (n-1)) frest
− Streaming/Prelude.hs
@@ -1,2787 +0,0 @@-{-| This module is very closely modeled on Pipes.Prelude, Pipes.Group and Pipes.Parse. It- maybe said to give independent expression to the conception of Producer manipulation - articulated in the latter two modules. Because we dispense with piping and - conduiting, the distinction between all of these modules collapses. - The leading type is chosen to permit an api that is as close as possible to that - of Data.List and the Prelude. -- Import qualified thus:--> import Streaming-> import qualified Streaming.Prelude as S-- For the examples below, one sometimes needs--> import Streaming.Prelude (each, yield, next, mapped, stdoutLn, stdinLn)-> import Data.Function ((&)) -- Other libraries that come up in passing are--> import qualified Control.Foldl as L -- cabal install foldl-> import qualified Pipes as P-> import qualified Pipes.Prelude as P-> import qualified System.IO as IO-- Here are some correspondences between the types employed here and elsewhere:--> streaming | pipes | conduit | io-streams-> --------------------------------------------------------------------------------------------------------------------> Stream (Of a) m () | Producer a m () | Source m a | InputStream a-> | ListT m a | ConduitM () o m () | Generator r ()-> --------------------------------------------------------------------------------------------------------------------> Stream (Of a) m r | Producer a m r | ConduitM () o m r | Generator a r-> --------------------------------------------------------------------------------------------------------------------> Stream (Of a) m (Stream (Of a) m r) | Producer a m (Producer a m r) | -> ---------------------------------------------------------------------------------------------------------------------> Stream (Stream (Of a) m) r | FreeT (Producer a m) m r |-> ---------------------------------------------------------------------------------------------------------------------> ---------------------------------------------------------------------------------------------------------------------> ByteString m () | Producer ByteString m () | Source m ByteString | InputStream ByteString-> ---------------------------------------------------------------------------------------------------------------------> --}-{-# LANGUAGE RankNTypes, BangPatterns, DeriveDataTypeable, TypeFamilies,- DeriveFoldable, DeriveFunctor, DeriveTraversable, CPP, Trustworthy #-}- -module Streaming.Prelude (- -- * Types- Of (..)-- -- * Introducing streams of elements- -- $producers- , yield- , each- , each'- , unfoldr- , stdinLn- , readLn- , fromHandle- , readFile- , iterate- , iterateM- , repeat- , repeatM- , replicate- , untilRight- , cycle- , replicateM- , enumFrom- , enumFromThen- , seconds- - - -- * Consuming streams of elements- -- $consumers- , stdoutLn- , stdoutLn'- , mapM_- , print- , toHandle- , writeFile- , effects- , erase- , drained- -- -- * Stream transformers- -- $pipes- , map- , mapM- , maps- , mapped- , for- , with- , subst- , copy- , copy'- , store- , chain- , sequence- , filter- , filterM- , delay- , intersperse- , take- , takeWhile--- , takeWhile'- , drop- , dropWhile- , concat- -- , elemIndices- -- , findIndices- , scan- , scanM- , scanned- , read- , show- , cons- , duplicate- , duplicate'--- -- * Splitting and inspecting streams of elements- , next- , uncons- , splitAt- , split--- , breaks- , break- , breakWhen- , span- , group- , groupBy- -- , groupedBy- -- , split- -- -- * Sum and Compose manipulation- - , distinguish - , switch- , separate- , unseparate- , eitherToSum- , sumToEither- , sumToCompose- , composeToSum- - -- * Folds- -- $folds- , fold- , fold_- , foldM- , foldM_- , all- , all_- , any- , any_- , sum- , sum_- , product- , product_- , head- , head_- , last- , last_- , elem- , elem_- , notElem- , notElem_- , length- , length_- , toList- , toList_- , mconcat- , mconcat_- , minimum- , minimum_- , maximum- , maximum_- , foldrM- , foldrT- - - -- , all- -- , any- -- , and- -- , or- -- , elem-- -- , find- -- , findIndex- -- , head- -- , index- -- , last- -- , length- -- , maximum- -- , minimum- -- , null-- -- * Zips and unzips- , zip- , zipWith- , zip3- , zipWith3- , unzip- , partitionEithers- , partition- - -- * Maybes- -- $maybes- , catMaybes- , mapMaybe-- -- * Pair manipulation- , lazily- , strictly- , fst'- , snd'- , mapOf- , _first- , _second- - -- * Interoperation- , reread- - -- * Basic Type- , Stream-- ) where-import Streaming.Internal--import Control.Monad hiding (filterM, mapM, mapM_, foldM, foldM_, replicateM, sequence)-import Data.Data ( Data, Typeable )-import Data.Functor.Identity-import Data.Functor.Sum-import Control.Monad.Trans-import Control.Applicative (Applicative (..))-import Data.Functor (Functor (..), (<$))--import qualified Prelude as Prelude -import Data.Foldable (Foldable)-import Data.Traversable (Traversable)-import qualified Data.Foldable as Foldable-import Text.Read (readMaybe)-import Prelude hiding (map, mapM, mapM_, filter, drop, dropWhile, take, mconcat- , sum, product, iterate, repeat, cycle, replicate, splitAt- , takeWhile, enumFrom, enumFromTo, enumFromThen, length- , print, zipWith, zip, zipWith3, zip3, unzip, seq, show, read- , readLn, sequence, concat, span, break, readFile, writeFile- , minimum, maximum, elem, notElem, intersperse, all, any, head- , last)--import qualified GHC.IO.Exception as G-import qualified System.IO as IO-import Foreign.C.Error (Errno(Errno), ePIPE)-import Control.Exception (throwIO, try)-import Data.Monoid (Monoid (mappend, mempty))-import Data.String (IsString (..))-import Control.Concurrent (threadDelay)-import Data.Time (getCurrentTime, diffUTCTime, picosecondsToDiffTime)-import Data.Functor.Classes-import Data.Functor.Compose-import Control.Monad.Trans.Resource--import GHC.Magic-#if MIN_VERSION_base(4,8,0)-import Data.Bifunctor-#endif---- | A left-strict pair; the base functor for streams of individual elements.-data Of a b = !a :> b- deriving (Data, Eq, Foldable, Ord,- Read, Show, Traversable, Typeable)-infixr 5 :>--instance (Monoid a, Monoid b) => Monoid (Of a b) where- mempty = mempty :> mempty- {-#INLINE mempty #-}- mappend (m :> w) (m' :> w') = mappend m m' :> mappend w w'- {-#INLINE mappend #-}--instance Functor (Of a) where- fmap f (a :> x) = a :> f x- {-#INLINE fmap #-}- a <$ (b :> x) = b :> a- {-#INLINE (<$) #-}--#if MIN_VERSION_base(4,8,0)-instance Bifunctor Of where- bimap f g (a :> b) = f a :> g b - {-#INLINE bimap #-}- first f (a :> b) = f a :> b- {-#INLINE first #-}- second g (a :> b) = a :> g b - {-#INLINE second #-}-#endif--instance Monoid a => Applicative (Of a) where- pure x = mempty :> x- {-#INLINE pure #-}- m :> f <*> m' :> x = mappend m m' :> f x- {-#INLINE (<*>) #-}- m :> x *> m' :> y = mappend m m' :> y- {-#INLINE (*>) #-}- m :> x <* m' :> y = mappend m m' :> x - {-#INLINE (<*) #-}--instance Monoid a => Monad (Of a) where- return x = mempty :> x- {-#INLINE return #-}- m :> x >> m' :> y = mappend m m' :> y- {-#INLINE (>>) #-}- m :> x >>= f = let m' :> y = f x in mappend m m' :> y- {-#INLINE (>>=) #-}--instance (r ~ (), Monad m, f ~ Of Char) => IsString (Stream f m r) where- fromString = each---- instance (Eq a) => Eq1 (Of a) where eq1 = (==)--- instance (Ord a) => Ord1 (Of a) where compare1 = compare--- instance (Read a) => Read1 (Of a) where readsPrec1 = readsPrec--- instance (Show a) => Show1 (Of a) where showsPrec1 = showsPrec--{-| Note that 'lazily', 'strictly', 'fst'', and 'mapOf' are all so-called /natural transformations/ on the primitive @Of a@ functor- If we write - -> type f ~~> g = forall x . f x -> g x- - then we can restate some types as follows:- -> mapOf :: (a -> b) -> Of a ~~> Of b -- Bifunctor first-> lazily :: Of a ~~> (,) a-> Identity . fst' :: Of a ~~> Identity a-- Manipulation of a @Stream f m r@ by mapping often turns on recognizing natural transformations of @f@.- Thus @maps@ is far more general the the @map@ of the @Streaming.Prelude@, which can be- defined thus:--> S.map :: (a -> b) -> Stream (Of a) m r -> Stream (Of b) m r-> S.map f = maps (mapOf f)-- i.e.--> S.map f = maps (\(a :> x) -> (f a :> x))- - This rests on recognizing that @mapOf@ is a natural transformation; note though- that it results in such a transformation as well:- -> S.map :: (a -> b) -> Stream (Of a) m ~> Stream (Of b) m -- Thus we can @maps@ it in turn.--- -}-lazily :: Of a b -> (a,b)-lazily = \(a:>b) -> (a,b)-{-# INLINE lazily #-}--{-| Convert a standard Haskell pair into a left-strict pair -}-strictly :: (a,b) -> Of a b-strictly = \(a,b) -> a :> b -{-# INLINE strictly #-}--{-| @fst'@ and @snd'@ extract the first and second element of a pair-->>> S.fst' (1:>"hi")-1->>> S.snd' (1:>"hi")-"hi"--- They are contained in the @_first@ and @_second@ lenses, - if any lens library is in scope- ->>> import Lens.Micro->>> (1:>"hi") ^. S._first-1->>> (1:>"hi") ^. S._second-"hi"-- -}--fst' :: Of a b -> a-fst' (a :> b) = a-{-#INLINE fst' #-}-snd' :: Of a b -> b-snd' (a :> b) = b-{-#INLINE snd' #-}--{-| Map a function over the first element of an @Of@ pair-->>> S.mapOf even (1:>"hi")-False :> "hi"-- @mapOf@ is just @first@ from the @Bifunctor@ instance - ->>> first even (1:>"hi")-False :> "hi" -- and is contained in the @_first@ lens- ->>> import Lens.Micro->>> over S._first even (1:>"hi")-False :> "hi"-- -}--mapOf :: (a -> b) -> Of a r -> Of b r-mapOf f (a:> b) = (f a :> b)-{-#INLINE mapOf #-}--{-| A lens into the first element of a left-strict pair -}-_first :: Functor f => (a -> f a') -> Of a b -> f (Of a' b)-_first afb (a:>b) = fmap (\c -> (c:>b)) (afb a)-{-# INLINE _first #-}--{-| A lens into the second element of a left-strict pair -}-_second :: Functor f => (b -> f b') -> Of a b -> f (Of a b')-_second afb (a:>b) = fmap (\c -> (a:>c)) (afb b)-{-#INLINABLE _second #-}--all :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m (Of Bool r)-all thus = loop True where- loop b str = case str of- Return r -> return (b :> r)- Effect m -> m >>= loop b- Step (a :> rest) -> if thus a- then loop True rest- else do - r <- effects rest- return (False :> r)-{-#INLINABLE all #-}--all_ :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m Bool-all_ thus = loop True where- loop b str = case str of- Return r -> return b- Effect m -> m >>= loop b- Step (a :> rest) -> if thus a- then loop True rest- else return False-{-#INLINABLE all_ #-}---any :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m (Of Bool r)-any thus = loop False where- loop b str = case str of- Return r -> return (b :> r)- Effect m -> m >>= loop b- Step (a :> rest) -> if thus a- then do - r <- effects rest- return (True :> r)- else loop False rest-{-#INLINABLE any #-}--any_ :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m Bool-any_ thus = loop False where- loop b str = case str of- Return r -> return b- Effect m -> m >>= loop b- Step (a :> rest) -> if thus a- then return True- else loop False rest -{-#INLINABLE any_ #-}- -{-| Break a sequence upon meeting element falls under a predicate, - keeping it and the rest of the stream as the return value.-->>> rest <- S.print $ S.break even $ each [1,1,2,3] -1-1->>> S.print rest-2-3---}--break :: Monad m => (a -> Bool) -> Stream (Of a) m r - -> Stream (Of a) m (Stream (Of a) m r)-break pred = loop where- loop str = case str of - Return r -> Return (Return r)- Effect m -> Effect $ liftM loop m- Step (a :> rest) -> if (pred a) - then Return (Step (a :> rest))- else Step (a :> loop rest)-{-# INLINABLE break #-}--{-| Yield elements, using a fold to maintain state, until the accumulated - value satifies the supplied predicate. The fold will then be short-circuited - and the element that breaks it will be put after the break.- This function is easiest to use with 'Control.Foldl.purely'-->>> rest <- each [1..10] & L.purely S.breakWhen L.sum (>10) & S.print -1-2-3-4->>> S.print rest-5-6-7-8-9-10---}-breakWhen :: Monad m => (x -> a -> x) -> x -> (x -> b) -> (b -> Bool) -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r)-breakWhen step begin done pred = loop0 begin- where- loop0 x stream = case stream of - Return r -> return (return r)- Effect mn -> Effect $ liftM (loop0 x) mn- Step (a :> rest) -> loop a (step x a) rest- loop a !x stream = do- if pred (done x) - then return (yield a >> stream) - else case stream of - Return r -> yield a >> return (return r)- Effect mn -> Effect $ liftM (loop a x) mn- Step (a' :> rest) -> do- yield a- loop a' (step x a') rest-{-# INLINABLE breakWhen #-}---- -- Break during periods where the predicate is not satisfied, grouping the periods when it is.------ >>> S.print $ mapped S.toList $ S.breaks not $ S.each [False,True,True,False,True,True,False]--- [True,True]--- [True,True]--- >>> S.print $ mapped S.toList $ S.breaks id $ S.each [False,True,True,False,True,True,False]--- [False]--- [False]--- [False]------ -}--- breaks--- :: Monad m =>--- (a -> Bool) -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r--- breaks thus = loop where--- loop stream = Effect $ do--- e <- next stream--- return $ case e of--- Left r -> Return r--- Right (a, p') ->--- if not (thus a)--- then Step $ fmap loop (yield a >> break thus p')--- else loop p'--- {-#INLINABLE breaks #-}--{-| Apply an action to all values, re-yielding each-->>> S.product $ S.chain Prelude.print $ S.each [1..5]-1-2-3-4-5-120 :> ()--}--chain :: Monad m => (a -> m ()) -> Stream (Of a) m r -> Stream (Of a) m r-chain f = loop where - loop str = case str of - Return r -> return r- Effect mn -> Effect (liftM loop mn)- Step (a :> rest) -> Effect $ do- f a- return (Step (a :> loop rest))-{-# INLINABLE chain #-}--{-| Make a stream of traversable containers into a stream of their separate elements.- This is just --> concat str = for str each-->>> S.print $ S.concat (each ["xy","z"])-'x'-'y'-'z'-- Note that it also has the effect of 'Data.Maybe.catMaybes', 'Data.Either.rights'- 'map snd' and such-like operations.-->>> S.print $ S.concat $ S.each [Just 1, Nothing, Just 2]-1-2->>> S.print $ S.concat $ S.each [Right 1, Left "Error!", Right 2]-1-2->>> S.print $ S.concat $ S.each [('A',1), ('B',2)]-1-2---}--concat :: (Monad m, Foldable.Foldable f) => Stream (Of (f a)) m r -> Stream (Of a) m r-concat str = for str each-{-# INLINE concat #-}--{-| The natural @cons@ for a @Stream (Of a)@. --> cons a stream = yield a >> stream-- Useful for interoperation:--> Data.Text.foldr S.cons (return ()) :: Text -> Stream (Of Char) m ()-> Lazy.foldrChunks S.cons (return ()) :: Lazy.ByteString -> Stream (Of Strict.ByteString) m ()-- and so on.--}--cons :: (Monad m) => a -> Stream (Of a) m r -> Stream (Of a) m r-cons a str = Step (a :> str)-{-# INLINE cons #-}--{- | Cycle repeatedly through the layers of a stream, /ad inf./ This- function is functor-general--> cycle = forever-->>> rest <- S.print $ S.splitAt 3 $ S.cycle (yield 0 >> yield 1)-True-False-True->>> S.print $ S.take 3 rest-False-True-False---}--cycle :: (Monad m, Functor f) => Stream f m r -> Stream f m s-cycle str = loop where loop = str >> loop-{-#INLINABLE cycle #-}---{-| Interpolate a delay of n seconds between yields.--}-delay :: MonadIO m => Double -> Stream (Of a) m r -> Stream (Of a) m r-delay seconds = loop where- pico = truncate (seconds * 1000000)- loop str = do - e <- lift $ next str- case e of- Left r -> Return r- Right (a,rest) -> do- yield a- liftIO $ threadDelay pico- loop rest-{-#INLINABLE delay #-}--- -{-| Where a transformer returns a stream, run the effects of the stream, keeping- the return value. This is usually used at the type--> drained :: Monad m => Stream (Of a) m (Stream (Of b) m r) -> Stream (Of a) m r-> drained = join . fmap (lift . effects)- - Here, for example, we split a stream in two places and throw out the middle segment:- ->>> rest <- S.print $ S.drained $ S.splitAt 2 $ S.splitAt 5 $ each [1..7]-1-2->>> S.print rest-6-7-- In particular, we can define versions of @take@ and @takeWhile@ which - retrieve the return value of the rest of the stream - and which can - thus be used with 'maps':--> take' n = S.drained . S.splitAt n-> takeWhile' thus = S.drained . S.span thus---}-drained :: (Monad m, Monad (t m), Functor (t m), MonadTrans t) => t m (Stream (Of a) m r) -> t m r-drained = join . fmap (lift . effects)-{-#INLINE drained #-}---- ------------------ drop--- ----------------{-| Ignore the first n elements of a stream, but carry out the actions-->>> S.toList $ S.drop 2 $ S.replicateM 5 getLine -a<Enter>-b<Enter>-c<Enter>-d<Enter>-e<Enter>-["c","d","e"] :> ()-- Because it retains the final return value, @drop n@ is a suitable argument - for @maps@:-->>> S.toList $ concats $ maps (S.drop 4) $ chunksOf 5 $ each [1..20]-[5,10,15,20] :> ()-- -- -}--drop :: (Monad m) => Int -> Stream (Of a) m r -> Stream (Of a) m r-drop n str | n <= 0 = str-drop n str = loop n str where- loop 0 stream = stream- loop n stream = case stream of- Return r -> Return r- Effect ma -> Effect (liftM (loop n) ma)- Step (a :> as) -> loop (n-1) as-{-# INLINABLE drop #-}---- ------------------ dropWhile--- -----------------{- | Ignore elements of a stream until a test succeeds, retaining the rest.-->>> S.print $ S.dropWhile ((< 5) . length) S.stdinLn -one<Enter>-two<Enter>-three<Enter>-"three"-four<Enter>-"four"-^CInterrupted.----}-dropWhile :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m r-dropWhile pred = loop where - loop stream = case stream of- Return r -> Return r- Effect ma -> Effect (liftM loop ma)- Step (a :> as) -> if pred a - then loop as- else Step (a :> as)-{-# INLINABLE dropWhile #-}---- ------------------ each --- -----------------{- | Stream the elements of a pure, foldable container.-->>> S.print $ each [1..3] -1-2-3->>> S.print $ mapped S.toList $ chunksOf 3 $ S.replicateM 5 getLine-s<Enter>-t<Enter>-u<Enter>-["s","t","u"]-v<Enter>-w<Enter>-["v","w"]---}-each :: (Monad m, Foldable.Foldable f) => f a -> Stream (Of a) m ()-each = Foldable.foldr (\a p -> Step (a :> p)) (Return ())-{-# INLINABLE each #-}--each' :: (Monad m, Foldable.Foldable f) => f a -> Stream (Of a) m ()-each' = Foldable.foldr (\a p -> Effect (return (Step (a :> p)))) (Return ())-{-# INLINABLE each' #-}---- ------------------ effects--- -----------------{- | Reduce a stream, performing its actions but ignoring its elements. - ->>> rest <- S.effects $ S.splitAt 2 $ each [1..5]->>> S.print rest-3-4-5- 'effects' should be understood together with 'copy' and is subject to the rules--> S.effects . S.copy = id-> hoist S.effects . S.copy = id-- The similar @effects@ and @copy@ operations in @Data.ByteString.Streaming@ obey the same rules. ---}-effects :: Monad m => Stream (Of a) m r -> m r-effects = loop where- loop stream = case stream of - Return r -> return r- Effect m -> m >>= loop - Step (_ :> rest) -> loop rest-{-#INLINABLE effects #-}--{-| Exhaust a stream remembering only whether @a@ was an element.---}--elem :: (Monad m, Eq a) => a -> Stream (Of a) m r -> m (Of Bool r)-elem a' = loop False where- loop True str = liftM (True :>) (effects str)- loop False str = case str of- Return r -> return (False :> r)- Effect m -> m >>= loop False- Step (a:> rest) -> - if a == a' - then liftM (True :>) (effects rest)- else loop False rest-{-#INLINABLE elem #-}--elem_ :: (Monad m, Eq a) => a -> Stream (Of a) m r -> m Bool-elem_ a' = loop False where- loop True str = return True- loop False str = case str of- Return r -> return False- Effect m -> m >>= loop False- Step (a:> rest) -> - if a == a' - then return True- else loop False rest-{-#INLINABLE elem_ #-}---- -------- enumFrom--- --------{-| An infinite stream of enumerable values, starting from a given value.- It is the same as `S.iterate succ`. - Because their return type is polymorphic, @enumFrom@ and @enumFromThen@- (and @iterate@ are useful for example with @zip@- and @zipWith@, which require the same return type in the zipped streams. - With @each [1..]@ the following bit of connect-and-resume would be impossible:-->>> rest <- S.print $ S.zip (S.enumFrom 'a') $ S.splitAt 3 $ S.enumFrom 1-('a',1)-('b',2)-('c',3)->>> S.print $ S.take 3 rest-4-5-6---}-enumFrom :: (Monad m, Enum n) => n -> Stream (Of n) m r-enumFrom = loop where- loop !n = Effect (return (Step (n :> loop (succ n))))-{-# INLINABLE enumFrom #-}---{-| An infinite sequence of enumerable values at a fixed distance, determined- by the first and second values. See the discussion of 'Streaming.enumFrom'-->>> S.print $ S.take 3 $ S.enumFromThen 100 200-100-200-300---}-enumFromThen:: (Monad m, Enum a) => a -> a -> Stream (Of a) m r-enumFromThen first second = Streaming.Prelude.map toEnum (loop _first)- where- _first = fromEnum first- _second = fromEnum second- diff = _second - _first- loop !s = Step (s :> loop (s+diff))-{-# INLINABLE enumFromThen #-}---- ------------------ erase--- ----------------{- | Remove the elements from a stream of values, retaining the structure of layers.--}-erase :: Monad m => Stream (Of a) m r -> Stream Identity m r-erase = loop where- loop str = case str of- Return r -> Return r- Effect m -> Effect (liftM loop m)- Step (a:>rest) -> Step (Identity (loop rest))-{-# INLINABLE erase #-} ---- ------------------ filter --- ------------------- | Skip elements of a stream that fail a predicate-filter :: (Monad m) => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m r-filter pred = loop where- loop str = case str of- Return r -> Return r- Effect m -> Effect (liftM loop m)- Step (a :> as) -> if pred a - then Step (a :> loop as)- else loop as-{-# INLINABLE filter #-}---- ------------------ filterM--- ------------------- | Skip elements of a stream that fail a monadic test-filterM :: (Monad m) => (a -> m Bool) -> Stream (Of a) m r -> Stream (Of a) m r-filterM pred = loop where- loop str = case str of- Return r -> Return r- Effect m -> Effect $ liftM loop m- Step (a :> as) -> Effect $ do - bool <- pred a- if bool - then return $ Step (a :> loop as)- else return $ loop as-{-# INLINABLE filterM #-}----- -- ------------------ -- first--- -- ------------------ {- | Take either the first item in a stream or the return value, if it is empty.--- The typical mark of an infinite stream is a polymorphic return value; in--- that case, 'first' is a sort of @safeHead@------ To iterate an action returning a 'Maybe', until it succeeds.------ -}--- first :: Monad m => Stream (Of r) m r -> m r--- first = loop where--- loop str = case str of--- Return r -> return r--- Effect m -> m >>= loop--- Step (r :> rest) -> return r--- {-# INLINABLE first #-}- --- ------------------ fold--- -----------------{- $folds- Use these to fold the elements of a 'Stream'. -->>> S.fold_ (+) 0 id $ S.each [1..0]-50-- The general folds 'fold', fold_', 'foldM' and 'foldM_' are arranged - for use with @Control.Foldl@ 'Control.Foldl.purely' and 'Control.Foldl.impurely'-->>> L.purely fold_ L.sum $ each [1..10]-55->>> L.purely fold_ (liftA3 (,,) L.sum L.product L.list) $ each [1..10]-(55,3628800,[1,2,3,4,5,6,7,8,9,10])-- All functions marked with an underscore omit - (e.g. @fold_@, @sum_@) the stream's return value in a left-strict pair.- They are good for exiting streaming completely, - but when you are, e.g. @mapped@-ing over a @Stream (Stream (Of a) m) m r@, - which is to be compared with @[[a]]@. Specializing, we have e.g.--> mapped sum :: (Monad m, Num n) => Stream (Stream (Of Int)) IO () -> Stream (Of n) IO ()-> mapped (fold mappend mempty id) :: Stream (Stream (Of Int)) IO () -> Stream (Of Int) IO ()-->>> S.print $ mapped S.sum $ chunksOf 3 $ S.each [1..10]-6-15-24-10-->>> let three_folds = L.purely S.fold (liftA3 (,,) L.sum L.product L.list)->>> S.print $ mapped three_folds $ chunksOf 3 (each [1..10])-(6,6,[1,2,3])-(15,120,[4,5,6])-(24,504,[7,8,9])-(10,10,[10])--}--{-| Strict fold of a 'Stream' of elements, preserving only the result of the fold, not- the return value of the stream. The third parameter will often be 'id' where a fold- is written by hand:-->>> S.fold_ (+) 0 id $ each [1..10]-55 - - It can be used to replace a standard Haskell type with one more suited to - writing a strict accumulation function. It is also crucial to the - Applicative instance for @Control.Foldl.Fold@--> Control.Foldl.purely fold :: Monad m => Fold a b -> Stream (Of a) m () -> m b--}-fold_ :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> m b-fold_ step begin done = liftM (\(a:>rest) -> a) . fold step begin done-{-#INLINE fold_ #-}--{-| Strict fold of a 'Stream' of elements that preserves the return value. - The third parameter will often be 'id' where a fold is written by hand:-->>> S.fold (+) 0 id $ each [1..10]-55 :> ()-->>> S.fold (*) 1 id $ S.fold (+) 0 id $ S.copy $ each [1..10]-3628800 :> (55 :> ())-- It can be used to replace a standard Haskell type with one more suited to - writing a strict accumulation function. It is also crucial to the - Applicative instance for @Control.Foldl.Fold@ We can apply such a fold- @purely@--> Control.Foldl.purely S.fold :: Monad m => Fold a b -> Stream (Of a) m r -> m (Of b r)-- Thus, specializing a bit:--> L.purely S.fold L.sum :: Stream (Of Int) Int r -> m (Of Int r)-> mapped (L.purely S.fold L.sum) :: Stream (Stream (Of Int)) IO r -> Stream (Of Int) IO r-- Here we use the Applicative instance for @Control.Foldl.Fold@ to - stream three-item segments of a stream together with their sums and products.-->>> S.print $ mapped (L.purely S.fold (liftA3 (,,) L.list L.product L.sum)) $ chunksOf 3 $ each [1..10]-([1,2,3],6,6)-([4,5,6],120,15)-([7,8,9],504,24)-([10],10,10)---}--fold :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> m (Of b r)-fold step begin done str = fold_loop str begin- where- fold_loop stream !x = case stream of- Return r -> return (done x :> r)- Effect m -> m >>= \str' -> fold_loop str' x- Step (a :> rest) -> fold_loop rest $! step x a-{-# INLINE fold #-}---{-| Strict, monadic fold of the elements of a 'Stream (Of a)'--> Control.Foldl.impurely foldM :: Monad m => FoldM a b -> Stream (Of a) m () -> m b--}-foldM_- :: Monad m- => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r -> m b-foldM_ step begin done = liftM (\(a :> rest) -> a) . foldM step begin done-{-#INLINE foldM_ #-}--{-| Strict, monadic fold of the elements of a 'Stream (Of a)'--> Control.Foldl.impurely foldM' :: Monad m => FoldM a b -> Stream (Of a) m r -> m (b, r)-- Thus to accumulate the elements of a stream as a vector, together with a random- element we might write:-->>> L.impurely S.foldM (liftA2 (,) L.vector L.random) $ each [1..10::Int] :: IO (Of (U.Vector Int,Maybe Int) ())-([1,2,3,4,5,6,7,8,9,10],Just 9) :> ()---}-foldM- :: Monad m- => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r ->m (Of b r)--foldM step begin done str = do- x0 <- begin- loop str x0- where- loop stream !x = case stream of- Return r -> done x >>= \b -> return (b :> r)- Effect m -> m >>= \s -> loop s x- Step (a :> rest) -> do- x' <- step x a- loop rest x'-{-# INLINABLE foldM #-}---- the following requires GHC.Magic.oneShot:--- foldM step begin done str = do--- x <- begin--- (x' :> r) <- streamFold--- (\r x -> return (x :> r))--- (\mx2mx -> oneShot (\x -> x `seq` mx2mx >>= ($ x) ))--- (\(a :> x2mx') -> oneShot (\x -> x `seq` (step x a >>= x2mx')) )--- ( str)--- x--- b <- done x'--- return (b :> r)--- where seq = Prelude.seq--- {-#INLINE foldM #-}--{-| A natural right fold for consuming a stream of elements. - See also the more general 'iterTM' in the 'Streaming' module - and the still more general 'destroy'--> foldrT (\a p -> Streaming.yield a >> p) = id-> foldrT (\a p -> Pipes.yield a >> p) :: Monad m => Stream (Of a) m r -> Producer a m r-> foldrT (\a p -> Conduit.yield a >> p) :: Monad m => Stream (Of a) m r -> Conduit a m r---}--foldrT :: (Monad m, MonadTrans t, Monad (t m)) - => (a -> t m r -> t m r) -> Stream (Of a) m r -> t m r-foldrT step = loop where- loop stream = case stream of- Return r -> return r- Effect m -> lift m >>= loop- Step (a :> as) -> step a (loop as)-{-# INLINABLE foldrT #-} --{-| A natural right fold for consuming a stream of elements.- See also the more general 'iterT' in the 'Streaming' module and the- still more general 'destroy'--}-foldrM :: Monad m - => (a -> m r -> m r) -> Stream (Of a) m r -> m r-foldrM step = loop where- loop stream = case stream of- Return r -> return r- Effect m -> m >>= loop- Step (a :> as) -> step a (loop as)-{-# INLINABLE foldrM #-} ---- ------------------ for--- ------------------- | @for@ replaces each element of a stream with an associated stream. Note that the--- associated stream may layer any functor. -for :: (Monad m, Functor f) => Stream (Of a) m r -> (a -> Stream f m x) -> Stream f m r-for str0 act = loop str0 where- loop str = case str of- Return r -> Return r - Effect m -> Effect $ liftM loop m- Step (a :> rest) -> do- act a- loop rest-{-# INLINABLE for #-}---- -| Group layers of any functor by comparisons on a preliminary annotation ---- groupedBy--- :: (Monad m, Functor f) =>--- (a -> a -> Bool)--- -> Stream (Compose (Of a) f) m r--- -> Stream (Stream (Compose (Of a) f) m) m r--- groupedBy equals = loop where--- loop stream = Effect $ do--- e <- inspect stream--- return $ case e of--- Left r -> Return r--- Right s@(Compose (a :> p')) -> Step $--- fmap loop (Step $ Compose (a :> fmap (span' (equals a)) p'))--- span' :: (Monad m, Functor f) => (a -> Bool) -> Stream (Compose (Of a) f) m r--- -> Stream (Compose (Of a) f) m (Stream (Compose (Of a) f) m r)--- span' pred = loop where--- loop str = case str of--- Return r -> Return (Return r)--- Effect m -> Effect $ liftM loop m--- Step s@(Compose (a :> rest)) -> case pred a of--- True -> Step (Compose (a :> fmap loop rest))--- False -> Return (Step s)--- {-# INLINABLE groupedBy #-}--{-| Group elements of a stream in accordance with the supplied comparison. --->>> S.print $ mapped S.toList $ S.groupBy (>=) $ each [1,2,3,1,2,3,4,3,2,4,5,6,7,6,5]-[1]-[2]-[3,1,2,3]-[4,3,2,4]-[5]-[6]-[7,6,5]---}-groupBy :: Monad m - => (a -> a -> Bool)- -> Stream (Of a) m r - -> Stream (Stream (Of a) m) m r-groupBy equals = loop where- loop stream = Effect $ do- e <- next stream- return $ case e of- Left r -> Return r- Right (a, p') -> Step $- fmap loop (yield a >> span (equals a) p')-{-# INLINABLE groupBy #-} ---{-| Group successive equal items together-->>> S.toList $ mapped S.toList $ S.group $ each "baaaaad"-["b","aaaaa","d"] :> ()-->>> S.toList $ concats $ maps (S.drained . S.splitAt 1) $ S.group $ each "baaaaaaad"-"bad" :> ()---}-group :: (Monad m, Eq a) => Stream (Of a) m r -> Stream (Stream (Of a) m) m r -group = groupBy (==)-{-#INLINE group #-}---head :: Monad m => Stream (Of a) m r -> m (Of (Maybe a) r)-head str = case str of- Return r -> return (Nothing :> r)- Effect m -> m >>= head- Step (a :> rest) -> effects rest >>= \r -> return (Just a :> r)-{-#INLINABLE head #-}- -head_ :: Monad m => Stream (Of a) m r -> m (Maybe a) -head_ str = case str of- Return r -> return Nothing- Effect m -> m >>= head_- Step (a :> rest) -> effects rest >> return (Just a)-{-#INLINABLE head_ #-}- -intersperse :: Monad m => a -> Stream (Of a) m r -> Stream (Of a) m r-intersperse x str = case str of- Return r -> Return r- Effect m -> Effect (liftM (intersperse x) m)- Step (a :> rest) -> loop a rest- where- loop !a str = case str of- Return r -> Step (a :> Return r)- Effect m -> Effect (liftM (loop a) m)- Step (b :> rest) -> Step (a :> Step (x :> loop b rest))-{-#INLINABLE intersperse #-}- - - ---- ------------------ iterate--- -----------------{-| Iterate a pure function from a seed value, streaming the results forever- ----}-iterate :: Monad m => (a -> a) -> a -> Stream (Of a) m r-iterate f = loop where- loop a' = Effect (return (Step (a' :> loop (f a'))))-{-# INLINABLE iterate #-}---- | Iterate a monadic function from a seed value, streaming the results forever-iterateM :: Monad m => (a -> m a) -> m a -> Stream (Of a) m r-iterateM f = loop where- loop ma = Effect $ do - a <- ma- return (Step (a :> loop (f a)))-{-# INLINABLE iterateM #-}----last :: Monad m => Stream (Of a) m r -> m (Of (Maybe a) r)-last = loop Nothing_ where- loop m str = case str of- Return r -> case m of - Nothing_ -> return (Nothing :> r)- Just_ a -> return (Just a :> r)- Effect m -> m >>= last- Step (a :> rest) -> loop (Just_ a) rest-{-#INLINABLE last #-}- ---last_ :: Monad m => Stream (Of a) m r -> m (Maybe a) -last_ = loop Nothing_ where- loop m str = case str of- Return r -> case m of - Nothing_ -> return Nothing - Just_ a -> return (Just a)- Effect m -> m >>= last_- Step (a :> rest) -> loop (Just_ a) rest-{-#INLINABLE last_ #-}---- ------------------ length--- -----------------{-| Run a stream, remembering only its length:-->>> S.length $ S.each [1..10]-10---}-length_ :: Monad m => Stream (Of a) m r -> m Int-length_ = fold_ (\n _ -> n + 1) 0 id-{-#INLINE length_#-}--{-| Run a stream, keeping its length and its return value. -->>> S.print $ mapped S.length $ chunksOf 3 $ S.each [1..10]-3-3-3-1---}--length :: Monad m => Stream (Of a) m r -> m (Of Int r)-length = fold (\n _ -> n + 1) 0 id-{-#INLINE length #-}--- ------------------ map--- -----------------{-| Standard map on the elements of a stream.-->>> S.stdoutLn $ S.map reverse $ each (words "alpha beta")-ahpla-ateb--}--map :: Monad m => (a -> b) -> Stream (Of a) m r -> Stream (Of b) m r-map f = maps (\(x :> rest) -> f x :> rest)--- loop where --- -- loop stream = case stream of- -- Return r -> Return r- -- Effect m -> Effect (liftM loop m)- -- Step (a :> as) -> Step (f a :> loop as)-{-# INLINABLE map #-}--- {-# NOINLINE [1] map #-}--- {-# RULES--- "map/map" [~1] forall f g bs . map f (map g bs) =--- map (f . g) bs--- #-}--{-| Replace each element of a stream with the result of a monadic action-->>> S.print $ S.mapM readIORef $ S.chain (\ior -> modifyIORef ior (*100)) $ S.mapM newIORef $ each [1..6]-100-200-300-400-500-600--}-mapM :: Monad m => (a -> m b) -> Stream (Of a) m r -> Stream (Of b) m r-mapM f = loop where- loop str = case str of - Return r -> Return r - Effect m -> Effect (liftM loop m)- Step (a :> as) -> Effect $ do - a' <- f a - return (Step (a' :> loop as) )-{-# INLINABLE mapM #-}----{-| Reduce a stream to its return value with a monadic action.-->>> S.mapM_ Prelude.print $ each [1..5]-1-2-3-4-5->>> rest <- S.mapM_ Prelude.print $ S.splitAt 3 $ each [1..10]-1-2-3->>> S.sum rest-49 :> ()---}-mapM_ :: Monad m => (a -> m b) -> Stream (Of a) m r -> m r-mapM_ f = loop where- loop str = case str of - Return r -> return r - Effect m -> m >>= loop- Step (a :> as) -> do - f a - loop as -{-# INLINABLE mapM_ #-}----{- | Map layers of one functor to another with a transformation involving the base monad.- This could be trivial, e.g.--> let noteBeginning text x = putStrLn text >> return text-- this puts the- is completely functor-general -- @maps@ and @mapped@ obey these rules:--> maps id = id-> mapped return = id-> maps f . maps g = maps (f . g)-> mapped f . mapped g = mapped (f <=< g)-> maps f . mapped g = mapped (liftM f . g)-> mapped f . maps g = mapped (f <=< liftM g)-- @maps@ is more fundamental than @mapped@, which is best understood as a convenience- for effecting this frequent composition:--> mapped phi = decompose . maps (Compose . phi) ----}--mapped :: (Monad m, Functor f) => (forall x . f x -> m (g x)) -> Stream f m r -> Stream g m r-mapped = mapsM-{-#INLINE mapped #-}---{-| Fold streamed items into their monoidal sum-->>> S.mconcat $ S.take 2 $ S.map (Data.Monoid.Last . Just) (S.stdinLn)-first<Enter>-last<Enter>-Last {getLast = Just "last"} :> ()-- -}-mconcat :: (Monad m, Monoid w) => Stream (Of w) m r -> m (Of w r)-mconcat = fold mappend mempty id-{-#INLINE mconcat #-}--data Maybe_ a = Just_ !a | Nothing_-mconcat_ :: (Monad m, Monoid w) => Stream (Of w) m r -> m w-mconcat_ = fold_ mappend mempty id--minimum :: (Monad m, Ord a) => Stream (Of a) m r -> m (Of (Maybe a) r)-minimum = fold (\m a -> case m of Nothing_ -> Just_ a ; Just_ a' -> Just_ (min a a')) - Nothing_- (\m -> case m of Nothing_ -> Nothing; Just_ r -> Just r)-{-#INLINE minimum #-}--minimum_ :: (Monad m, Ord a) => Stream (Of a) m r -> m (Maybe a) -minimum_ = fold_ (\m a -> case m of Nothing_ -> Just_ a ; Just_ a' -> Just_ (min a a')) - Nothing_- (\m -> case m of Nothing_ -> Nothing; Just_ r -> Just r)-{-#INLINE minimum_ #-}--maximum :: (Monad m, Ord a) => Stream (Of a) m r -> m (Of (Maybe a) r)-maximum = fold (\m a -> case m of Nothing_ -> Just_ a ; Just_ a' -> Just_ (max a a')) - Nothing_- (\m -> case m of Nothing_ -> Nothing; Just_ r -> Just r)-{-#INLINE maximum #-}--maximum_ :: (Monad m, Ord a) => Stream (Of a) m r -> m (Maybe a)-maximum_ = fold_ (\m a -> case m of Nothing_ -> Just_ a ; Just_ a' -> Just_ (max a a')) - Nothing_- (\m -> case m of Nothing_ -> Nothing; Just_ r -> Just r)-{-#INLINE maximum_ #-}--{-| The standard way of inspecting the first item in a stream of elements, if the- stream is still \'running\'. The @Right@ case contains a - Haskell pair, where the more general @inspect@ would return a left-strict pair. - There is no reason to prefer @inspect@ since, if the @Right@ case is exposed, - the first element in the pair will have been evaluated to whnf.--> next :: Monad m => Stream (Of a) m r -> m (Either r (a, Stream (Of a) m r))-> inspect :: Monad m => Stream (Of a) m r -> m (Either r (Of a (Stream (Of a) m r)))-- Interoperate with @pipes@ producers thus:--> Pipes.unfoldr Stream.next :: Stream (Of a) m r -> Producer a m r-> Stream.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r - - Similarly: --> IOStreams.unfoldM (liftM (either (const Nothing) Just) . next) :: Stream (Of a) IO b -> IO (InputStream a)-> Conduit.unfoldM (liftM (either (const Nothing) Just) . next) :: Stream (Of a) m r -> Source a m r-- But see 'uncons', which is better fitted to these @unfoldM@s--}-next :: Monad m => Stream (Of a) m r -> m (Either r (a, Stream (Of a) m r))-next = loop where- loop stream = case stream of- Return r -> return (Left r)- Effect m -> m >>= loop- Step (a :> rest) -> return (Right (a,rest))-{-# INLINABLE next #-}---{-| Exhaust a stream deciding whether @a@ was an element.---}--notElem :: (Monad m, Eq a) => a -> Stream (Of a) m r -> m (Of Bool r)-notElem a' = loop True where- loop False str = liftM (False :>) (effects str)- loop True str = case str of- Return r -> return (True:> r)- Effect m -> m >>= loop True- Step (a:> rest) -> - if a == a' - then liftM (False :>) (effects rest)- else loop True rest-{-#INLINABLE notElem #-}--notElem_ :: (Monad m, Eq a) => a -> Stream (Of a) m r -> m Bool-notElem_ a' = loop True where- loop False str = return False- loop True str = case str of- Return r -> return True- Effect m -> m >>= loop True- Step (a:> rest) -> - if a == a' - then return False - else loop True rest-{-#INLINABLE notElem_ #-}---{-| -> filter p = hoist effects (partition p)- - -}-partition :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r-partition thus = loop where- loop str = case str of - Return r -> Return r- Effect m -> Effect (liftM loop (lift m))- Step (a :> rest) -> if thus a - then Step (a :> loop rest)- else Effect $ do- yield a- return (loop rest)---{-| Separate left and right values in distinct streams. ('separate' is - a more powerful, functor-general, equivalent using 'Sum' in place of 'Either'). - So, for example, to permit unlimited user- input of @Int@s on condition of only two errors, we might write:-->>> S.toList $ S.print $ S.take 2 $ partitionEithers $ S.map readEither $ S.stdinLn :: IO (Of [Int] ())-1<Enter>-2<Enter>-qqqqqqqqqq<Enter>-"Prelude.read: no parse"-3<Enter>-rrrrrrrrrr<Enter>-"Prelude.read: no parse"-[1,2,3] :> ()--> partitionEithers = separate . maps S.eitherToSum -> lefts = hoist S.effects . partitionEithers-> rights = S.effects . partitionEithers-> rights = S.concat --}-partitionEithers :: Monad m => Stream (Of (Either a b)) m r -> Stream (Of a) (Stream (Of b) m) r-partitionEithers = loop where- loop str = case str of - Return r -> Return r- Effect m -> Effect (liftM loop (lift m))- Step (Left a :> rest) -> Step (a :> loop rest)- Step (Right b :> rest) -> Effect $ do- yield b- return (loop rest)----- | Fold a 'Stream' of numbers into their product-product_ :: (Monad m, Num a) => Stream (Of a) m () -> m a-product_ = fold_ (*) 1 id-{-# INLINE product_ #-}--{-| Fold a 'Stream' of numbers into their product with the return value--> maps' product' :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r--}-product :: (Monad m, Num a) => Stream (Of a) m r -> m (Of a r)-product = fold (*) 1 id-{-# INLINE product #-}----- ------------------ read--- -----------------{- | Make a stream of strings into a stream of parsed values, skipping bad cases-->>> S.sum_ $ S.read $ S.takeWhile (/= "total") S.stdinLn :: IO Int-1000<Enter>-2000<Enter>-total<Enter>-3000----}-read :: (Monad m, Read a) => Stream (Of String) m r -> Stream (Of a) m r-read stream = for stream $ \str -> case readMaybe str of - Nothing -> return ()- Just r -> yield r-{-# INLINE read #-}---- ------------------ repeat--- ----------------{-| Repeat an element /ad inf./ .-->>> S.print $ S.take 3 $ S.repeat 1-1-1-1--}--repeat :: Monad m => a -> Stream (Of a) m r-repeat a = loop where loop = Effect (return (Step (a :> loop)))-{-# INLINE repeat #-}---{-| Repeat a monadic action /ad inf./, streaming its results.-->>> S.toList $ S.take 2 $ repeatM getLine-one<Enter>-two<Enter>-["one","two"]--}--repeatM :: Monad m => m a -> Stream (Of a) m r-repeatM ma = loop where- loop = do - a <- lift ma - yield a - loop-{-# INLINABLE repeatM #-}---- ------------------ replicate --- ------------------- | Repeat an element several times-replicate :: Monad m => Int -> a -> Stream (Of a) m ()-replicate n a | n <= 0 = return ()-replicate n a = loop n where- loop 0 = Return ()- loop m = Effect (return (Step (a :> loop (m-1))))-{-# INLINABLE replicate #-}--{-| Repeat an action several times, streaming the results.-->>> S.print $ S.replicateM 2 getCurrentTime-2015-08-18 00:57:36.124508 UTC-2015-08-18 00:57:36.124785 UTC---}-replicateM :: Monad m => Int -> m a -> Stream (Of a) m ()-replicateM n ma | n <= 0 = return () -replicateM n ma = loop n where - loop 0 = Return ()- loop n = Effect $ do - a <- ma - return (Step $ a :> loop (n-1))-{-# INLINABLE replicateM #-}--{-| Read an @IORef (Maybe a)@ or a similar device until it reads @Nothing@.- @reread@ provides convenient exit from the @io-streams@ library--> reread readIORef :: IORef (Maybe a) -> Stream (Of a) IO ()-> reread Streams.read :: System.IO.Streams.InputStream a -> Stream (Of a) IO ()--}-reread :: Monad m => (s -> m (Maybe a)) -> s -> Stream (Of a) m ()-reread step s = loop where - loop = Effect $ do - m <- step s- case m of - Nothing -> return (Return ())- Just a -> return (Step (a :> loop))-{-# INLINABLE reread #-}--{-| Strict left scan, streaming, e.g. successive partial results.--->>> S.print $ S.scan (++) "" id $ each (words "a b c d")-""-"a"-"ab"-"abc"-"abcd"-- 'scan' is fitted for use with @Control.Foldl@, thus:-->>> S.print $ L.purely S.scan L.list $ each [3..5]-[]-[3]-[3,4]-[3,4,5]---}-scan :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> Stream (Of b) m r-scan step begin done = loop begin- where- loop !acc stream = do- case stream of - Return r -> Step (done acc :> Return r)- Effect m -> Effect (liftM (loop acc) m)- Step (a :> rest) -> Step (done acc :> loop (step acc a) rest)-{-#INLINABLE scan #-}--{-| Strict left scan, accepting a monadic function. It can be used with- 'FoldM's from @Control.Foldl@ using 'impurely'. Here we yield- a succession of vectors each recording -->>> let v = L.impurely scanM L.vector $ each [1..4::Int] :: Stream (Of (U.Vector Int)) IO ()->>> S.print v-fromList []-fromList [1]-fromList [1,2]-fromList [1,2,3]-fromList [1,2,3,4]---}-scanM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r -> Stream (Of b) m r-scanM step begin done str = do- x <- lift begin- loop x str- where- loop !x stream = do - b <- lift (done x)- yield b- case stream of - Return r -> Return r- Effect m -> Effect (do - stream' <- m- return (loop x stream')- )- Step (a :> rest) -> Effect (do- x' <- step x a- return (loop x' rest)- )-{-# INLINABLE scanM #-}--{- Label each element in a stream with a value accumulated according to a fold.--->>> S.print $ S.scanned (*) 1 id $ S.each [100,200,300]-(100,100)-(200,20000)-(300,6000000)-->>> S.print $ L.purely S.scanned L.product $ S.each [100,200,300]-(100,100)-(200,20000)-(300,6000000)---}--data Maybe' a = Just' a | Nothing'--scanned :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> Stream (Of (a,b)) m r-scanned step begin done = loop Nothing' begin- where- loop !m !x stream = do - case stream of - Return r -> return r- Effect mn -> Effect $ liftM (loop m x) mn- Step (a :> rest) -> do- case m of - Nothing' -> do - let !acc = step x a- yield (a, done acc)- loop (Just' a) acc rest- Just' _ -> do- let !acc = done (step x a)- yield (a, acc) - loop (Just' a) (step x a) rest-{-# INLINABLE scanned #-}---{-| Streams the number of seconds from the beginning of action- - Thus, to mark times of user input we might write something like:-->>> S.toList $ S.take 3 $ S.zip S.seconds S.stdinLn -a<Enter>-b<Enter>-c<Enter>-[(0.0,"a"),(1.088711,"b"),(3.7289649999999996,"c")] :> ()- - To restrict user input to some number of seconds, we might write:- ->>> S.toList $ S.map fst $ S.zip S.stdinLn $ S.takeWhile (< 3) S.seconds-one<Enter>-two<Enter>-three<Enter>-four<Enter>-five<Enter>-["one","two","three","four","five"] :> ()-- This is of course does not interrupt an action that has already begun.-- -}- -seconds :: Stream (Of Double) IO r-seconds = do - e <- lift $ next preseconds- case e of- Left r -> return r- Right (t, rest) -> do- yield 0- map (subtract t) rest- where - preseconds :: Stream (Of Double) IO r- preseconds = do- utc <- liftIO getCurrentTime- map ((/1000000000) . nice utc) (repeatM getCurrentTime)- where- nice u u' = fromIntegral $ truncate (1000000000 * diffUTCTime u' u)---- ------------------ sequence--- -----------------{-| Like the 'Data.List.sequence' but streaming. The result type is a- stream of a\'s, /but is not accumulated/; the effects of the elements- of the original stream are interleaved in the resulting stream. Compare:--> sequence :: Monad m => [m a] -> m [a]-> sequence :: Monad m => Stream (Of (m a)) m r -> Stream (Of a) m r-- This obeys the rule---}-sequence :: Monad m => Stream (Of (m a)) m r -> Stream (Of a) m r-sequence = loop where- loop stream = case stream of- Return r -> Return r- Effect m -> Effect $ liftM loop m- Step (ma :> rest) -> Effect $ do- a <- ma- return (Step (a :> loop rest))-{-# INLINABLE sequence #-}---- ------------------ show--- -----------------show :: (Monad m, Show a) => Stream (Of a) m r -> Stream (Of String) m r-show = map Prelude.show-{-# INLINE show #-}--- ------------------ sum --- ------------------- | Fold a 'Stream' of numbers into their sum-sum_ :: (Monad m, Num a) => Stream (Of a) m () -> m a-sum_ = fold_ (+) 0 id-{-# INLINE sum_ #-}--{-| Fold a 'Stream' of numbers into their sum with the return value--> mapped S.sum :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r--->>> S.sum $ each [1..10]-55 :> ()-->>> (n :> rest) <- S.sum $ S.splitAt 3 $ each [1..10]->>> print n-6->>> (m :> rest') <- S.sum $ S.splitAt 3 rest->>> print m-15->>> S.print rest'-7-8-9---}-sum :: (Monad m, Num a) => Stream (Of a) m r -> m (Of a r)-sum = fold (+) 0 id-{-# INLINABLE sum #-}---- ------------------ span--- ------------------- | Stream elements until one fails the condition, return the rest.-span :: Monad m => (a -> Bool) -> Stream (Of a) m r - -> Stream (Of a) m (Stream (Of a) m r)-span pred = loop where- loop str = case str of - Return r -> Return (Return r)- Effect m -> Effect $ liftM loop m- Step (a :> rest) -> if pred a - then Step (a :> loop rest)- else Return (Step (a :> rest))-{-# INLINABLE span #-}-- -{-| Split a stream of elements wherever a given element arises.- The action is like that of 'Prelude.words'. -->>> S.stdoutLn $ mapped S.toList $ S.split ' ' $ each "hello world "-hello-world---}--split :: (Eq a, Monad m) =>- a -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r-split t = loop where- loop stream = case stream of - Return r -> Return r- Effect m -> Effect (liftM loop m)- Step (a :> rest) -> - if a /= t- then Step (fmap loop (yield a >> break (== t) rest))- else loop rest-{-#INLINABLE split #-}--{-| Split a succession of layers after some number, returning a streaming or--- effectful pair. This function is the same as the 'splitsAt' exported by the--- @Streaming@ module, but since this module is imported qualified, it can --- usurp a Prelude name. It specializes to:--> splitAt :: (Monad m, Functor f) => Int -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r)---}-splitAt :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Stream f m r)-splitAt = splitsAt-{-# INLINE splitAt #-}---- ---------- subst--- --------{-| Replace each element in a stream of individual values with a functorial- layer of any sort. @subst = flip with@ and is more convenient in - a sequence of compositions that transform a stream.- -> with = flip subst-> for str f = concats $ subst f str-> subst f = maps (\(a:>r) -> r <$ f a)-> S.concat = concats . subst each--}-subst :: (Monad m, Functor f) => (a -> f x) -> Stream (Of a) m r -> Stream f m r-subst f s = loop s where- loop str = case str of - Return r -> Return r- Effect m -> Effect (liftM loop m)- Step (a :> rest) -> Step (loop rest <$ f a)-{-#INLINABLE subst #-}--- ------------------ take--- -----------------{-| End a stream after n elements; the original return value is thus lost.- 'splitAt' preserves this information. Note that, like @splitAt@, this- function is functor-general, so that, for example, you can @take@ not- just a number of items from a stream of elements, but a number - of substreams and the like.-->>> S.toList $ S.take 3 $ each "with"-"wit" :> ()-->>> runResourceT $ S.stdoutLn $ S.take 3 $ S.readFile "stream.hs"-import Streaming -import qualified Streaming.Prelude as S-import Streaming.Prelude (each, next, yield)----}--take :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m ()-take n0 str | n0 <= 0 = return ()-take n0 str = loop n0 str where- loop 0 p = return ()- loop n p = - case p of Step fas -> Step (fmap (loop (n-1)) fas)- Effect m -> Effect (liftM (loop n) m)- Return r -> Return ()-{-# INLINABLE take #-}---- ------------------ takeWhile--- -----------------{-| End stream when an element fails a condition; the original return value is lost.- By contrast 'span' preserves this information, and is generally more desirable.--> S.takeWhile thus = void . S.span thus-- To preserve the information - but thus also force the rest of the stream to be - developed - write --> S.drained . S.span thus-- as @dropWhile thus@ is--> S.effects . S.span thus---}-takeWhile :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m ()-takeWhile pred = loop where- loop str = case str of - Step (a :> as) -> when (pred a) (Step (a :> loop as))- Effect m -> Effect (liftM loop m)- Return r -> Return ()-{-# INLINE takeWhile #-}---{-| Convert an effectful 'Stream (Of a)' into a list of @as@-- Note: Needless to say, this function does not stream properly.- It is basically the same as Prelude 'mapM' which, like 'replicateM',- 'sequence' and similar operations on traversable containers- is a leading cause of space leaks.- --}-toList_ :: Monad m => Stream (Of a) m () -> m [a]-toList_ = fold_ (\diff a ls -> diff (a: ls)) id (\diff -> diff [])-{-# INLINE toList_ #-}---{-| Convert an effectful 'Stream' into a list alongside the return value--> mapped toList :: Stream (Stream (Of a)) m r -> Stream (Of [a]) m -- Like 'toList_', it breaks streaming; unlike 'toList_' it preserves- the return value and thus is frequently useful with e.g. 'mapped'-->>> S.print $ mapped S.toList $ chunksOf 3 $ each [1..9]-[1,2,3]-[4,5,6]-[7,8,9]---}-toList :: Monad m => Stream (Of a) m r -> m (Of [a] r)-toList = fold (\diff a ls -> diff (a: ls)) id (\diff -> diff [])-{-# INLINE toList #-}---{-| Inspect the first item in a stream of elements, without a return value. - @uncons@ provides convenient exit into another streaming type:--> IOStreams.unfoldM uncons :: Stream (Of a) IO b -> IO (InputStream a)-> Conduit.unfoldM uncons :: Stream (Of a) m r -> Conduit.Source m a---}-uncons :: Monad m => Stream (Of a) m () -> m (Maybe (a, Stream (Of a) m ()))-uncons = loop where- loop stream = case stream of- Return () -> return Nothing- Effect m -> m >>= loop- Step (a :> rest) -> return (Just (a,rest))-{-# INLINABLE uncons #-}---{-| Build a @Stream@ by unfolding steps starting from a seed. -- The seed can of course be anything, but this is one natural way - to consume a @pipes@ 'Pipes.Producer'. Consider:-->>> S.stdoutLn $ S.take 2 $ S.unfoldr P.next P.stdinLn-hello<Enter>-hello-goodbye<Enter>-goodbye-->>> S.stdoutLn $ S.unfoldr P.next (P.stdinLn P.>-> P.take 2)-hello<Enter>-hello-goodbye<Enter>-goodbye-->>> S.effects $ S.unfoldr P.next (P.stdinLn P.>-> P.take 2 P.>-> P.stdoutLn)-hello<Enter>-hello-goodbye<Enter>-goodbye---}-unfoldr :: Monad m - => (s -> m (Either r (a, s))) -> s -> Stream (Of a) m r-unfoldr step = loop where- loop s0 = Effect (do - e <- step s0- case e of- Left r -> return (Return r)- Right (a,s) -> return (Step (a :> loop s)))-{-# INLINABLE unfoldr #-}---- ------------------------------------------ untilRight--- ----------------------------------------untilRight :: Monad m => m (Either a r) -> Stream (Of a) m r-untilRight act = Effect loop where- loop = do - e <- act- case e of- Right r -> return (Return r)- Left a -> return (Step (a :> Effect loop))-{-#INLINABLE untilRight #-}- --- ------------------------------------------ with--- -----------------------------------------{-| Replace each element in a stream of individual Haskell values (a @Stream (Of a) m r@) with an associated 'functorial' step. - -> for str f = concats (with str f) -> with str f = for str (yields . f)-> with str f = maps (\(a:>r) -> r <$ f a) str-> with = flip subst-> subst = flip with-->>> with (each [1..3]) (yield . show) & intercalates (yield "--") & S.stdoutLn-1----2----3- -}-with :: (Monad m, Functor f) => Stream (Of a) m r -> (a -> f x) -> Stream f m r-with s f = loop s where- loop str = case str of - Return r -> Return r- Effect m -> Effect (liftM loop m)- Step (a :> rest) -> Step (loop rest <$ f a)-{-#INLINABLE with #-}- --- ------------------------------------------ yield--- -----------------------------------------{-| A singleton stream-->>> stdoutLn $ yield "hello"-hello-->>> S.sum $ do {yield 1; yield 2}-3- ->>> let prompt = putStrLn "Enter a number:" ->>> let number = lift (prompt >> readLn) >>= yield :: Stream (Of Int) IO ()->>> S.toList $ do {number; number; number}-Enter a number:-1-Enter a number:-2-Enter a number:-3-[1,2,3] :> ()---}--yield :: Monad m => a -> Stream (Of a) m ()-yield a = Step (a :> Return ())-{-# INLINE yield #-}---- | Zip two 'Streams's -zip :: Monad m- => (Stream (Of a) m r)- -> (Stream (Of b) m r)- -> (Stream (Of (a,b)) m r)-zip = zipWith (,)-{-# INLINE zip #-}---- | Zip two 'Streams's using the provided combining function-zipWith :: Monad m- => (a -> b -> c)- -> (Stream (Of a) m r)- -> (Stream (Of b) m r)- -> (Stream (Of c) m r)-zipWith f = loop- where- loop str0 str1 = case str0 of- Return r -> Return r- Effect m -> Effect $ liftM (\str -> loop str str1) m - Step (a :> rest0) -> case str1 of- Return r -> Return r- Effect m -> Effect $ liftM (loop str0) m- Step (b :> rest1) -> Step (f a b :>loop rest0 rest1)-{-# INLINABLE zipWith #-}----- | Zip three 'Stream's with a combining function-zipWith3 :: Monad m =>- (a -> b -> c -> d)- -> Stream (Of a) m r- -> Stream (Of b) m r- -> Stream (Of c) m r- -> Stream (Of d) m r-zipWith3 op = loop where- loop str0 str1 str2 = do- e0 <- lift (next str0)- case e0 of - Left r0 -> return r0- Right (a0,rest0) -> do - e1 <- lift (next str1)- case e1 of- Left r1 -> return r1- Right (a1,rest1) -> do - e2 <- lift (next str2)- case e2 of- Left r2 -> return r2- Right (a2,rest2) -> do - yield (op a0 a1 a2)- loop rest0 rest1 rest2-{-# INLINABLE zipWith3 #-} - - --- | Zip three streams together -zip3 :: Monad m- => (Stream (Of a) m r)- -> (Stream (Of b) m r)- -> (Stream (Of c) m r)- -> (Stream (Of (a,b,c)) m r)-zip3 = zipWith3 (,,)-{-# INLINABLE zip3 #-}---- ----------------- IO fripperies --- ----------------{-| View standard input as a 'Stream (Of String) m r'. 'stdoutLn', by- contrast, renders a 'Stream (Of String) m r' to standard output. The names- follow @Pipes.Prelude@-->>> stdoutLn stdinLn -hello<Enter>-hello-world<Enter>-world-^CInterrupted.--->>> stdoutLn $ S.map reverse stdinLn -hello<Enter>-olleh-world<Enter>-dlrow-^CInterrupted.---}-stdinLn :: MonadIO m => Stream (Of String) m ()-stdinLn = fromHandle IO.stdin-{-# INLINABLE stdinLn #-}--{-| Read values from 'IO.stdin', ignoring failed parses-->>> S.sum_ $ S.take 2 S.readLn :: IO Int-10<Enter>-12<Enter>-22-->>> S.toList $ S.take 3 (S.readLn :: Stream (Of Int) IO ())-1<Enter>-2<Enter>-1@#$%^&*\<Enter>-3<Enter>-[1,2,3] :> ()---}--readLn :: (MonadIO m, Read a) => Stream (Of a) m ()-readLn = for stdinLn $ \str -> case readMaybe str of - Nothing -> return ()- Just n -> yield n-{-# INLINABLE readLn #-}---{-| Read 'String's from a 'IO.Handle' using 'IO.hGetLine'-- Terminates on end of input-->>> IO.withFile "/usr/share/dict/words" IO.ReadMode $ S.stdoutLn . S.take 3 . S.drop 50000 . S.fromHandle-deflagrator-deflate-deflation---}-fromHandle :: MonadIO m => IO.Handle -> Stream (Of String) m ()-fromHandle h = go- where- go = do- eof <- liftIO $ IO.hIsEOF h- unless eof $ do- str <- liftIO $ IO.hGetLine h- yield str- go-{-# INLINABLE fromHandle #-} --{-| Write a succession of strings to a handle as separate lines.-->>> S.toHandle IO.stdout $ each $ words "one two three"-one-two-three--}-toHandle :: MonadIO m => IO.Handle -> Stream (Of String) m r -> m r-toHandle handle = loop where- loop str = case str of- Return r -> return r- Effect m -> m >>= loop - Step (s :> rest) -> do - liftIO (IO.hPutStrLn handle s)- loop rest-{-# INLINABLE toHandle #-} --{-| Print the elements of a stream as they arise.-->>> S.print $ S.take 2 S.stdinLn -hello-"hello"-world-"world"->>> ---}-print :: (MonadIO m, Show a) => Stream (Of a) m r -> m r-print = loop where- loop stream = case stream of - Return r -> return r - Effect m -> m >>= loop- Step (a :> rest) -> do - liftIO (Prelude.print a)- loop rest---{-| Write 'String's to 'IO.stdout' using 'putStrLn'; terminates on a broken output pipe- (This operation is modelled on 'Pipes.Prelude.stdoutLn').-->>> S.stdoutLn $ S.take 3 $ S.each $ words "one two three four five"-one-two-three--}-stdoutLn :: MonadIO m => Stream (Of String) m () -> m ()-stdoutLn = loop- where- loop stream = case stream of - Return _ -> return () - Effect m -> m >>= loop- Step (s :> rest) -> do- x <- liftIO $ try (putStrLn s)- case x of- Left (G.IOError { G.ioe_type = G.ResourceVanished- , G.ioe_errno = Just ioe })- | Errno ioe == ePIPE- -> return ()- Left e -> liftIO (throwIO e)- Right () -> loop rest-{-# INLINABLE stdoutLn #-}-----{-| Write 'String's to 'IO.stdout' using 'putStrLn'-- This does not handle a broken output pipe, but has a polymorphic return- value, which makes this possible:-->>> rest <- S.stdoutLn' $ S.show $ S.splitAt 3 (each [1..5])-1-2-3->>> S.print rest-4-5---}--stdoutLn' :: MonadIO m => Stream (Of String) m r -> m r-stdoutLn' = loop where - loop stream = case stream of - Return r -> return r - Effect m -> m >>= loop- Step (s :> rest) -> liftIO (putStrLn s) >> loop rest-{-# INLINE stdoutLn' #-}--{-| Read the lines of a file as Haskell 'String's-->>> runResourceT $ S.writeFile "lines.txt" $ S.take 2 S.stdinLn-hello<Enter>-world<Enter>->>> runResourceT $ S.print $ S.readFile "lines.txt" -"hello"-"world"-- 'runResourceT', as it is used here, means something like 'closing_all_handles'.- It makes it possible to write convenient, fairly sensible versions of - 'readFile', 'writeFile' and 'appendFile'. @IO.withFile IO.ReadMode ...@- is more complicated but is generally to be preferred. Its use is explained - <https://www.fpcomplete.com/user/snoyberg/library-documentation/resourcet here>.---}--readFile :: MonadResource m => FilePath -> Stream (Of String) m ()-readFile f = bracketStream (IO.openFile f IO.ReadMode) (IO.hClose) fromHandle--{-| Write a series of strings as lines to a file. The handle is crudely - managed with 'ResourceT':-->>> runResourceT $ S.writeFile "lines.txt" $ S.take 2 S.stdinLn-hello<Enter>-world<Enter>->>> runResourceT $ S.print $ S.readFile "lines.txt" -"hello"-"world"---}-writeFile :: MonadResource m => FilePath -> Stream (Of String) m r -> m r-writeFile f str = do - (key, handle) <- allocate (IO.openFile f IO.WriteMode) (IO.hClose) - r <- toHandle handle str- release key- return r---- -- * Producers--- -- $producers--- stdinLn -- --- , readLn -- --- , fromHandle -- --- , repeatM -- --- , replicateM -------- -- * Consumers--- -- $consumers--- , stdoutLn ----- , stdoutLn' ----- , mapM_ ----- , print -- --- , toHandle ----- , effects -------- -- * Pipes--- -- $pipes--- , map -- --- , mapM ----- , sequence -- --- , mapFoldable -- --- , filter ----- , filterM ----- , take ----- , takeWhile ----- , takeWhile' ----- , drop ----- , dropWhile -- --- , concat ----- , elemIndices--- , findIndices--- , scan ----- , scanM ----- , chain ----- , read ----- , show -- --- , seq -------- -- * Folds--- -- $folds--- , fold ----- , fold' ----- , foldM ----- , foldM' ----- , all--- , any--- , and--- , or--- , elem--- , notElem--- , find--- , findIndex--- , head--- , index--- , last--- , length--- , maximum--- , minimum--- , null--- , sum ----- , product ----- , toList ----- , toListM ----- , toListM' -------- -- * Zips--- , zip ----- , zipWith -------distinguish :: (a -> Bool) -> Of a r -> Sum (Of a) (Of a) r-distinguish predicate (a :> b) = if predicate a then InR (a :> b) else InL (a :> b)-{-#INLINE distinguish #-}--sumToEither ::Sum (Of a) (Of b) r -> Of (Either a b) r -sumToEither s = case s of - InL (a :> r) -> Left a :> r - InR (b :> r) -> Right b :> r-{-#INLINE sumToEither #-}--eitherToSum :: Of (Either a b) r -> Sum (Of a) (Of b) r-eitherToSum s = case s of - Left a :> r -> InL (a :> r)- Right b :> r -> InR (b :> r)-{-#INLINE eitherToSum #-}--composeToSum :: Compose (Of Bool) f r -> Sum f f r-composeToSum x = case x of - Compose (True :> f) -> InR f- Compose (False :> f) -> InL f-{-#INLINE composeToSum #-}--sumToCompose :: Sum f f r -> Compose (Of Bool) f r -sumToCompose x = case x of- InR f -> Compose (True :> f) - InL f -> Compose (False :> f)-{-#INLINE sumToCompose #-}--{-| Store the result of any suitable fold over a stream, keeping the stream for- further manipulation. @store f = f . copy@ :-->>> S.print $ S.store S.product $ each [1..4]-1-2-3-4-24 :> ()-->>> S.print $ S.store S.sum $ S.store S.product $ each [1..4]-1-2-3-4-10 :> (24 :> ())-- Here the sum (10) and the product (24) have been \'stored\' for use when- finally we have traversed the stream with 'print' . Needless to say,- a second 'pass' is excluded conceptually, so the- folds that you apply successively with @store@ are performed- simultaneously, and in constant memory -- as they would be if,- say, you linked them together with @Control.Fold@:-->>> L.impurely S.foldM (liftA3 (\a b c -> (b,c)) (L.sink print) (L.generalize L.sum) (L.generalize L.product)) $ each [1..4]-1-2-3-4-(10,24) :> ()-- Fusing folds after the fashion of @Control.Foldl@ will generally be a bit faster- than the corresponding succession of uses of 'store', but by- constant factor that will be completely dwarfed when any IO is at issue.-- But 'store' / 'copy' is /much/ more powerful, as you can see by reflecting on- uses like this:-->>> S.sum $ S.store (S.sum . mapped S.product . chunksOf 2) $ S.store (S.product . mapped S.sum . chunksOf 2 )$ each [1..6]-21 :> (44 :> (231 :> ()))-- It will be clear that this cannot be reproduced with any combination of lenses,- @Control.Fold@ folds, or the like. (See also the discussion of 'copy'.)-- It would conceivable be clearer to import a series of specializations of 'store'.- It is intended to be used at types like these:--> storeM :: (forall s m . Monad m => Stream (Of a) m s -> m (Of b s))-> -> (Monad n => Stream (Of a) n r -> Stream (Of a) n (Of b r))-> storeM = store->-> storeMIO :: (forall s m . MonadIO m => Stream (Of a) m s -> m (Of b s))-> -> ( MonadIO n => Stream (Of a) n r -> Stream (Of a) n (Of b r)-> storeMIO = store-- It is clear from these types that we are just using the general instances:--> instance (Functor f, Monad m ) => Monad (Stream f m)-> instance (Functor f, MonadIO m) => MonadIO (Stream f m)-- We thus can't be touching the elements of the stream, or the final return value.- It it is the same with other constraints that @Stream (Of a)@ inherits,- like 'MonadResource'. Thus I can filter and write to one file, but- nub and write to another, or to a database or the like:-->>> runResourceT $ (S.writeFile "hello2.txt" . S.nub) $ store (S.writeFile "hello.txt" . S.filter (/= "world")) $ each ["hello", "world", "goodbye", "world"]->>> :! cat hello.txt-hello-goodbye->>> :! cat hello2.txt-hello-world-goodbye----}-store- :: Monad m =>- (Stream (Of a) (Stream (Of a) m) r -> t) -> Stream (Of a) m r -> t-store f x = f (copy x)-{-#INLINE store #-}--{-| Duplicate the content of stream, so that it can be acted on twice in different ways,- but without breaking streaming. Thus, with @each [1,2]@ I might do:-->>> S.print $ each ["one","two"]-"one"-"two"->>> S.stdoutLn $ each ["one","two"]-one-two-- With copy, I can as well do:-->>> S.print $ S.stdoutLn $ S.copy $ each ["one","two"]-one-"one"-two-"two"-- 'copy' should be understood together with 'effects' and is subject to the rules--> S.effects . S.copy = id-> hoist S.effects . S.copy = id-- The similar operations in 'Data.ByteString.Streaming' obey the same rules.-- Where the actions you are contemplating are each simple folds over- the elements, or a selection of elements, then the coupling of the- folds is often more straightforwardly effected with `Control.Foldl`,- e.g.-->>> L.purely S.fold (liftA2 (,) L.sum L.product) $ each [1..10]-(55,3628800) :> ()-- rather than-->>> S.sum $ S.product . S.copy $ each [1..10]-55 :> (3628800 :> ())-- A @Control.Foldl@ fold can be altered to act on a selection of elements by- using 'Control.Foldl.handles' on an appropriate lens. Some such- manipulations are simpler and more 'Data.List'-like, using 'copy':-->>> L.purely S.fold (liftA2 (,) (L.handles (filtered odd) L.sum) (L.handles (filtered even) L.product)) $ each [1..10]-(25,3840) :> ()-- becomes-->>> S.sum $ S.filter odd $ S.product $ S.filter even $ S.copy $ each [1..10]-25 :> (3840 :> ())-- or using 'store'-->>> S.sum $ S.filter odd $ S.store (S.product . S.filter even) $ each [1..10]-25 :> (3840 :> ())-- But anything that fold of a @Stream (Of a) m r@ into e.g. an @m (Of b r)@- that has a constraint on @m@ that is carried over into @Stream f m@ -- e.g. @Monad@, @MonadIO@, @MonadResource@, etc. can be used on the stream.- Thus, I can fold over different groupings of the original stream:-->>> (S.toList . mapped S.toList . chunksOf 5) $ (S.toList . mapped S.toList . chunksOf 3) $ S.copy $ each [1..10]-[[1,2,3,4,5],[6,7,8,9,10]] :> ([[1,2,3],[4,5,6],[7,8,9],[10]] :> ())-- The procedure can be iterated as one pleases, as one can see from this (otherwise unadvisable!) example:-->>> (S.toList . mapped S.toList . chunksOf 4) $ (S.toList . mapped S.toList . chunksOf 3) $ S.copy $ (S.toList . mapped S.toList . chunksOf 2) $ S.copy $ each [1..12]-[[1,2,3,4],[5,6,7,8],[9,10,11,12]] :> ([[1,2,3],[4,5,6],[7,8,9],[10,11,12]] :> ([[1,2],[3,4],[5,6],[7,8],[9,10],[11,12]] :> ()))---}-copy- :: Monad m =>- Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r-copy = Effect . return . loop where- loop str = case str of- Return r -> Return r- Effect m -> Effect (liftM loop (lift m))- Step (a :> rest) -> Effect (Step (a :> Return (Step (a :> loop rest))))- -{-#INLINABLE copy#-}--duplicate- :: Monad m =>- Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r-duplicate = copy-{-#INLINE duplicate #-}-- -{-| @copy'@ is the same as @copy@ but reverses the order of interleaved effects.- The difference should not be observable at all for pure folds over the data.---}-copy'- :: Monad m =>- Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r-copy' = Effect . return . loop where- loop str = case str of- Return r -> Return r- Effect m -> Effect (liftM loop (lift m))- Step (a :> rest) -> Step (a :> Effect (Step (a :> Return (loop rest)))) -{-#INLINABLE copy' #-}--duplicate'- :: Monad m =>- Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r-duplicate' = copy'-{-#INLINE duplicate' #-}--{-| The type--> Data.List.unzip :: [(a,b)] -> ([a],[b])-- might lead us to expect --> Streaming.unzip :: Stream (Of (a,b)) m r -> Stream (Of a) m (Stream (Of b) m r)- - which would not stream, since it would have to accumulate the second stream (of @b@s).- Of course, @Data.List@ 'Data.List.unzip' doesn't stream either. -- This @unzip@ does- stream, though of course you can spoil this by using e.g. 'toList':- ->>> let xs = map (\x-> (x,show x)) [1..5::Int]-->>> S.toList $ S.toList $ S.unzip (S.each xs)-["1","2","3","4","5"] :> ([1,2,3,4,5] :> ())-->>> Prelude.unzip xs-([1,2,3,4,5],["1","2","3","4","5"])-- Note the difference of order in the results. It may be of some use to think why.- The first application of 'toList' was applied to a stream of integers:-->>> :t S.unzip $ S.each xs-S.unzip $ S.each xs :: Monad m => Stream (Of Int) (Stream (Of String) m) ()-- Like any fold, 'toList' takes no notice of the monad of effects.--> toList :: Monad m => Stream (Of a) m r -> m (Of [a] r)- - In the case at hand (since I am in @ghci@) @m = Stream (Of String) IO@. - So when I apply 'toList', I exhaust that stream of integers, folding - it into a list:-->>> :t S.toList $ S.unzip $ S.each xs-S.toList $ S.unzip $ S.each xs- :: Monad m => Stream (Of String) m (Of [Int] ())-- When I apply 'toList' to /this/, I reduce everything to an ordinary action in @IO@,- and return a list of strings:-->>> S.toList $ S.toList $ S.unzip (S.each xs)-["1","2","3","4","5"] :> ([1,2,3,4,5] :> ())---}-unzip :: Monad m => Stream (Of (a,b)) m r -> Stream (Of a) (Stream (Of b) m) r-unzip = loop where- loop str = case str of - Return r -> Return r- Effect m -> Effect (liftM loop (lift m))- Step ((a,b):> rest) -> Step (a :> Effect (Step (b :> Return (loop rest))))-{-#INLINABLE unzip #-}----- "fold/map" forall step begin done f str .--- fold step begin done (map f str) = fold (\x a -> step x $! f a) begin done str;------ "fold/filter" forall step begin done pred str .--- fold step begin done (filter pred str) = fold (\x a -> if pred a then step x a else x) begin done str;------ "scan/map" forall step begin done f str .--- scan step begin done (map f str) = scan (\x a -> step x $! f a) begin done str-----{- $maybes- These functions discard the 'Nothing's that they encounter. They are analogous - to the functions from @Data.Maybe@ that share their names.--}--{-| The 'catMaybes' function takes a 'Stream' of 'Maybe's and returns- a 'Stream' of all of the 'Just' values.--}-catMaybes :: Monad m => Stream (Of (Maybe a)) m r -> Stream (Of a) m r -catMaybes = loop where - loop stream = case stream of - Return r -> Return r - Effect m -> Effect (liftM loop m) - Step (ma :> snext) -> case ma of - Nothing -> loop snext - Just a -> Step (a :> loop snext) -{-#INLINABLE catMaybes #-}--{-| The 'mapMaybe' function is a version of 'map' which can throw out elements. In particular, - the functional argument returns something of type @'Maybe' b@. If this is 'Nothing', no element - is added on to the result 'Stream'. If it is @'Just' b@, then @b@ is included in the result 'Stream'.--}-mapMaybe :: Monad m => (a -> Maybe b) -> Stream (Of a) m r -> Stream (Of b) m r -mapMaybe phi = loop where - loop stream = case stream of - Return r -> Return r - Effect m -> Effect (liftM loop m) - Step (a :> snext) -> case phi a of - Nothing -> loop snext - Just b -> Step (b :> loop snext) -{-#INLINABLE mapMaybe #-}-
+ src/Streaming.hs view
@@ -0,0 +1,176 @@+{-#LANGUAGE RankNTypes, CPP, Trustworthy #-}+module Streaming+ (+ -- * An iterable streaming monad transformer+ -- $stream+ Stream,+ -- * Constructing a 'Stream' on a given functor+ yields,+ effect,+ wrap,+ replicates,+ repeats,+ repeatsM,+ unfold,+ never,+ untilJust,+ streamBuild,+ delays,++ -- * Transforming streams+ maps,+ mapsM,+ mapped,+ distribute,+ groups,++ -- * Inspecting a stream+ inspect,++ -- * Splitting and joining 'Stream's+ splitsAt,+ takes,+ chunksOf,+ concats,+ intercalates,+ cutoff,+ -- period,+ -- periods,+++ -- * Zipping, unzipping, separating and unseparating streams+ zipsWith,+ zips,+ unzips,+ interleaves,+ separate,+ unseparate,+ decompose,+++ -- * Eliminating a 'Stream'+ mapsM_,+ run,+ streamFold,+ iterTM,+ iterT,+ destroy,++ -- * Base functor for streams of individual items+ Of (..),+ lazily,+ strictly,++ -- * ResourceT help++ bracketStream,++ -- * re-exports+ MFunctor(..),+ MMonad(..),+ MonadTrans(..),+ MonadIO(..),+ Compose(..),+ Sum(..),+ Identity(..),+ Alternative((<|>)),+ MonadThrow(..),+ MonadResource(..),+ MonadBase(..),+ ResourceT(..),+ runResourceT,+#if MIN_VERSION_base(4,8,0)+ Bifunctor(..),+#endif++ join,+ liftM,+ liftM2,+ liftA2,+ liftA3,+ void,+ (<>)+ )+ where+import Streaming.Internal+import Streaming.Prelude+import Control.Monad.Morph+import Control.Monad+import Data.Monoid ((<>))+import Control.Applicative+import Control.Monad.Trans+import Data.Functor.Compose+import Data.Functor.Sum+import Data.Functor.Identity++import Control.Monad.Base+import Control.Monad.Trans.Resource+#if MIN_VERSION_base(4,8,0)+import Data.Bifunctor+#endif++{- $stream++ The 'Stream' data type can be used to represent any effectful+ succession of steps arising in some monad.+ The form of the steps is specified by the first (\"functor\")+ parameter in @Stream f m r@. The monad of the underlying effects+ is expressed by the second parameter.++ This module exports combinators that pertain to that general case.+ Some of these are quite abstract and pervade any use of the library,+ e.g.++> maps :: (forall x . f x -> g x) -> Stream f m r -> Stream g m r+> mapped :: (forall x . f x -> m (g x)) -> Stream f m r -> Stream g m r+> hoist :: (forall x . m x -> n x) -> Stream f m r -> Stream f n r -- from the MFunctor instance+> concats :: Stream (Stream f m) m r -> Stream f m r ++ (assuming here and thoughout that @m@ or @n@ satisfies a @Monad@ constraint, and+ @f@ or @g@ a @Functor@ constraint.)++ Others are surprisingly determinate in content:++> chunksOf :: Int -> Stream f m r -> Stream (Stream f m) m r+> splitsAt :: Int -> Stream f m r -> Stream f m (Stream f m r)+> zipsWith :: (forall x y. f x -> g y -> h (x, y)) -> Stream f m r -> Stream g m r -> Stream h m r+> intercalates :: Stream f m () -> Stream (Stream f m) m r -> Stream f m r+> unzips :: Stream (Compose f g) m r -> Stream f (Stream g m) r+> separate :: Stream (Sum f g) m r -> Stream f (Stream g) m r -- cp. partitionEithers+> unseparate :: Stream f (Stream g) m r -> Stream (Sum f g) m r+> groups :: Stream (Sum f g) m r -> Stream (Sum (Stream f m) (Stream g m)) m r++ One way to see that /any/ streaming library needs some such general type is+ that it is required to represent the segmentation of a stream, and to+ express the equivalents of @Prelude/Data.List@ combinators that involve+ 'lists of lists' and the like. See for example this+ <http://www.haskellforall.com/2013/09/perfect-streaming-using-pipes-bytestring.html post>+ on the correct expression of a streaming \'lines\' function.++ The module @Streaming.Prelude@ exports combinators relating to++> Stream (Of a) m r++ where @Of a r = !a :> r@ is a left-strict pair.+++ This expresses the concept of a 'Producer' or 'Source' or 'Generator' and+ easily inter-operates with types with such names in e.g. 'conduit',+ 'iostreams' and 'pipes'.+-}++{-| Map a stream to its church encoding; compare @Data.List.foldr@+ This is the @safe_destroy@ exported by the @Internal@ module.++ Typical @FreeT@ operators can be defined in terms of @destroy@+ e.g.++> iterT :: (Functor f, Monad m) => (f (m a) -> m a) -> Stream f m a -> m a+> iterT out stream = destroy stream out join return+> iterTM :: (Functor f, Monad m, MonadTrans t, Monad (t m)) => (f (t m a) -> t m a) -> Stream f m a -> t m a+> iterTM out stream = destroy stream out (join . lift) return+> concats :: (Monad m, MonadTrans t, Monad (t m)) => Stream (t m) m a -> t m a+> concats stream = destroy stream join (join . lift) return+-}++
+ src/Streaming/Internal.hs view
@@ -0,0 +1,1158 @@+{-# LANGUAGE RankNTypes, StandaloneDeriving,DeriveDataTypeable, BangPatterns #-}+{-# LANGUAGE UndecidableInstances, CPP, FlexibleInstances, MultiParamTypeClasses #-}+{-#LANGUAGE Trustworthy #-}+module Streaming.Internal (+ -- * The free monad transformer+ -- $stream+ Stream (..)+ + -- * Introducing a stream+ , unfold+ , replicates+ , repeats+ , repeatsM+ , effect+ , wrap+ , yields+ , streamBuild+ , cycles+ , delays+ , never+ , untilJust+ + -- * Eliminating a stream+ , intercalates+ , concats+ , iterT+ , iterTM+ , destroy+ , streamFold+ + -- * Inspecting a stream wrap by wrap+ , inspect+ + -- * Transforming streams+ , maps+ , mapsM+ , decompose+ , mapsM_+ , run+ , distribute+ , groups+-- , groupInL+ + -- * Splitting streams+ , chunksOf+ , splitsAt+ , takes+ , cutoff+ -- , period+ -- , periods+ + -- * Zipping and unzipping streams+ , zipsWith+ , zips+ , unzips+ , interleaves+ , separate+ , unseparate++ + -- * Assorted Data.Functor.x help+ + , switch+ + -- * ResourceT help+ + , bracketStream+ + -- * For use in implementation+ , unexposed+ , hoistExposed+ , mapsExposed+ , mapsMExposed+ , destroyExposed+ + ) where++import Control.Monad+import Control.Monad.Trans+import Control.Monad.Trans.Class+import Control.Monad.Reader.Class+import Control.Monad.Writer.Class+import Control.Monad.State.Class+import Control.Monad.Error.Class+import Control.Monad.Cont.Class+import Control.Applicative+import Data.Foldable ( Foldable(..) )+import Data.Traversable+import Control.Monad.Morph+import Data.Monoid (Monoid (..), (<>))+import Data.Functor.Identity+import Data.Data ( Data, Typeable )+import Prelude hiding (splitAt)+import Data.Functor.Compose+import Data.Functor.Sum+import Control.Concurrent (threadDelay)+import Control.Monad.Base+import Control.Monad.Trans.Resource+import Control.Monad.Catch (MonadCatch (..))+import Control.Monad.Trans.Control+++{- $stream++ The 'Stream' data type is equivalent to @FreeT@ and can represent any effectful+ succession of steps, where the form of the steps or 'commands' is+ specified by the first (functor) parameter.++> data Stream f m r = Step !(f (Stream f m r)) | Effect (m (Stream f m r)) | Return r++ The /producer/ concept uses the simple functor @ (a,_) @ \- or the stricter+ @ Of a _ @. Then the news at each step or layer is just: an individual item of type @a@.+ Since @Stream (Of a) m r@ is equivalent to @Pipe.Producer a m r@, much of+ the @pipes@ @Prelude@ can easily be mirrored in a @streaming@ @Prelude@. Similarly,+ a simple @Consumer a m r@ or @Parser a m r@ concept arises when the base functor is+ @ (a -> _) @ . @Stream ((->) input) m result@ consumes @input@ until it returns a+ @result@.++ To avoid breaking reasoning principles, the constructors+ should not be used directly. A pattern-match should go by way of 'inspect' \+ \- or, in the producer case, 'Streaming.Prelude.next'+ The constructors are exported by the 'Internal' module.+-}+data Stream f m r = Step !(f (Stream f m r))+ | Effect (m (Stream f m r))+ | Return r+#if __GLASGOW_HASKELL__ >= 710+ deriving (Typeable)+#endif+deriving instance (Show r, Show (m (Stream f m r))+ , Show (f (Stream f m r))) => Show (Stream f m r)+deriving instance (Eq r, Eq (m (Stream f m r))+ , Eq (f (Stream f m r))) => Eq (Stream f m r)+#if __GLASGOW_HASKELL__ >= 710+deriving instance (Typeable f, Typeable m, Data r, Data (m (Stream f m r))+ , Data (f (Stream f m r))) => Data (Stream f m r)+#endif+instance (Functor f, Monad m) => Functor (Stream f m) where+ fmap f = loop where+ loop stream = case stream of+ Return r -> Return (f r)+ Effect m -> Effect (do {stream' <- m; return (loop stream')})+ Step f -> Step (fmap loop f)+ {-# INLINABLE fmap #-}+ a <$ stream0 = loop stream0 where+ loop stream = case stream of+ Return r -> Return a+ Effect m -> Effect (do {stream' <- m; return (loop stream')})+ Step f -> Step (fmap loop f)+ {-# INLINABLE (<$) #-} ++instance (Functor f, Monad m) => Monad (Stream f m) where+ return = Return+ {-# INLINE return #-}+ stream1 >> stream2 = loop stream1 where+ loop stream = case stream of+ Return _ -> stream2+ Effect m -> Effect (liftM loop m)+ Step f -> Step (fmap loop f) + {-# INLINABLE (>>) #-}+ -- (>>=) = _bind+ -- {-#INLINE (>>=) #-}+ --+ stream >>= f =+ loop stream where+ loop stream0 = case stream0 of+ Step fstr -> Step (fmap loop fstr)+ Effect m -> Effect (liftM loop m)+ Return r -> f r+ {-# INLINABLE (>>=) #-} ++ fail = lift . fail+ {-#INLINE fail #-}+++-- _bind+-- :: (Functor f, Monad m)+-- => Stream f m r+-- -> (r -> Stream f m s)+-- -> Stream f m s+-- _bind p0 f = go p0 where+-- go p = case p of+-- Step fstr -> Step (fmap go fstr)+-- Effect m -> Effect (m >>= \s -> return (go s))+-- Return r -> f r+-- {-#INLINABLE _bind #-}+--+-- see https://github.com/Gabriel439/Haskell-Pipes-Library/pull/163+-- for a plan to delay inlining and manage interaction with other operations.++-- {-# RULES+ -- "_bind (Step fstr) f" forall fstr f .+ -- _bind (Step fstr) f = Step (fmap (\p -> _bind p f) fstr);+ -- "_bind (Effect m) f" forall m f .+ -- _bind (Effect m) f = Effect (m >>= \p -> return (_bind p f));+ -- "_bind (Return r) f" forall r f .+ -- _bind (Return r) f = f r;+-- #-}++instance (Functor f, Monad m) => Applicative (Stream f m) where+ pure = Return+ {-# INLINE pure #-}+ streamf <*> streamx = do {f <- streamf; x <- streamx; return (f x)} + {-# INLINE (<*>) #-} ++{- | The 'Alternative' instance glues streams together stepwise.++> empty = never+> (<|>) = zipsWith (liftA2 (,))++ See also 'never', 'untilJust' and 'delays'+-}+instance (Applicative f, Monad m) => Alternative (Stream f m) where+ empty = never+ {-#INLINE empty #-}++ str <|> str' = zipsWith (liftA2 (,)) str str'+ {-#INLINE (<|>) #-}++instance (Functor f, Monad m, Monoid w) => Monoid (Stream f m w) where+ mempty = return mempty+ {-#INLINE mempty #-}+ mappend a b = a >>= \w -> fmap (w <>) b+ {-#INLINE mappend #-}++instance (Applicative f, Monad m) => MonadPlus (Stream f m) where+ mzero = empty+ mplus = (<|>)++instance Functor f => MonadTrans (Stream f) where+ lift = Effect . liftM Return+ {-# INLINE lift #-}++instance Functor f => MFunctor (Stream f) where+ hoist trans = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> Effect (trans (liftM loop m))+ Step f -> Step (fmap loop f)+ {-# INLINABLE hoist #-} ++instance Functor f => MMonad (Stream f) where+ embed phi = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> phi m >>= loop+ Step f -> Step (fmap loop f)+ {-# INLINABLE embed #-}++instance (MonadIO m, Functor f) => MonadIO (Stream f m) where+ liftIO = Effect . liftM Return . liftIO+ {-# INLINE liftIO #-}++instance (MonadBase b m, Functor f) => MonadBase b (Stream f m) where+ liftBase = effect . fmap return . liftBase+ {-#INLINE liftBase #-}++instance (MonadThrow m, Functor f) => MonadThrow (Stream f m) where+ throwM = lift . throwM+ {-#INLINE throwM #-}++instance (MonadCatch m, Functor f) => MonadCatch (Stream f m) where+ catch str f = go str+ where+ go p = case p of+ Step f -> Step (fmap go f)+ Return r -> Return r+ Effect m -> Effect (catch (do+ p' <- m+ return (go p'))+ (\e -> return (f e)) )+ {-#INLINABLE catch #-}+ +instance (MonadResource m, Functor f) => MonadResource (Stream f m) where+ liftResourceT = lift . liftResourceT+ {-#INLINE liftResourceT #-}+++instance (Functor f, MonadReader r m) => MonadReader r (Stream f m) where+ ask = lift ask+ {-# INLINE ask #-}+ local f = hoist (local f)+ {-# INLINE local #-}++-- instance (Functor f, MonadWriter w m) => MonadWriter w (Stream f m) where+-- tell = lift . tell+-- {-# INLINE tell #-}+-- -- listen (FreeT m) = FreeT $ liftM concat' $ listen (fmap listen `liftM` m)+-- where+-- concat' (Pure x, w) = Pure (x, w)+-- concat' (Free y, w) = Free $ fmap (second (w <>)) <$> y+-- pass m = FreeT . pass' . runFreeT . hoist clean $ listen m+-- where+-- clean = pass . liftM (\x -> (x, const mempty))+-- pass' = join . liftM g+-- g (Pure ((x, f), w)) = tell (f w) >> return (Pure x)+-- g (Free f) = return . Free . fmap (FreeT . pass' . runFreeT) $ f+-- #if MIN_VERSION_mtl(2,1,1)+-- writer w = lift (writer w)+-- {-# INLINE writer #-}+-- #endif+--+instance (Functor f, MonadState s m) => MonadState s (Stream f m) where+ get = lift get+ {-# INLINE get #-}+ put = lift . put+ {-# INLINE put #-}+#if MIN_VERSION_mtl(2,1,1)+ state f = lift (state f)+ {-# INLINE state #-}+#endif++instance (Functor f, MonadError e m) => MonadError e (Stream f m) where+ throwError = lift . throwError+ {-# INLINE throwError #-}+ str `catchError` f = loop str where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect $ liftM loop m `catchError` (return . f)+ Step f -> Step (fmap loop f)+ {-# INLINABLE catchError #-}++bracketStream :: (Functor f, MonadResource m) =>+ IO a -> (a -> IO ()) -> (a -> Stream f m b) -> Stream f m b+bracketStream alloc free inside = do+ (key, seed) <- lift (allocate alloc free)+ clean key (inside seed)+ where+ clean key = loop where+ loop str = case str of+ Return r -> Effect (release key >> return (Return r))+ Effect m -> Effect (liftM loop m)+ Step f -> Step (fmap loop f)+{-#INLINABLE bracketStream #-}+++{-| Map a stream directly to its church encoding; compare @Data.List.foldr@+-}+destroy+ :: (Functor f, Monad m) =>+ Stream f m r -> (f b -> b) -> (m b -> b) -> (r -> b) -> b+destroy stream0 construct effect done = loop stream0 where+ loop stream = case stream of+ Return r -> done r+ Effect m -> effect (liftM loop m)+ Step fs -> construct (fmap loop fs)+{-# INLINABLE destroy #-}+++{-| 'streamFold' reorders the arguments of 'destroy' to be more akin+ to @foldr@ It is more convenient to query in ghci to figure out+ what kind of \'algebra\' you need to write.++>>> :t streamFold return join+(Monad m, Functor f) =>+ (f (m a) -> m a) -> Stream f m a -> m a -- iterT++>>> :t streamFold return (join . lift)+(Monad m, Monad (t m), Functor f, MonadTrans t) =>+ (f (t m a) -> t m a) -> Stream f m a -> t m a -- iterTM++>>> :t streamFold return effect+(Monad m, Functor f, Functor g) =>+ (f (Stream g m r) -> Stream g m r) -> Stream f m r -> Stream g m r++>>> :t \f -> streamFold return effect (wrap . f)+(Monad m, Functor f, Functor g) =>+ (f (Stream g m a) -> g (Stream g m a))+ -> Stream f m a -> Stream g m a -- maps++>>> :t \f -> streamFold return effect (effect . liftM wrap . f)+(Monad m, Functor f, Functor g) =>+ (f (Stream g m a) -> m (g (Stream g m a)))+ -> Stream f m a -> Stream g m a -- mapped++-}+streamFold+ :: (Functor f, Monad m) =>+ (r -> b) -> (m b -> b) -> (f b -> b) -> Stream f m r -> b+streamFold done effect construct stream = destroy stream construct effect done+{-#INLINE streamFold #-}++{- | Reflect a church-encoded stream; cp. @GHC.Exts.build@++> streamFold return_ effect_ step_ (streamBuild psi) = psi return_ effect_ step_+-}+streamBuild+ :: (forall b . (r -> b) -> (m b -> b) -> (f b -> b) -> b) -> Stream f m r+streamBuild = \phi -> phi Return Effect Step+{-# INLINE streamBuild #-}+++{-| Inspect the first stage of a freely layered sequence.+ Compare @Pipes.next@ and the replica @Streaming.Prelude.next@.+ This is the 'uncons' for the general 'unfold'.++> unfold inspect = id+> Streaming.Prelude.unfoldr StreamingPrelude.next = id+-}+inspect :: (Functor f, Monad m) =>+ Stream f m r -> m (Either r (f (Stream f m r)))+inspect = loop where+ loop stream = case stream of+ Return r -> return (Left r)+ Effect m -> m >>= loop+ Step fs -> return (Right fs)+{-# INLINABLE inspect #-}+ +{-| Build a @Stream@ by unfolding steps starting from a seed. See also+ the specialized 'Streaming.Prelude.unfoldr' in the prelude.++> unfold inspect = id -- modulo the quotient we work with+> unfold Pipes.next :: Monad m => Producer a m r -> Stream ((,) a) m r+> unfold (curry (:>) . Pipes.next) :: Monad m => Producer a m r -> Stream (Of a) m r++-}+unfold :: (Monad m, Functor f)+ => (s -> m (Either r (f s)))+ -> s -> Stream f m r+unfold step = loop where+ loop s0 = Effect $ do+ e <- step s0+ case e of+ Left r -> return (Return r)+ Right fs -> return (Step (fmap loop fs))+{-# INLINABLE unfold #-}+++{- | Map layers of one functor to another with a transformation. Compare+ hoist, which has a similar effect on the 'monadic' parameter.++> maps id = id+> maps f . maps g = maps (f . g)++-}+maps :: (Monad m, Functor f)+ => (forall x . f x -> g x) -> Stream f m r -> Stream g m r+maps phi = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step f -> Step (phi (fmap loop f))+{-# INLINABLE maps #-}+++{- | Map layers of one functor to another with a transformation involving the base monad+ @maps@ is more fundamental than @mapsM@, which is best understood as a convenience+ for effecting this frequent composition:++> mapsM phi = decompose . maps (Compose . phi)++ The streaming prelude exports the same function under the better name @mapped@,+ which overlaps with the lens libraries.++-}+mapsM :: (Monad m, Functor f) => (forall x . f x -> m (g x)) -> Stream f m r -> Stream g m r+mapsM phi = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step f -> Effect (liftM Step (phi (fmap loop f)))+{-# INLINABLE mapsM #-}+++{-| Rearrange a succession of layers of the form @Compose m (f x)@.++ we could as well define @decompose@ by @mapsM@:++> decompose = mapped getCompose++ but @mapped@ is best understood as:++> mapped phi = decompose . maps (Compose . phi)++ since @maps@ and @hoist@ are the really fundamental operations that preserve the+ shape of the stream:++> maps :: (Monad m, Functor f) => (forall x. f x -> g x) -> Stream f m r -> Stream g m r+> hoist :: (Monad m, Functor f) => (forall a. m a -> n a) -> Stream f m r -> Stream f n r++-}+decompose :: (Monad m, Functor f) => Stream (Compose m f) m r -> Stream f m r+decompose = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step (Compose mstr) -> Effect $ do+ str <- mstr+ return (Step (fmap loop str))+++{-| Run the effects in a stream that merely layers effects.+-}+run :: Monad m => Stream m m r -> m r+run = loop where+ loop stream = case stream of+ Return r -> return r+ Effect m -> m >>= loop+ Step mrest -> mrest >>= loop+{-# INLINABLE run #-}+++{-| Map each layer to an effect, and run them all.+-}+mapsM_ :: (Functor f, Monad m) => (forall x . f x -> m x) -> Stream f m r -> m r+mapsM_ f = run . maps f+{-# INLINE mapsM_ #-}+++{-| Interpolate a layer at each segment. This specializes to e.g.++> intercalates :: (Monad m, Functor f) => Stream f m () -> Stream (Stream f m) m r -> Stream f m r+-}+intercalates :: (Monad m, Monad (t m), MonadTrans t) =>+ t m x -> Stream (t m) m r -> t m r+intercalates sep = go0+ where+ go0 f = case f of+ Return r -> return r+ Effect m -> lift m >>= go0+ Step fstr -> do+ f' <- fstr+ go1 f'+ go1 f = case f of+ Return r -> return r+ Effect m -> lift m >>= go1+ Step fstr -> do+ sep+ f' <- fstr+ go1 f'+{-# INLINABLE intercalates #-}++{-| Specialized fold following the usage of @Control.Monad.Trans.Free@++> iterTM alg = streamFold return (join . lift)+-}+iterTM ::+ (Functor f, Monad m, MonadTrans t,+ Monad (t m)) =>+ (f (t m a) -> t m a) -> Stream f m a -> t m a+iterTM out stream = destroyExposed stream out (join . lift) return+{-# INLINE iterTM #-}++{-| Specialized fold following the usage of @Control.Monad.Trans.Free@++> iterT alg = streamFold return join alg+-}+iterT ::+ (Functor f, Monad m) => (f (m a) -> m a) -> Stream f m a -> m a+iterT out stream = destroyExposed stream out join return+{-# INLINE iterT #-}++{-| Dissolves the segmentation into layers of @Stream f m@ layers.++-}+concats :: (Monad m, Functor f) => Stream (Stream f m) m r -> Stream f m r+concats = loop where+ loop stream = case stream of+ Return r -> return r+ Effect m -> join $ lift (liftM loop m)+ Step fs -> join (fmap loop fs)+{-# INLINE concats #-}++{-| Split a succession of layers after some number, returning a streaming or+ effectful pair.++>>> rest <- S.print $ S.splitAt 1 $ each [1..3]+1+>>> S.print rest+2+3++> splitAt 0 = return+> splitAt n >=> splitAt m = splitAt (m+n)++ Thus, e.g.++>>> rest <- S.print $ splitsAt 2 >=> splitsAt 2 $ each [1..5]+1+2+3+4+>>> S.print rest+5++-}+splitsAt :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Stream f m r)+splitsAt = loop where+ loop !n stream+ | n <= 0 = Return stream+ | otherwise = case stream of+ Return r -> Return (Return r)+ Effect m -> Effect (liftM (loop n) m)+ Step fs -> case n of+ 0 -> Return (Step fs)+ _ -> Step (fmap (loop (n-1)) fs)+{-# INLINABLE splitsAt #-}++{- Functor-general take.++ @takes 3@ can take three individual values++>>> S.print $ takes 3 $ each [1..]+1+2+3+++ or three sub-streams++>>> S.print $ mapped S.toList $ takes 3 $ chunksOf 2 $ each [1..]+[1,2]+[3,4]+[5,6]++ Or, using 'Data.ByteString.Streaming.Char' (here called @Q@),+ three byte streams.++>>> Q.stdout $ Q.unlines $ takes 3 $ Q.lines $ Q.chunk "a\nb\nc\nd\ne\nf"+a+b+c++-}+takes :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m ()+takes n = void . splitsAt n+{-# INLINE takes #-} ++{-| Break a stream into substreams each with n functorial layers.++>>> S.print $ mapped S.sum $ chunksOf 2 $ each [1,1,1,1,1]+2+2+1+-}+chunksOf :: (Monad m, Functor f) => Int -> Stream f m r -> Stream (Stream f m) m r+chunksOf n0 = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step fs -> Step (Step (fmap (fmap loop . splitsAt (n0-1)) fs))+{-# INLINABLE chunksOf #-} ++{- | Make it possible to \'run\' the underlying transformed monad.+-}+distribute :: (Monad m, Functor f, MonadTrans t, MFunctor t, Monad (t (Stream f m)))+ => Stream f (t m) r -> t (Stream f m) r+distribute = loop where+ loop stream = case stream of+ Return r -> lift (Return r)+ Effect tmstr -> hoist lift tmstr >>= loop+ Step fstr -> join (lift (Step (fmap (Return . loop) fstr)))+{-#INLINABLE distribute #-}+ +-- | Repeat a functorial layer (a \"command\" or \"instruction\") forever.+repeats :: (Monad m, Functor f) => f () -> Stream f m r+repeats f = loop where+ loop = Effect (return (Step (fmap (\_ -> loop) f)))++-- | Repeat an effect containing a functorial layer, command or instruction forever.+repeatsM :: (Monad m, Functor f) => m (f ()) -> Stream f m r+repeatsM mf = loop where+ loop = Effect $ do+ f <- mf+ return $ Step $ fmap (\_ -> loop) f++{- | Repeat a functorial layer, command or instruction a fixed number of times.++> replicates n = takes n . repeats+-}+replicates :: (Monad m, Functor f) => Int -> f () -> Stream f m ()+replicates n f = splitsAt n (repeats f) >> return ()++{-| Construct an infinite stream by cycling a finite one++> cycles = forever++>>>+-}++cycles :: (Monad m, Functor f) => Stream f m () -> Stream f m r+cycles = forever++++hoistExposed trans = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> Effect (trans (liftM loop m))+ Step f -> Step (fmap loop f)++mapsExposed :: (Monad m, Functor f)+ => (forall x . f x -> g x) -> Stream f m r -> Stream g m r+mapsExposed phi = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step f -> Step (phi (fmap loop f))+{-# INLINABLE mapsExposed #-}++mapsMExposed phi = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step f -> Effect (liftM Step (phi (fmap loop f)))+{-# INLINABLE mapsMExposed #-}++-- Map a stream directly to its church encoding; compare @Data.List.foldr@+-- It permits distinctions that should be hidden, as can be seen from+-- e.g.+--+-- isPure stream = destroy (const True) (const False) (const True)+--+-- and similar nonsense. The crucial+-- constraint is that the @m x -> x@ argument is an /Eilenberg-Moore algebra/.+-- See Atkey "Reasoning about Stream Processing with Effects"+++destroyExposed stream0 construct effect done = loop stream0 where+ loop stream = case stream of+ Return r -> done r+ Effect m -> effect (liftM loop m)+ Step fs -> construct (fmap loop fs)+{-# INLINABLE destroyExposed #-}+++{-| This is akin to the @observe@ of @Pipes.Internal@ . It reeffects the layering+ in instances of @Stream f m r@ so that it replicates that of+ @FreeT@.++-}+unexposed :: (Functor f, Monad m) => Stream f m r -> Stream f m r+unexposed = Effect . loop where+ loop stream = case stream of+ Return r -> return (Return r)+ Effect m -> m >>= loop+ Step f -> return (Step (fmap (Effect . loop) f))+{-# INLINABLE unexposed #-} +++{- Wrap a new layer of a stream. So, e.g.++> S.cons :: Monad m => a -> Stream (Of a) m r -> Stream (Of a) m r+> S.cons a str = wrap (a :> str)++ and, recursively:++> S.each :: (Monad m, Foldable t) => t a -> Stream (Of a) m ()+> S.each = foldr (\a b -> wrap (a :> b)) (return ())++ The two operations++> wrap :: (Monad m, Functor f ) => f (Stream f m r) -> Stream f m r+> effect :: (Monad m, Functor f ) => m (Stream f m r) -> Stream f m r++ are fundamental. We can define the parallel operations @yields@ and @lift@ in+ terms of them++> yields :: (Monad m, Functor f ) => f r -> Stream f m r+> yields = wrap . fmap return+> lift :: (Monad m, Functor f ) => m r -> Stream f m r+> lift = effect . fmap return++-}+wrap :: (Monad m, Functor f ) => f (Stream f m r) -> Stream f m r+wrap = Step+{-#INLINE wrap #-}+++{- | Wrap an effect that returns a stream++> effect = join . lift++-}+effect :: (Monad m, Functor f ) => m (Stream f m r) -> Stream f m r+effect = Effect+{-#INLINE effect #-}+++{-| @yields@ is like @lift@ for items in the streamed functor.+ It makes a singleton or one-layer succession.++> lift :: (Monad m, Functor f) => m r -> Stream f m r+> yields :: (Monad m, Functor f) => f r -> Stream f m r++ Viewed in another light, it is like a functor-general version of @yield@:++> S.yield a = yields (a :> ())++-}++yields :: (Monad m, Functor f) => f r -> Stream f m r+yields fr = Step (fmap Return fr)+{-#INLINE yields #-}+++zipsWith :: (Monad m, Functor f, Functor g, Functor h)+ => (forall x y . f x -> g y -> h (x,y))+ -> Stream f m r -> Stream g m r -> Stream h m r+zipsWith phi s t = loop (s,t) where+ loop (s1, s2) = Effect (go s1 s2)+ go s1 s2 = do + e <- inspect s1+ case e of+ Left r -> return (Return r)+ Right fstr -> do + e <- inspect s2+ case e of+ Left r -> return (Return r)+ Right gstr -> return $ Step $ fmap loop (phi fstr gstr)+{-# INLINABLE zipsWith #-} ++zips :: (Monad m, Functor f, Functor g)+ => Stream f m r -> Stream g m r -> Stream (Compose f g) m r+zips = zipsWith go where+ go fx gy = Compose (fmap (\x -> fmap (\y -> (x,y)) gy) fx)+{-# INLINE zips #-} ++++{-| Interleave functor layers, with the effects of the first preceding+ the effects of the second.++> interleaves = zipsWith (liftA2 (,))++>>> let paste = \a b -> interleaves (Q.lines a) (maps (Q.cons' '\t') (Q.lines b))+>>> Q.stdout $ Q.unlines $ paste "hello\nworld\n" "goodbye\nworld\n"+hello goodbye+world world++-}++interleaves+ :: (Monad m, Applicative h) =>+ Stream h m r -> Stream h m r -> Stream h m r+interleaves = zipsWith (liftA2 (,))+{-# INLINE interleaves #-} +++{-| Swap the order of functors in a sum of functors.+++>>> S.toListM' $ S.print $ separate $ maps S.switch $ maps (S.distinguish (=='a')) $ S.each "banana"+'a'+'a'+'a'+"bnn" :> ()+>>> S.toListM' $ S.print $ separate $ maps (S.distinguish (=='a')) $ S.each "banana"+'b'+'n'+'n'+"aaa" :> ()+-}+switch :: Sum f g r -> Sum g f r+switch s = case s of InL a -> InR a; InR a -> InL a+{-#INLINE switch #-}++++{-| Given a stream on a sum of functors, make it a stream on the left functor,+ with the streaming on the other functor as the governing monad. This is+ useful for acting on one or the other functor with a fold. It generalizes+ 'Data.Either.partitionEithers' massively, but actually streams properly.++>>> let odd_even = S.maps (S.distinguish even) $ S.each [1..10::Int]+>>> :t separate odd_even+separate odd_even+ :: Monad m => Stream (Of Int) (Stream (Of Int) m) ()++ Now, for example, it is convenient to fold on the left and right values separately:++>>> S.toList $ S.toList $ separate odd_even+[2,4,6,8,10] :> ([1,3,5,7,9] :> ())+++ Or we can write them to separate files or whatever:++>>> runResourceT $ S.writeFile "even.txt" . S.show $ S.writeFile "odd.txt" . S.show $ S.separate odd_even+>>> :! cat even.txt+2+4+6+8+10+>>> :! cat odd.txt+1+3+5+7+9++ Of course, in the special case of @Stream (Of a) m r@, we can achieve the above+ effects more simply by using 'Streaming.Prelude.copy'++>>> S.toList . S.filter even $ S.toList . S.filter odd $ S.copy $ each [1..10::Int]+[2,4,6,8,10] :> ([1,3,5,7,9] :> ())+++ But 'separate' and 'unseparate' are functor-general.++-}++separate :: (Monad m, Functor f, Functor g) => Stream (Sum f g) m r -> Stream f (Stream g m) r+separate str = destroyExposed+ str+ (\x -> case x of InL fss -> wrap fss; InR gss -> effect (yields gss))+ (effect . lift)+ return+{-#INLINABLE separate #-}++unseparate :: (Monad m, Functor f, Functor g) => Stream f (Stream g m) r -> Stream (Sum f g) m r+unseparate str = destroyExposed+ str+ (wrap . InL)+ (join . maps InR)+ return+{-#INLINABLE unseparate #-}+++unzips :: (Monad m, Functor f, Functor g) =>+ Stream (Compose f g) m r -> Stream f (Stream g m) r+unzips str = destroyExposed+ str+ (\(Compose fgstr) -> Step (fmap (Effect . yields) fgstr))+ (Effect . lift)+ return+{-#INLINABLE unzips #-}++{-| Group layers in an alternating stream into adjoining sub-streams+ of one type or another.+-}+groups :: (Monad m, Functor f, Functor g)+ => Stream (Sum f g) m r+ -> Stream (Sum (Stream f m) (Stream g m)) m r+groups = loop+ where+ loop str = do+ e <- lift $ inspect str+ case e of+ Left r -> return r+ Right ostr -> case ostr of+ InR gstr -> wrap $ InR (fmap loop (cleanR (wrap (InR gstr))))+ InL fstr -> wrap $ InL (fmap loop (cleanL (wrap (InL fstr))))++ cleanL :: (Monad m, Functor f, Functor g) =>+ Stream (Sum f g) m r -> Stream f m (Stream (Sum f g) m r)+ cleanL = loop where+ loop s = do+ e <- lift $ inspect s+ case e of+ Left r -> return (return r)+ Right (InL fstr) -> wrap (fmap loop fstr)+ Right (InR gstr) -> return (wrap (InR gstr))++ cleanR :: (Monad m, Functor f, Functor g) =>+ Stream (Sum f g) m r -> Stream g m (Stream (Sum f g) m r)+-- cleanR = fmap (maps switch) . cleanL . maps switch+ cleanR = loop where+ loop s = do+ e <- lift $ inspect s+ case e of+ Left r -> return (return r)+ Right (InL fstr) -> return (wrap (InL fstr))+ Right (InR gstr) -> wrap (fmap loop gstr)+{-#INLINABLE groups #-}+ +-- groupInL :: (Monad m, Functor f, Functor g)+-- => Stream (Sum f g) m r+-- -> Stream (Sum (Stream f m) g) m r+-- groupInL = loop+-- where+-- loop str = do+-- e <- lift $ inspect str+-- case e of+-- Left r -> return r+-- Right ostr -> case ostr of+-- InR gstr -> wrap $ InR (fmap loop gstr)+-- InL fstr -> wrap $ InL (fmap loop (cleanL (wrap (InL fstr))))+-- cleanL :: (Monad m, Functor f, Functor g) =>+-- Stream (Sum f g) m r -> Stream f m (Stream (Sum f g) m r)+-- cleanL = loop where+-- loop s = dos+-- e <- lift $ inspect s+-- case e of+-- Left r -> return (return r)+-- Right (InL fstr) -> wrap (fmap loop fstr)+-- Right (InR gstr) -> return (wrap (InR gstr))++{- | 'never' interleaves the pure applicative action with the return of the monad forever.+ It is the 'empty' of the 'Alternative' instance, thus++> never <|> a = a+> a <|> never = a++ and so on. If w is a monoid then @never :: Stream (Of w) m r@ is+ the infinite sequence of 'mempty', and+ @str1 \<|\> str2@ appends the elements monoidally until one of streams ends.+ Thus we have, e.g.++>>> S.stdoutLn $ S.take 2 $ S.stdinLn <|> S.repeat " " <|> S.stdinLn <|> S.repeat " " <|> S.stdinLn+1<Enter>+2<Enter>+3<Enter>+1 2 3+4<Enter>+5<Enter>+6<Enter>+4 5 6++ This is equivalent to++>>> S.stdoutLn $ S.take 2 $ foldr (<|>) never [S.stdinLn, S.repeat " ", S.stdinLn, S.repeat " ", S.stdinLn ]++ Where 'f' is a monad, @(\<|\>)@ sequences the conjoined streams stepwise. See the+ definition of @paste@ <https://gist.github.com/michaelt/6c6843e6dd8030e95d58 here>,+ where the separate steps are bytestreams corresponding to the lines of a file.++ Given, say,++> data Branch r = Branch r r deriving Functor -- add obvious applicative instance++ then @never :: Stream Branch Identity r@ is the pure infinite binary tree with+ (inaccessible) @r@s in its leaves. Given two binary trees, @tree1 \<|\> tree2@+ intersects them, preserving the leaves that came first,+ so @tree1 \<|\> never = tree1@++ @Stream Identity m r@ is an action in @m@ that is indefinitely delayed. Such an+ action can be constructed with e.g. 'untilJust'.++> untilJust :: (Monad m, Applicative f) => m (Maybe r) -> Stream f m r++ Given two such items, @\<|\>@ instance races them.+ It is thus the iterative monad transformer specially defined in+ <https://hackage.haskell.org/package/free-4.12.1/docs/Control-Monad-Trans-Iter.html Control.Monad.Trans.Iter>++ So, for example, we might write++>>> let justFour str = if length str == 4 then Just str else Nothing+>>> let four = untilJust (liftM justFour getLine)+>>> run four+one<Enter>+two<Enter>+three<Enter>+four<Enter>+"four"+++ The 'Alternative' instance in+ <https://hackage.haskell.org/package/free-4.12.1/docs/Control-Monad-Trans-Free.html Control.Monad.Trans.Free>+ is avowedly wrong, though no explanation is given for this.+++-}+never :: (Monad m, Applicative f) => Stream f m r+never = let loop = Effect $ return $ Step $ pure loop in loop+{-#INLINABLE never #-}+++delays :: (MonadIO m, Applicative f) => Double -> Stream f m r+delays seconds = loop where+ loop = Effect $ liftIO (threadDelay delay) >> return (Step (pure loop))+ delay = fromInteger (truncate (1000000 * seconds))+{-#INLINABLE delays #-}++-- {-| Permit streamed actions to proceed unless the clock has run out.+--+-- -}+-- period :: (MonadIO m, Functor f) => Double -> Stream f m r -> Stream f m (Stream f m r)+-- period seconds str = do+-- utc <- liftIO getCurrentTime+-- let loop s = do+-- utc' <- liftIO getCurrentTime+-- if diffUTCTime utc' utc > (cutoff / 1000000000)+-- then return s+-- else case s of+-- Return r -> Return (Return r)+-- Effect m -> Effect (liftM loop m)+-- Step f -> Step (fmap loop f)+-- loop str+-- where+-- cutoff = fromInteger (truncate (1000000000 * seconds))+-- {-#INLINABLE period #-}+--+--+-- {-| Divide a succession of phases according to a specified time interval. If time runs out+-- while an action is proceeding, it is allowed to run to completion. The clock is only then+-- restarted.+-- -}+-- periods :: (MonadIO m, Functor f) => Double -> Stream f m r -> Stream (Stream f m) m r+-- periods seconds s = do+-- utc <- liftIO getCurrentTime+-- loop (addUTCTime cutoff utc) s+--+-- where+-- cutoff = fromInteger (truncate (1000000000 * seconds)) / 1000000000+-- loop final stream = do+-- utc <- liftIO getCurrentTime+-- if utc > final+-- then loop (addUTCTime cutoff utc) stream+-- else case stream of+-- Return r -> Return r+-- Effect m -> Effect $ liftM (loop final) m+-- Step fstr -> Step $ fmap (periods seconds) (cutoff_ final (Step fstr))+--+-- -- do+-- -- let sloop s = do+-- -- utc' <- liftIO getCurrentTime+-- -- if final < utc'+-- -- then return s+-- -- else case s of+-- -- Return r -> Return (Return r)+-- -- Effect m -> Effect (liftM sloop m)+-- -- Step f -> Step (fmap sloop f)+-- -- Step (Step (fmap (fmap (periods seconds) . sloop) fstr))+-- -- str <- m+-- -- utc' <- liftIO getCurrentTime+-- -- if diffUTCTime utc' utc > (cutoff / 1000000000)+-- -- then return (loop utc' str)+-- -- else return (loop utc str)+-- -- Step fs -> do+-- -- let check str = do+-- -- utc' <- liftIO getCurrentTime+-- -- loop utc' str+-- --+-- {-# INLINABLE periods #-}+--+-- cutoff_ final str = do+-- let loop s = do+-- utc' <- liftIO getCurrentTime+-- if utc' > final+-- then Return s+-- else case s of+-- Return r -> Return (Return r)+-- Effect m -> Effect (liftM loop m)+-- Step f -> Step (fmap loop f)+-- loop str++{- | Repeat a++-}++untilJust :: (Monad m, Applicative f) => m (Maybe r) -> Stream f m r+untilJust act = loop where+ loop = Effect $ do+ m <- act+ case m of+ Nothing -> return $ Step $ pure loop+ Just a -> return $ Return a+ + +cutoff :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Maybe r)+cutoff = loop where+ loop 0 str = return Nothing+ loop n str = do+ e <- lift $ inspect str+ case e of+ Left r -> return (Just r)+ Right (frest) -> Step $ fmap (loop (n-1)) frest
+ src/Streaming/Prelude.hs view
@@ -0,0 +1,2805 @@+{-| This names exported by this module are closely modeled on those in @Prelude@ and @Data.List@,+ but also on+ <http://hackage.haskell.org/package/pipes-4.1.9/docs/Pipes-Prelude.html @Pipes.Prelude@>,+ <http://hackage.haskell.org/package/pipes-group-1.0.3/docs/Pipes-Group.html @Pipes.Group@>+ and <http://hackage.haskell.org/package/pipes-parse-3.0.6/docs/Pipes-Parse.html @Pipes.Parse@>.+ The module may be said to give independent expression to the conception of+ Producer / Source / Generator manipulation+ articulated in the latter two modules. Because we dispense with piping and+ conduiting, the distinction between all of these modules collapses. Some things are+ lost but much is gained in that everything comes much closer to ordinary+ beginning Haskell programming. The leading type is chosen to permit an api+ that is as close as possible to that of @Data.List@ and the @Prelude@.++ Import qualified thus:++> import Streaming+> import qualified Streaming.Prelude as S++ For the examples below, one sometimes needs++> import Streaming.Prelude (each, yield, next, mapped, stdoutLn, stdinLn)+> import Data.Function ((&))++ Other libraries that come up in passing are++> import qualified Control.Foldl as L -- cabal install foldl+> import qualified Pipes as P+> import qualified Pipes.Prelude as P+> import qualified System.IO as IO++ Here are some correspondences between the types employed here and elsewhere:++> streaming | pipes | conduit | io-streams+> -------------------------------------------------------------------------------------------------------------------+> Stream (Of a) m () | Producer a m () | Source m a | InputStream a+> | ListT m a | ConduitM () o m () | Generator r ()+> -------------------------------------------------------------------------------------------------------------------+> Stream (Of a) m r | Producer a m r | ConduitM () o m r | Generator a r+> -------------------------------------------------------------------------------------------------------------------+> Stream (Of a) m (Stream (Of a) m r) | Producer a m (Producer a m r) |+> --------------------------------------------------------------------------------------------------------------------+> Stream (Stream (Of a) m) r | FreeT (Producer a m) m r |+> --------------------------------------------------------------------------------------------------------------------+> --------------------------------------------------------------------------------------------------------------------+> ByteString m () | Producer ByteString m () | Source m ByteString | InputStream ByteString+> --------------------------------------------------------------------------------------------------------------------+>+-}+{-# LANGUAGE RankNTypes, BangPatterns, DeriveDataTypeable, TypeFamilies,+ DeriveFoldable, DeriveFunctor, DeriveTraversable, CPP, Trustworthy #-}++module Streaming.Prelude (+ -- * Types+ Of (..)++ -- * Introducing streams of elements+ -- $producers+ , yield+ , each+ , each'+ , unfoldr+ , stdinLn+ , readLn+ , fromHandle+ , readFile+ , iterate+ , iterateM+ , repeat+ , repeatM+ , replicate+ , untilRight+ , cycle+ , replicateM+ , enumFrom+ , enumFromThen+ , seconds+++ -- * Consuming streams of elements+ -- $consumers+ , stdoutLn+ , stdoutLn'+ , mapM_+ , print+ , toHandle+ , writeFile+ , effects+ , erase+ , drained+++ -- * Stream transformers+ -- $pipes+ , map+ , mapM+ , maps+ , mapped+ , for+ , with+ , subst+ , copy+ , copy'+ , store+ , chain+ , sequence+ , filter+ , filterM+ , delay+ , intersperse+ , take+ , takeWhile+-- , takeWhile'+ , takeWhileM+ , drop+ , dropWhile+ , concat+ -- , elemIndices+ -- , findIndices+ , scan+ , scanM+ , scanned+ , read+ , show+ , cons+ , duplicate+ , duplicate'+++ -- * Splitting and inspecting streams of elements+ , next+ , uncons+ , splitAt+ , split+-- , breaks+ , break+ , breakWhen+ , span+ , group+ , groupBy+ -- , groupedBy+ -- , split+++ -- * Sum and Compose manipulation++ , distinguish+ , switch+ , separate+ , unseparate+ , eitherToSum+ , sumToEither+ , sumToCompose+ , composeToSum++ -- * Folds+ -- $folds+ , fold+ , fold_+ , foldM+ , foldM_+ , all+ , all_+ , any+ , any_+ , sum+ , sum_+ , product+ , product_+ , head+ , head_+ , last+ , last_+ , elem+ , elem_+ , notElem+ , notElem_+ , length+ , length_+ , toList+ , toList_+ , mconcat+ , mconcat_+ , minimum+ , minimum_+ , maximum+ , maximum_+ , foldrM+ , foldrT+++ -- , all+ -- , any+ -- , and+ -- , or+ -- , elem++ -- , find+ -- , findIndex+ -- , head+ -- , index+ -- , last+ -- , length+ -- , maximum+ -- , minimum+ -- , null++ -- * Zips and unzips+ , zip+ , zipWith+ , zip3+ , zipWith3+ , unzip+ , partitionEithers+ , partition++ -- * Maybes+ -- $maybes+ , catMaybes+ , mapMaybe++ -- * Pair manipulation+ , lazily+ , strictly+ , fst'+ , snd'+ , mapOf+ , _first+ , _second++ -- * Interoperation+ , reread++ -- * Basic Type+ , Stream++ ) where+import Streaming.Internal++import Control.Monad hiding (filterM, mapM, mapM_, foldM, foldM_, replicateM, sequence)+import Data.Data ( Data, Typeable )+import Data.Functor.Identity+import Data.Functor.Sum+import Control.Monad.Trans+import Control.Applicative (Applicative (..))+import Data.Functor (Functor (..), (<$))++import qualified Prelude as Prelude+import Data.Foldable (Foldable)+import Data.Traversable (Traversable)+import qualified Data.Foldable as Foldable+import Text.Read (readMaybe)+import Prelude hiding (map, mapM, mapM_, filter, drop, dropWhile, take, mconcat+ , sum, product, iterate, repeat, cycle, replicate, splitAt+ , takeWhile, enumFrom, enumFromTo, enumFromThen, length+ , print, zipWith, zip, zipWith3, zip3, unzip, seq, show, read+ , readLn, sequence, concat, span, break, readFile, writeFile+ , minimum, maximum, elem, notElem, intersperse, all, any, head+ , last)++import qualified GHC.IO.Exception as G+import qualified System.IO as IO+import Foreign.C.Error (Errno(Errno), ePIPE)+import Control.Exception (throwIO, try)+import Data.Monoid (Monoid (mappend, mempty))+import Data.String (IsString (..))+import Control.Concurrent (threadDelay)+import Data.Time (getCurrentTime, diffUTCTime, picosecondsToDiffTime)+import Data.Functor.Classes+import Data.Functor.Compose+import Control.Monad.Trans.Resource++import GHC.Magic+#if MIN_VERSION_base(4,8,0)+import Data.Bifunctor+#endif++-- | A left-strict pair; the base functor for streams of individual elements.+data Of a b = !a :> b+ deriving (Data, Eq, Foldable, Ord,+ Read, Show, Traversable, Typeable)+infixr 5 :>++instance (Monoid a, Monoid b) => Monoid (Of a b) where+ mempty = mempty :> mempty+ {-#INLINE mempty #-}+ mappend (m :> w) (m' :> w') = mappend m m' :> mappend w w'+ {-#INLINE mappend #-}++instance Functor (Of a) where+ fmap f (a :> x) = a :> f x+ {-#INLINE fmap #-}+ a <$ (b :> x) = b :> a+ {-#INLINE (<$) #-}++#if MIN_VERSION_base(4,8,0)+instance Bifunctor Of where+ bimap f g (a :> b) = f a :> g b+ {-#INLINE bimap #-}+ first f (a :> b) = f a :> b+ {-#INLINE first #-}+ second g (a :> b) = a :> g b+ {-#INLINE second #-}+#endif++instance Monoid a => Applicative (Of a) where+ pure x = mempty :> x+ {-#INLINE pure #-}+ m :> f <*> m' :> x = mappend m m' :> f x+ {-#INLINE (<*>) #-}+ m :> x *> m' :> y = mappend m m' :> y+ {-#INLINE (*>) #-}+ m :> x <* m' :> y = mappend m m' :> x+ {-#INLINE (<*) #-}++instance Monoid a => Monad (Of a) where+ return x = mempty :> x+ {-#INLINE return #-}+ m :> x >> m' :> y = mappend m m' :> y+ {-#INLINE (>>) #-}+ m :> x >>= f = let m' :> y = f x in mappend m m' :> y+ {-#INLINE (>>=) #-}++instance (r ~ (), Monad m, f ~ Of Char) => IsString (Stream f m r) where+ fromString = each++-- instance (Eq a) => Eq1 (Of a) where eq1 = (==)+-- instance (Ord a) => Ord1 (Of a) where compare1 = compare+-- instance (Read a) => Read1 (Of a) where readsPrec1 = readsPrec+-- instance (Show a) => Show1 (Of a) where showsPrec1 = showsPrec++{-| Note that 'lazily', 'strictly', 'fst'', and 'mapOf' are all so-called /natural transformations/ on the primitive @Of a@ functor+ If we write++> type f ~~> g = forall x . f x -> g x++ then we can restate some types as follows:++> mapOf :: (a -> b) -> Of a ~~> Of b -- Bifunctor first+> lazily :: Of a ~~> (,) a+> Identity . fst' :: Of a ~~> Identity a++ Manipulation of a @Stream f m r@ by mapping often turns on recognizing natural transformations of @f@.+ Thus @maps@ is far more general the the @map@ of the @Streaming.Prelude@, which can be+ defined thus:++> S.map :: (a -> b) -> Stream (Of a) m r -> Stream (Of b) m r+> S.map f = maps (mapOf f)++ i.e.++> S.map f = maps (\(a :> x) -> (f a :> x))++ This rests on recognizing that @mapOf@ is a natural transformation; note though+ that it results in such a transformation as well:++> S.map :: (a -> b) -> Stream (Of a) m ~> Stream (Of b) m++ Thus we can @maps@ it in turn.+++ -}+lazily :: Of a b -> (a,b)+lazily = \(a:>b) -> (a,b)+{-# INLINE lazily #-}++{-| Convert a standard Haskell pair into a left-strict pair -}+strictly :: (a,b) -> Of a b+strictly = \(a,b) -> a :> b+{-# INLINE strictly #-}++{-| @fst'@ and @snd'@ extract the first and second element of a pair++>>> S.fst' (1:>"hi")+1+>>> S.snd' (1:>"hi")+"hi"+++ They are contained in the @_first@ and @_second@ lenses,+ if any lens library is in scope++>>> import Lens.Micro+>>> (1:>"hi") ^. S._first+1+>>> (1:>"hi") ^. S._second+"hi"++ -}++fst' :: Of a b -> a+fst' (a :> b) = a+{-#INLINE fst' #-}+snd' :: Of a b -> b+snd' (a :> b) = b+{-#INLINE snd' #-}++{-| Map a function over the first element of an @Of@ pair++>>> S.mapOf even (1:>"hi")+False :> "hi"++ @mapOf@ is just @first@ from the @Bifunctor@ instance++>>> first even (1:>"hi")+False :> "hi"++ and is contained in the @_first@ lens++>>> import Lens.Micro+>>> over S._first even (1:>"hi")+False :> "hi"++ -}++mapOf :: (a -> b) -> Of a r -> Of b r+mapOf f (a:> b) = (f a :> b)+{-#INLINE mapOf #-}++{-| A lens into the first element of a left-strict pair -}+_first :: Functor f => (a -> f a') -> Of a b -> f (Of a' b)+_first afb (a:>b) = fmap (\c -> (c:>b)) (afb a)+{-# INLINE _first #-}++{-| A lens into the second element of a left-strict pair -}+_second :: Functor f => (b -> f b') -> Of a b -> f (Of a b')+_second afb (a:>b) = fmap (\c -> (a:>c)) (afb b)+{-#INLINABLE _second #-}++all :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m (Of Bool r)+all thus = loop True where+ loop b str = case str of+ Return r -> return (b :> r)+ Effect m -> m >>= loop b+ Step (a :> rest) -> if thus a+ then loop True rest+ else do+ r <- effects rest+ return (False :> r)+{-#INLINABLE all #-}++all_ :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m Bool+all_ thus = loop True where+ loop b str = case str of+ Return r -> return b+ Effect m -> m >>= loop b+ Step (a :> rest) -> if thus a+ then loop True rest+ else return False+{-#INLINABLE all_ #-}+++any :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m (Of Bool r)+any thus = loop False where+ loop b str = case str of+ Return r -> return (b :> r)+ Effect m -> m >>= loop b+ Step (a :> rest) -> if thus a+ then do+ r <- effects rest+ return (True :> r)+ else loop False rest+{-#INLINABLE any #-}++any_ :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m Bool+any_ thus = loop False where+ loop b str = case str of+ Return r -> return b+ Effect m -> m >>= loop b+ Step (a :> rest) -> if thus a+ then return True+ else loop False rest+{-#INLINABLE any_ #-}++{-| Break a sequence upon meeting element falls under a predicate,+ keeping it and the rest of the stream as the return value.++>>> rest <- S.print $ S.break even $ each [1,1,2,3]+1+1+>>> S.print rest+2+3++-}++break :: Monad m => (a -> Bool) -> Stream (Of a) m r+ -> Stream (Of a) m (Stream (Of a) m r)+break pred = loop where+ loop str = case str of+ Return r -> Return (Return r)+ Effect m -> Effect $ liftM loop m+ Step (a :> rest) -> if (pred a)+ then Return (Step (a :> rest))+ else Step (a :> loop rest)+{-# INLINABLE break #-}++{-| Yield elements, using a fold to maintain state, until the accumulated+ value satifies the supplied predicate. The fold will then be short-circuited+ and the element that breaks it will be put after the break.+ This function is easiest to use with 'Control.Foldl.purely'++>>> rest <- each [1..10] & L.purely S.breakWhen L.sum (>10) & S.print+1+2+3+4+>>> S.print rest+5+6+7+8+9+10++-}+breakWhen :: Monad m => (x -> a -> x) -> x -> (x -> b) -> (b -> Bool) -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r)+breakWhen step begin done pred = loop0 begin+ where+ loop0 x stream = case stream of+ Return r -> return (return r)+ Effect mn -> Effect $ liftM (loop0 x) mn+ Step (a :> rest) -> loop a (step x a) rest+ loop a !x stream = do+ if pred (done x)+ then return (yield a >> stream)+ else case stream of+ Return r -> yield a >> return (return r)+ Effect mn -> Effect $ liftM (loop a x) mn+ Step (a' :> rest) -> do+ yield a+ loop a' (step x a') rest+{-# INLINABLE breakWhen #-}++-- -- Break during periods where the predicate is not satisfied, grouping the periods when it is.+--+-- >>> S.print $ mapped S.toList $ S.breaks not $ S.each [False,True,True,False,True,True,False]+-- [True,True]+-- [True,True]+-- >>> S.print $ mapped S.toList $ S.breaks id $ S.each [False,True,True,False,True,True,False]+-- [False]+-- [False]+-- [False]+--+-- -}+-- breaks+-- :: Monad m =>+-- (a -> Bool) -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r+-- breaks thus = loop where+-- loop stream = Effect $ do+-- e <- next stream+-- return $ case e of+-- Left r -> Return r+-- Right (a, p') ->+-- if not (thus a)+-- then Step $ fmap loop (yield a >> break thus p')+-- else loop p'+-- {-#INLINABLE breaks #-}++{-| Apply an action to all values, re-yielding each++>>> S.product $ S.chain Prelude.print $ S.each [1..5]+1+2+3+4+5+120 :> ()+-}++chain :: Monad m => (a -> m ()) -> Stream (Of a) m r -> Stream (Of a) m r+chain f = loop where+ loop str = case str of+ Return r -> return r+ Effect mn -> Effect (liftM loop mn)+ Step (a :> rest) -> Effect $ do+ f a+ return (Step (a :> loop rest))+{-# INLINABLE chain #-}++{-| Make a stream of traversable containers into a stream of their separate elements.+ This is just++> concat str = for str each++>>> S.print $ S.concat (each ["xy","z"])+'x'+'y'+'z'++ Note that it also has the effect of 'Data.Maybe.catMaybes', 'Data.Either.rights'+ 'map snd' and such-like operations.++>>> S.print $ S.concat $ S.each [Just 1, Nothing, Just 2]+1+2+>>> S.print $ S.concat $ S.each [Right 1, Left "Error!", Right 2]+1+2+>>> S.print $ S.concat $ S.each [('A',1), ('B',2)]+1+2++-}++concat :: (Monad m, Foldable.Foldable f) => Stream (Of (f a)) m r -> Stream (Of a) m r+concat str = for str each+{-# INLINE concat #-}++{-| The natural @cons@ for a @Stream (Of a)@.++> cons a stream = yield a >> stream++ Useful for interoperation:++> Data.Text.foldr S.cons (return ()) :: Text -> Stream (Of Char) m ()+> Lazy.foldrChunks S.cons (return ()) :: Lazy.ByteString -> Stream (Of Strict.ByteString) m ()++ and so on.+-}++cons :: (Monad m) => a -> Stream (Of a) m r -> Stream (Of a) m r+cons a str = Step (a :> str)+{-# INLINE cons #-}++{- | Cycle repeatedly through the layers of a stream, /ad inf./ This+ function is functor-general++> cycle = forever++>>> rest <- S.print $ S.splitAt 3 $ S.cycle (yield 0 >> yield 1)+True+False+True+>>> S.print $ S.take 3 rest+False+True+False++-}++cycle :: (Monad m, Functor f) => Stream f m r -> Stream f m s+cycle str = loop where loop = str >> loop+{-#INLINABLE cycle #-}+++{-| Interpolate a delay of n seconds between yields.+-}+delay :: MonadIO m => Double -> Stream (Of a) m r -> Stream (Of a) m r+delay seconds = loop where+ pico = truncate (seconds * 1000000)+ loop str = do+ e <- lift $ next str+ case e of+ Left r -> Return r+ Right (a,rest) -> do+ yield a+ liftIO $ threadDelay pico+ loop rest+{-#INLINABLE delay #-}++++{-| Where a transformer returns a stream, run the effects of the stream, keeping+ the return value. This is usually used at the type++> drained :: Monad m => Stream (Of a) m (Stream (Of b) m r) -> Stream (Of a) m r+> drained = join . fmap (lift . effects)++ Here, for example, we split a stream in two places and throw out the middle segment:++>>> rest <- S.print $ S.drained $ S.splitAt 2 $ S.splitAt 5 $ each [1..7]+1+2+>>> S.print rest+6+7++ In particular, we can define versions of @take@ and @takeWhile@ which+ retrieve the return value of the rest of the stream - and which can+ thus be used with 'maps':++> take' n = S.drained . S.splitAt n+> takeWhile' thus = S.drained . S.span thus++-}+drained :: (Monad m, Monad (t m), Functor (t m), MonadTrans t) => t m (Stream (Of a) m r) -> t m r+drained = join . fmap (lift . effects)+{-#INLINE drained #-}++-- ---------------+-- drop+-- ---------------+{-| Ignore the first n elements of a stream, but carry out the actions++>>> S.toList $ S.drop 2 $ S.replicateM 5 getLine+a<Enter>+b<Enter>+c<Enter>+d<Enter>+e<Enter>+["c","d","e"] :> ()++ Because it retains the final return value, @drop n@ is a suitable argument+ for @maps@:++>>> S.toList $ concats $ maps (S.drop 4) $ chunksOf 5 $ each [1..20]+[5,10,15,20] :> ()++++ -}++drop :: (Monad m) => Int -> Stream (Of a) m r -> Stream (Of a) m r+drop n str | n <= 0 = str+drop n str = loop n str where+ loop 0 stream = stream+ loop n stream = case stream of+ Return r -> Return r+ Effect ma -> Effect (liftM (loop n) ma)+ Step (a :> as) -> loop (n-1) as+{-# INLINABLE drop #-}++-- ---------------+-- dropWhile+-- ---------------++{- | Ignore elements of a stream until a test succeeds, retaining the rest.++>>> S.print $ S.dropWhile ((< 5) . length) S.stdinLn+one<Enter>+two<Enter>+three<Enter>+"three"+four<Enter>+"four"+^CInterrupted.+++-}+dropWhile :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m r+dropWhile pred = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect ma -> Effect (liftM loop ma)+ Step (a :> as) -> if pred a+ then loop as+ else Step (a :> as)+{-# INLINABLE dropWhile #-}++-- ---------------+-- each+-- ---------------++{- | Stream the elements of a pure, foldable container.++>>> S.print $ each [1..3]+1+2+3+>>> S.print $ mapped S.toList $ chunksOf 3 $ S.replicateM 5 getLine+s<Enter>+t<Enter>+u<Enter>+["s","t","u"]+v<Enter>+w<Enter>+["v","w"]++-}+each :: (Monad m, Foldable.Foldable f) => f a -> Stream (Of a) m ()+each = Foldable.foldr (\a p -> Step (a :> p)) (Return ())+{-# INLINABLE each #-}++each' :: (Monad m, Foldable.Foldable f) => f a -> Stream (Of a) m ()+each' = Foldable.foldr (\a p -> Effect (return (Step (a :> p)))) (Return ())+{-# INLINABLE each' #-}++-- ---------------+-- effects+-- ---------------++{- | Reduce a stream, performing its actions but ignoring its elements.++>>> rest <- S.effects $ S.splitAt 2 $ each [1..5]+>>> S.print rest+3+4+5+ 'effects' should be understood together with 'copy' and is subject to the rules++> S.effects . S.copy = id+> hoist S.effects . S.copy = id++ The similar @effects@ and @copy@ operations in @Data.ByteString.Streaming@ obey the same rules.++-}+effects :: Monad m => Stream (Of a) m r -> m r+effects = loop where+ loop stream = case stream of+ Return r -> return r+ Effect m -> m >>= loop+ Step (_ :> rest) -> loop rest+{-#INLINABLE effects #-}++{-| Exhaust a stream remembering only whether @a@ was an element.++-}++elem :: (Monad m, Eq a) => a -> Stream (Of a) m r -> m (Of Bool r)+elem a' = loop False where+ loop True str = liftM (True :>) (effects str)+ loop False str = case str of+ Return r -> return (False :> r)+ Effect m -> m >>= loop False+ Step (a:> rest) ->+ if a == a'+ then liftM (True :>) (effects rest)+ else loop False rest+{-#INLINABLE elem #-}++elem_ :: (Monad m, Eq a) => a -> Stream (Of a) m r -> m Bool+elem_ a' = loop False where+ loop True str = return True+ loop False str = case str of+ Return r -> return False+ Effect m -> m >>= loop False+ Step (a:> rest) ->+ if a == a'+ then return True+ else loop False rest+{-#INLINABLE elem_ #-}++-- -----+-- enumFrom+-- ------++{-| An infinite stream of enumerable values, starting from a given value.+ It is the same as `S.iterate succ`.+ Because their return type is polymorphic, @enumFrom@ and @enumFromThen@+ (and @iterate@ are useful for example with @zip@+ and @zipWith@, which require the same return type in the zipped streams.+ With @each [1..]@ the following bit of connect-and-resume would be impossible:++>>> rest <- S.print $ S.zip (S.enumFrom 'a') $ S.splitAt 3 $ S.enumFrom 1+('a',1)+('b',2)+('c',3)+>>> S.print $ S.take 3 rest+4+5+6++-}+enumFrom :: (Monad m, Enum n) => n -> Stream (Of n) m r+enumFrom = loop where+ loop !n = Effect (return (Step (n :> loop (succ n))))+{-# INLINABLE enumFrom #-}+++{-| An infinite sequence of enumerable values at a fixed distance, determined+ by the first and second values. See the discussion of 'Streaming.enumFrom'++>>> S.print $ S.take 3 $ S.enumFromThen 100 200+100+200+300++-}+enumFromThen:: (Monad m, Enum a) => a -> a -> Stream (Of a) m r+enumFromThen first second = Streaming.Prelude.map toEnum (loop _first)+ where+ _first = fromEnum first+ _second = fromEnum second+ diff = _second - _first+ loop !s = Step (s :> loop (s+diff))+{-# INLINABLE enumFromThen #-}++-- ---------------+-- erase+-- ---------------+{- | Remove the elements from a stream of values, retaining the structure of layers.+-}+erase :: Monad m => Stream (Of a) m r -> Stream Identity m r+erase = loop where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step (a:>rest) -> Step (Identity (loop rest))+{-# INLINABLE erase #-}++-- ---------------+-- filter+-- ---------------++-- | Skip elements of a stream that fail a predicate+filter :: (Monad m) => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m r+filter pred = loop where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step (a :> as) -> if pred a+ then Step (a :> loop as)+ else loop as+{-# INLINABLE filter #-}++-- ---------------+-- filterM+-- ---------------++-- | Skip elements of a stream that fail a monadic test+filterM :: (Monad m) => (a -> m Bool) -> Stream (Of a) m r -> Stream (Of a) m r+filterM pred = loop where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect $ liftM loop m+ Step (a :> as) -> Effect $ do+ bool <- pred a+ if bool+ then return $ Step (a :> loop as)+ else return $ loop as+{-# INLINABLE filterM #-}+++-- -- ---------------+-- -- first+-- -- ---------------+-- {- | Take either the first item in a stream or the return value, if it is empty.+-- The typical mark of an infinite stream is a polymorphic return value; in+-- that case, 'first' is a sort of @safeHead@+--+-- To iterate an action returning a 'Maybe', until it succeeds.+--+-- -}+-- first :: Monad m => Stream (Of r) m r -> m r+-- first = loop where+-- loop str = case str of+-- Return r -> return r+-- Effect m -> m >>= loop+-- Step (r :> rest) -> return r+-- {-# INLINABLE first #-}++-- ---------------+-- fold+-- ---------------++{- $folds+ Use these to fold the elements of a 'Stream'.++>>> S.fold_ (+) 0 id $ S.each [1..0]+50++ The general folds 'fold', fold_', 'foldM' and 'foldM_' are arranged+ for use with @Control.Foldl@ 'Control.Foldl.purely' and 'Control.Foldl.impurely'++>>> L.purely fold_ L.sum $ each [1..10]+55+>>> L.purely fold_ (liftA3 (,,) L.sum L.product L.list) $ each [1..10]+(55,3628800,[1,2,3,4,5,6,7,8,9,10])++ All functions marked with an underscore omit+ (e.g. @fold_@, @sum_@) the stream's return value in a left-strict pair.+ They are good for exiting streaming completely,+ but when you are, e.g. @mapped@-ing over a @Stream (Stream (Of a) m) m r@,+ which is to be compared with @[[a]]@. Specializing, we have e.g.++> mapped sum :: (Monad m, Num n) => Stream (Stream (Of Int)) IO () -> Stream (Of n) IO ()+> mapped (fold mappend mempty id) :: Stream (Stream (Of Int)) IO () -> Stream (Of Int) IO ()++>>> S.print $ mapped S.sum $ chunksOf 3 $ S.each [1..10]+6+15+24+10++>>> let three_folds = L.purely S.fold (liftA3 (,,) L.sum L.product L.list)+>>> S.print $ mapped three_folds $ chunksOf 3 (each [1..10])+(6,6,[1,2,3])+(15,120,[4,5,6])+(24,504,[7,8,9])+(10,10,[10])+-}++{-| Strict fold of a 'Stream' of elements, preserving only the result of the fold, not+ the return value of the stream. The third parameter will often be 'id' where a fold+ is written by hand:++>>> S.fold_ (+) 0 id $ each [1..10]+55++ It can be used to replace a standard Haskell type with one more suited to+ writing a strict accumulation function. It is also crucial to the+ Applicative instance for @Control.Foldl.Fold@++> Control.Foldl.purely fold :: Monad m => Fold a b -> Stream (Of a) m () -> m b+-}+fold_ :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> m b+fold_ step begin done = liftM (\(a:>rest) -> a) . fold step begin done+{-#INLINE fold_ #-}++{-| Strict fold of a 'Stream' of elements that preserves the return value.+ The third parameter will often be 'id' where a fold is written by hand:++>>> S.fold (+) 0 id $ each [1..10]+55 :> ()++>>> S.fold (*) 1 id $ S.fold (+) 0 id $ S.copy $ each [1..10]+3628800 :> (55 :> ())++ It can be used to replace a standard Haskell type with one more suited to+ writing a strict accumulation function. It is also crucial to the+ Applicative instance for @Control.Foldl.Fold@ We can apply such a fold+ @purely@++> Control.Foldl.purely S.fold :: Monad m => Fold a b -> Stream (Of a) m r -> m (Of b r)++ Thus, specializing a bit:++> L.purely S.fold L.sum :: Stream (Of Int) Int r -> m (Of Int r)+> mapped (L.purely S.fold L.sum) :: Stream (Stream (Of Int)) IO r -> Stream (Of Int) IO r++ Here we use the Applicative instance for @Control.Foldl.Fold@ to+ stream three-item segments of a stream together with their sums and products.++>>> S.print $ mapped (L.purely S.fold (liftA3 (,,) L.list L.product L.sum)) $ chunksOf 3 $ each [1..10]+([1,2,3],6,6)+([4,5,6],120,15)+([7,8,9],504,24)+([10],10,10)++-}++fold :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> m (Of b r)+fold step begin done str = fold_loop str begin+ where+ fold_loop stream !x = case stream of+ Return r -> return (done x :> r)+ Effect m -> m >>= \str' -> fold_loop str' x+ Step (a :> rest) -> fold_loop rest $! step x a+{-# INLINE fold #-}+++{-| Strict, monadic fold of the elements of a 'Stream (Of a)'++> Control.Foldl.impurely foldM :: Monad m => FoldM a b -> Stream (Of a) m () -> m b+-}+foldM_+ :: Monad m+ => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r -> m b+foldM_ step begin done = liftM (\(a :> rest) -> a) . foldM step begin done+{-#INLINE foldM_ #-}++{-| Strict, monadic fold of the elements of a 'Stream (Of a)'++> Control.Foldl.impurely foldM' :: Monad m => FoldM a b -> Stream (Of a) m r -> m (b, r)++ Thus to accumulate the elements of a stream as a vector, together with a random+ element we might write:++>>> L.impurely S.foldM (liftA2 (,) L.vector L.random) $ each [1..10::Int] :: IO (Of (U.Vector Int,Maybe Int) ())+([1,2,3,4,5,6,7,8,9,10],Just 9) :> ()++-}+foldM+ :: Monad m+ => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r ->m (Of b r)++foldM step begin done str = do+ x0 <- begin+ loop str x0+ where+ loop stream !x = case stream of+ Return r -> done x >>= \b -> return (b :> r)+ Effect m -> m >>= \s -> loop s x+ Step (a :> rest) -> do+ x' <- step x a+ loop rest x'+{-# INLINABLE foldM #-}++-- the following requires GHC.Magic.oneShot:+-- foldM step begin done str = do+-- x <- begin+-- (x' :> r) <- streamFold+-- (\r x -> return (x :> r))+-- (\mx2mx -> oneShot (\x -> x `seq` mx2mx >>= ($ x) ))+-- (\(a :> x2mx') -> oneShot (\x -> x `seq` (step x a >>= x2mx')) )+-- ( str)+-- x+-- b <- done x'+-- return (b :> r)+-- where seq = Prelude.seq+-- {-#INLINE foldM #-}++{-| A natural right fold for consuming a stream of elements.+ See also the more general 'iterTM' in the 'Streaming' module+ and the still more general 'destroy'++> foldrT (\a p -> Streaming.yield a >> p) = id+> foldrT (\a p -> Pipes.yield a >> p) :: Monad m => Stream (Of a) m r -> Producer a m r+> foldrT (\a p -> Conduit.yield a >> p) :: Monad m => Stream (Of a) m r -> Conduit a m r++-}++foldrT :: (Monad m, MonadTrans t, Monad (t m))+ => (a -> t m r -> t m r) -> Stream (Of a) m r -> t m r+foldrT step = loop where+ loop stream = case stream of+ Return r -> return r+ Effect m -> lift m >>= loop+ Step (a :> as) -> step a (loop as)+{-# INLINABLE foldrT #-}++{-| A natural right fold for consuming a stream of elements.+ See also the more general 'iterT' in the 'Streaming' module and the+ still more general 'destroy'+-}+foldrM :: Monad m+ => (a -> m r -> m r) -> Stream (Of a) m r -> m r+foldrM step = loop where+ loop stream = case stream of+ Return r -> return r+ Effect m -> m >>= loop+ Step (a :> as) -> step a (loop as)+{-# INLINABLE foldrM #-}++-- ---------------+-- for+-- ---------------++-- | @for@ replaces each element of a stream with an associated stream. Note that the+-- associated stream may layer any functor.+for :: (Monad m, Functor f) => Stream (Of a) m r -> (a -> Stream f m x) -> Stream f m r+for str0 act = loop str0 where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect $ liftM loop m+ Step (a :> rest) -> do+ act a+ loop rest+{-# INLINABLE for #-}++-- -| Group layers of any functor by comparisons on a preliminary annotation++-- groupedBy+-- :: (Monad m, Functor f) =>+-- (a -> a -> Bool)+-- -> Stream (Compose (Of a) f) m r+-- -> Stream (Stream (Compose (Of a) f) m) m r+-- groupedBy equals = loop where+-- loop stream = Effect $ do+-- e <- inspect stream+-- return $ case e of+-- Left r -> Return r+-- Right s@(Compose (a :> p')) -> Step $+-- fmap loop (Step $ Compose (a :> fmap (span' (equals a)) p'))+-- span' :: (Monad m, Functor f) => (a -> Bool) -> Stream (Compose (Of a) f) m r+-- -> Stream (Compose (Of a) f) m (Stream (Compose (Of a) f) m r)+-- span' pred = loop where+-- loop str = case str of+-- Return r -> Return (Return r)+-- Effect m -> Effect $ liftM loop m+-- Step s@(Compose (a :> rest)) -> case pred a of+-- True -> Step (Compose (a :> fmap loop rest))+-- False -> Return (Step s)+-- {-# INLINABLE groupedBy #-}++{-| Group elements of a stream in accordance with the supplied comparison.+++>>> S.print $ mapped S.toList $ S.groupBy (>=) $ each [1,2,3,1,2,3,4,3,2,4,5,6,7,6,5]+[1]+[2]+[3,1,2,3]+[4,3,2,4]+[5]+[6]+[7,6,5]++-}+groupBy :: Monad m+ => (a -> a -> Bool)+ -> Stream (Of a) m r+ -> Stream (Stream (Of a) m) m r+groupBy equals = loop where+ loop stream = Effect $ do+ e <- next stream+ return $ case e of+ Left r -> Return r+ Right (a, p') -> Step $+ fmap loop (yield a >> span (equals a) p')+{-# INLINABLE groupBy #-}+++{-| Group successive equal items together++>>> S.toList $ mapped S.toList $ S.group $ each "baaaaad"+["b","aaaaa","d"] :> ()++>>> S.toList $ concats $ maps (S.drained . S.splitAt 1) $ S.group $ each "baaaaaaad"+"bad" :> ()++-}+group :: (Monad m, Eq a) => Stream (Of a) m r -> Stream (Stream (Of a) m) m r+group = groupBy (==)+{-#INLINE group #-}+++head :: Monad m => Stream (Of a) m r -> m (Of (Maybe a) r)+head str = case str of+ Return r -> return (Nothing :> r)+ Effect m -> m >>= head+ Step (a :> rest) -> effects rest >>= \r -> return (Just a :> r)+{-#INLINABLE head #-}++head_ :: Monad m => Stream (Of a) m r -> m (Maybe a)+head_ str = case str of+ Return r -> return Nothing+ Effect m -> m >>= head_+ Step (a :> rest) -> effects rest >> return (Just a)+{-#INLINABLE head_ #-}++intersperse :: Monad m => a -> Stream (Of a) m r -> Stream (Of a) m r+intersperse x str = case str of+ Return r -> Return r+ Effect m -> Effect (liftM (intersperse x) m)+ Step (a :> rest) -> loop a rest+ where+ loop !a str = case str of+ Return r -> Step (a :> Return r)+ Effect m -> Effect (liftM (loop a) m)+ Step (b :> rest) -> Step (a :> Step (x :> loop b rest))+{-#INLINABLE intersperse #-}+++++-- ---------------+-- iterate+-- ---------------++{-| Iterate a pure function from a seed value, streaming the results forever++++-}+iterate :: Monad m => (a -> a) -> a -> Stream (Of a) m r+iterate f = loop where+ loop a' = Effect (return (Step (a' :> loop (f a'))))+{-# INLINABLE iterate #-}++-- | Iterate a monadic function from a seed value, streaming the results forever+iterateM :: Monad m => (a -> m a) -> m a -> Stream (Of a) m r+iterateM f = loop where+ loop ma = Effect $ do+ a <- ma+ return (Step (a :> loop (f a)))+{-# INLINABLE iterateM #-}++++last :: Monad m => Stream (Of a) m r -> m (Of (Maybe a) r)+last = loop Nothing_ where+ loop m str = case str of+ Return r -> case m of+ Nothing_ -> return (Nothing :> r)+ Just_ a -> return (Just a :> r)+ Effect m -> m >>= last+ Step (a :> rest) -> loop (Just_ a) rest+{-#INLINABLE last #-}++++last_ :: Monad m => Stream (Of a) m r -> m (Maybe a)+last_ = loop Nothing_ where+ loop m str = case str of+ Return r -> case m of+ Nothing_ -> return Nothing+ Just_ a -> return (Just a)+ Effect m -> m >>= last_+ Step (a :> rest) -> loop (Just_ a) rest+{-#INLINABLE last_ #-}++-- ---------------+-- length+-- ---------------++{-| Run a stream, remembering only its length:++>>> S.length $ S.each [1..10]+10++-}+length_ :: Monad m => Stream (Of a) m r -> m Int+length_ = fold_ (\n _ -> n + 1) 0 id+{-#INLINE length_#-}++{-| Run a stream, keeping its length and its return value.++>>> S.print $ mapped S.length $ chunksOf 3 $ S.each [1..10]+3+3+3+1++-}++length :: Monad m => Stream (Of a) m r -> m (Of Int r)+length = fold (\n _ -> n + 1) 0 id+{-#INLINE length #-}+-- ---------------+-- map+-- ---------------++{-| Standard map on the elements of a stream.++>>> S.stdoutLn $ S.map reverse $ each (words "alpha beta")+ahpla+ateb+-}++map :: Monad m => (a -> b) -> Stream (Of a) m r -> Stream (Of b) m r+map f = maps (\(x :> rest) -> f x :> rest)+-- loop where --+ -- loop stream = case stream of+ -- Return r -> Return r+ -- Effect m -> Effect (liftM loop m)+ -- Step (a :> as) -> Step (f a :> loop as)+{-# INLINABLE map #-}+-- {-# NOINLINE [1] map #-}+-- {-# RULES+-- "map/map" [~1] forall f g bs . map f (map g bs) =+-- map (f . g) bs+-- #-}++{-| Replace each element of a stream with the result of a monadic action++>>> S.print $ S.mapM readIORef $ S.chain (\ior -> modifyIORef ior (*100)) $ S.mapM newIORef $ each [1..6]+100+200+300+400+500+600+-}+mapM :: Monad m => (a -> m b) -> Stream (Of a) m r -> Stream (Of b) m r+mapM f = loop where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step (a :> as) -> Effect $ do+ a' <- f a+ return (Step (a' :> loop as) )+{-# INLINABLE mapM #-}++++{-| Reduce a stream to its return value with a monadic action.++>>> S.mapM_ Prelude.print $ each [1..5]+1+2+3+4+5+>>> rest <- S.mapM_ Prelude.print $ S.splitAt 3 $ each [1..10]+1+2+3+>>> S.sum rest+49 :> ()++-}+mapM_ :: Monad m => (a -> m b) -> Stream (Of a) m r -> m r+mapM_ f = loop where+ loop str = case str of+ Return r -> return r+ Effect m -> m >>= loop+ Step (a :> as) -> do+ f a+ loop as+{-# INLINABLE mapM_ #-}++++{- | Map layers of one functor to another with a transformation involving the base monad.+ This could be trivial, e.g.++> let noteBeginning text x = putStrLn text >> return text++ this puts the+ is completely functor-general++ @maps@ and @mapped@ obey these rules:++> maps id = id+> mapped return = id+> maps f . maps g = maps (f . g)+> mapped f . mapped g = mapped (f <=< g)+> maps f . mapped g = mapped (liftM f . g)+> mapped f . maps g = mapped (f <=< liftM g)++ @maps@ is more fundamental than @mapped@, which is best understood as a convenience+ for effecting this frequent composition:++> mapped phi = decompose . maps (Compose . phi)+++-}++mapped :: (Monad m, Functor f) => (forall x . f x -> m (g x)) -> Stream f m r -> Stream g m r+mapped = mapsM+{-#INLINE mapped #-}+++{-| Fold streamed items into their monoidal sum++>>> S.mconcat $ S.take 2 $ S.map (Data.Monoid.Last . Just) (S.stdinLn)+first<Enter>+last<Enter>+Last {getLast = Just "last"} :> ()++ -}+mconcat :: (Monad m, Monoid w) => Stream (Of w) m r -> m (Of w r)+mconcat = fold mappend mempty id+{-#INLINE mconcat #-}++data Maybe_ a = Just_ !a | Nothing_+mconcat_ :: (Monad m, Monoid w) => Stream (Of w) m r -> m w+mconcat_ = fold_ mappend mempty id++minimum :: (Monad m, Ord a) => Stream (Of a) m r -> m (Of (Maybe a) r)+minimum = fold (\m a -> case m of Nothing_ -> Just_ a ; Just_ a' -> Just_ (min a a'))+ Nothing_+ (\m -> case m of Nothing_ -> Nothing; Just_ r -> Just r)+{-#INLINE minimum #-}++minimum_ :: (Monad m, Ord a) => Stream (Of a) m r -> m (Maybe a)+minimum_ = fold_ (\m a -> case m of Nothing_ -> Just_ a ; Just_ a' -> Just_ (min a a'))+ Nothing_+ (\m -> case m of Nothing_ -> Nothing; Just_ r -> Just r)+{-#INLINE minimum_ #-}++maximum :: (Monad m, Ord a) => Stream (Of a) m r -> m (Of (Maybe a) r)+maximum = fold (\m a -> case m of Nothing_ -> Just_ a ; Just_ a' -> Just_ (max a a'))+ Nothing_+ (\m -> case m of Nothing_ -> Nothing; Just_ r -> Just r)+{-#INLINE maximum #-}++maximum_ :: (Monad m, Ord a) => Stream (Of a) m r -> m (Maybe a)+maximum_ = fold_ (\m a -> case m of Nothing_ -> Just_ a ; Just_ a' -> Just_ (max a a'))+ Nothing_+ (\m -> case m of Nothing_ -> Nothing; Just_ r -> Just r)+{-#INLINE maximum_ #-}++{-| The standard way of inspecting the first item in a stream of elements, if the+ stream is still \'running\'. The @Right@ case contains a+ Haskell pair, where the more general @inspect@ would return a left-strict pair.+ There is no reason to prefer @inspect@ since, if the @Right@ case is exposed,+ the first element in the pair will have been evaluated to whnf.++> next :: Monad m => Stream (Of a) m r -> m (Either r (a, Stream (Of a) m r))+> inspect :: Monad m => Stream (Of a) m r -> m (Either r (Of a (Stream (Of a) m r)))++ Interoperate with @pipes@ producers thus:++> Pipes.unfoldr Stream.next :: Stream (Of a) m r -> Producer a m r+> Stream.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r++ Similarly:++> IOStreams.unfoldM (liftM (either (const Nothing) Just) . next) :: Stream (Of a) IO b -> IO (InputStream a)+> Conduit.unfoldM (liftM (either (const Nothing) Just) . next) :: Stream (Of a) m r -> Source a m r++ But see 'uncons', which is better fitted to these @unfoldM@s+-}+next :: Monad m => Stream (Of a) m r -> m (Either r (a, Stream (Of a) m r))+next = loop where+ loop stream = case stream of+ Return r -> return (Left r)+ Effect m -> m >>= loop+ Step (a :> rest) -> return (Right (a,rest))+{-# INLINABLE next #-}+++{-| Exhaust a stream deciding whether @a@ was an element.++-}++notElem :: (Monad m, Eq a) => a -> Stream (Of a) m r -> m (Of Bool r)+notElem a' = loop True where+ loop False str = liftM (False :>) (effects str)+ loop True str = case str of+ Return r -> return (True:> r)+ Effect m -> m >>= loop True+ Step (a:> rest) ->+ if a == a'+ then liftM (False :>) (effects rest)+ else loop True rest+{-#INLINABLE notElem #-}++notElem_ :: (Monad m, Eq a) => a -> Stream (Of a) m r -> m Bool+notElem_ a' = loop True where+ loop False str = return False+ loop True str = case str of+ Return r -> return True+ Effect m -> m >>= loop True+ Step (a:> rest) ->+ if a == a'+ then return False+ else loop True rest+{-#INLINABLE notElem_ #-}+++{-|+> filter p = hoist effects (partition p)++ -}+partition :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r+partition thus = loop where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect (liftM loop (lift m))+ Step (a :> rest) -> if thus a+ then Step (a :> loop rest)+ else Effect $ do+ yield a+ return (loop rest)+++{-| Separate left and right values in distinct streams. ('separate' is+ a more powerful, functor-general, equivalent using 'Sum' in place of 'Either').+ So, for example, to permit unlimited user+ input of @Int@s on condition of only two errors, we might write:++>>> S.toList $ S.print $ S.take 2 $ partitionEithers $ S.map readEither $ S.stdinLn :: IO (Of [Int] ())+1<Enter>+2<Enter>+qqqqqqqqqq<Enter>+"Prelude.read: no parse"+3<Enter>+rrrrrrrrrr<Enter>+"Prelude.read: no parse"+[1,2,3] :> ()++> partitionEithers = separate . maps S.eitherToSum+> lefts = hoist S.effects . partitionEithers+> rights = S.effects . partitionEithers+> rights = S.concat+-}+partitionEithers :: Monad m => Stream (Of (Either a b)) m r -> Stream (Of a) (Stream (Of b) m) r+partitionEithers = loop where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect (liftM loop (lift m))+ Step (Left a :> rest) -> Step (a :> loop rest)+ Step (Right b :> rest) -> Effect $ do+ yield b+ return (loop rest)+++-- | Fold a 'Stream' of numbers into their product+product_ :: (Monad m, Num a) => Stream (Of a) m () -> m a+product_ = fold_ (*) 1 id+{-# INLINE product_ #-}++{-| Fold a 'Stream' of numbers into their product with the return value++> maps' product' :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r+-}+product :: (Monad m, Num a) => Stream (Of a) m r -> m (Of a r)+product = fold (*) 1 id+{-# INLINE product #-}+++-- ---------------+-- read+-- ---------------++{- | Make a stream of strings into a stream of parsed values, skipping bad cases++>>> S.sum_ $ S.read $ S.takeWhile (/= "total") S.stdinLn :: IO Int+1000<Enter>+2000<Enter>+total<Enter>+3000+++-}+read :: (Monad m, Read a) => Stream (Of String) m r -> Stream (Of a) m r+read stream = for stream $ \str -> case readMaybe str of+ Nothing -> return ()+ Just r -> yield r+{-# INLINE read #-}++-- ---------------+-- repeat+-- ---------------+{-| Repeat an element /ad inf./ .++>>> S.print $ S.take 3 $ S.repeat 1+1+1+1+-}++repeat :: Monad m => a -> Stream (Of a) m r+repeat a = loop where loop = Effect (return (Step (a :> loop)))+{-# INLINE repeat #-}+++{-| Repeat a monadic action /ad inf./, streaming its results.++>>> S.toList $ S.take 2 $ repeatM getLine+one<Enter>+two<Enter>+["one","two"]+-}++repeatM :: Monad m => m a -> Stream (Of a) m r+repeatM ma = loop where+ loop = do+ a <- lift ma+ yield a+ loop+{-# INLINABLE repeatM #-}++-- ---------------+-- replicate+-- ---------------++-- | Repeat an element several times+replicate :: Monad m => Int -> a -> Stream (Of a) m ()+replicate n a | n <= 0 = return ()+replicate n a = loop n where+ loop 0 = Return ()+ loop m = Effect (return (Step (a :> loop (m-1))))+{-# INLINABLE replicate #-}++{-| Repeat an action several times, streaming the results.++>>> S.print $ S.replicateM 2 getCurrentTime+2015-08-18 00:57:36.124508 UTC+2015-08-18 00:57:36.124785 UTC++-}+replicateM :: Monad m => Int -> m a -> Stream (Of a) m ()+replicateM n ma | n <= 0 = return ()+replicateM n ma = loop n where+ loop 0 = Return ()+ loop n = Effect $ do+ a <- ma+ return (Step $ a :> loop (n-1))+{-# INLINABLE replicateM #-}++{-| Read an @IORef (Maybe a)@ or a similar device until it reads @Nothing@.+ @reread@ provides convenient exit from the @io-streams@ library++> reread readIORef :: IORef (Maybe a) -> Stream (Of a) IO ()+> reread Streams.read :: System.IO.Streams.InputStream a -> Stream (Of a) IO ()+-}+reread :: Monad m => (s -> m (Maybe a)) -> s -> Stream (Of a) m ()+reread step s = loop where+ loop = Effect $ do+ m <- step s+ case m of+ Nothing -> return (Return ())+ Just a -> return (Step (a :> loop))+{-# INLINABLE reread #-}++{-| Strict left scan, streaming, e.g. successive partial results.+++>>> S.print $ S.scan (++) "" id $ each (words "a b c d")+""+"a"+"ab"+"abc"+"abcd"++ 'scan' is fitted for use with @Control.Foldl@, thus:++>>> S.print $ L.purely S.scan L.list $ each [3..5]+[]+[3]+[3,4]+[3,4,5]++-}+scan :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> Stream (Of b) m r+scan step begin done = loop begin+ where+ loop !acc stream = do+ case stream of+ Return r -> Step (done acc :> Return r)+ Effect m -> Effect (liftM (loop acc) m)+ Step (a :> rest) -> Step (done acc :> loop (step acc a) rest)+{-#INLINABLE scan #-}++{-| Strict left scan, accepting a monadic function. It can be used with+ 'FoldM's from @Control.Foldl@ using 'impurely'. Here we yield+ a succession of vectors each recording++>>> let v = L.impurely scanM L.vector $ each [1..4::Int] :: Stream (Of (U.Vector Int)) IO ()+>>> S.print v+fromList []+fromList [1]+fromList [1,2]+fromList [1,2,3]+fromList [1,2,3,4]++-}+scanM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r -> Stream (Of b) m r+scanM step begin done str = do+ x <- lift begin+ loop x str+ where+ loop !x stream = do+ b <- lift (done x)+ yield b+ case stream of+ Return r -> Return r+ Effect m -> Effect (do+ stream' <- m+ return (loop x stream')+ )+ Step (a :> rest) -> Effect (do+ x' <- step x a+ return (loop x' rest)+ )+{-# INLINABLE scanM #-}++{- Label each element in a stream with a value accumulated according to a fold.+++>>> S.print $ S.scanned (*) 1 id $ S.each [100,200,300]+(100,100)+(200,20000)+(300,6000000)++>>> S.print $ L.purely S.scanned L.product $ S.each [100,200,300]+(100,100)+(200,20000)+(300,6000000)++-}++data Maybe' a = Just' a | Nothing'++scanned :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> Stream (Of (a,b)) m r+scanned step begin done = loop Nothing' begin+ where+ loop !m !x stream = do+ case stream of+ Return r -> return r+ Effect mn -> Effect $ liftM (loop m x) mn+ Step (a :> rest) -> do+ case m of+ Nothing' -> do+ let !acc = step x a+ yield (a, done acc)+ loop (Just' a) acc rest+ Just' _ -> do+ let !acc = done (step x a)+ yield (a, acc)+ loop (Just' a) (step x a) rest+{-# INLINABLE scanned #-}+++{-| Streams the number of seconds from the beginning of action++ Thus, to mark times of user input we might write something like:++>>> S.toList $ S.take 3 $ S.zip S.seconds S.stdinLn+a<Enter>+b<Enter>+c<Enter>+[(0.0,"a"),(1.088711,"b"),(3.7289649999999996,"c")] :> ()++ To restrict user input to some number of seconds, we might write:++>>> S.toList $ S.map fst $ S.zip S.stdinLn $ S.takeWhile (< 3) S.seconds+one<Enter>+two<Enter>+three<Enter>+four<Enter>+five<Enter>+["one","two","three","four","five"] :> ()++ This is of course does not interrupt an action that has already begun.++ -}++seconds :: Stream (Of Double) IO r+seconds = do+ e <- lift $ next preseconds+ case e of+ Left r -> return r+ Right (t, rest) -> do+ yield 0+ map (subtract t) rest+ where+ preseconds :: Stream (Of Double) IO r+ preseconds = do+ utc <- liftIO getCurrentTime+ map ((/1000000000) . nice utc) (repeatM getCurrentTime)+ where+ nice u u' = fromIntegral $ truncate (1000000000 * diffUTCTime u' u)++-- ---------------+-- sequence+-- ---------------++{-| Like the 'Data.List.sequence' but streaming. The result type is a+ stream of a\'s, /but is not accumulated/; the effects of the elements+ of the original stream are interleaved in the resulting stream. Compare:++> sequence :: Monad m => [m a] -> m [a]+> sequence :: Monad m => Stream (Of (m a)) m r -> Stream (Of a) m r++ This obeys the rule++-}+sequence :: Monad m => Stream (Of (m a)) m r -> Stream (Of a) m r+sequence = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> Effect $ liftM loop m+ Step (ma :> rest) -> Effect $ do+ a <- ma+ return (Step (a :> loop rest))+{-# INLINABLE sequence #-}++-- ---------------+-- show+-- ---------------++show :: (Monad m, Show a) => Stream (Of a) m r -> Stream (Of String) m r+show = map Prelude.show+{-# INLINE show #-}+-- ---------------+-- sum+-- ---------------++-- | Fold a 'Stream' of numbers into their sum+sum_ :: (Monad m, Num a) => Stream (Of a) m () -> m a+sum_ = fold_ (+) 0 id+{-# INLINE sum_ #-}++{-| Fold a 'Stream' of numbers into their sum with the return value++> mapped S.sum :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r+++>>> S.sum $ each [1..10]+55 :> ()++>>> (n :> rest) <- S.sum $ S.splitAt 3 $ each [1..10]+>>> print n+6+>>> (m :> rest') <- S.sum $ S.splitAt 3 rest+>>> print m+15+>>> S.print rest'+7+8+9++-}+sum :: (Monad m, Num a) => Stream (Of a) m r -> m (Of a r)+sum = fold (+) 0 id+{-# INLINABLE sum #-}++-- ---------------+-- span+-- ---------------++-- | Stream elements until one fails the condition, return the rest.+span :: Monad m => (a -> Bool) -> Stream (Of a) m r+ -> Stream (Of a) m (Stream (Of a) m r)+span pred = loop where+ loop str = case str of+ Return r -> Return (Return r)+ Effect m -> Effect $ liftM loop m+ Step (a :> rest) -> if pred a+ then Step (a :> loop rest)+ else Return (Step (a :> rest))+{-# INLINABLE span #-}+++{-| Split a stream of elements wherever a given element arises.+ The action is like that of 'Prelude.words'.++>>> S.stdoutLn $ mapped S.toList $ S.split ' ' $ each "hello world "+hello+world++-}++split :: (Eq a, Monad m) =>+ a -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r+split t = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step (a :> rest) ->+ if a /= t+ then Step (fmap loop (yield a >> break (== t) rest))+ else loop rest+{-#INLINABLE split #-}++{-| Split a succession of layers after some number, returning a streaming or+-- effectful pair. This function is the same as the 'splitsAt' exported by the+-- @Streaming@ module, but since this module is imported qualified, it can+-- usurp a Prelude name. It specializes to:++> splitAt :: (Monad m, Functor f) => Int -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r)++-}+splitAt :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Stream f m r)+splitAt = splitsAt+{-# INLINE splitAt #-}++-- -------+-- subst+-- -------+{-| Replace each element in a stream of individual values with a functorial+ layer of any sort. @subst = flip with@ and is more convenient in+ a sequence of compositions that transform a stream.++> with = flip subst+> for str f = concats $ subst f str+> subst f = maps (\(a:>r) -> r <$ f a)+> S.concat = concats . subst each+-}+subst :: (Monad m, Functor f) => (a -> f x) -> Stream (Of a) m r -> Stream f m r+subst f s = loop s where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step (a :> rest) -> Step (loop rest <$ f a)+{-#INLINABLE subst #-}+-- ---------------+-- take+-- ---------------++{-| End a stream after n elements; the original return value is thus lost.+ 'splitAt' preserves this information. Note that, like @splitAt@, this+ function is functor-general, so that, for example, you can @take@ not+ just a number of items from a stream of elements, but a number+ of substreams and the like.++>>> S.toList $ S.take 3 $ each "with"+"wit" :> ()++>>> runResourceT $ S.stdoutLn $ S.take 3 $ S.readFile "stream.hs"+import Streaming+import qualified Streaming.Prelude as S+import Streaming.Prelude (each, next, yield)+++-}++take :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m ()+take n0 str | n0 <= 0 = return ()+take n0 str = loop n0 str where+ loop 0 p = return ()+ loop n p =+ case p of Step fas -> Step (fmap (loop (n-1)) fas)+ Effect m -> Effect (liftM (loop n) m)+ Return r -> Return ()+{-# INLINABLE take #-}++-- ---------------+-- takeWhile+-- ---------------++{-| End stream when an element fails a condition; the original return value is lost.+ By contrast 'span' preserves this information, and is generally more desirable.++> S.takeWhile thus = void . S.span thus++ To preserve the information - but thus also force the rest of the stream to be+ developed - write++> S.drained . S.span thus++ as @dropWhile thus@ is++> S.effects . S.span thus++-}+takeWhile :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m ()+takeWhile pred = loop where+ loop str = case str of+ Step (a :> as) -> when (pred a) (Step (a :> loop as))+ Effect m -> Effect (liftM loop m)+ Return r -> Return ()+{-# INLINE takeWhile #-}++{-| Like 'takeWhile', but takes a monadic predicate. -}+takeWhileM :: Monad m => (a -> m Bool) -> Stream (Of a) m r -> Stream (Of a) m ()+takeWhileM pred = loop where+ loop str = case str of+ Step (a :> as) -> do+ b <- lift (pred a)+ when b (Step (a :> loop as))+ Effect m -> Effect (liftM loop m)+ Return r -> Return ()+{-# INLINE takeWhileM #-}+++{-| Convert an effectful 'Stream (Of a)' into a list of @as@++ Note: Needless to say, this function does not stream properly.+ It is basically the same as Prelude 'mapM' which, like 'replicateM',+ 'sequence' and similar operations on traversable containers+ is a leading cause of space leaks.++-}+toList_ :: Monad m => Stream (Of a) m () -> m [a]+toList_ = fold_ (\diff a ls -> diff (a: ls)) id (\diff -> diff [])+{-# INLINE toList_ #-}+++{-| Convert an effectful 'Stream' into a list alongside the return value++> mapped toList :: Stream (Stream (Of a)) m r -> Stream (Of [a]) m++ Like 'toList_', it breaks streaming; unlike 'toList_' it preserves+ the return value and thus is frequently useful with e.g. 'mapped'++>>> S.print $ mapped S.toList $ chunksOf 3 $ each [1..9]+[1,2,3]+[4,5,6]+[7,8,9]++-}+toList :: Monad m => Stream (Of a) m r -> m (Of [a] r)+toList = fold (\diff a ls -> diff (a: ls)) id (\diff -> diff [])+{-# INLINE toList #-}+++{-| Inspect the first item in a stream of elements, without a return value.+ @uncons@ provides convenient exit into another streaming type:++> IOStreams.unfoldM uncons :: Stream (Of a) IO b -> IO (InputStream a)+> Conduit.unfoldM uncons :: Stream (Of a) m r -> Conduit.Source m a++-}+uncons :: Monad m => Stream (Of a) m () -> m (Maybe (a, Stream (Of a) m ()))+uncons = loop where+ loop stream = case stream of+ Return () -> return Nothing+ Effect m -> m >>= loop+ Step (a :> rest) -> return (Just (a,rest))+{-# INLINABLE uncons #-}+++{-| Build a @Stream@ by unfolding steps starting from a seed.++ The seed can of course be anything, but this is one natural way+ to consume a @pipes@ 'Pipes.Producer'. Consider:++>>> S.stdoutLn $ S.take 2 $ S.unfoldr P.next P.stdinLn+hello<Enter>+hello+goodbye<Enter>+goodbye++>>> S.stdoutLn $ S.unfoldr P.next (P.stdinLn P.>-> P.take 2)+hello<Enter>+hello+goodbye<Enter>+goodbye++>>> S.effects $ S.unfoldr P.next (P.stdinLn P.>-> P.take 2 P.>-> P.stdoutLn)+hello<Enter>+hello+goodbye<Enter>+goodbye++-}+unfoldr :: Monad m+ => (s -> m (Either r (a, s))) -> s -> Stream (Of a) m r+unfoldr step = loop where+ loop s0 = Effect (do+ e <- step s0+ case e of+ Left r -> return (Return r)+ Right (a,s) -> return (Step (a :> loop s)))+{-# INLINABLE unfoldr #-}++-- ---------------------------------------+-- untilRight+-- ---------------------------------------+untilRight :: Monad m => m (Either a r) -> Stream (Of a) m r+untilRight act = Effect loop where+ loop = do+ e <- act+ case e of+ Right r -> return (Return r)+ Left a -> return (Step (a :> Effect loop))+{-#INLINABLE untilRight #-}++-- ---------------------------------------+-- with+-- ---------------------------------------++{-| Replace each element in a stream of individual Haskell values (a @Stream (Of a) m r@) with an associated 'functorial' step.++> for str f = concats (with str f)+> with str f = for str (yields . f)+> with str f = maps (\(a:>r) -> r <$ f a) str+> with = flip subst+> subst = flip with++>>> with (each [1..3]) (yield . show) & intercalates (yield "--") & S.stdoutLn+1+--+2+--+3+ -}+with :: (Monad m, Functor f) => Stream (Of a) m r -> (a -> f x) -> Stream f m r+with s f = loop s where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step (a :> rest) -> Step (loop rest <$ f a)+{-#INLINABLE with #-}++-- ---------------------------------------+-- yield+-- ---------------------------------------++{-| A singleton stream++>>> stdoutLn $ yield "hello"+hello++>>> S.sum $ do {yield 1; yield 2}+3++>>> let prompt = putStrLn "Enter a number:"+>>> let number = lift (prompt >> readLn) >>= yield :: Stream (Of Int) IO ()+>>> S.toList $ do {number; number; number}+Enter a number:+1+Enter a number:+2+Enter a number:+3+[1,2,3] :> ()++-}++yield :: Monad m => a -> Stream (Of a) m ()+yield a = Step (a :> Return ())+{-# INLINE yield #-}++-- | Zip two 'Streams's+zip :: Monad m+ => (Stream (Of a) m r)+ -> (Stream (Of b) m r)+ -> (Stream (Of (a,b)) m r)+zip = zipWith (,)+{-# INLINE zip #-}++-- | Zip two 'Streams's using the provided combining function+zipWith :: Monad m+ => (a -> b -> c)+ -> (Stream (Of a) m r)+ -> (Stream (Of b) m r)+ -> (Stream (Of c) m r)+zipWith f = loop+ where+ loop str0 str1 = case str0 of+ Return r -> Return r+ Effect m -> Effect $ liftM (\str -> loop str str1) m+ Step (a :> rest0) -> case str1 of+ Return r -> Return r+ Effect m -> Effect $ liftM (loop str0) m+ Step (b :> rest1) -> Step (f a b :>loop rest0 rest1)+{-# INLINABLE zipWith #-}+++-- | Zip three 'Stream's with a combining function+zipWith3 :: Monad m =>+ (a -> b -> c -> d)+ -> Stream (Of a) m r+ -> Stream (Of b) m r+ -> Stream (Of c) m r+ -> Stream (Of d) m r+zipWith3 op = loop where+ loop str0 str1 str2 = do+ e0 <- lift (next str0)+ case e0 of+ Left r0 -> return r0+ Right (a0,rest0) -> do+ e1 <- lift (next str1)+ case e1 of+ Left r1 -> return r1+ Right (a1,rest1) -> do+ e2 <- lift (next str2)+ case e2 of+ Left r2 -> return r2+ Right (a2,rest2) -> do+ yield (op a0 a1 a2)+ loop rest0 rest1 rest2+{-# INLINABLE zipWith3 #-}+++-- | Zip three streams together+zip3 :: Monad m+ => (Stream (Of a) m r)+ -> (Stream (Of b) m r)+ -> (Stream (Of c) m r)+ -> (Stream (Of (a,b,c)) m r)+zip3 = zipWith3 (,,)+{-# INLINABLE zip3 #-}++-- --------------+-- IO fripperies+-- --------------++{-| View standard input as a 'Stream (Of String) m r'. 'stdoutLn', by+ contrast, renders a 'Stream (Of String) m r' to standard output. The names+ follow @Pipes.Prelude@++>>> stdoutLn stdinLn+hello<Enter>+hello+world<Enter>+world+^CInterrupted.+++>>> stdoutLn $ S.map reverse stdinLn+hello<Enter>+olleh+world<Enter>+dlrow+^CInterrupted.++-}+stdinLn :: MonadIO m => Stream (Of String) m ()+stdinLn = fromHandle IO.stdin+{-# INLINABLE stdinLn #-}++{-| Read values from 'IO.stdin', ignoring failed parses++>>> S.sum_ $ S.take 2 S.readLn :: IO Int+10<Enter>+12<Enter>+22++>>> S.toList $ S.take 3 (S.readLn :: Stream (Of Int) IO ())+1<Enter>+2<Enter>+1@#$%^&*\<Enter>+3<Enter>+[1,2,3] :> ()++-}++readLn :: (MonadIO m, Read a) => Stream (Of a) m ()+readLn = for stdinLn $ \str -> case readMaybe str of+ Nothing -> return ()+ Just n -> yield n+{-# INLINABLE readLn #-}+++{-| Read 'String's from a 'IO.Handle' using 'IO.hGetLine'++ Terminates on end of input++>>> IO.withFile "/usr/share/dict/words" IO.ReadMode $ S.stdoutLn . S.take 3 . S.drop 50000 . S.fromHandle+deflagrator+deflate+deflation++-}+fromHandle :: MonadIO m => IO.Handle -> Stream (Of String) m ()+fromHandle h = go+ where+ go = do+ eof <- liftIO $ IO.hIsEOF h+ unless eof $ do+ str <- liftIO $ IO.hGetLine h+ yield str+ go+{-# INLINABLE fromHandle #-}++{-| Write a succession of strings to a handle as separate lines.++>>> S.toHandle IO.stdout $ each $ words "one two three"+one+two+three+-}+toHandle :: MonadIO m => IO.Handle -> Stream (Of String) m r -> m r+toHandle handle = loop where+ loop str = case str of+ Return r -> return r+ Effect m -> m >>= loop+ Step (s :> rest) -> do+ liftIO (IO.hPutStrLn handle s)+ loop rest+{-# INLINABLE toHandle #-}++{-| Print the elements of a stream as they arise.++>>> S.print $ S.take 2 S.stdinLn+hello+"hello"+world+"world"+>>>++-}+print :: (MonadIO m, Show a) => Stream (Of a) m r -> m r+print = loop where+ loop stream = case stream of+ Return r -> return r+ Effect m -> m >>= loop+ Step (a :> rest) -> do+ liftIO (Prelude.print a)+ loop rest+++{-| Write 'String's to 'IO.stdout' using 'putStrLn'; terminates on a broken output pipe+ (This operation is modelled on 'Pipes.Prelude.stdoutLn').++>>> S.stdoutLn $ S.take 3 $ S.each $ words "one two three four five"+one+two+three+-}+stdoutLn :: MonadIO m => Stream (Of String) m () -> m ()+stdoutLn = loop+ where+ loop stream = case stream of+ Return _ -> return ()+ Effect m -> m >>= loop+ Step (s :> rest) -> do+ x <- liftIO $ try (putStrLn s)+ case x of+ Left (G.IOError { G.ioe_type = G.ResourceVanished+ , G.ioe_errno = Just ioe })+ | Errno ioe == ePIPE+ -> return ()+ Left e -> liftIO (throwIO e)+ Right () -> loop rest+{-# INLINABLE stdoutLn #-}+++++{-| Write 'String's to 'IO.stdout' using 'putStrLn'++ This does not handle a broken output pipe, but has a polymorphic return+ value, which makes this possible:++>>> rest <- S.stdoutLn' $ S.show $ S.splitAt 3 (each [1..5])+1+2+3+>>> S.print rest+4+5++-}++stdoutLn' :: MonadIO m => Stream (Of String) m r -> m r+stdoutLn' = loop where+ loop stream = case stream of+ Return r -> return r+ Effect m -> m >>= loop+ Step (s :> rest) -> liftIO (putStrLn s) >> loop rest+{-# INLINE stdoutLn' #-}++{-| Read the lines of a file as Haskell 'String's++>>> runResourceT $ S.writeFile "lines.txt" $ S.take 2 S.stdinLn+hello<Enter>+world<Enter>+>>> runResourceT $ S.print $ S.readFile "lines.txt"+"hello"+"world"++ 'runResourceT', as it is used here, means something like 'closing_all_handles'.+ It makes it possible to write convenient, fairly sensible versions of+ 'readFile', 'writeFile' and 'appendFile'. @IO.withFile IO.ReadMode ...@+ is more complicated but is generally to be preferred. Its use is explained+ <https://www.fpcomplete.com/user/snoyberg/library-documentation/resourcet here>.++-}++readFile :: MonadResource m => FilePath -> Stream (Of String) m ()+readFile f = bracketStream (IO.openFile f IO.ReadMode) (IO.hClose) fromHandle++{-| Write a series of strings as lines to a file. The handle is crudely+ managed with 'ResourceT':++>>> runResourceT $ S.writeFile "lines.txt" $ S.take 2 S.stdinLn+hello<Enter>+world<Enter>+>>> runResourceT $ S.print $ S.readFile "lines.txt"+"hello"+"world"++-}+writeFile :: MonadResource m => FilePath -> Stream (Of String) m r -> m r+writeFile f str = do+ (key, handle) <- allocate (IO.openFile f IO.WriteMode) (IO.hClose)+ r <- toHandle handle str+ release key+ return r++-- -- * Producers+-- -- $producers+-- stdinLn --+-- , readLn --+-- , fromHandle --+-- , repeatM --+-- , replicateM --+--+-- -- * Consumers+-- -- $consumers+-- , stdoutLn --+-- , stdoutLn' --+-- , mapM_ --+-- , print --+-- , toHandle --+-- , effects --+--+-- -- * Pipes+-- -- $pipes+-- , map --+-- , mapM --+-- , sequence --+-- , mapFoldable --+-- , filter --+-- , filterM --+-- , take --+-- , takeWhile --+-- , takeWhile' --+-- , drop --+-- , dropWhile --+-- , concat --+-- , elemIndices+-- , findIndices+-- , scan --+-- , scanM --+-- , chain --+-- , read --+-- , show --+-- , seq --+--+-- -- * Folds+-- -- $folds+-- , fold --+-- , fold' --+-- , foldM --+-- , foldM' --+-- , all+-- , any+-- , and+-- , or+-- , elem+-- , notElem+-- , find+-- , findIndex+-- , head+-- , index+-- , last+-- , length+-- , maximum+-- , minimum+-- , null+-- , sum --+-- , product --+-- , toList --+-- , toListM --+-- , toListM' --+--+-- -- * Zips+-- , zip --+-- , zipWith --+--++distinguish :: (a -> Bool) -> Of a r -> Sum (Of a) (Of a) r+distinguish predicate (a :> b) = if predicate a then InR (a :> b) else InL (a :> b)+{-#INLINE distinguish #-}++sumToEither ::Sum (Of a) (Of b) r -> Of (Either a b) r+sumToEither s = case s of+ InL (a :> r) -> Left a :> r+ InR (b :> r) -> Right b :> r+{-#INLINE sumToEither #-}++eitherToSum :: Of (Either a b) r -> Sum (Of a) (Of b) r+eitherToSum s = case s of+ Left a :> r -> InL (a :> r)+ Right b :> r -> InR (b :> r)+{-#INLINE eitherToSum #-}++composeToSum :: Compose (Of Bool) f r -> Sum f f r+composeToSum x = case x of+ Compose (True :> f) -> InR f+ Compose (False :> f) -> InL f+{-#INLINE composeToSum #-}++sumToCompose :: Sum f f r -> Compose (Of Bool) f r+sumToCompose x = case x of+ InR f -> Compose (True :> f)+ InL f -> Compose (False :> f)+{-#INLINE sumToCompose #-}++{-| Store the result of any suitable fold over a stream, keeping the stream for+ further manipulation. @store f = f . copy@ :++>>> S.print $ S.store S.product $ each [1..4]+1+2+3+4+24 :> ()++>>> S.print $ S.store S.sum $ S.store S.product $ each [1..4]+1+2+3+4+10 :> (24 :> ())++ Here the sum (10) and the product (24) have been \'stored\' for use when+ finally we have traversed the stream with 'print' . Needless to say,+ a second 'pass' is excluded conceptually, so the+ folds that you apply successively with @store@ are performed+ simultaneously, and in constant memory -- as they would be if,+ say, you linked them together with @Control.Fold@:++>>> L.impurely S.foldM (liftA3 (\a b c -> (b,c)) (L.sink print) (L.generalize L.sum) (L.generalize L.product)) $ each [1..4]+1+2+3+4+(10,24) :> ()++ Fusing folds after the fashion of @Control.Foldl@ will generally be a bit faster+ than the corresponding succession of uses of 'store', but by+ constant factor that will be completely dwarfed when any IO is at issue.++ But 'store' / 'copy' is /much/ more powerful, as you can see by reflecting on+ uses like this:++>>> S.sum $ S.store (S.sum . mapped S.product . chunksOf 2) $ S.store (S.product . mapped S.sum . chunksOf 2 )$ each [1..6]+21 :> (44 :> (231 :> ()))++ It will be clear that this cannot be reproduced with any combination of lenses,+ @Control.Fold@ folds, or the like. (See also the discussion of 'copy'.)++ It would conceivable be clearer to import a series of specializations of 'store'.+ It is intended to be used at types like these:++> storeM :: (forall s m . Monad m => Stream (Of a) m s -> m (Of b s))+> -> (Monad n => Stream (Of a) n r -> Stream (Of a) n (Of b r))+> storeM = store+>+> storeMIO :: (forall s m . MonadIO m => Stream (Of a) m s -> m (Of b s))+> -> ( MonadIO n => Stream (Of a) n r -> Stream (Of a) n (Of b r)+> storeMIO = store++ It is clear from these types that we are just using the general instances:++> instance (Functor f, Monad m ) => Monad (Stream f m)+> instance (Functor f, MonadIO m) => MonadIO (Stream f m)++ We thus can't be touching the elements of the stream, or the final return value.+ It it is the same with other constraints that @Stream (Of a)@ inherits,+ like 'MonadResource'. Thus I can filter and write to one file, but+ nub and write to another, or to a database or the like:++>>> runResourceT $ (S.writeFile "hello2.txt" . S.nub) $ store (S.writeFile "hello.txt" . S.filter (/= "world")) $ each ["hello", "world", "goodbye", "world"]+>>> :! cat hello.txt+hello+goodbye+>>> :! cat hello2.txt+hello+world+goodbye+++-}+store+ :: Monad m =>+ (Stream (Of a) (Stream (Of a) m) r -> t) -> Stream (Of a) m r -> t+store f x = f (copy x)+{-#INLINE store #-}++{-| Duplicate the content of stream, so that it can be acted on twice in different ways,+ but without breaking streaming. Thus, with @each [1,2]@ I might do:++>>> S.print $ each ["one","two"]+"one"+"two"+>>> S.stdoutLn $ each ["one","two"]+one+two++ With copy, I can as well do:++>>> S.print $ S.stdoutLn $ S.copy $ each ["one","two"]+one+"one"+two+"two"++ 'copy' should be understood together with 'effects' and is subject to the rules++> S.effects . S.copy = id+> hoist S.effects . S.copy = id++ The similar operations in 'Data.ByteString.Streaming' obey the same rules.++ Where the actions you are contemplating are each simple folds over+ the elements, or a selection of elements, then the coupling of the+ folds is often more straightforwardly effected with `Control.Foldl`,+ e.g.++>>> L.purely S.fold (liftA2 (,) L.sum L.product) $ each [1..10]+(55,3628800) :> ()++ rather than++>>> S.sum $ S.product . S.copy $ each [1..10]+55 :> (3628800 :> ())++ A @Control.Foldl@ fold can be altered to act on a selection of elements by+ using 'Control.Foldl.handles' on an appropriate lens. Some such+ manipulations are simpler and more 'Data.List'-like, using 'copy':++>>> L.purely S.fold (liftA2 (,) (L.handles (filtered odd) L.sum) (L.handles (filtered even) L.product)) $ each [1..10]+(25,3840) :> ()++ becomes++>>> S.sum $ S.filter odd $ S.product $ S.filter even $ S.copy $ each [1..10]+25 :> (3840 :> ())++ or using 'store'++>>> S.sum $ S.filter odd $ S.store (S.product . S.filter even) $ each [1..10]+25 :> (3840 :> ())++ But anything that fold of a @Stream (Of a) m r@ into e.g. an @m (Of b r)@+ that has a constraint on @m@ that is carried over into @Stream f m@ -+ e.g. @Monad@, @MonadIO@, @MonadResource@, etc. can be used on the stream.+ Thus, I can fold over different groupings of the original stream:++>>> (S.toList . mapped S.toList . chunksOf 5) $ (S.toList . mapped S.toList . chunksOf 3) $ S.copy $ each [1..10]+[[1,2,3,4,5],[6,7,8,9,10]] :> ([[1,2,3],[4,5,6],[7,8,9],[10]] :> ())++ The procedure can be iterated as one pleases, as one can see from this (otherwise unadvisable!) example:++>>> (S.toList . mapped S.toList . chunksOf 4) $ (S.toList . mapped S.toList . chunksOf 3) $ S.copy $ (S.toList . mapped S.toList . chunksOf 2) $ S.copy $ each [1..12]+[[1,2,3,4],[5,6,7,8],[9,10,11,12]] :> ([[1,2,3],[4,5,6],[7,8,9],[10,11,12]] :> ([[1,2],[3,4],[5,6],[7,8],[9,10],[11,12]] :> ()))++-}+copy+ :: Monad m =>+ Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r+copy = Effect . return . loop where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect (liftM loop (lift m))+ Step (a :> rest) -> Effect (Step (a :> Return (Step (a :> loop rest))))++{-#INLINABLE copy#-}++duplicate+ :: Monad m =>+ Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r+duplicate = copy+{-#INLINE duplicate #-}+++{-| @copy'@ is the same as @copy@ but reverses the order of interleaved effects.+ The difference should not be observable at all for pure folds over the data.++-}+copy'+ :: Monad m =>+ Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r+copy' = Effect . return . loop where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect (liftM loop (lift m))+ Step (a :> rest) -> Step (a :> Effect (Step (a :> Return (loop rest))))+{-#INLINABLE copy' #-}++duplicate'+ :: Monad m =>+ Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r+duplicate' = copy'+{-#INLINE duplicate' #-}++{-| The type++> Data.List.unzip :: [(a,b)] -> ([a],[b])++ might lead us to expect++> Streaming.unzip :: Stream (Of (a,b)) m r -> Stream (Of a) m (Stream (Of b) m r)++ which would not stream, since it would have to accumulate the second stream (of @b@s).+ Of course, @Data.List@ 'Data.List.unzip' doesn't stream either.++ This @unzip@ does+ stream, though of course you can spoil this by using e.g. 'toList':++>>> let xs = map (\x-> (x,show x)) [1..5::Int]++>>> S.toList $ S.toList $ S.unzip (S.each xs)+["1","2","3","4","5"] :> ([1,2,3,4,5] :> ())++>>> Prelude.unzip xs+([1,2,3,4,5],["1","2","3","4","5"])++ Note the difference of order in the results. It may be of some use to think why.+ The first application of 'toList' was applied to a stream of integers:++>>> :t S.unzip $ S.each xs+S.unzip $ S.each xs :: Monad m => Stream (Of Int) (Stream (Of String) m) ()++ Like any fold, 'toList' takes no notice of the monad of effects.++> toList :: Monad m => Stream (Of a) m r -> m (Of [a] r)++ In the case at hand (since I am in @ghci@) @m = Stream (Of String) IO@.+ So when I apply 'toList', I exhaust that stream of integers, folding+ it into a list:++>>> :t S.toList $ S.unzip $ S.each xs+S.toList $ S.unzip $ S.each xs+ :: Monad m => Stream (Of String) m (Of [Int] ())++ When I apply 'toList' to /this/, I reduce everything to an ordinary action in @IO@,+ and return a list of strings:++>>> S.toList $ S.toList $ S.unzip (S.each xs)+["1","2","3","4","5"] :> ([1,2,3,4,5] :> ())++-}+unzip :: Monad m => Stream (Of (a,b)) m r -> Stream (Of a) (Stream (Of b) m) r+unzip = loop where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect (liftM loop (lift m))+ Step ((a,b):> rest) -> Step (a :> Effect (Step (b :> Return (loop rest))))+{-#INLINABLE unzip #-}+++-- "fold/map" forall step begin done f str .+-- fold step begin done (map f str) = fold (\x a -> step x $! f a) begin done str;+--+-- "fold/filter" forall step begin done pred str .+-- fold step begin done (filter pred str) = fold (\x a -> if pred a then step x a else x) begin done str;+--+-- "scan/map" forall step begin done f str .+-- scan step begin done (map f str) = scan (\x a -> step x $! f a) begin done str+--++{- $maybes+ These functions discard the 'Nothing's that they encounter. They are analogous+ to the functions from @Data.Maybe@ that share their names.+-}++{-| The 'catMaybes' function takes a 'Stream' of 'Maybe's and returns+ a 'Stream' of all of the 'Just' values.+-}+catMaybes :: Monad m => Stream (Of (Maybe a)) m r -> Stream (Of a) m r+catMaybes = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step (ma :> snext) -> case ma of+ Nothing -> loop snext+ Just a -> Step (a :> loop snext)+{-#INLINABLE catMaybes #-}++{-| The 'mapMaybe' function is a version of 'map' which can throw out elements. In particular,+ the functional argument returns something of type @'Maybe' b@. If this is 'Nothing', no element+ is added on to the result 'Stream'. If it is @'Just' b@, then @b@ is included in the result 'Stream'.+-}+mapMaybe :: Monad m => (a -> Maybe b) -> Stream (Of a) m r -> Stream (Of b) m r+mapMaybe phi = loop where+ loop stream = case stream of+ Return r -> Return r+ Effect m -> Effect (liftM loop m)+ Step (a :> snext) -> case phi a of+ Nothing -> loop snext+ Just b -> Step (b :> loop snext)+{-#INLINABLE mapMaybe #-}+
streaming.cabal view
@@ -1,5 +1,5 @@ name: streaming-version: 0.1.4.2+version: 0.1.4.3 cabal-version: >=1.10 build-type: Simple synopsis: an elementary streaming prelude and general stream type.@@ -8,55 +8,60 @@ a simple \"source\" or \"producer\" type, namely @Stream (Of a) m r@. @Stream (Of a) m r@ is a sort of effectful version of @([a],r)@ in which successive elements arise from some sort of monadic- action. Everything is the library is organized to make - programming with this type as simple as possible- by making it as close to @Prelude@ and @Data.List@. Thus for example+ action. Everything in the library is organized to make+ programming with this type as simple as possible,+ by the simple expedient of making it as close to @Prelude@+ and @Data.List@ as possible. Thus for example the trivial program .- > S.sum (S.take 3 (S.readLn :: Stream (Of Int) IO ()))+ > S.sum (S.take 3 (S.readLn :: Stream (Of Integer) IO ())) . sums the first three valid integers from user input. Similarly, .- > S.stdoutLn (S.map reverse (S.take 3 S.stdinLn)) + > S.stdoutLn (S.map reverse (S.take 3 S.stdinLn)) .- reverses the first three lines from stdin as they arise, + reverses the first three lines from stdin as they arise, and sends them to stdout. And so on,- with filtering, mapping, breaking, chunking and so forth. - We program with streams of @Int@s or @String@s directly as - if they constituted something like a list rather than \"extracting a list from IO\",- which is the origin of typical Haskell memory catastrophes. - Basically any case where you are + with filtering, mapping, breaking, chunking and so forth.+ We program with streams of @Int@s or @String@s directly as+ if they constituted something like a list. And we everywhere+ oppose \"extracting a list from IO\",+ which is the origin of typical Haskell memory catastrophes.+ Basically any case where you are tempted to use @mapM@, @replicateM@, @traverse@ or @sequence@ with Haskell lists, you would do better to use something like- @Stream (Of a) m r@. The type signatures are a little fancier, but - the programs themselves are mostly the same or simpler. Thus, - the little demo program from+ @Stream (Of a) m r@. The type signatures are a little fancier, but+ the programs themselves are mostly the same or simpler. Thus,+ consider the trivial demo program mentioned in <http://stackoverflow.com/questions/24068399/haskell-performance-of-iorefs this SO question> . > main = mapM newIORef [1..10^8::Int] >>= mapM readIORef >>= mapM_ print .- quickly exhausts memory; this of course has nothing to do with @IORefs@ - and is cured by+ It quickly exhausts memory, of course, and this has nothing to do with+ the efficiency of @IORefs@. It is immediately cured by writing . > import qualified Streaming.Prelude as S > main = S.print (S.mapM readIORef (S.mapM newIORef (S.each [1..10^8::Int]))) .- which uses no more memory than @hello-world@, and is simpler anyway, since it+ which really does what the other program was meant to do,+ uses no more memory than @hello-world@, and is simpler anyway, since it doesn't involve \"extracting a list from IO\". Almost every use of list @mapM@, @replicateM@, @traverse@ and @sequence@ produces this problem on a smaller scale. People get used to it, as if it were characteristic of Haskell programs to use a lot of memory, when \"extracting a list or sequence from IO\" is just bad practice pure and simple.- List @mapM@, @replicateM@, @traverse@ and @sequence@ make sense under certain - conditions. Similarly, @unsafePerformIO@ makes sense under certain conditions.+ Of course, @mapM@, @replicateM@, @traverse@ and @sequence@ make sense for lists,+ under certain conditions. Similarly, @unsafePerformIO@ makes sense under+ certain conditions. . The @Streaming@ module exports the general type, @Stream f m r@, which can be used to stream successive distinct steps characterized by /any/ functor @f@, though we are mostly interested in organizing computations- of the form @Stream (Of a) m r@. The streaming-IO libraries have + of the form @Stream (Of a) m r@. The streaming-IO libraries have various devices for dealing- with effectful variants of @[a]@ or @([a],r)@. But it is only with+ with effectful variants of @[a]@ or @([a],r)@ in which the emergence of+ successive elements somehow depends on IO. But it is only with the general type @Stream f m r@, or some equivalent, that one can envisage (for example) the connected streaming of their sorts of stream - as one makes lists of lists in the Haskell@@ -67,41 +72,43 @@ > chunksOf :: Int -> [a] -> [[a]] > lines :: [Char] -> [[Char]] -- but similarly with bytestring, etc. .- to mention a few obviously desirable operations. (This is explained more elaborately in the <https://hackage.haskell.org/package/streaming#readme readme> below.) One could throw something+ to mention a few obviously desirable operations.+ (This is explained more elaborately in the <https://hackage.haskell.org/package/streaming#readme readme> below.)+ One could throw something like @Stream@ on top of a prior stream concept: this is how @pipes@ and @pipes-group@ (which are very much our model here) use @FreeT@.- But once one grasps- the iterable stream concept needed to express those functions - - here given a somewhat optimized implementation as @Stream f m r@ - (following, as usual, models derived from the @pipes@ library) - + But once one grasps the iterable stream concept needed to express+ those functions -+ the one here given a somewhat optimized implementation as @Stream f m r@+ (the specific optimization again follows the model of the @pipes@ library) - then one will also see that, with it, one is /already/ in possession of a complete elementary streaming library - since one possesses @Stream ((,) a) m r@ or equivalently @Stream (Of a) m r@. This is the type of a \'generator\' or \'producer\' or whatever you call an effectful stream of items.- /The present @Streaming.Prelude@ is thus the simplest streaming library that can replicate anything like the API of the @Prelude@ and @Data.List@/. + /The present Streaming.Prelude is thus the simplest streaming library that can replicate anything like the API of the Prelude and Data.List/. . The emphasis of the library is on interoperation; for the rest its advantages are: extreme simplicity, re-use of intuitions the user has gathered from mastery of @Prelude@ and- @Data.List@, and a total and systematic rejection of type synonyms. + @Data.List@, and a total and systematic rejection of type synonyms. The two conceptual pre-requisites are some comprehension of monad transformers and some familiarity with \'rank 2 types\'. It is hoped that experimentation with this- simple material, starting with the ghci examples in @Streaming.Prelude@, - will give people who are new to these concepts some + simple material, starting with the ghci examples in @Streaming.Prelude@,+ will give people who are new to these concepts some intuition about their importance. The most fundamental purpose of the- library is to express elementary streaming ideas without reliance on + library is to express elementary streaming ideas without reliance on a complex framework, but in a way that integrates transparently with the rest of Haskell, using ideas - e.g. rank 2 types, which are here implicit or explicit in most mapping - that the user can carry elsewhere,- rather than binding her intelligence to a so-called streaming IO framework (as + rather than binding her intelligence to a so-called streaming IO framework (as necessary as that is for certain purposes.) . See the- <https://hackage.haskell.org/package/streaming#readme readme> - below for further explanation, including the examples linked there. + <https://hackage.haskell.org/package/streaming#readme readme>+ below for further explanation, including the examples linked there. Elementary usage can be divined from the ghci examples in @Streaming.Prelude@ and perhaps from this rough beginning of a <https://github.com/michaelt/streaming-tutorial/blob/master/tutorial.md tutorial>.@@ -110,37 +117,37 @@ and <https://hackage.haskell.org/package/streaming-utils streaming utils> packages. Questions about usage can be put- raised on StackOverflow with the tag @[haskell-streaming]@, - or as an issue on Github, or on the + raised on StackOverflow with the tag @[haskell-streaming]@,+ or as an issue on Github, or on the <https://groups.google.com/forum/#!forum/haskell-pipes pipes list> (the package understands itself as part of the pipes \'ecosystem\'.) .- The simplest form of interoperation with + The simplest form of interoperation with <http://hackage.haskell.org/package/pipes pipes> is accomplished with this isomorphism: . > Pipes.unfoldr Streaming.next :: Stream (Of a) m r -> Producer a m r > Streaming.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r .- Interoperation with - <http://hackage.haskell.org/package/io-streams io-streams> + Interoperation with+ <http://hackage.haskell.org/package/io-streams io-streams> is thus: . > Streaming.reread IOStreams.read :: InputStream a -> Stream (Of a) IO () > IOStreams.unfoldM Streaming.uncons :: Stream (Of a) IO () -> IO (InputStream a) .- With - <http://hackage.haskell.org/package/conduit conduit> + With+ <http://hackage.haskell.org/package/conduit conduit> one might use, e.g.: .- > Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a- > Streaming.mapM_ Conduit.yield . hoist lift :: Stream (Of o) m r -> ConduitM i o m r- > ($$ Conduit.mapM_ Streaming.yield) . hoist lift :: Source m a -> Stream (Of a) m ()+ > Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a+ > \str -> Streaming.mapM_ Conduit.yield (hoist lift str) :: Stream (Of o) m r -> ConduitM i o m r+ > \src -> hoist lift str $$ Conduit.mapM_ Streaming.yield :: Source m a -> Stream (Of a) m () .- These conversions should never be more expensive than a single @>->@ or @=$=@. + These conversions should never be more expensive than a single @>->@ or @=$=@. The simplest interoperation with regular Haskell lists is provided by, say .- > Streaming.each :: [a] -> Stream (Of a) m ()+ > Streaming.each :: [a] -> Stream (Of a) m () > Streaming.toList_ :: Stream (Of a) m r -> m [a] . The latter of course accumulates the whole list in memory, and is mostly what we are trying@@ -149,14 +156,14 @@ considering whether the accumulation required by @Streaming.toList_@ is really necessary. . Here are the results of some- <https://gist.github.com/michaelt/f19bef01423b17f29ffd microbenchmarks>+ <https://gist.github.com/michaelt/96606bbf05b29bf43a05aba081dc9bd4#file-benchmachines-hs microbenchmarks> based on the <https://github.com/ekmett/machines/blob/master/benchmarks/Benchmarks.hs benchmarks> included in the machines package: .- <<http://i.imgur.com/sSG5MvH.png>>+ <<http://i.imgur.com/YbQtlXm.png>> .- Because these are microbenchmarks for individual functions, + Because these are microbenchmarks for individual functions, they represent a sort of \"worst case\"; many other factors can influence the speed of a complex program. .@@ -199,9 +206,10 @@ , monad-control >=0.3.1 && <1.1 , time , ghc-prim- + hs-source-dirs: src default-language: Haskell2010+