@@ -44,11 +44,12 @@ import Streamly.Prelude (SerialT, IsStream, serial, fromSerial)
4444#ifndef COVERAGE_BUILD
4545import Streamly.Prelude (avgRate, maxBuffer)
4646#endif
47−import qualified Streamly.Prelude as S
47+import qualified Streamly.Data.Stream.Prelude as S
4848import qualified Streamly.Internal.Data.Fold as FL
4949import qualified Streamly.Internal.Data.Unfold as UF
50−import qualified Streamly.Internal.Data.Stream.IsStream as IS
50+import qualified Streamly.Internal.Data.Stream as IS
5151import qualified Streamly.Data.Array as A
52+import qualified Streamly.Internal.Data.Parser as Parser
5253
5354import Streamly.Internal.Data.Time.Units
5455 (AbsTime, NanoSecond64(..), toRelTime64, diffAbsTime64)
@@ -59,31 +60,66 @@ import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
5960
6061import Streamly.Test.Common
6162import Streamly.Test.Prelude.Common
63+import Control.Monad.IO.Class (MonadIO)
64+
65+toList :: Monad m => S.Stream m a -> m [a]
66+toList = S.fold FL.toList
67+
68+splitOn :: Monad m =>
69+ (a -> Bool) -> FL.Fold m a b -> S.Stream m a -> S.Stream m b
70+splitOn predicate f = IS.foldManyPost (FL.takeEndBy_ predicate f)
71+
72+splitOnSuffix :: Monad m =>
73+ (a -> Bool) -> FL.Fold m a b -> S.Stream m a -> S.Stream m b
74+splitOnSuffix predicate f = S.foldMany (FL.takeEndBy_ predicate f)
75+
76+splitOnSeq' :: (MonadIO m, Unbox a, Storable a, Enum a, Eq a) =>
77+ A.Array a -> FL.Fold m a b -> S.Stream m a -> S.Stream m b
78+splitOnSeq' patt f m = IS.foldManyPost (FL.takeEndBySeq_ patt f) m
79+
80+splitOnSuffixSeq' :: (MonadIO m, Unbox a, Storable a, Enum a, Eq a) =>
81+ A.Array a -> FL.Fold m a b -> S.Stream m a -> S.Stream m b
82+splitOnSuffixSeq' patt f m = S.foldMany (FL.takeEndBySeq_ patt f) m
83+
84+groupsBy :: Monad m =>
85+ (a -> a -> Bool) -> FL.Fold m a b -> S.Stream m a -> S.Stream m b
86+groupsBy cmp f m = S.catRights $ S.parseMany (Parser.groupBy cmp f) m
87+
88+groupsByRolling :: Monad m =>
89+ (a -> a -> Bool) -> FL.Fold m a b -> S.Stream m a -> S.Stream m b
90+groupsByRolling cmp f m =
91+ S.catRights $ S.parseMany (Parser.groupByRolling cmp f) m
92+
93+drainWhile :: Monad m => (a -> Bool) -> S.Stream m a -> m ()
94+drainWhile p = S.fold FL.drain . S.takeWhile p
95+
96+drainMapM :: Monad m => (a -> m b) -> S.Stream m a -> m ()
97+drainMapM f = S.fold (FL.drainMapM f)
6298
6399splitOnSeq :: Spec
64100splitOnSeq = do
65101 describe "Tests for splitOnSeq" $ do
66− it "splitOnSeq' \"hello\" \"\" = [\"\"]"
67− $ splitOnSeq' "hello" "" `shouldReturn` [""]
68− it "splitOnSeq' \"hello\" \"hello\" = [\"\", \"\"]"
69− $ splitOnSeq' "hello" "hello" `shouldReturn` ["", ""]
70− it "splitOnSeq' \"x\" \"hello\" = [\"hello\"]"
71− $ splitOnSeq' "x" "hello" `shouldReturn` ["hello"]
72− it "splitOnSeq' \"h\" \"hello\" = [\"\", \"ello\"]"
73− $ splitOnSeq' "h" "hello" `shouldReturn` ["", "ello"]
74− it "splitOnSeq' \"o\" \"hello\" = [\"hell\", \"\"]"
75− $ splitOnSeq' "o" "hello" `shouldReturn` ["hell", ""]
76− it "splitOnSeq' \"e\" \"hello\" = [\"h\", \"llo\"]"
77− $ splitOnSeq' "e" "hello" `shouldReturn` ["h", "llo"]
78− it "splitOnSeq' \"l\" \"hello\" = [\"he\", \"\", \"o\"]"
79− $ splitOnSeq' "l" "hello" `shouldReturn` ["he", "", "o"]
80− it "splitOnSeq' \"ll\" \"hello\" = [\"he\", \"o\"]"
81− $ splitOnSeq' "ll" "hello" `shouldReturn` ["he", "o"]
102+ it "splitOnSeq_ \"hello\" \"\" = [\"\"]"
103+ $ splitOnSeq_ "hello" "" `shouldReturn` [""]
104+ it "splitOnSeq_ \"hello\" \"hello\" = [\"\", \"\"]"
105+ $ splitOnSeq_ "hello" "hello" `shouldReturn` ["", ""]
106+ it "splitOnSeq_ \"x\" \"hello\" = [\"hello\"]"
107+ $ splitOnSeq_ "x" "hello" `shouldReturn` ["hello"]
108+ it "splitOnSeq_ \"h\" \"hello\" = [\"\", \"ello\"]"
109+ $ splitOnSeq_ "h" "hello" `shouldReturn` ["", "ello"]
110+ it "splitOnSeq_ \"o\" \"hello\" = [\"hell\", \"\"]"
111+ $ splitOnSeq_ "o" "hello" `shouldReturn` ["hell", ""]
112+ it "splitOnSeq_ \"e\" \"hello\" = [\"h\", \"llo\"]"
113+ $ splitOnSeq_ "e" "hello" `shouldReturn` ["h", "llo"]
114+ it "splitOnSeq_ \"l\" \"hello\" = [\"he\", \"\", \"o\"]"
115+ $ splitOnSeq_ "l" "hello" `shouldReturn` ["he", "", "o"]
116+ it "splitOnSeq_ \"ll\" \"hello\" = [\"he\", \"o\"]"
117+ $ splitOnSeq_ "ll" "hello" `shouldReturn` ["he", "o"]
82118
83119 where
84120
85− splitOnSeq' pat xs =
86− S.toList $ IS.splitOnSeq (A.fromList pat) FL.toList (S.fromList xs)
121+ splitOnSeq_ pat xs = toList $
122+ splitOnSeq' (A.fromList pat) FL.toList (S.fromList xs)
87123
88124splitOnSuffixSeq :: Spec
89125splitOnSuffixSeq = do
@@ -107,9 +143,8 @@ splitOnSuffixSeq = do
107143
108144 where
109145
110− splitSuffixOn_ pat xs =
111− S.toList
112− $ IS.splitOnSuffixSeq (A.fromList pat) FL.toList (S.fromList xs)
146+ splitSuffixOn_ pat xs = toList $
147+ splitOnSuffixSeq' (A.fromList pat) FL.toList (S.fromList xs)
113148
114149splitterProperties ::
115150 forall a. (Arbitrary a, Eq a, Show a, Storable a, Unbox a, Enum a)
@@ -180,16 +215,16 @@ splitterProperties sep desc = do
180215 where
181216
182217 splitOnSeq_ xs ys =
183− S.toList $ IS.splitOnSeq (A.fromList ys) FL.toList (S.fromList xs)
218+ toList $ splitOnSeq' (A.fromList ys) FL.toList (S.fromList xs)
184219
185220 splitOnSuffixSeq_ xs ys =
186− S.toList $ IS.splitOnSuffixSeq (A.fromList ys) FL.toList (S.fromList xs)
221+ toList $ splitOnSuffixSeq' (A.fromList ys) FL.toList (S.fromList xs)
187222
188223 splitOn_ xs ys =
189− S.toList $ IS.splitOn (== (head ys)) FL.toList (S.fromList xs)
224+ toList $ splitOn (== (head ys)) FL.toList (S.fromList xs)
190225
191226 splitOnSuffix_ xs ys =
192− S.toList $ IS.splitOnSuffix (== (head ys)) FL.toList (S.fromList xs)
227+ toList $ splitOnSuffix (== (head ys)) FL.toList (S.fromList xs)
193228
194229 intercalateSuffix xs yss = intercalate xs yss ++ xs
195230
@@ -221,7 +256,7 @@ splitterProperties sep desc = do
221256 testCase xs = do
222257 ys <- splitter xs (replicate i sep)
223258 szs <-
224− IS.toList
259+ toList
225260 $ sIntercalater UF.fromList (replicate i sep)
226261 $ IS.fromList ys
227262 let lzs = lIntercalater (replicate i sep) ys
@@ -243,7 +278,7 @@ splitterProperties sep desc = do
243278 testCase xs = do
244279 ys <- splitter xs (replicate i sep)
245280 szs <-
246− IS.toList
281+ toList
247282 $ sIntercalater UF.fromList (replicate i sep)
248283 $ IS.fromList ys
249284 let lzs = lIntercalater (replicate i sep) ys
@@ -265,7 +300,7 @@ splitterProperties sep desc = do
265300 let lxs = lIntercalater (replicate i sep) xss
266301 lys <- splitter lxs (replicate i sep)
267302 sxs <-
268− S.toList
303+ toList
269304 $ sIntercalater UF.fromList (replicate i sep)
270305 $ S.fromList xss
271306 sys <- splitter sxs (replicate i sep)
@@ -286,7 +321,7 @@ splitterProperties sep desc = do
286321 testCase xss = do
287322 let lxs = lIntercalater [sep] xss
288323 lys <- splitter lxs [sep]
289− sxs <- S.toList $ sIntercalater UF.fromList [sep] $ S.fromList xss
324+ sxs <- toList $ sIntercalater UF.fromList [sep] $ S.fromList xss
290325 sys <- splitter sxs [sep]
291326 listEquals (==) lys xss
292327 listEquals (==) sys xss
@@ -299,7 +334,7 @@ intercalateSplitOnId x desc =
299334 forAll listWithZeroes $ \xs -> do
300335 withMaxSuccess maxTestCount $
301336 monadicIO $ do
302− ys <- S.toList $ S.splitOn (== x) toListFL (S.fromList xs)
337+ ys <- toList $ splitOn (== x) toListFL (S.fromList xs)
303338 listEquals (==) (intercalate [x] ys) xs
304339
305340 where
@@ -320,17 +355,17 @@ groupSplitOps desc = do
320355 intercalateSplitOnId (0 :: Word8) desc
321356
322357-- |
323−-- After grouping (and folding) Int stream using @>@ operation,
358+-- After grouping (and folding) Int stream using @<@ operation,
324359-- the first @Int@ of every @[Int]@ in the @[Int]@ stream should be the minimum.
325360testGroupsBy :: Property
326361testGroupsBy =
327362 forAll (choose (0, maxStreamLen)) $ \len ->
328363 forAll (vectorOf len (arbitrary :: Gen Int)) $ \vec -> monadicIO $ do
329− r <- run $ S.all (\ls ->
364+ r <- run $ S.fold (FL.all (\ls ->
330365 case ls of
331366 [] -> True
332− (x:_) -> x == minimum ls)
333− $ S.groupsBy (>) FL.toList
367+ (x:_) -> x == minimum ls))
368+ $ groupsBy (<) FL.toList
334369 $ S.fromList vec
335370 assert r
336371
@@ -338,14 +373,14 @@ testGroups :: Property
338373testGroups =
339374 forAll (choose (0, maxStreamLen)) $ \len ->
340375 forAll (vectorOf len (arbitrary :: Gen Int)) $ \vec -> monadicIO $ do
341− r <- run $ S.toList $ S.groups FL.toList $ S.fromList vec
376+ r <- toList $ groupsBy (==) FL.toList $ S.fromList vec
342377 assert $ r == group vec
343378
344379testGroupsByRolling :: Property
345380testGroupsByRolling =
346381 forAll (choose (0, maxStreamLen)) $ \len ->
347382 forAll (vectorOf len (arbitrary :: Gen Int)) $ \vec -> monadicIO $ do
348− r <- run $ S.toList $ S.groupsByRolling (==) FL.toList $ S.fromList vec
383+ r <- toList $ groupsByRolling (==) FL.toList $ S.fromList vec
349384 assert $ r == group vec
350385
351386-- |
@@ -362,7 +397,7 @@ decreasing [] = True
362397decreasing xs = all (== True) $ zipWith (<=) (tail xs) xs
363398
364399-- |
365−-- To check if the minimum elements (after grouping on @>@)
400+-- To check if the minimum elements (after grouping on @<@)
366401-- are non-increasing (either decrease or remain the same).
367402-- Had an element been strictly greater, it would have been grouped
368403-- with that element only.
@@ -370,10 +405,10 @@ testGroupsBySep :: Property
370405testGroupsBySep =
371406 forAll (choose (0, maxStreamLen)) $ \len ->
372407 forAll (vectorOf len (arbitrary :: Gen Int)) $ \vec -> monadicIO $ do
373− a <- run $ S.toList
374− $ S.map maybeMinimum
375− $ S.groupsBy (>) FL.toList
376− $ S.fromList vec
408+ a <- toList
409+ $ fmap maybeMinimum
410+ $ groupsBy (<) FL.toList
411+ $ S.fromList vec
377412 assert $ decreasing a
378413
379414groupingOps :: Spec
@@ -396,8 +431,8 @@ associativityCheck desc t = prop desc assocCheckProp
396431 yStream = S.fromList ys
397432 zStream = S.fromList zs
398433 infixAssocstream <-
399− run $ S.toList $ t $ xStream `serial` yStream `serial` zStream
400− assocStream <- run $ S.toList $ t $ xStream <> yStream <> zStream
434+ run $ toList $ t $ xStream `serial` yStream `serial` zStream
435+ assocStream <- run $ toList $ t $ xStream <> yStream <> zStream
401436 listEquals (==) infixAssocstream assocStream
402437
403438max_length :: Int
@@ -436,7 +471,7 @@ testTakeInterval :: IO Bool
436471testTakeInterval = do
437472 r <-
438473 S.fold (FL.tee FL.head FL.last)
439− $ IS.takeInterval takeDropTime
474+ $ S.takeInterval (toRelTime64 takeDropTime)
440475 $ S.repeatM (threadDelay 1000 >> getTime Monotonic)
441476 checkTakeDropTime r
442477
@@ -444,8 +479,8 @@ testDropInterval :: IO Bool
444479testDropInterval = do
445480 t0 <- getTime Monotonic
446481 mt1 <-
447− S.head
448− $ IS.dropInterval takeDropTime
482+ S.fold FL.head
483+ $ S.dropInterval (toRelTime64 takeDropTime)
449484 $ S.repeatM (threadDelay 1000 >> getTime Monotonic)
450485 checkTakeDropTime (Just t0, mt1)
451486#endif
@@ -455,23 +490,18 @@ unfold = monadicIO $ do
455490 a <- pick $ choose (0, max_length `div` 2)
456491 b <- pick $ choose (0, max_length)
457492 let unf = UF.second b UF.enumerateFromToIntegral
458− ls <- S.toList $ S.unfold unf a
459− return $ ls == [a..b]
460−
461−unfold0 :: Property
462−unfold0 = monadicIO $ do
463− a <- pick $ choose (0, max_length `div` 2)
464− b <- pick $ choose (0, max_length)
465− let unf = UF.both a (UF.second b UF.enumerateFromToIntegral)
466− ls <- S.toList $ IS.unfold0 unf
493+ ls <- toList $ S.unfold unf a
467494 return $ ls == [a..b]
468495
469496testFromCallback :: IO Int
470497testFromCallback = do
471498 ref <- newIORef Nothing
472− let stream = S.map Just (IS.fromCallback (setCallback ref))
473− `S.parallel` runCallback ref
474− S.sum $ S.map fromJust $ S.takeWhile isJust stream
499+ let stream =
500+ S.parConcatList (S.eager True)
501+ [ fmap Just (S.fromCallback (setCallback ref))
502+ , runCallback ref
503+ ]
504+ S.fold FL.sum $ fmap fromJust $ S.takeWhile isJust stream
475505
476506 where
477507
@@ -483,11 +513,11 @@ testFromCallback = do
483513 S.repeatM (readIORef ref)
484514 & IS.delayPost 0.1
485515 & S.mapMaybe id
486− & S.head
516+ & S.fold FL.head
487517
488518 S.fromList [1..100]
489519 & IS.delayPost 0.001
490− & S.mapM_ (fromJust cb)
520+ & drainMapM (fromJust cb)
491521 threadDelay 100000
492522 return Nothing
493523
@@ -497,12 +527,12 @@ foldIterateM =
497527 let s1 = Prelude.sum lst
498528 strm = S.fromList lst
499529 ms2 <-
500− S.last
501− $ S.map getSum
530+ S.fold FL.last
531+ $ fmap getSum
502532 $ IS.foldIterateM
503533 (return . FL.take 1 . FL.sconcat)
504534 (return (Sum 0))
505− $ S.map Sum strm
535+ $ fmap Sum strm
506536 case ms2 of
507537 Nothing -> assert $ s1 == 0
508538 Just s2 -> assert $ s1 == s2
@@ -510,7 +540,7 @@ foldIterateM =
510540sortBy :: Property
511541sortBy = forAll (listOf (chooseInt (0, max_length))) $ \lst -> monadicIO $ do
512542 let s1 = sort lst
513− s2 <- run $ S.toList $ IS.sortBy compare $ S.fromList lst
543+ s2 <- toList $ IS.sortBy compare $ S.fromList lst
514544 assert $ s1 == s2
515545
516546moduleName :: String
@@ -530,14 +560,16 @@ main = hspec
530560 <> [("maxBuffer -1", fromSerial . maxBuffer (-1))]
531561#endif
532562 let toListSerial :: SerialT IO a -> IO [a]
533− toListSerial = S.toList . fromSerial
563+ toListSerial = toList . fromSerial
534564
535565 describe "Runners" $ do
536566 -- XXX use an IORef to store and check the side effects
537567 it "simple serially" $
538− (S.drain . fromSerial) (return (0 :: Int)) `shouldReturn` ()
568+ (S.fold FL.drain . fromSerial)
569+ (return (0 :: Int)) `shouldReturn` ()
539570 it "simple serially with IO" $
540− (S.drain . fromSerial) (S.fromEffect $ putStrLn "hello") `shouldReturn` ()
571+ (S.fold FL.drain . fromSerial)
572+ (S.fromEffect $ putStrLn "hello") `shouldReturn` ()
541573
542574 describe "Empty" $
543575 it "Monoid - mempty" $
@@ -571,12 +603,9 @@ main = hspec
571603 serialOps $ prop "serially fromEffect" . constructWithFromEffect id
572604 serialOps $ prop "serially cons" . constructWithCons S.cons
573605 serialOps $ prop "serially consM" . constructWithConsM S.consM id
574− serialOps $ prop "serially (.:)" . constructWithCons (S..:)
575− serialOps $ prop "serially (|:)" . constructWithConsM (S.|:) id
576606
577607 describe "From Generators" $ do
578608 prop "unfold" unfold
579− prop "unfold0" unfold0
580609
581610 describe "Simple Operations" $ serialOps simpleOps
582611
@@ -649,13 +678,13 @@ main = hspec
649678 -- Just some basic sanity tests for now
650679 let input = [[1,1] :: [Int],[2,2],[3,3],[4,4],[5,5],[6,6],[7,7],[8,8]]
651680 mustBe g inp out =
652− S.toList (IS.concatPairsWith g S.fromList (S.fromList inp))
681+ toList (S.mergeMapWith g S.fromList (S.fromList inp))
653682 `shouldReturn` out
654683 in do
655684 it "concatPairsWith serial"
656− $ mustBe S.serial input [1,1,2,2,3,3,4,4,5,5,6,6,7,7,8,8]
685+ $ mustBe S.append input [1,1,2,2,3,3,4,4,5,5,6,6,7,7,8,8]
657686 it "concatPairsWith wSerial"
658− $ mustBe S.wSerial input [1,5,3,7,2,6,4,8,1,5,3,7,2,6,4,8]
687+ $ mustBe S.interleave input [1,5,3,7,2,6,4,8,1,5,3,7,2,6,4,8]
659688 it "concatPairsWith mergeBy sorted"
660689 $ mustBe
661690 (S.mergeBy compare) input [1,1,2,2,3,3,4,4,5,5,6,6,7,7,8,8]
@@ -681,7 +710,7 @@ main = hspec
681710 let xs = [1..n]
682711 ioRef <- run $ newIORef ([] :: [Int])
683712 run $
684− S.drainWhile (> 0) . t $
713+ drainWhile (> 0) . t $
685714 S.mapM (\a -> modifyIORef' ioRef (a :) >> return a) $
686715 S.fromList xs
687716 strm <- run $ readIORef ioRef