@@ -23,8 +23,9 @@ import Test.QuickCheck (Testable, Property, choose, forAll, withMaxSuccess)
2323import Test.QuickCheck.Monadic (monadicIO, run)
2424import Test.Hspec as H
2525
26−import qualified Streamly.Data.Fold as Fold
26+import qualified Streamly.Data.Fold as Fold ( toList )
2727import qualified Streamly.Data.Stream as Stream
28+ ( replicate, fromEffect, fromPure, fromList, fold, take, nil )
2829import qualified Streamly.Internal.Data.Stream.Concurrent as Async
2930
3031import Streamly.Test.Common (listEquals)
@@ -39,10 +40,10 @@ moduleName = "Data.Stream.Concurrent"
3940sortEq :: Ord a => [a] -> [a] -> Bool
4041sortEq a b = sort a == sort b
4142
42−cmp :: Stream IO Int -> ([Int] -> [Int] -> Bool) -> [Int] -> Property
43−cmp s eq list =
43+cmp :: (Show a, Ord a) => ([a] -> [a] -> Bool) -> [a] -> Stream IO a -> Property
44+cmp eq list s =
4445 monadicIO $ do
45− stream <- run $ Stream.fold Fold.toList s
46+ stream <- run $ sort <$> Stream.fold Fold.toList s
4647 listEquals eq stream list
4748
4849prop1 :: Testable prop => String -> prop -> SpecWith ()
@@ -226,94 +227,137 @@ main = hspec
226227 $ describe moduleName $ do
227228 let transform = transformCombineFromList Stream.fromList sortEq
228229
229− prop "eval" $
230+ prop "parEval" $
230231 transform
231232 (fmap (+2))
232233 (fmap (+1) . Async.parEval id . fmap (+1))
233234
234235 asyncSpec $ prop "parSequence" . sequenceReplicate
235− -- XXX Need to use asyncSpec in all tests
236− prop "mapM (+1)" $
237− transform (fmap (+1)) (Async.parMapM id (\x -> return (x + 1)))
236+
237+ asyncSpec $
238+ prop "parMapM (+1)"
239+ . transform (fmap (+1))
240+ . (`Async.parMapM` (\x -> return (x + 1)))
238241
239242 -- XXX Need to use eq instead of sortEq for ahead oeprations
240243 -- Binary append
241− prop1 "append [] []"
242− $ cmp (Async.parList id [Stream.nil, Stream.nil]) sortEq []
243− prop1 "append [] [1]"
244− $ cmp (Async.parList id [Stream.nil, Stream.fromPure 1]) sortEq [1]
245− prop1 "append [1] []"
246− $ cmp (Async.parList id [Stream.fromPure 1, Stream.nil]) sortEq [1]
247− prop1 "append [0] [1]"
248− $ let stream = Async.parList id [Stream.fromPure 0, Stream.fromPure 1]
249− in cmp stream sortEq [0, 1]
250−
251− prop1 "append [0] [] [1]"
252− $ let stream =
253− Async.parList id
254− [Stream.fromPure 0, Stream.nil, Stream.fromPure 1]
255− in cmp stream sortEq [0, 1]
256−
257− let async = Async.parTwo id
258− prop1 "append2 left associated"
259− $ let stream =
260− Stream.fromPure 0
261− `async` Stream.fromPure 1
262− `async` Stream.fromPure 2
263− `async` Stream.fromPure 3
264− in cmp stream sortEq [0, 1, 2, 3]
265−
266− prop1 "append right associated"
267− $ let stream =
268− Stream.fromPure 0
269− `async` (Stream.fromPure 1
270− `async` (Stream.fromPure 2
271− `async` Stream.fromPure 3))
272− in cmp stream sortEq [0, 1, 2, 3]
273−
274− prop1 "append balanced"
275− $ let leaf x y = Stream.fromPure x `async` Stream.fromPure y
276− leaf11 = leaf 0 1 `async` leaf 2 (3 :: Int)
277− leaf12 = leaf 4 5 `async` leaf 6 7
278− stream = leaf11 `async` leaf12
279− in cmp stream sortEq [0, 1, 2, 3, 4, 5, 6,7]
280−
281− prop1 "combineWith (maxThreads 1)"
282− $ let stream =
283− Async.parTwo (Async.maxThreads 1)
284− (Stream.fromList [1,2,3,4,5])
244+ asyncSpec $
245+ let appWith cfg = Async.parList cfg [Stream.nil, Stream.nil]
246+ in prop1 "parList [] []" . cmp sortEq ([] :: [Int]) . appWith
247+
248+ asyncSpec $
249+ let appWith cfg = Async.parList cfg [Stream.nil, Stream.fromPure 1]
250+ in prop1 "parList [] [1]" . cmp sortEq [1 :: Int] . appWith
251+
252+ asyncSpec $
253+ let appWith cfg = Async.parList cfg [Stream.fromPure 1, Stream.nil]
254+ in prop1 "parList [1] []" . cmp sortEq [1 :: Int] . appWith
255+
256+ asyncSpec $
257+ let appWith cfg =
258+ Async.parList cfg [Stream.fromPure 0, Stream.fromPure 1]
259+ in prop1 "parList [0] [1]" . cmp sortEq [0, 1 :: Int] . appWith
260+
261+ asyncSpec $
262+ let appWith cfg =
263+ Async.parList
264+ cfg [Stream.fromPure 0, Stream.nil, Stream.fromPure 1]
265+ in prop1 "parList [0] [] [1]" . cmp sortEq [0, 1 :: Int] . appWith
266+
267+ asyncSpec $
268+ let appWith cfg =
269+ Async.parTwo cfg
270+ (Async.parTwo cfg
271+ (Async.parTwo cfg
272+ (Stream.fromPure 0) (Stream.fromPure 1))
273+ (Stream.fromPure 2))
274+ (Stream.fromPure 3)
275+ in prop1 "parTwo left associated"
276+ . cmp sortEq [0, 1, 2, 3 :: Int] . appWith
277+
278+ asyncSpec $
279+ let appWith cfg =
280+ Async.parTwo cfg
281+ (Stream.fromPure 0)
282+ (Async.parTwo cfg
283+ (Stream.fromPure 1)
284+ (Async.parTwo cfg
285+ (Stream.fromPure 2) (Stream.fromPure 3))
286+ )
287+ in prop1 "parTwo right associated"
288+ . cmp sortEq [0, 1, 2, 3 :: Int] . appWith
289+
290+ asyncSpec $
291+ let leaf x y cfg =
292+ Async.parTwo cfg (Stream.fromPure x)
293+ (Stream.fromPure y)
294+ leaf11 cfg =
295+ Async.parTwo cfg (leaf 0 1 cfg) $ leaf 2 (3 :: Int) cfg
296+ leaf12 cfg =
297+ Async.parTwo cfg (leaf 4 5 cfg) $ leaf 6 7 cfg
298+ appWith cfg =
299+ Async.parTwo cfg (leaf11 cfg) (leaf12 cfg)
300+ in prop1 "parTwo balanced"
301+ . cmp sortEq [0, 1, 2, 3, 4, 5, 6,7] . appWith
302+
303+ asyncSpec $
304+ let appWith cfg =
305+ Async.parTwo cfg
306+ (Stream.fromList [1,2,3,4,5 :: Int])
285307 (Stream.fromList [6,7,8,9,10])
286− in cmp stream (==) [1,2,3,4,5,6,7,8,9,10]
287−
288− prop1 "apply (async arg1)"
289− $ let s1 = Async.parApply id (Stream.fromPure (,)) (Stream.fromPure 1 `async` Stream.fromPure 2)
290− s2 = Async.parApply id s1 (Stream.fromPure 3) :: Stream IO (Int, Int)
291− xs = Stream.fold Fold.toList s2
292− in sort <$> xs `shouldReturn` [(1, 3), (2, 3)]
293−
294− prop1 "apply (async arg2)"
295− $ let s1 = Stream.fromPure (1,)
296− s2 = Async.parApply id s1 (Stream.fromPure 2 `async` Stream.fromPure 3)
297− xs = Stream.fold Fold.toList s2 :: IO [(Int, Int)]
298− in sort <$> xs `shouldReturn` [(1, 2), (1, 3)]
308+ in prop1 "parTwo" . cmp (==) [1,2,3,4,5,6,7,8,9,10] . appWith
309+
310+ asyncSpec $
311+ let par2 cfg =
312+ Async.parTwo
313+ cfg
314+ (Stream.fromPure 1)
315+ (Stream.fromPure 2)
316+ s1 cfg =
317+ Async.parApply
318+ cfg
319+ (Stream.fromPure (,))
320+ (par2 cfg)
321+ s2 cfg =
322+ Async.parApply
323+ cfg
324+ (s1 cfg)
325+ (Stream.fromPure 3) :: Stream IO (Int, Int)
326+ in prop1
327+ "parApply (async arg1)" . cmp (==) ( [(1, 3), (2, 3)]) . s2
328+
329+ asyncSpec $
330+ let par2 cfg =
331+ Async.parTwo
332+ cfg
333+ (Stream.fromPure (2 :: Int))
334+ (Stream.fromPure 3)
335+ s1 = Stream.fromPure (1 :: Int,)
336+ s2 cfg = Async.parApply cfg s1 (par2 cfg)
337+ in prop1 "apply (async arg2)" . cmp (==) ([(1, 2), (1, 3)]) . s2
299338
300339 -- concat
301− prop1 "concat"
302− $ let stream =
303− Async.parConcat id
340+ asyncSpec $
341+ let stream cfg =
342+ Async.parConcat cfg
304343 $ fmap Stream.fromPure
305− $ Stream.fromList [1..100]
306− in cmp stream sortEq [1..100]
307−
308− prop "concatMap" $
309− forAll (choose (0, 100)) $ \n ->
310− transform
311− (concatMap (const [1..n]))
312− (Async.parConcatMap id (const (Stream.fromList [1..n])))
344+ $ Stream.fromList [1..100 :: Int]
345+ in prop1 "parConcat" . cmp sortEq [1..100] . stream
346+
347+ asyncSpec $
348+ let f cfg =
349+ forAll (choose (0, 100)) $ \n ->
350+ transform
351+ (concatMap (const [1..n]))
352+ (Async.parConcatMap
353+ cfg (const (Stream.fromList [1..n]))
354+ )
355+ in prop "parConcatMap" . f
313356
314357#ifdef DEVBUILD
315358 describe "Time ordering" $ timeOrdering (Async.parList id)
316359#endif
360+ let async = Async.parTwo id
317361 describe "Exception propagation" $ exceptionPropagation async
318362 -- Ad-hoc tests
319363 it "takes n from stream of streams" $ takeCombined 2