@@ -0,0 +1,322 @@
1+-- |
2+-- Module : Streamly.Test.Data.Stream.Async
3+-- Copyright : (c) 2020 Composewell Technologies
4+--
5+-- License : BSD-3-Clause
6+-- Maintainer : streamly@composewell.com
7+-- Stability : experimental
8+-- Portability : GHC
9+
10+module Streamly.Test.Data.Stream.Async (main) where
11+
12+#ifdef DEVBUILD
13+import Control.Concurrent (threadDelay)
14+#endif
15+import Control.Exception (Exception, try)
16+import Control.Monad (replicateM)
17+import Control.Monad.Catch (throwM)
18+import Data.List (sort)
19+#if __GLASGOW_HASKELL__ < 808
20+import Data.Semigroup ((<>))
21+#endif
22+import Data.Word (Word8)
23+import Streamly.Data.Stream (Stream)
24+import Test.Hspec.QuickCheck
25+import Test.QuickCheck (Testable, Property, choose, forAll, withMaxSuccess)
26+import Test.QuickCheck.Monadic (monadicIO, run)
27+import Test.Hspec as H
28+
29+import qualified Streamly.Data.Fold as Fold
30+import qualified Streamly.Data.Stream as Stream
31+import qualified Streamly.Internal.Data.Stream.Async as Async
32+
33+import Streamly.Test.Common (listEquals)
34+
35+moduleName :: String
36+moduleName = "Data.Stream.Async"
37+
38+-------------------------------------------------------------------------------
39+-- Utilities
40+-------------------------------------------------------------------------------
41+
42+sortEq :: Ord a => [a] -> [a] -> Bool
43+sortEq a b = sort a == sort b
44+
45+cmp :: Stream IO Int -> ([Int] -> [Int] -> Bool) -> [Int] -> Property
46+cmp s eq list =
47+ monadicIO $ do
48+ stream <- run $ Stream.fold Fold.toList s
49+ listEquals eq stream list
50+
51+prop1 :: Testable prop => String -> prop -> SpecWith ()
52+prop1 x y = modifyMaxSuccess (const 1) $ prop x y
53+
54+-- Coverage build takes too long with default number of tests
55+maxTestCount :: Int
56+#ifdef DEVBUILD
57+maxTestCount = 100
58+#else
59+maxTestCount = 10
60+#endif
61+
62+transformCombineFromList ::
63+ ([Int] -> Stream IO Int)
64+ -> ([Int] -> [Int] -> Bool)
65+ -> ([Int] -> [Int])
66+ -> (Stream IO Int -> Stream IO Int)
67+ -> [Int]
68+ -> [Int]
69+ -> [Int]
70+ -> Property
71+transformCombineFromList constr eq listOp op a b c =
72+ withMaxSuccess maxTestCount $
73+ monadicIO $ do
74+ stream <- run (Stream.fold Fold.toList $
75+ constr a `Async.append` op (constr b `Async.append` constr c))
76+ let list = a <> listOp (b <> c)
77+ listEquals eq stream list
78+
79+commonOpConfigs :: [(String, Async.Config -> Async.Config)]
80+commonOpConfigs =
81+ [ ("default", id)
82+#ifndef COVERAGE_BUILD
83+ , ("rate AvgRate 10000", Async.avgRate 10000)
84+ , ("rate Nothing", Async.rate Nothing)
85+ , ("maxBuffer 0", Async.maxBuffer 0)
86+ , ("maxThreads 0", Async.maxThreads 0)
87+ , ("maxThreads 1", Async.maxThreads 1)
88+#ifdef USE_LARGE_MEMORY
89+ , ("maxThreads -1", Async.maxThreads (-1))
90+#endif
91+#endif
92+ ]
93+
94+opConfigs :: [(String, Async.Config -> Async.Config)]
95+opConfigs = commonOpConfigs
96+ ++ [
97+#ifndef COVERAGE_BUILD
98+ ("maxBuffer 1", Async.maxBuffer 1)
99+#endif
100+ ]
101+
102+makeSpec :: [(String, a)] -> (a -> Spec) -> Spec
103+makeSpec cfg spec = mapM_ (\(desc, arg) -> describe desc $ spec arg) cfg
104+asyncSpec :: ((Async.Config -> Async.Config) -> Spec) -> Spec
105+asyncSpec =
106+ makeSpec $ opConfigs
107+#ifndef COVERAGE_BUILD
108+ <> [("maxBuffer (-1)", Async.maxBuffer (-1))]
109+#endif
110+
111+-------------------------------------------------------------------------------
112+-- Compose with MonadThrow
113+-------------------------------------------------------------------------------
114+
115+newtype ExampleException = ExampleException String deriving (Eq, Show, Ord)
116+
117+instance Exception ExampleException
118+
119+exceptionPropagation ::
120+ (Stream IO Int -> Stream IO Int -> Stream IO Int) -> Spec
121+exceptionPropagation f = do
122+ it "append throwM, nil" $
123+ try (tl (throwM (ExampleException "E") `f` Stream.nil))
124+ `shouldReturn`
125+ (Left (ExampleException "E") :: Either ExampleException [Int])
126+ it "append nil, throwM" $
127+ try (tl (Stream.nil `f` throwM (ExampleException "E")))
128+ `shouldReturn`
129+ (Left (ExampleException "E") :: Either ExampleException [Int])
130+ it "append nested throwM" $ do
131+ let nested =
132+ Stream.fromFoldable [1..10]
133+ `f` throwM (ExampleException "E")
134+ `f` Stream.fromFoldable [1..10]
135+ try (tl (Stream.nil `f` nested `f` Stream.fromFoldable [1..10]))
136+ `shouldReturn`
137+ (Left (ExampleException "E")
138+ :: Either ExampleException [Int])
139+ it "sequence throwM" $
140+ let stream = Stream.fromList [throwM (ExampleException "E")]
141+ in try (tl (Stream.nil `f` Async.sequence stream))
142+ `shouldReturn`
143+ (Left (ExampleException "E") :: Either ExampleException [Int])
144+
145+ it "concatMap throwM" $ do
146+ let s1 = Async.concatList $ fmap Stream.fromPure [1..4]
147+ s2 = Async.concatList $ fmap Stream.fromPure [5..8]
148+ try $ tl (
149+ let bind = flip Async.concatMap
150+ in bind s1 $ \x ->
151+ bind s2 $ \y ->
152+ if x + y > 10
153+ then throwM (ExampleException "E")
154+ else return (x + y)
155+ )
156+ `shouldReturn`
157+ (Left (ExampleException "E") :: Either ExampleException [Int])
158+
159+ where
160+
161+ tl = Stream.fold Fold.toList
162+
163+---------------------------------------------------------------------------
164+-- Time ordering
165+---------------------------------------------------------------------------
166+
167+#ifdef DEVBUILD
168+timeOrdering :: (Stream IO Int -> Stream IO Int -> Stream IO Int) -> Spec
169+timeOrdering f = do
170+ it "Parallel ordering left associated" $
171+ Stream.fold Fold.toList (((event 4 `f` event 3) `f` event 2) `f` event 1)
172+ `shouldReturn` [1..4]
173+
174+ it "Parallel ordering right associated" $
175+ Stream.fold Fold.toList (event 4 `f` (event 3 `f` (event 2 `f` event 1)))
176+ `shouldReturn` [1..4]
177+
178+ where event n = Stream.fromEffect (threadDelay (n * 200000)) >> return n
179+#endif
180+
181+-------------------------------------------------------------------------------
182+-- Some ad-hoc tests that failed at times
183+-------------------------------------------------------------------------------
184+
185+takeCombined :: Int -> IO ()
186+takeCombined n = do
187+ let constr = Stream.fromFoldable
188+ r <- Stream.fold Fold.toList $
189+ Stream.take n
190+ (constr ([] :: [Int]) `Async.append` constr ([] :: [Int]))
191+ r `shouldBe` []
192+
193+---------------------------------------------------------------------------
194+-- Main
195+---------------------------------------------------------------------------
196+
197+constructWithLenM
198+ :: (Int -> Stream IO Int)
199+ -> (Int -> IO [Int])
200+ -> Word8
201+ -> Property
202+constructWithLenM mkStream mkList len =
203+ withMaxSuccess maxTestCount
204+ $ monadicIO $ do
205+ stream <-
206+ run
207+ $ Stream.fold Fold.toList
208+ $ mkStream (fromIntegral len)
209+ list <- run $ mkList (fromIntegral len)
210+ listEquals (==) stream list
211+
212+sequenceReplicate
213+ :: (Async.Config -> Async.Config)
214+ -> Word8
215+ -> Property
216+sequenceReplicate cfg = constructWithLenM stream list
217+
218+ where
219+
220+ list = flip replicateM (return 1 :: IO Int)
221+ stream = Async.sequenceWith cfg . flip Stream.replicate (return 1 :: IO Int)
222+
223+main :: IO ()
224+main = hspec
225+ $ H.parallel
226+#ifdef COVERAGE_BUILD
227+ $ modifyMaxSuccess (const 10)
228+#endif
229+ $ describe moduleName $ do
230+ let transform = transformCombineFromList Stream.fromList sortEq
231+
232+ prop "eval" $
233+ transform
234+ (fmap (+2))
235+ (fmap (+1) . Async.eval . fmap (+1))
236+
237+ asyncSpec $ prop "sequenceWith" . sequenceReplicate
238+ -- XXX Need to use asyncSpec in all tests
239+ prop "mapM (+1)" $
240+ transform (fmap (+1)) (Async.mapM (\x -> return (x + 1)))
241+
242+ -- Binary append
243+ prop1 "append [] []"
244+ $ cmp (Stream.nil `Async.append` Stream.nil) sortEq []
245+ prop1 "append [] [1]"
246+ $ cmp (Stream.nil `Async.append` Stream.fromPure 1) sortEq [1]
247+ prop1 "append [1] []"
248+ $ cmp (Stream.fromPure 1 `Async.append` Stream.nil) sortEq [1]
249+ prop1 "append [0] [1]"
250+ $ let stream =
251+ Stream.fromPure 0 `Async.append` Stream.fromPure 1
252+ in cmp stream sortEq [0, 1]
253+
254+ prop1 "append [0] [] [1]"
255+ $ let stream =
256+ Stream.fromPure 0
257+ `Async.append` Stream.nil
258+ `Async.append` Stream.fromPure 1
259+ in cmp stream sortEq [0, 1]
260+
261+ prop1 "append left associated"
262+ $ let stream =
263+ Stream.fromPure 0
264+ `Async.append` Stream.fromPure 1
265+ `Async.append` Stream.fromPure 2
266+ `Async.append` Stream.fromPure 3
267+ in cmp stream sortEq [0, 1, 2, 3]
268+
269+ prop1 "append right associated"
270+ $ let stream =
271+ Stream.fromPure 0
272+ `Async.append` (Stream.fromPure 1
273+ `Async.append` (Stream.fromPure 2
274+ `Async.append` Stream.fromPure 3))
275+ in cmp stream sortEq [0, 1, 2, 3]
276+
277+ prop1 "append balanced"
278+ $ let leaf x y = Stream.fromPure x `Async.append` Stream.fromPure y
279+ leaf11 = leaf 0 1 `Async.append` leaf 2 (3 :: Int)
280+ leaf12 = leaf 4 5 `Async.append` leaf 6 7
281+ stream = leaf11 `Async.append` leaf12
282+ in cmp stream sortEq [0, 1, 2, 3, 4, 5, 6,7]
283+
284+ prop1 "appendWith (maxThreads 1)"
285+ $ let stream =
286+ Async.appendWith (Async.maxThreads 1)
287+ (Stream.fromList [1,2,3,4,5])
288+ (Stream.fromList [6,7,8,9,10])
289+ in cmp stream (==) [1,2,3,4,5,6,7,8,9,10]
290+
291+ prop1 "apply (async arg1)"
292+ $ let s1 = Async.apply (pure (,)) (pure 1 `Async.append` pure 2)
293+ s2 = Async.apply s1 (pure 3) :: Stream IO (Int, Int)
294+ xs = Stream.fold Fold.toList s2
295+ in sort <$> xs `shouldReturn` [(1, 3), (2, 3)]
296+
297+ prop1 "apply (async arg2)"
298+ $ let s1 = pure (1,)
299+ s2 = Async.apply s1 (pure 2 `Async.append` pure 3)
300+ xs = Stream.fold Fold.toList s2 :: IO [(Int, Int)]
301+ in sort <$> xs `shouldReturn` [(1, 2), (1, 3)]
302+
303+ -- concat
304+ prop1 "concat"
305+ $ let stream =
306+ Async.concat
307+ $ fmap Stream.fromPure
308+ $ Stream.fromList [1..100]
309+ in cmp stream sortEq [1..100]
310+
311+ prop "concatMap" $
312+ forAll (choose (0, 100)) $ \n ->
313+ transform
314+ (concatMap (const [1..n]))
315+ (Async.concatMap (const (Stream.fromList [1..n])))
316+
317+#ifdef DEVBUILD
318+ describe "Time ordering" $ timeOrdering Async.append
319+#endif
320+ describe "Exception propagation" $ exceptionPropagation Async.append
321+ -- Ad-hoc tests
322+ it "takes n from stream of streams" $ takeCombined 2