@@ -181,7 +181,9 @@ module Streamly.Prelude.Internal
181181 , minimumBy
182182 , minimum
183183 , the
184− -- , toListRev -- experimental
184+ , toListRev -- experimental
185+ , toStream
186+ , toStreamRev
185187
186188 -- ** Lazy Folds
187189 --
@@ -222,7 +224,7 @@ module Streamly.Prelude.Internal
222224
223225 -- * Transformation
224226
225− --, transform
227+ , transform
226228
227229 -- ** Mapping
228230 -- | In imperative terms a map operation can be considered as a loop over
@@ -298,8 +300,8 @@ module Streamly.Prelude.Internal
298300 , scanlM'
299301 , postscanl'
300302 , postscanlM'
301− -- , prescanl'
302− -- , prescanlM'
303+ , prescanl'
304+ , prescanlM'
303305 , scanl1'
304306 , scanl1M'
305307
@@ -368,13 +370,13 @@ module Streamly.Prelude.Internal
368370 , insertBy
369371 , intersperseM
370372 , intersperse
371− -- , insertAfterEach
373+ , insertAfterEach
372374 -- , intersperseBySpan
373− -- , intersperseByTime
375+ , intersperseByTime
374376
375377 -- ** Reordering
376378 , reverse
377− -- , reverse'
379+ , reverse'
378380
379381 -- * Multi-Stream Operations
380382 -- | New streams can be constructed by appending, merging or zipping
@@ -442,20 +444,20 @@ module Streamly.Prelude.Internal
442444 -- -- ** Breaking
443445
444446 -- By chunks
445− -- , splitAt -- spanN
447+ , splitAt -- spanN
446448 -- , splitIn -- sessionN
447449
448450 -- By elements
449− -- , span -- spanWhile
450− -- , break -- breakBefore
451+ , span -- spanWhile
452+ , break -- breakBefore
451453 -- , breakAfter
452454 -- , breakOn
453455 -- , breakAround
454− -- , spanBy
455− -- , spanByRolling
456+ , spanBy
457+ , spanByRolling
456458
457459 -- By sequences
458− -- breakOnSeq
460+ -- , breakOnSeq
459461
460462 -- ** Splitting
461463 -- | Streams can be split into segments in space or in time. We use the
@@ -473,9 +475,6 @@ module Streamly.Prelude.Internal
473475 , chunksOf
474476 , sessionsOf
475477
476− -- , lchunksOf
477− -- , lsessionsOf
478−
479478 -- -- *** Using Element Separators
480479 -- On == Dropping the separator
481480 , splitOn
@@ -489,14 +488,14 @@ module Streamly.Prelude.Internal
489488 , wordsBy -- stripAndCompactBy
490489
491490 -- -- *** Using Sequence Separators
492− -- , splitOnSeq
493− -- , splitOnSuffixSeq
491+ , splitOnSeq
492+ , splitOnSuffixSeq
494493 -- , splitOnPrefixSeq
495494
496495 -- Keeping the delimiters
497− -- , splitBySeq
498− -- , splitBySeqSuffix
499− -- , splitBySeqPrefix
496+ , splitBySeq
497+ , splitBySuffixSeq
498+ -- , splitByPrefixSeq
500499 -- , wordsBySeq
501500
502501 -- Splitting using multiple sequence separators
@@ -513,7 +512,6 @@ module Streamly.Prelude.Internal
513512 , trace
514513 , tap
515514
516− {-
517515 -- * Windowed Classification
518516 -- | Split the stream into windows or chunks in space or time. Each window
519517 -- can be associated with a key, all events associated with a particular
@@ -527,13 +525,14 @@ module Streamly.Prelude.Internal
527525 -- ** Tumbling Windows
528526 -- | A new window starts after the previous window is finished.
529527 -- , classifyChunksOf
530− -- , classifySessionsOf
528+ , classifySessionsBy
529+ , classifySessionsOf
531530
532531 -- ** Keep Alive Windows
533532 -- | The window size is extended if an event arrives within the specified
534533 -- window size. This can represent sessions with idle or inactive timeout.
535534 -- , classifyKeepAliveChunks
536− -- , classifyKeepAliveSessions
535+ , classifyKeepAliveSessions
537536
538537 {-
539538 -- ** Sliding Windows
@@ -545,7 +544,6 @@ module Streamly.Prelude.Internal
545544 -- ** Sliding Window Buffers
546545 -- , slidingChunkBuffer
547546 -- , slidingSessionBuffer
548− -}
549547
550548 -- ** Containers of Streams
551549 -- | These are variants of standard 'Foldable' fold functions that use a
@@ -605,7 +603,7 @@ import Prelude
605603 foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem,
606604 notElem, maximum, minimum, head, last, tail, length, null,
607605 reverse, iterate, init, and, or, lookup, foldr1, (!!),
608− scanl, scanl1, replicate, concatMap, span)
606+ scanl, scanl1, replicate, concatMap, span, splitAt, break)
609607
610608import qualified Data.Heap as H
611609import qualified Data.Map.Strict as Map
@@ -621,7 +619,7 @@ import Streamly.Streams.Combinators (maxYields)
621619import Streamly.Streams.Prelude
622620 (fromStreamS, toStreamS, foldWith, foldMapWith, forEachWith)
623621import Streamly.Streams.StreamD (fromStreamD, toStreamD)
624−import Streamly.Streams.StreamK (IsStream(..))
622+import Streamly.Streams.StreamK (IsStream((|:), consM))
625623import Streamly.Streams.Serial (SerialT)
626624import Streamly.Pipe.Types (Pipe (..))
627625import Streamly.Time.Units
@@ -901,7 +899,7 @@ repeatM = go
901899--
902900-- @since 0.1.2
903901iterate :: IsStream t => (a -> a) -> a -> t m a
904−iterate step = fromStream . go
902+iterate step = K.fromStream . go
905903 where
906904 go s = K.cons s (go (step s))
907905
@@ -1762,9 +1760,9 @@ toList = P.toList
17621760-- very inefficient, consider using "Streamly.Array" instead.
17631761--
17641762-- @since 0.7.0
1765−{-# INLINE _toListRev #-}
1766−_toListRev :: Monad m => SerialT m a -> m [a]
1767−_toListRev = D.toListRev . toStreamD
1763+{-# INLINE toListRev #-}
1764+toListRev :: Monad m => SerialT m a -> m [a]
1765+toListRev = D.toListRev . toStreamD
17681766
17691767-- |
17701768-- @
@@ -1791,9 +1789,9 @@ toHandle h m = go m
17911789-- be very inefficient, consider using "Streamly.Array" instead.
17921790--
17931791-- @since 0.7.0
1794−{-# INLINE _toStream #-}
1795−_toStream :: Monad m => Fold m a (SerialT Identity a)
1796−_toStream = Fold (\f x -> return $ f . (x `K.cons`))
1792+{-# INLINE toStream #-}
1793+toStream :: Monad m => Fold m a (SerialT Identity a)
1794+toStream = Fold (\f x -> return $ f . (x `K.cons`))
17971795 (return id)
17981796 (return . ($ K.nil))
17991797
@@ -1809,18 +1807,18 @@ _toStream = Fold (\f x -> return $ f . (x `K.cons`))
18091807-- @since 0.7.0
18101808
18111809-- xn : ... : x2 : x1 : []
1812−{-# INLINABLE _toStreamRev #-}
1813−_toStreamRev :: Monad m => Fold m a (SerialT Identity a)
1814−_toStreamRev = Fold (\xs x -> return $ x `K.cons` xs) (return K.nil) return
1810+{-# INLINABLE toStreamRev #-}
1811+toStreamRev :: Monad m => Fold m a (SerialT Identity a)
1812+toStreamRev = Fold (\xs x -> return $ x `K.cons` xs) (return K.nil) return
18151813
18161814------------------------------------------------------------------------------
18171815-- General Transformation
18181816------------------------------------------------------------------------------
18191817
18201818-- | Use a 'Pipe' to transform a stream.
1821−{-# INLINE _transform #-}
1822−_transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
1823−_transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
1819+{-# INLINE transform #-}
1820+transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
1821+transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
18241822
18251823------------------------------------------------------------------------------
18261824-- Transformation by Folding (Scans)
@@ -1929,17 +1927,17 @@ postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
19291927-- | Like scanl' but does not stream the final value of the accumulator.
19301928--
19311929-- @since 0.6.0
1932−{-# INLINE _prescanl' #-}
1933−_prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1934−_prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
1930+{-# INLINE prescanl' #-}
1931+prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1932+prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
19351933
19361934-- XXX this needs to be concurrent
19371935-- | Like postscanl' but with a monadic step function.
19381936--
19391937-- @since 0.6.0
1940−{-# INLINE _prescanlM' #-}
1941−_prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
1942−_prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
1938+{-# INLINE prescanlM' #-}
1939+prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
1940+prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
19431941
19441942-- XXX this needs to be concurrent
19451943-- | Like 'scanl1'' but with a monadic step function.
@@ -2180,9 +2178,9 @@ reverse s = fromStreamS $ S.reverse $ toStreamS s
21802178-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
21812179--
21822180-- @since 0.7.0
2183−{-# INLINE _reverse' #-}
2184−_reverse' :: (IsStream t, MonadIO m, Storable a) => t m a -> t m a
2185−_reverse' s = fromStreamD $ D.reverse' $ toStreamD s
2181+{-# INLINE reverse' #-}
2182+reverse' :: (IsStream t, MonadIO m, Storable a) => t m a -> t m a
2183+reverse' s = fromStreamD $ D.reverse' $ toStreamD s
21862184
21872185------------------------------------------------------------------------------
21882186-- Transformation by Inserting
@@ -2221,9 +2219,9 @@ intersperse a = fromStreamS . S.intersperse a . toStreamS
22212219-- | Insert a monadic action after each element in the stream.
22222220--
22232221-- @since 0.7.0
2224−{-# INLINE _insertAfterEach #-}
2225−_insertAfterEach :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2226−_insertAfterEach m = fromStreamD . D.insertAfterEach m . toStreamD
2222+{-# INLINE insertAfterEach #-}
2223+insertAfterEach :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2224+insertAfterEach m = fromStreamD . D.insertAfterEach m . toStreamD
22272225
22282226{-
22292227-- | Intersperse a monadic action into the input stream after every @n@
@@ -2571,14 +2569,14 @@ intercalate :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
25712569-- the segments instead of discarding the leftover.
25722570--
25732571-- @since 0.7.0
2574−{-# INLINE _splitAt #-}
2575−_splitAt
2572+{-# INLINE splitAt #-}
2573+splitAt
25762574 :: Monad m
25772575 => Int
25782576 -> Fold m a b
25792577 -> Fold m a c
25802578 -> Fold m a (b, c)
2581−_splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2579+splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
25822580 Fold step initial extract
25832581 where
25842582 initial = Tuple3' <$> return n <*> initialL <*> initialR
@@ -2678,13 +2676,13 @@ sessionsOf n f xs =
26782676-- long as the predicate applied to the first element of the stream and next
26792677-- input element holds 'True', the second group takes the rest of the input.
26802678--
2681−_spanBy
2679+spanBy
26822680 :: Monad m
26832681 => (a -> a -> Bool)
26842682 -> Fold m a b
26852683 -> Fold m a c
26862684 -> Fold m a (b, c)
2687−_spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2685+spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
26882686 Fold step initial extract
26892687
26902688 where
@@ -2767,25 +2765,25 @@ span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
27672765-- > ([],[3,2,1])
27682766--
27692767-- @since 0.7.0
2770−{-# INLINE _break #-}
2771−_break
2768+{-# INLINE break #-}
2769+break
27722770 :: Monad m
27732771 => (a -> Bool)
27742772 -> Fold m a b
27752773 -> Fold m a c
27762774 -> Fold m a (b, c)
2777−_break p = span (not . p)
2775+break p = span (not . p)
27782776
27792777-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
27802778-- predicate is applied to the previous and the next input elements.
2781−{-# INLINE _spanRollingBy #-}
2782−_spanRollingBy
2779+{-# INLINE spanByRolling #-}
2780+spanByRolling
27832781 :: Monad m
27842782 => (a -> a -> Bool)
27852783 -> Fold m a b
27862784 -> Fold m a c
27872785 -> Fold m a (b, c)
2788−_spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2786+spanByRolling cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
27892787 Fold step initial extract
27902788
27912789 where
@@ -3187,11 +3185,11 @@ splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.to
31873185-- > lines = splitSuffixOn "\n"
31883186--
31893187-- @since 0.7.0
3190−{-# INLINE _splitOnSuffixSeq #-}
3191−_splitOnSuffixSeq
3188+{-# INLINE splitOnSuffixSeq #-}
3189+splitOnSuffixSeq
31923190 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
31933191 => Array a -> Fold m a b -> t m a -> t m b
3194−_splitOnSuffixSeq patt f m =
3192+splitOnSuffixSeq patt f m =
31953193 D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)
31963194
31973195{-
@@ -3238,11 +3236,11 @@ wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStream
32383236-- > ["he","ll","o"]
32393237--
32403238-- @since 0.7.0
3241−{-# INLINE _splitBySeq #-}
3242−_splitBySeq
3239+{-# INLINE splitBySeq #-}
3240+splitBySeq
32433241 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
32443242 => Array a -> Fold m a b -> t m a -> t m b
3245−_splitBySeq patt f m = intersperseM (runFold f (A.read patt)) $ splitOnSeq patt f m
3243+splitBySeq patt f m = intersperseM (runFold f (A.read patt)) $ splitOnSeq patt f m
32463244
32473245-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
32483246--
@@ -3273,11 +3271,11 @@ _splitBySeq patt f m = intersperseM (runFold f (A.read patt)) $ splitOnSeq patt
32733271-- > ["a.",".","b.","."]
32743272--
32753273-- @since 0.7.0
3276−{-# INLINE _splitBySuffixSeq #-}
3277−_splitBySuffixSeq
3274+{-# INLINE splitBySuffixSeq #-}
3275+splitBySuffixSeq
32783276 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
32793277 => Array a -> Fold m a b -> t m a -> t m b
3280−_splitBySuffixSeq patt f m =
3278+splitBySuffixSeq patt f m =
32813279 D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)
32823280
32833281{-
@@ -3608,14 +3606,14 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =
36083606-- only if no event is received within the specified session window size.
36093607--
36103608-- @since 0.7.0
3611−{-# INLINABLE _classifyKeepAliveSessions #-}
3612−_classifyKeepAliveSessions
3609+{-# INLINABLE classifyKeepAliveSessions #-}
3610+classifyKeepAliveSessions
36133611 :: (IsStream t, MonadAsync m, Ord k)
36143612 => Double -- ^ session inactive timeout
36153613 -> Fold m a b -- ^ Fold to be applied to session payload data
36163614 -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
36173615 -> t m (k, b)
3618−_classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
3616+classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
36193617
36203618------------------------------------------------------------------------------
36213619-- Keyed tumbling windows
@@ -3657,14 +3655,15 @@ classifyChunksOf wsize = classifyChunksBy wsize False
36573655-- the timestamps with a clock resolution of 1 second.
36583656--
36593657-- @since 0.7.0
3660−{-# INLINABLE _classifySessionsOf #-}
3661−_classifySessionsOf
3658+{-# INLINABLE classifySessionsOf #-}
3659+classifySessionsOf
36623660 :: (IsStream t, MonadAsync m, Ord k)
36633661 => Double -- ^ time window size
36643662 -> Fold m a b -- ^ Fold to be applied to window events
36653663 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
36663664 -> t m (k, b)
3667−_classifySessionsOf interval = classifySessionsBy 1 interval False
3665+classifySessionsOf interval = classifySessionsBy 1 interval False
3666+
36683667------------------------------------------------------------------------------
36693668-- Exceptions
36703669------------------------------------------------------------------------------