streaming 0.1.3.4 → 0.1.4.0
raw patch · 4 files changed
+541/−184 lines, 4 files
Files
- Streaming.hs +8/−8
- Streaming/Internal.hs +6/−22
- Streaming/Prelude.hs +403/−106
- streaming.cabal +124/−48
Streaming.hs view
@@ -99,21 +99,21 @@ 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 -- via MFunctor+> 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+> 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-> groups: Stream (Sum f g) m r -> Stream (Sum (Stream f m) (Stream g m)) 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
Streaming/Internal.hs view
@@ -30,7 +30,6 @@ -- * Transforming streams , maps , mapsM - , mapped , decompose , mapsM_ , run@@ -393,6 +392,10 @@ 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@@ -402,26 +405,7 @@ Step f -> Effect (liftM Step (phi (fmap loop f))) {-# INLINABLE mapsM #-} -{- | Map layers of one functor to another with a transformation involving the base monad- @maps@ is more fundamental than @mapped@, which is best understood as a convenience- for effecting this frequent composition: -> mapped = mapsM -> mapsM phi = decompose . maps (Compose . phi) -- @mapped@ obeys these rules:--> mapped return = id-> mapped f . mapped g = mapped (f <=< g)-> map f . mapped g = mapped (liftM f . g)-> mapped f . map g = mapped (f . g)---}--mapped :: (Monad m, Functor f) => (forall x . f x -> m (g x)) -> Stream f m r -> Stream g m r-mapped = mapsM-{-#INLINE mapped #-}- {-| Resort a succession of layers of the form @m (f x)@. Though @mapsM@ is best understood as: @@ -600,9 +584,9 @@ => Stream f (t m) r -> t (Stream f m) r distribute = loop where loop stream = case stream of - Return r -> lift $ Return r+ Return r -> lift $ Return r Effect tmstr -> hoist lift tmstr >>= distribute- Step fstr -> join $ lift (Step (fmap (Return . distribute) fstr))+ Step fstr -> join $ lift (Step (fmap (Return . distribute) fstr)) -- | Repeat a functorial layer, command or instruction forever. repeats :: (Monad m, Functor f) => f () -> Stream f m r
Streaming/Prelude.hs view
@@ -3,9 +3,7 @@ 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. Thecan - be used with any - rational \"streaming IO\" system. + of Data.List and the Prelude. Import qualified thus: @@ -14,7 +12,7 @@ For the examples below, one sometimes needs -> import Streaming.Prelude (each, yield, stdoutLn, stdinLn)+> import Streaming.Prelude (each, yield, next, mapped, stdoutLn, stdinLn) > import Data.Function ((&)) Other libraries that come up in passing are@@ -63,12 +61,14 @@ , repeat , repeatM , replicate+ , untilRight , cycle , replicateM , enumFrom , enumFromThen , seconds + -- * Consuming streams of elements -- $consumers , stdoutLn@@ -77,7 +77,9 @@ , print , toHandle , writeFile+ , first , effects+ , erase , drained @@ -85,14 +87,19 @@ -- $pipes , map , mapM- , chain , maps+ , mapped+ , for+ , with+ , subst+ , copy+ , copy'+ , store+ , chain , sequence , nub , filter , filterM- , for- , with , delay , intersperse , take@@ -110,8 +117,9 @@ , show , cons , duplicate- , store+ , duplicate' + -- * Splitting and inspecting streams of elements , next , uncons@@ -158,6 +166,8 @@ , last_ , elem , elem_+ , notElem+ , notElem_ , length , length_ , toList@@ -177,7 +187,7 @@ -- , and -- , or -- , elem- -- , notElem+ -- , find -- , findIndex -- , head@@ -194,6 +204,8 @@ , zip3 , zipWith3 , unzip+ , partitionEithers+ , partition -- * Pair manipulation , lazily@@ -223,12 +235,13 @@ 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+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, intersperse, all, any, head, last)+ , minimum, maximum, elem, notElem, intersperse, all, any, head+ , last) import qualified GHC.IO.Exception as G import qualified System.IO as IO@@ -489,7 +502,7 @@ {-| Make a stream of traversable containers into a stream of their separate elements. This is just -> concat = for str each+> concat str = for str each >>> S.print $ S.concat (each ["xy","z"]) 'x'@@ -552,9 +565,7 @@ {-#INLINABLE cycle #-} -{-| A--+{-| 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@@ -580,8 +591,13 @@ 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@@ -642,7 +658,8 @@ -} drop :: (Monad m) => Int -> Stream (Of a) m r -> Stream (Of a) m r-drop = loop where+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@@ -683,17 +700,17 @@ {- | Stream the elements of a pure, foldable container. ->>> each [1..3] & S.print+>>> S.print $ each [1..3] 1 2 3->>> S.replicateM 5 getLine & chunksOf 3 & mapped S.toList & S.print-s-t-u+>>> S.print $ mapped S.toList $ chunksOf 3 $ S.replicateM 5 getLine+s<Enter>+t<Enter>+u<Enter> ["s","t","u"]-v-w+v<Enter>+w<Enter> ["v","w"] -}@@ -706,17 +723,27 @@ -} elem :: (Monad m, Eq a) => a -> Stream (Of a) m r -> m (Of Bool r)-elem a = fold op False id where- op True _ = True- op False a' | a == a' = True- op _ _ = False+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 = fold_ op False id where- op True _ = True- op False a' | a == a' = True- op _ _ = False+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_ #-} -- -----@@ -730,7 +757,7 @@ 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+>>> rest <- S.print $ S.zip (S.enumFrom 'a') $ S.splitAt 3 $ S.enumFrom 1 ('a',1) ('b',2) ('c',3)@@ -765,6 +792,19 @@ {-# 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 -- --------------- @@ -796,7 +836,26 @@ 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 -- --------------- @@ -807,7 +866,7 @@ 50 The general folds 'fold', fold_', 'foldM' and 'foldM_' are arranged - for use with 'Control.Foldl'+ for use with @Control.Foldl@ 'Control.Foldl.purely' and 'Control.Foldl.impurely' >>> L.purely fold_ L.sum $ each [1..10] 55@@ -860,10 +919,9 @@ >>> S.fold (+) 0 id $ each [1..10] 55 :> () ->>> S.fold (*) 1 id $ S.fold (+) 0 id $ S.duplicate $ each [1..10]+>>> 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@@ -874,7 +932,7 @@ Thus, specializing a bit: > L.purely S.fold L.sum :: Stream (Of Int) Int r -> m (Of Int r)-> maps (L.purely S.fold L.sum) :: Stream (Stream (Of Int)) IO r -> Stream (Of Int) IO 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.@@ -904,7 +962,7 @@ 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+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)'@@ -939,8 +997,9 @@ See also the more general 'iterTM' in the 'Streaming' module and the still more general 'destroy' -> 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 (\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 -} @@ -949,7 +1008,7 @@ foldrT step = loop where loop stream = case stream of Return r -> return r- Effect m -> lift m >>= loop+ Effect m -> lift m >>= loop Step (a :> as) -> step a (loop as) {-# INLINABLE foldrT #-} @@ -962,7 +1021,7 @@ foldrM step = loop where loop stream = case stream of Return r -> return r- Effect m -> m >>= loop+ Effect m -> m >>= loop Step (a :> as) -> step a (loop as) {-# INLINABLE foldrM #-} @@ -1122,7 +1181,7 @@ Effect m -> m >>= last_ Step (a :> rest) -> loop (Just_ a) rest {-#INLINABLE last_ #-}- + -- --------------- -- length -- ---------------@@ -1160,15 +1219,20 @@ 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+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 @@ -1218,6 +1282,29 @@ loop as {-# INLINABLE mapM_ #-} +++{- | Map layers of one functor to another with a transformation involving the base monad+ @maps@ is more fundamental than @mapped@, which is best understood as a convenience+ for effecting this frequent composition:++> mapped = mapsM +> mapsM phi = decompose . maps (Compose . phi) ++ @mapped@ obeys these rules:++> mapped return = id+> mapped f . mapped g = mapped (f <=< g)+> map f . mapped g = mapped (liftM f . g)+> mapped f . map g = mapped (f . g)++-}++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)@@ -1287,6 +1374,35 @@ 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_ #-}+ {-| Remove repeated elements from a Stream. 'nub' of course accumulates a 'Data.Set.Set' of elements that have already been seen and should thus be used with care. @@ -1299,6 +1415,7 @@ [1,2,3] -}+ nub :: (Monad m, Ord a) => Stream (Of a) m r -> Stream (Of a) m r nub = loop Set.empty where loop !set stream = case stream of @@ -1308,6 +1425,53 @@ then loop set rest else Step (a :> loop (Set.insert a set) rest) +{-| +> 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@@ -1380,6 +1544,7 @@ -- | 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 = Step (a :> loop (m-1))@@ -1393,6 +1558,7 @@ -} 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 @@ -1437,12 +1603,15 @@ 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 !x stream = Step $ done x :> - case stream of - Return r -> Return r- Effect m -> Effect $ liftM (loop x) m- Step (a :> rest) -> loop (step x a) rest-{-# INLINABLE scan #-}+ loop !acc stream = do+ case stream of + Return r -> yield (done acc) >> return r+ Effect m -> Effect (liftM (loop acc) m)+ Step (a :> rest) -> do+ yield (done acc) + let !acc' = step acc a+ loop acc' 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@@ -1666,7 +1835,25 @@ 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 -- ---------------@@ -1677,20 +1864,20 @@ just a number of items from a stream of elements, but a number of substreams and the like. ->>> S.toList $ S.take 3 $ each "pennsylvania"-"pen" :> ()+>>> S.toList $ S.take 3 $ each "with"+"wit" :> () ->>> total <- S.sum_ $ S.take 3 S.readLn :: IO Int-1<Enter>-10<Enter>-100<Enter>->>> print total-111+>>> 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 = loop where+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)@@ -1703,9 +1890,19 @@ -- --------------- {-| End stream when an element fails a condition; the original return value is lost.- By contrast 'span' preserves this information.+ 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@@ -1719,7 +1916,7 @@ {-| 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 'mapM' which, like 'replicateM',+ 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. @@ -1731,7 +1928,16 @@ {-| Convert an effectful 'Stream' into a list alongside the return value -> mapped toListM :: Stream (Stream (Of a)) m r -> Stream (Of [a]) m +> 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 [])@@ -1789,6 +1995,18 @@ {-# 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 -- --------------------------------------- @@ -1797,6 +2015,8 @@ > 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@@ -1945,7 +2165,6 @@ 3<Enter> [1,2,3] :> () - -} readLn :: (MonadIO m, Read a) => Stream (Of a) m ()@@ -2064,7 +2283,7 @@ Step (s :> rest) -> liftIO (putStrLn s) >> loop rest {-# INLINE stdoutLn' #-} -{-| Read a series of strings as lines to a file.+{-| Read the lines of a file as Haskell 'String's >>> runResourceT $ S.writeFile "lines.txt" $ S.take 2 S.stdinLn hello<Enter>@@ -2073,9 +2292,10 @@ "hello" "world" - 'runResourceT', as it is used here, means something like 'closing_handles';- it makes it possible to write convenient, fairly sensible versions of - 'readFile', 'writeFile' and 'appendFile'. Its use is explained + '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>. -}@@ -2202,7 +2422,7 @@ {-#INLINE sumToCompose #-} {-| Store the result of any suitable fold over a stream, keeping the stream for- further manipulation. @store f = f . duplicate@ :+ further manipulation. @store f = f . copy@ : >>> S.print $ S.store S.product $ each [1..4] 1@@ -2218,11 +2438,11 @@ 4 10 :> (24 :> ()) - Here the sum (10) and the product (24) have been \'stored\' for use when + 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, + 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]@@ -2236,29 +2456,36 @@ 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' / 'duplicate' is /much/ more powerful, as you can see by reflecting on + 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 'duplicate'.)+ 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'.) - 'store' is intended to be used at types like these+ It would conceivable be clearer to import a series of specializations of 'store'.+ It is intended to be used at types like these: -> storeM :: (Monad m => Stream (Of a) m r -> m (Of b r)) +> 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 :: (MonadIO m => Stream (Of a) m r -> m (Of b r)) +> 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 - And similarly for other constraints that @Stream (Of a)@ inherits, - like 'MonadResource'. Thus I can filter and write to one file, but - nub and write to another: + 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@@ -2273,11 +2500,11 @@ store :: Monad m => (Stream (Of a) (Stream (Of a) m) r -> t) -> Stream (Of a) m r -> t-store f x = f (duplicate x)+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, given: +{-| 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"@@ -2286,17 +2513,24 @@ one two - I can as well do:+ With copy, I can as well do: ->>> S.print $ S.stdoutLn $ S.duplicate $ each ["one","two"]+>>> S.print $ S.stdoutLn $ S.copy $ each ["one","two"] one "one" two "two" - 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`, + '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]@@ -2304,50 +2538,77 @@ rather than ->>> S.sum $ S.product . S.duplicate $ each [1..10]+>>> 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 'duplicate':+ 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.duplicate $ each [1..10]+>>> S.sum $ S.filter odd $ S.product $ S.filter even $ S.copy $ each [1..10] 25 :> (3840 :> ()) - or using 'store' + 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@ - + 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.duplicate $ each [1..10]+>>> (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.duplicate $ (S.toList . mapped S.toList . chunksOf 2) $ S.duplicate $ each [1..12]+>>> (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 = 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 = loop where+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' = 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 duplicate#-}+ 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' #-}+ {-| The type > Data.List.unzip :: [(a,b)] -> ([a],[b])@@ -2356,7 +2617,43 @@ > Streaming.unzip :: Stream (Of (a,b)) m r -> Stream (Of a) m (Stream (Of b) m r) - which would not stream. Of course, neither does 'Data.List.unzip'+ 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
streaming.cabal view
@@ -1,61 +1,66 @@ name: streaming-version: 0.1.3.4+version: 0.1.4.0 cabal-version: >=1.10 build-type: Simple-synopsis: an elementary streaming prelude and a general stream type.+synopsis: an elementary streaming prelude and general stream type. -description: @Streaming.Prelude@ exports an elementary streaming prelude relating to - an elementary source\/generator\/producer type, namely @Stream (Of a) m r@. - The main module, @Streaming@, exports a much more general type, - @Stream f m r@, which- can be used to stream successive distinct steps characterized by /any/ - functor @f@, though we are here interested only in a limited range of - cases. +description: @Streaming.Prelude@ exports an elementary streaming prelude focussed on+ a simple \"source\" or \"producer\" type, namely @Stream (Of a) m r@.+ This is a sort of effectful version of @([a],r)@ in which monadic action + is interleaved between successive elements.+ The main module, @Streaming@, exports a much more general type,+ @Stream f m r@, which can be used to stream successive distinct+ steps characterized by /any/+ functor @f@, though we are here interested only in a limited range of+ cases. .- The streaming-io libraries have various devices for dealing+ The streaming-IO libraries have various devices for dealing with effectful variants of @[a]@ or @([a],r)@. But it is only with- the general type @Stream f m r@, or some equivalent, - that one can hope to stream their sorts of stream, as one - makes lists of lists in the Haskell @Prelude@ and @Data.List@. - One needs some such type if we are to- express a properly streaming equivalent of e.g.+ 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+ @Prelude@ and @Data.List@. One needs some such type if we are+ to express properly streaming equivalents of e.g. . > group :: Ord a => [a] -> [[a]] > chunksOf :: Int -> [a] -> [[a]] > lines :: [Char] -> [[Char]] -- but similarly with bytestring, etc. .- to mention a few obviously desirable operations. But once one grasps - the concept needed to express those functions - i.e. @Stream f m r@ or- some equivalent - then one will also see that, + to mention a few obviously desirable operations. But once one grasps+ the iterable stream concept needed to express those functions - to wit,+ @Stream f m r@ or some equivalent - then one will also see that, with it, one is already in possession of a complete- elementary streaming library - since one possesses @Stream (Of a) m r@, which- is the type of a \'generator\' or \'producer\' or whatever you call an - effectful stream of items. The present @Streaming.Prelude@ is the- simplest streaming library that can replicate anything like the - API of the @Prelude@ and @Data.List@. + 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 emphasis of the library is on interoperation; for- the rest its advantages are: extreme simplicity and re-use of + the rest its advantages are: extreme simplicity and re-use of intuitions the user has gathered from mastery of @Prelude@ and- @Data.List@. The two conceptual pre-requisites are some - comprehension of monad transformers and some familiarity + @Data.List@. The two conceptual pre-requisites are some+ comprehension of monad transformers and some familiarity with \'rank 2 types\'.- . - See the + .+ See the <https://hackage.haskell.org/package/streaming#readme readme> below- for an 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> Note also the - <https://hackage.haskell.org/package/streaming-bytestring streaming bytestring> - and - <https://hackage.haskell.org/package/streaming-utils streaming utils> + for an 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>.+ Note also the+ <https://hackage.haskell.org/package/streaming-bytestring streaming bytestring>+ and+ <https://hackage.haskell.org/package/streaming-utils streaming utils> packages. . 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 + > Streaming.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r . Interoperation with <http://hackage.haskell.org/package/io-streams io-streams> is thus: .@@ -66,21 +71,92 @@ . > Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a .- These conversions should never be more expensive than a single @>->@ or @=$=@. Further- points of comparison are discussed in the + These conversions should never be more expensive than a single @>->@ or @=$=@. + Here is a simple example that runs a single underlying stream with several+ streaming-io libraries at once, superimposing their effects + without any accumulation:+ .+ > module Main (main) where+ > import Streaming + > import Pipes + > import Data.Conduit+ > import qualified Streaming.Prelude as S+ > import qualified Data.Conduit.List as CL+ > import qualified Pipes.Prelude as P+ > import qualified System.IO.Streams as IOS+ > import Data.ByteString.Char8 (pack)+ > import Data.Function ((&))+ >+ > mkConduit = CL.unfoldM S.uncons+ > mkPipe = P.unfoldr S.next+ > mkIOStream = IOS.unfoldM S.uncons+ >+ > main = iostreamed where+ > urstream = S.take 4 S.readLn :: Stream (Of Int) IO () + > streamed = S.copy urstream+ > & S.map (\n -> "streaming says: " ++ show n) + > & S.stdoutLn + > piped = runEffect $ + > mkPipe (S.copy streamed)+ > >-> P.map (\n -> "pipes says: " ++ show n)+ > >-> P.stdoutLn + > conduited = mkConduit (S.copy piped) + > $$ CL.map (\n -> "conduit says: " ++ show n)+ > =$ CL.mapM_ (liftIO . putStrLn)+ > iostreamed = do+ > str0 <- mkIOStream conduited+ > str1 <- IOS.map (\n -> pack $ "io-streams says: " ++ show n ++ "\n") str0 + > IOS.supply str1 IOS.stdout+ .+ This program successively parses four @Int@s from standard input, + and /simulaneously/ passes them to (here trivial) stream-consuming + processes from four different libraries, using the @copy@ function from+ @Streaming.Prelude@. I mark my own input with @/<Enter/>@:+ .+ > >>> main+ > 1 <Enter>+ > streaming says: 1+ > pipes says: 1+ > conduit says: 1+ > io-streams says: 1+ > 2 <Enter>+ > streaming says: 2+ > pipes says: 2+ > conduit says: 2+ > io-streams says: 2+ > 3 <Enter>+ > streaming says: 3+ > pipes says: 3+ > conduit says: 3+ > io-streams says: 3+ > 4 <Enter>+ > streaming says: 4+ > pipes says: 4+ > conduit says: 4+ > io-streams says: 4+ .+ Of course, I could as well have passed the stream to several+ independent conduits, for example. Further+ points of comparison with the going streaming-IO libraries+ are discussed in the <https://hackage.haskell.org/package/streaming#readme readme> below. . Here are the results of some- <https://gist.github.com/michaelt/f19bef01423b17f29ffd microbenchmarks> + <https://gist.github.com/michaelt/f19bef01423b17f29ffd microbenchmarks> based on the- <https://github.com/ekmett/machines/blob/master/benchmarks/Benchmarks.hs benchmarks> + <https://github.com/ekmett/machines/blob/master/benchmarks/Benchmarks.hs benchmarks> included in the machines package: . <<http://i.imgur.com/sSG5MvH.png>> .+ 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.+ + . - + license: BSD3 license-file: LICENSE author: michaelt@@ -97,17 +173,17 @@ library- exposed-modules: Streaming, + exposed-modules: Streaming, Streaming.Prelude, Streaming.Internal - -- other-modules: + -- other-modules: other-extensions: RankNTypes, CPP,- StandaloneDeriving, FlexibleContexts, - DeriveDataTypeable, DeriveFoldable, - DeriveFunctor, DeriveTraversable, + StandaloneDeriving, FlexibleContexts,+ DeriveDataTypeable, DeriveFoldable,+ DeriveFunctor, DeriveTraversable, UndecidableInstances- + build-depends: base >=4.6 && <5 , mtl >=2.1 && <2.3 , mmorph >=1.0 && <1.2@@ -118,8 +194,8 @@ , resourcet , exceptions , containers- + default-language: Haskell2010- +