git.iko.soy/streamly
on e9828b3 help sign in
streamly/test/Main.hs
1101 lines 45 kB history raw
1module Main (main) where
3import Control.Concurrent (threadDelay)
4import Control.Exception (Exception, try, ErrorCall(..), catch)
5import Control.Monad (void)
6import Control.Monad.Catch (throwM, MonadThrow)
7import Control.Monad.Error.Class (throwError, MonadError)
8import Control.Monad.IO.Class (MonadIO(liftIO))
9import Control.Monad.State (MonadState, get, modify, runStateT
10 , StateT(..), evalStateT)
11import Control.Monad.Trans.Except (runExceptT, ExceptT)
12import Data.Foldable (forM_, fold)
13import Data.Function ((&))
14import Data.List (sort)
15import Data.Maybe (fromJust, isJust)
16import System.Mem (performMajorGC)
18import Data.IORef
19import Test.Hspec as H
21import Streamly
22import Streamly.Prelude ((.:), nil)
23import qualified Streamly as S
24import qualified Streamly.Prelude as S
25import qualified Streamly.Internal.Data.Stream.IsStream as IP
27singleton :: IsStream t => a -> t m a
28singleton a = a .: nil
30toListSerial :: SerialT IO a -> IO [a]
31toListSerial = S.toList . serially
33toListInterleaved :: WSerialT IO a -> IO [a]
34toListInterleaved = S.toList . wSerially
36toListAsync :: AsyncT IO a -> IO [a]
37toListAsync = S.toList . asyncly
39main :: IO ()
40main = hspec $ do
41 parallelTests
43 describe "restricts concurrency and cleans up extra tasks" $ do
44 it "take 1 asyncly" $ checkCleanup 2 asyncly (S.take 1)
45 it "take 1 wAsyncly" $ checkCleanup 2 wAsyncly (S.take 1)
46 it "take 1 aheadly" $ checkCleanup 2 aheadly (S.take 1)
48 it "takeWhile (< 0) asyncly" $ checkCleanup 2 asyncly (S.takeWhile (< 0))
49 it "takeWhile (< 0) wAsyncly" $ checkCleanup 2 wAsyncly (S.takeWhile (< 0))
50 it "takeWhile (< 0) aheadly" $ checkCleanup 2 aheadly (S.takeWhile (< 0))
52#ifdef DEVBUILD
53 let timed :: (IsStream t, Monad (t IO)) => Int -> t IO Int
54 timed x = S.yieldM (threadDelay (x * 100000)) >> return x
56 -- These are not run parallely because the timing gets affected
57 -- unpredictably when other tests are running on the same machine.
58 --
59 -- Also, they fail intermittently due to scheduling delays, so not run on
60 -- CI machines.
61 describe "Nested parallel and serial compositions" $ do
62 let t = timed
63 p = wAsyncly
64 s = serially
65 {-
66 -- This is not correct, the result can also be [4,4,8,0,8,0,2,2]
67 -- because of parallelism of [8,0] and [8,0].
68 it "Nest <|>, <>, <|> (1)" $
69 let t = timed
70 in toListSerial (
71 ((t 8 <|> t 4) <> (t 2 <|> t 0))
72 <|> ((t 8 <|> t 4) <> (t 2 <|> t 0)))
73 `shouldReturn` ([4,4,8,8,0,0,2,2])
74 -}
75 it "Nest <|>, <>, <|> (2)" $
76 (S.toList . wAsyncly) (
77 s (p (t 4 <> t 8) <> p (t 1 <> t 2))
78 <> s (p (t 4 <> t 8) <> p (t 1 <> t 2)))
79 `shouldReturn` ([4,4,8,8,1,1,2,2])
80 -- FIXME: These two keep failing intermittently on Mac OS X
81 -- Need to examine and fix the tests.
82 {-
83 it "Nest <|>, <=>, <|> (1)" $
84 let t = timed
85 in toListSerial (
86 ((t 8 <|> t 4) <=> (t 2 <|> t 0))
87 <|> ((t 9 <|> t 4) <=> (t 2 <|> t 0)))
88 `shouldReturn` ([4,4,0,0,8,2,9,2])
89 it "Nest <|>, <=>, <|> (2)" $
90 let t = timed
91 in toListSerial (
92 ((t 4 <|> t 8) <=> (t 1 <|> t 2))
93 <|> ((t 4 <|> t 9) <=> (t 1 <|> t 2)))
94 `shouldReturn` ([4,4,1,1,8,2,9,2])
95 -}
96 it "Nest <|>, <|>, <|>" $
97 (S.toList . wAsyncly) (
98 ((t 4 <> t 8) <> (t 0 <> t 2))
99 <> ((t 4 <> t 8) <> (t 0 <> t 2)))
100 `shouldReturn` ([0,0,2,2,4,4,8,8])
102 -- parallely fails on CI machines, may need more difference in times of
103 -- the events, but that would make tests even slower.
104 it "take 1 parallely" $ checkCleanup 3 parallely (S.take 1)
105 it "takeWhile (< 0) parallely" $ checkCleanup 3 parallely (S.takeWhile (< 0))
107 testFoldOpsCleanup "head" S.head
108 testFoldOpsCleanup "null" S.null
109 testFoldOpsCleanup "elem" (S.elem 0)
110 testFoldOpsCleanup "notElem" (S.notElem 0)
111 testFoldOpsCleanup "elemIndex" (S.elemIndex 0)
112 -- S.lookup
113 testFoldOpsCleanup "notElem" (S.notElem 0)
114 testFoldOpsCleanup "find" (S.find (==0))
115 testFoldOpsCleanup "findIndex" (S.findIndex (==0))
116 testFoldOpsCleanup "all" (S.all (==1))
117 testFoldOpsCleanup "any" (S.any (==0))
118 testFoldOpsCleanup "and" (S.and . S.map (==1))
119 testFoldOpsCleanup "or" (S.or . S.map (==0))
120#endif
122 ---------------------------------------------------------------------------
123 -- Semigroup/Monoidal Composition strict ordering checks
124 ---------------------------------------------------------------------------
126 -- test both (<>) and mappend to make sure we are using correct instance
127 -- for Monoid that is using the right version of semigroup. Instance
128 -- deriving can cause us to pick wrong instances sometimes.
130 describe "WSerial interleaved (<>) ordering check" $ interleaveCheck wSerially (<>)
131 describe "WSerial interleaved mappend ordering check" $ interleaveCheck wSerially mappend
133 -- describe "WAsync interleaved (<>) ordering check" $ interleaveCheck wAsyncly (<>)
134 -- describe "WAsync interleaved mappend ordering check" $ interleaveCheck wAsyncly mappend
136 describe "Async (<>) time order check" $ parallelCheck asyncly (<>)
137 describe "Async mappend time order check" $ parallelCheck asyncly mappend
139 -- XXX this keeps failing intermittently, need to investigate
140 -- describe "WAsync (<>) time order check" $ parallelCheck wAsyncly (<>)
141 -- describe "WAsync mappend time order check" $ parallelCheck wAsyncly mappend
143 describe "Parallel (<>) time order check" $ parallelCheck parallely (<>)
144 describe "Parallel mappend time order check" $ parallelCheck parallely mappend
145 it "fromCallback" $ testFromCallback `shouldReturn` (50*101)
147checkCleanup :: IsStream t
148 => Int
149 -> (t IO Int -> SerialT IO Int)
150 -> (t IO Int -> t IO Int)
151 -> IO ()
152checkCleanup d t op = do
153 r <- newIORef (-1 :: Int)
154 S.drain . serially $ do
155 _ <- t $ op $ delay r 0 S.|: delay r 1 S.|: delay r 2 S.|: S.nil
156 return ()
157 performMajorGC
158 threadDelay 500000
159 res <- readIORef r
160 res `shouldBe` 0
161 where
162 delay ref i = threadDelay (i*d*100000) >> writeIORef ref i >> return i
164#ifdef DEVBUILD
165checkCleanupFold :: IsStream t
166 => (t IO Int -> SerialT IO Int)
167 -> (SerialT IO Int -> IO (Maybe Int))
168 -> IO ()
169checkCleanupFold t op = do
170 r <- newIORef (-1 :: Int)
171 _ <- op $ t $ delay r 0 S.|: delay r 1 S.|: delay r 2 S.|: S.nil
172 performMajorGC
173 threadDelay 500000
174 res <- readIORef r
175 res `shouldBe` 0
176 where
177 delay ref i = threadDelay (i*200000) >> writeIORef ref i >> return i
179testFoldOpsCleanup :: String -> (SerialT IO Int -> IO a) -> Spec
180testFoldOpsCleanup name f = do
181 let testOp op x = op x >> return Nothing
182 it (name <> " asyncly") $ checkCleanupFold asyncly (testOp f)
183 it (name <> " wAsyncly") $ checkCleanupFold wAsyncly (testOp f)
184 it (name <> " aheadly") $ checkCleanupFold aheadly (testOp f)
185 it (name <> " parallely") $ checkCleanupFold parallely (testOp f)
186#endif
188parallelTests :: SpecWith ()
189parallelTests = H.parallel $ do
191 ---------------------------------------------------------------------------
192 -- TBD Monoidal composition combinations
193 ---------------------------------------------------------------------------
195 -- TBD need more such combinations to be tested.
196 describe "serial <> and serial <>" $
197 composeAndComposeSimple serially serially (repeat [1 .. 9])
198 describe "ahead <> and ahead <>" $
199 composeAndComposeSimple aheadly aheadly (repeat [1 .. 9])
200 describe "ahead <> and serial <>" $
201 composeAndComposeSimple aheadly serially (repeat [1 .. 9])
202 describe "serial <> and ahead <>" $
203 composeAndComposeSimple serially aheadly (repeat [1 .. 9])
205 describe "<> and <=>" $ composeAndComposeSimple
206 serially
207 wSerially
208 [ [1 .. 9]
209 , [1 .. 9]
210 , [1, 3, 2, 4, 6, 5, 7, 9, 8]
211 , [1, 3, 2, 4, 6, 5, 7, 9, 8]
212 ]
214 describe "<=> and <=>" $ composeAndComposeSimple
215 wSerially
216 wSerially
217 [ [1, 4, 2, 7, 3, 5, 8, 6, 9]
218 , [1, 7, 4, 8, 2, 9, 5, 3, 6]
219 , [1, 4, 3, 7, 2, 6, 9, 5, 8]
220 , [1, 7, 4, 9, 3, 8, 6, 2, 5]
221 ]
223 describe "<=> and <>" $ composeAndComposeSimple
224 wSerially
225 serially
226 [ [1, 4, 2, 7, 3, 5, 8, 6, 9]
227 , [1, 7, 4, 8, 2, 9, 5, 3, 6]
228 , [1, 4, 2, 7, 3, 5, 8, 6, 9]
229 , [1, 7, 4, 8, 2, 9, 5, 3, 6]
230 ]
232 ---------------------------------------------------------------------------
233 -- Monoidal composition recursion loops
234 ---------------------------------------------------------------------------
236 describe "Serial loops" $ loops serially id reverse
237 describe "Ahead loops" $ loops aheadly id reverse
238 describe "Async parallel loops" $ loops asyncly sort sort
239 describe "WAsync loops" $ loops wAsyncly sort sort
240 describe "parallel loops" $ loops parallely sort sort
242 ---------------------------------------------------------------------------
243 -- Bind and monoidal composition combinations
244 ---------------------------------------------------------------------------
246 describe "Bind and compose Stream 1" $ bindAndComposeSimple serially serially
247 describe "Bind and compose Stream 2" $ bindAndComposeSimple serially wSerially
248 describe "Bind and compose Stream 3" $ bindAndComposeSimple serially asyncly
249 describe "Bind and compose Stream 4" $ bindAndComposeSimple serially wAsyncly
250 describe "Bind and compose Stream 5" $ bindAndComposeSimple serially parallely
251 describe "Bind and compose Stream 6" $ bindAndComposeSimple serially aheadly
253 describe "Bind and compose Ahead Stream 0" $ bindAndComposeSimple aheadly aheadly
254 describe "Bind and compose Ahead Stream 1" $ bindAndComposeSimple aheadly serially
255 describe "Bind and compose Ahead Stream 2" $ bindAndComposeSimple aheadly wSerially
256 describe "Bind and compose Ahead Stream 3" $ bindAndComposeSimple aheadly asyncly
257 describe "Bind and compose Ahead Stream 4" $ bindAndComposeSimple aheadly wAsyncly
258 describe "Bind and compose Ahead Stream 5" $ bindAndComposeSimple aheadly parallely
260 describe "Bind and compose Costream 1" $ bindAndComposeSimple wSerially serially
261 describe "Bind and compose Costream 2" $ bindAndComposeSimple wSerially wSerially
262 describe "Bind and compose Costream 3" $ bindAndComposeSimple wSerially asyncly
263 describe "Bind and compose Costream 4" $ bindAndComposeSimple wSerially wAsyncly
264 describe "Bind and compose Costream 5" $ bindAndComposeSimple wSerially parallely
265 describe "Bind and compose Costream 6" $ bindAndComposeSimple wSerially aheadly
267 describe "Bind and compose Async 1" $ bindAndComposeSimple asyncly serially
268 describe "Bind and compose Async 2" $ bindAndComposeSimple asyncly wSerially
269 describe "Bind and compose Async 3" $ bindAndComposeSimple asyncly asyncly
270 describe "Bind and compose Async 4" $ bindAndComposeSimple asyncly wAsyncly
271 describe "Bind and compose Async 5" $ bindAndComposeSimple asyncly parallely
272 describe "Bind and compose Async 6" $ bindAndComposeSimple asyncly aheadly
274 describe "Bind and compose WAsync 1" $ bindAndComposeSimple wAsyncly serially
275 describe "Bind and compose WAsync 2" $ bindAndComposeSimple wAsyncly wSerially
276 describe "Bind and compose WAsync 3" $ bindAndComposeSimple wAsyncly asyncly
277 describe "Bind and compose WAsync 4" $ bindAndComposeSimple wAsyncly wAsyncly
278 describe "Bind and compose WAsync 5" $ bindAndComposeSimple wAsyncly parallely
279 describe "Bind and compose WAsync 6" $ bindAndComposeSimple wAsyncly aheadly
281 describe "Bind and compose Parallel 1" $ bindAndComposeSimple parallely serially
282 describe "Bind and compose Parallel 2" $ bindAndComposeSimple parallely wSerially
283 describe "Bind and compose Parallel 3" $ bindAndComposeSimple parallely asyncly
284 describe "Bind and compose Parallel 4" $ bindAndComposeSimple parallely wAsyncly
285 describe "Bind and compose Parallel 5" $ bindAndComposeSimple parallely parallely
286 describe "Bind and compose Parallel 6" $ bindAndComposeSimple parallely aheadly
288 let fldr, fldl :: (IsStream t, Semigroup (t IO Int)) => [t IO Int] -> t IO Int
289 fldr = foldr (<>) nil
290 fldl = foldl (<>) nil
292 forM_ [fldr, fldl] $ \k ->
293 describe "Bind and compose" $ bindAndComposeHierarchy serially serially k
294 forM_ [fldr, fldl] $ \k ->
295 describe "Bind and compose" $ bindAndComposeHierarchy serially wSerially k
296 forM_ [fldr, fldl] $ \k ->
297 describe "Bind and compose" $ bindAndComposeHierarchy serially asyncly k
298 forM_ [fldr, fldl] $ \k ->
299 describe "Bind and compose" $ bindAndComposeHierarchy serially wAsyncly k
300 forM_ [fldr, fldl] $ \k ->
301 describe "Bind and compose" $ bindAndComposeHierarchy serially parallely k
302 forM_ [fldr, fldl] $ \k ->
303 describe "Bind and compose serially aheadly" $ bindAndComposeHierarchy serially aheadly k
305 forM_ [fldr, fldl] $ \k ->
306 describe "Bind and compose aheadly serially" $ bindAndComposeHierarchy aheadly serially k
307 forM_ [fldr, fldl] $ \k ->
308 describe "Bind and compose aheadly wSerially" $ bindAndComposeHierarchy aheadly wSerially k
309 forM_ [fldr, fldl] $ \k ->
310 describe "Bind and compose aheadly asyncly" $ bindAndComposeHierarchy aheadly asyncly k
311 forM_ [fldr, fldl] $ \k ->
312 describe "Bind and compose aheadly wAsyncly" $ bindAndComposeHierarchy aheadly wAsyncly k
313 forM_ [fldr, fldl] $ \k ->
314 describe "Bind and compose aheadly parallely" $ bindAndComposeHierarchy aheadly parallely k
315 forM_ [fldr, fldl] $ \k ->
316 describe "Bind and compose serially aheadly" $ bindAndComposeHierarchy aheadly aheadly k
318 forM_ [fldr, fldl] $ \k ->
319 describe "Bind and compose" $ bindAndComposeHierarchy wSerially serially k
320 forM_ [fldr, fldl] $ \k ->
321 describe "Bind and compose" $ bindAndComposeHierarchy wSerially wSerially k
322 forM_ [fldr, fldl] $ \k ->
323 describe "Bind and compose" $ bindAndComposeHierarchy wSerially asyncly k
324 forM_ [fldr, fldl] $ \k ->
325 describe "Bind and compose" $ bindAndComposeHierarchy wSerially wAsyncly k
326 forM_ [fldr, fldl] $ \k ->
327 describe "Bind and compose" $ bindAndComposeHierarchy wSerially parallely k
328 forM_ [fldr, fldl] $ \k ->
329 describe "Bind and compose wserially aheadly" $ bindAndComposeHierarchy wSerially aheadly k
331 forM_ [fldr, fldl] $ \k ->
332 describe "Bind and compose" $ bindAndComposeHierarchy asyncly serially k
333 forM_ [fldr, fldl] $ \k ->
334 describe "Bind and compose" $ bindAndComposeHierarchy asyncly wSerially k
335 forM_ [fldr, fldl] $ \k ->
336 describe "Bind and compose" $ bindAndComposeHierarchy asyncly asyncly k
337 forM_ [fldr, fldl] $ \k ->
338 describe "Bind and compose" $ bindAndComposeHierarchy asyncly wAsyncly k
339 forM_ [fldr, fldl] $ \k ->
340 describe "Bind and compose" $ bindAndComposeHierarchy asyncly parallely k
341 forM_ [fldr, fldl] $ \k ->
342 describe "Bind and compose asyncly aheadly" $ bindAndComposeHierarchy asyncly aheadly k
344 forM_ [fldr, fldl] $ \k ->
345 describe "Bind and compose" $ bindAndComposeHierarchy wAsyncly serially k
346 forM_ [fldr, fldl] $ \k ->
347 describe "Bind and compose" $ bindAndComposeHierarchy wAsyncly wSerially k
348 forM_ [fldr, fldl] $ \k ->
349 describe "Bind and compose" $ bindAndComposeHierarchy wAsyncly asyncly k
350 forM_ [fldr, fldl] $ \k ->
351 describe "Bind and compose" $ bindAndComposeHierarchy wAsyncly wAsyncly k
352 forM_ [fldr, fldl] $ \k ->
353 describe "Bind and compose" $ bindAndComposeHierarchy wAsyncly parallely k
354 forM_ [fldr, fldl] $ \k ->
355 describe "Bind and compose wAsyncly aheadly" $ bindAndComposeHierarchy wAsyncly aheadly k
357 forM_ [fldr, fldl] $ \k ->
358 describe "Bind and compose" $ bindAndComposeHierarchy parallely serially k
359 forM_ [fldr, fldl] $ \k ->
360 describe "Bind and compose" $ bindAndComposeHierarchy parallely wSerially k
361 forM_ [fldr, fldl] $ \k ->
362 describe "Bind and compose" $ bindAndComposeHierarchy parallely asyncly k
363 forM_ [fldr, fldl] $ \k ->
364 describe "Bind and compose" $ bindAndComposeHierarchy parallely wAsyncly k
365 forM_ [fldr, fldl] $ \k ->
366 describe "Bind and compose" $ bindAndComposeHierarchy parallely parallely k
367 forM_ [fldr, fldl] $ \k ->
368 describe "Bind and compose parallely aheadly" $ bindAndComposeHierarchy parallely aheadly k
370 -- Nest two lists using different styles of product compositions
371 it "Nests two streams using monadic serial composition" nestTwoSerial
372 it "Nests two streams using monadic ahead composition" nestTwoAhead
373 it "Nests two streams using monadic interleaved composition" nestTwoInterleaved
374 it "Nests two streams using monadic Async composition" nestTwoAsync
375 it "Nests two streams using monadic WAsync composition" nestTwoWAsync
376 it "Nests two streams using monadic parallel composition" nestTwoParallel
378 it "Nests two streams using applicative serial composition" nestTwoSerialApp
379 it "Nests two streams using applicative ahead composition" nestTwoAheadApp
380 it "Nests two streams using applicative interleaved composition" nestTwoInterleavedApp
381 it "Nests two streams using applicative Async composition" nestTwoAsyncApp
382 it "Nests two streams using applicative WAsync composition" nestTwoWAsyncApp
383 it "Nests two streams using applicative parallel composition" nestTwoParallelApp
385 ---------------------------------------------------------------------------
386 -- TBD Bind and Bind combinations
387 ---------------------------------------------------------------------------
389 -- TBD combine all binds and all compose in one example
390 describe "Miscellaneous combined examples" mixedOps
391 describe "Miscellaneous combined examples aheadly" mixedOpsAheadly
392 describe "Simple MonadError and MonadThrow" simpleMonadError
394 {-
395 describe "Composed MonadError serially" $ composeWithMonadError serially
396 describe "Composed MonadError wSerially" $ composeWithMonadError wSerially
397 describe "Composed MonadError asyncly" $ composeWithMonadError asyncly
398 describe "Composed MonadError wAsyncly" $ composeWithMonadError wAsyncly
399 -}
401 describe "Composed MonadThrow serially" $ composeWithMonadThrow serially
402 describe "Composed MonadThrow wSerially" $ composeWithMonadThrow wSerially
403 describe "Composed MonadThrow asyncly" $ composeWithMonadThrow asyncly
404 describe "Composed MonadThrow wAsyncly" $ composeWithMonadThrow wAsyncly
405 describe "Composed MonadThrow parallely" $ composeWithMonadThrow parallely
406 describe "Composed MonadThrow aheadly" $ composeWithMonadThrow aheadly
408 describe "take on infinite concurrent stream" $ takeInfinite asyncly
409 describe "take on infinite concurrent stream" $ takeInfinite wAsyncly
410 describe "take on infinite concurrent stream" $ takeInfinite aheadly
412 ---------------------------------------------------------------------------
413 -- Some ad-hoc tests that failed at times
414 ---------------------------------------------------------------------------
416 it "takes n from stream of streams" (takeCombined 1 aheadly)
417 it "takes n from stream of streams" (takeCombined 2 asyncly)
418 it "takes n from stream of streams" (takeCombined 3 wAsyncly)
420 ---------------------------------------------------------------------------
421 -- Left folds are strict enough
422 ---------------------------------------------------------------------------
424#ifdef DEVBUILD
425 it "foldx is strict enough" checkFoldxStrictness
426 it "scanx is strict enough" checkScanxStrictness
427 it "foldxM is strict enough" (checkFoldMStrictness foldxMStrictCheck)
428#endif
429 it "foldl' is strict enough" checkFoldl'Strictness
430 it "scanl' is strict enough" checkScanl'Strictness
431 it "foldlM' is strict enough" (checkFoldMStrictness foldlM'StrictCheck)
432 it "scanlM' is strict enough" (checkScanlMStrictness scanlM'StrictCheck)
434 ---------------------------------------------------------------------------
435 -- Right folds are lazy enough
436 ---------------------------------------------------------------------------
438 it "foldrM is lazy enough" checkFoldrLaziness
440 ---------------------------------------------------------------------------
441 -- Monadic state transfer in concurrent tasks
442 ---------------------------------------------------------------------------
444 -- XXX Can we write better test cases to hit every case?
446 it "async: state is saved and used if the work is partially enqueued"
447 (checkMonadicStateTransfer async)
449 it "wAsync: state is saved and used if the work is partially enqueued"
450 (checkMonadicStateTransfer wAsync)
452 it "ahead: state is saved and used if the work is partially enqueued"
453 (checkMonadicStateTransfer ahead)
455 ---------------------------------------------------------------------------
456 -- Monadic state snapshot in concurrent tasks
457 ---------------------------------------------------------------------------
459 it "asyncly maintains independent states in concurrent tasks"
460 (monadicStateSnapshot asyncly)
461 it "asyncly limited maintains independent states in concurrent tasks"
462 (monadicStateSnapshot (asyncly . S.take 10000))
463 it "wAsyncly maintains independent states in concurrent tasks"
464 (monadicStateSnapshot wAsyncly)
465 it "wAsyncly limited maintains independent states in concurrent tasks"
466 (monadicStateSnapshot (wAsyncly . S.take 10000))
467 it "aheadly maintains independent states in concurrent tasks"
468 (monadicStateSnapshot aheadly)
469 it "aheadly limited maintains independent states in concurrent tasks"
470 (monadicStateSnapshot (aheadly . S.take 10000))
471 it "parallely maintains independent states in concurrent tasks"
472 (monadicStateSnapshot parallely)
474 it "async maintains independent states in concurrent tasks"
475 (monadicStateSnapshotOp async)
476 it "ahead maintains independent states in concurrent tasks"
477 (monadicStateSnapshotOp ahead)
478 it "wAsync maintains independent states in concurrent tasks"
479 (monadicStateSnapshotOp wAsync)
480 it "parallel maintains independent states in concurrent tasks"
481 (monadicStateSnapshotOp Streamly.parallel)
483 ---------------------------------------------------------------------------
484 -- Slower tests are at the end
485 ---------------------------------------------------------------------------
487 ---------------------------------------------------------------------------
488 -- Thread limits
489 ---------------------------------------------------------------------------
491 it "asyncly crosses thread limit (2000 threads)" $
492 S.drain (asyncly $ fold $
493 replicate 2000 $ S.yieldM $ threadDelay 1000000)
494 `shouldReturn` ()
496 it "aheadly crosses thread limit (4000 threads)" $
497 S.drain (aheadly $ fold $
498 replicate 4000 $ S.yieldM $ threadDelay 1000000)
499 `shouldReturn` ()
501-- Each snapshot carries an independent state. Multiple parallel tasks should
502-- not affect each other's state. This is especially important when we run
503-- multiple tasks in a single thread.
504snapshot :: (IsStream t, MonadAsync m, MonadState Int m) => t m ()
505snapshot =
506 -- We deliberately use a replicate count 1 here, because a lower count
507 -- catches problems that a higher count doesn't.
508 S.replicateM 1 $ do
509 -- Even though we modify the state here it should not reflect in other
510 -- parallel tasks, it is local to each concurrent task.
511 modify (+1) >> get >>= liftIO . (`shouldSatisfy` (==1))
512 modify (+1) >> get >>= liftIO . (`shouldSatisfy` (==2))
514snapshot1 :: (IsStream t, MonadAsync m, MonadState Int m) => t m ()
515snapshot1 = S.replicateM 1000 $
516 modify (+1) >> get >>= liftIO . (`shouldSatisfy` (==2))
518snapshot2 :: (IsStream t, MonadAsync m, MonadState Int m) => t m ()
519snapshot2 = S.replicateM 1000 $
520 modify (+1) >> get >>= liftIO . (`shouldSatisfy` (==2))
522stateComp
523 :: ( IsStream t
524 , MonadAsync m
525 , Semigroup (t m ())
526 , MonadIO (t m)
527 , MonadState Int m
528 , MonadState Int (t m)
529 )
530 => t m ()
531stateComp = do
532 -- Each task in a concurrent composition inherits the state and maintains
533 -- its own modifications to it, not affecting the parent computation.
534 snapshot <> (modify (+1) >> (snapshot1 <> snapshot2))
535 -- The above modify statement does not affect our state because that is
536 -- used in a parallel composition. In a serial composition it will affect
537 -- our state.
538 get >>= liftIO . (`shouldSatisfy` (== (0 :: Int)))
540monadicStateSnapshot
541 :: ( IsStream t
542 , Semigroup (t (StateT Int IO) ())
543 , MonadIO (t (StateT Int IO))
544 , MonadState Int (t (StateT Int IO))
545 )
546 => (t (StateT Int IO) () -> SerialT (StateT Int IO) ()) -> IO ()
547monadicStateSnapshot t = void $ runStateT (S.drain $ t stateComp) 0
549stateCompOp
550 :: ( AsyncT (StateT Int IO) ()
551 -> AsyncT (StateT Int IO) ()
552 -> AsyncT (StateT Int IO) ()
553 )
554 -> SerialT (StateT Int IO) ()
555stateCompOp op = do
556 -- Each task in a concurrent composition inherits the state and maintains
557 -- its own modifications to it, not affecting the parent computation.
558 asyncly (snapshot `op` (modify (+1) >> (snapshot1 `op` snapshot2)))
559 -- The above modify statement does not affect our state because that is
560 -- used in a parallel composition. In a serial composition it will affect
561 -- our state.
562 get >>= liftIO . (`shouldSatisfy` (== (0 :: Int)))
564monadicStateSnapshotOp
565 :: ( AsyncT (StateT Int IO) ()
566 -> AsyncT (StateT Int IO) ()
567 -> AsyncT (StateT Int IO) ()
568 )
569 -> IO ()
570monadicStateSnapshotOp op = void $ runStateT (S.drain $ stateCompOp op) 0
572takeCombined :: (Monad m, Semigroup (t m Int), Show a, Eq a, IsStream t)
573 => Int -> (t m Int -> SerialT IO a) -> IO ()
574takeCombined n t = do
575 let constr = S.fromFoldable
576 r <- (S.toList . t) $
577 S.take n (constr ([] :: [Int]) <> constr ([] :: [Int]))
578 r `shouldBe` []
580checkMonadicStateTransfer
581 :: (IsStream t1, IsStream t2)
582 => ( t1 (StateT Int IO) ()
583 -> t2 (StateT Int IO) ()
584 -> SerialT (StateT Int IO) a3 )
585 -> IO ()
586checkMonadicStateTransfer op = evalStateT str (0 :: Int)
587 where
588 str =
589 S.drain $
590 maxBuffer 1 $
591 (serially $ S.mapM snapshoti $ S.fromList [1..10]) `op`
592 (serially $ S.mapM snapshoti $ S.fromList [1..10])
593 snapshoti y = do
594 modify (+ 1)
595 x <- get
596 lift1 $ x `shouldBe` y
597 lift1 m = StateT $ \s -> do
598 a <- m
599 return (a, s)
601checkFoldrLaziness :: IO ()
602checkFoldrLaziness = do
603 S.foldrM (\x xs -> if odd x then return True else xs)
604 (return False) (S.fromList (2:4:5:undefined :: [Int]))
605 `shouldReturn` True
607 S.toList (IP.foldrS (\x xs -> if odd x then return True else xs)
608 (return False)
609 $ (S.fromList (2:4:5:undefined) :: SerialT IO Int))
610 `shouldReturn` [True]
612 S.toList (IP.foldrT (\x xs -> if odd x then return True else xs)
613 (return False)
614 $ (S.fromList (2:4:5:undefined) :: SerialT IO Int))
615 `shouldReturn` [True]
617#ifdef DEVBUILD
618checkFoldxStrictness :: IO ()
619checkFoldxStrictness = do
620 let s = return (1 :: Int) `S.consM` error "failure"
621 catch (S.foldx (\_ a -> if a == 1 then error "success" else "done")
622 "begin" id s)
623 (\(ErrorCall err) -> return err)
624 `shouldReturn` "success"
625#endif
627checkFoldl'Strictness :: IO ()
628checkFoldl'Strictness = do
629 let s = return (1 :: Int) `S.consM` error "failure"
630 catch (S.foldl' (\_ a -> if a == 1 then error "success" else "done")
631 "begin" s)
632 (\(ErrorCall err) -> return err)
633 `shouldReturn` "success"
635#ifdef DEVBUILD
636checkScanxStrictness :: IO ()
637checkScanxStrictness = do
638 let s = return (1 :: Int) `S.consM` error "failure"
639 catch
640 (S.drain (
641 S.scanx (\_ a ->
642 if a == 1
643 then error "success"
644 else "done")
645 "begin" id s
646 )
647 >> return "finished"
648 )
649 (\(ErrorCall err) -> return err)
650 `shouldReturn` "success"
651#endif
653checkScanl'Strictness :: IO ()
654checkScanl'Strictness = do
655 let s = return (1 :: Int) `S.consM` error "failure"
656 catch
657 (S.drain
658 (S.scanl'
659 (\_ a ->
660 if a == 1
661 then error "success"
662 else "done")
663 "begin"
664 s)
665 >> return "finished"
666 )
667 (\(ErrorCall err) -> return err)
668 `shouldReturn` "success"
670foldlM'StrictCheck :: IORef Int -> SerialT IO Int -> IO ()
671foldlM'StrictCheck ref = S.foldlM' (\_ _ -> writeIORef ref 1) (return ())
673#ifdef DEVBUILD
674foldxMStrictCheck :: IORef Int -> SerialT IO Int -> IO ()
675foldxMStrictCheck ref = S.foldxM (\_ _ -> writeIORef ref 1) (return ()) return
676#endif
678checkFoldMStrictness :: (IORef Int -> SerialT IO Int -> IO ()) -> IO ()
679checkFoldMStrictness f = do
680 ref <- newIORef 0
681 let s = return 1 `S.consM` error "x"
682 catch (f ref s) (\(_ :: ErrorCall) -> return ())
683 readIORef ref `shouldReturn` 1
685scanlM'StrictCheck :: IORef Int -> SerialT IO Int -> SerialT IO ()
686scanlM'StrictCheck ref = S.scanlM' (\_ _ -> writeIORef ref 1) ()
688checkScanlMStrictness :: (IORef Int -> SerialT IO Int -> SerialT IO ()) -> IO ()
689checkScanlMStrictness f = do
690 ref <- newIORef 0
691 let s = return 1 `S.consM` error "x"
692 catch (S.drain $ f ref s) (\(_ :: ErrorCall) -> return ())
693 readIORef ref `shouldReturn` 1
695takeInfinite :: IsStream t => (t IO Int -> SerialT IO Int) -> Spec
696takeInfinite t =
697 it "take 1" $
698 S.drain (t $ S.take 1 $ S.repeatM (print "hello" >> return (1::Int)))
699 `shouldReturn` ()
701-- XXX need to test that we have promptly cleaned up everything after the error
702-- XXX We can also check the output that we are expected to get before the
703-- error occurs.
705newtype ExampleException = ExampleException String deriving (Eq, Show)
707instance Exception ExampleException
709simpleMonadError :: Spec
710simpleMonadError = do
711{-
712 it "simple runExceptT" $ do
713 (runExceptT $ S.drain $ return ())
714 `shouldReturn` (Right () :: Either String ())
715 it "simple runExceptT with error" $ do
716 (runExceptT $ S.drain $ throwError "E") `shouldReturn` Left "E"
717 -}
718 it "simple try" $
719 try (S.drain $ return ())
720 `shouldReturn` (Right () :: Either ExampleException ())
721 it "simple try with throw error" $
722 try (S.drain $ throwM $ ExampleException "E")
723 `shouldReturn` (Left (ExampleException "E") :: Either ExampleException ())
725composeWithMonadThrow
726 :: ( IsStream t
727 , Semigroup (t IO Int)
728 , MonadThrow (t IO)
729 )
730 => (t IO Int -> SerialT IO Int) -> Spec
731composeWithMonadThrow t = do
732 it "Compose throwM, nil" $
733 try (tl (throwM (ExampleException "E") <> S.nil))
734 `shouldReturn` (Left (ExampleException "E") :: Either ExampleException [Int])
735 it "Compose nil, throwM" $
736 try (tl (S.nil <> throwM (ExampleException "E")))
737 `shouldReturn` (Left (ExampleException "E") :: Either ExampleException [Int])
738 oneLevelNestedSum "serially" serially
739 oneLevelNestedSum "wSerially" wSerially
740 oneLevelNestedSum "asyncly" asyncly
741 oneLevelNestedSum "wAsyncly" wAsyncly
742 -- XXX add two level nesting
744 oneLevelNestedProduct "serially" serially
745 oneLevelNestedProduct "wSerially" wSerially
746 oneLevelNestedProduct "asyncly" asyncly
747 oneLevelNestedProduct "wAsyncly" wAsyncly
749 where
750 tl = S.toList . t
751 oneLevelNestedSum desc t1 =
752 it ("One level nested sum " <> desc) $ do
753 let nested = S.fromFoldable [1..10] <> throwM (ExampleException "E")
754 <> S.fromFoldable [1..10]
755 try (tl (S.nil <> t1 nested <> S.fromFoldable [1..10]))
756 `shouldReturn` (Left (ExampleException "E") :: Either ExampleException [Int])
758 oneLevelNestedProduct desc t1 =
759 it ("One level nested product" <> desc) $ do
760 let s1 = t $ S.foldMapWith (<>) return [1..4]
761 s2 = t1 $ S.foldMapWith (<>) return [5..8]
762 try $ tl (do
763 x <- adapt s1
764 y <- s2
765 if x + y > 10
766 then throwM (ExampleException "E")
767 else return (x + y)
768 )
769 `shouldReturn` (Left (ExampleException "E") :: Either ExampleException [Int])
771_composeWithMonadError
772 :: ( IsStream t
773 , Semigroup (t (ExceptT String IO) Int)
774 , MonadError String (t (ExceptT String IO))
775 )
776 => (t (ExceptT String IO) Int -> SerialT (ExceptT String IO) Int) -> Spec
777_composeWithMonadError t = do
778 let tl = S.toList . t
779 it "Compose throwError, nil" $
780 runExceptT (tl (throwError "E" <> S.nil)) `shouldReturn` Left "E"
781 it "Compose nil, error" $
782 runExceptT (tl (S.nil <> throwError "E")) `shouldReturn` Left "E"
784nestTwoSerial :: Expectation
785nestTwoSerial =
786 let s1 = S.foldMapWith (<>) return [1..4]
787 s2 = S.foldMapWith (<>) return [5..8]
788 in toListSerial (do
789 x <- s1
790 y <- s2
791 return (x + y)
792 ) `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])
794nestTwoAhead :: Expectation
795nestTwoAhead =
796 let s1 = S.foldMapWith (<>) return [1..4]
797 s2 = S.foldMapWith (<>) return [5..8]
798 in (S.toList . aheadly) (do
799 x <- s1
800 y <- s2
801 return (x + y)
802 ) `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])
804nestTwoSerialApp :: Expectation
805nestTwoSerialApp =
806 let s1 = S.foldMapWith (<>) return [1..4]
807 s2 = S.foldMapWith (<>) return [5..8]
808 in toListSerial ((+) <$> s1 <*> s2)
809 `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])
811nestTwoAheadApp :: Expectation
812nestTwoAheadApp =
813 let s1 = S.foldMapWith (<>) return [1..4]
814 s2 = S.foldMapWith (<>) return [5..8]
815 in (S.toList . aheadly) ((+) <$> s1 <*> s2)
816 `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])
818nestTwoInterleaved :: Expectation
819nestTwoInterleaved =
820 let s1 = S.foldMapWith (<>) return [1..4]
821 s2 = S.foldMapWith (<>) return [5..8]
822 in toListInterleaved (do
823 x <- s1
824 y <- s2
825 return (x + y)
826 ) `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])
828nestTwoInterleavedApp :: Expectation
829nestTwoInterleavedApp =
830 let s1 = S.foldMapWith (<>) return [1..4]
831 s2 = S.foldMapWith (<>) return [5..8]
832 in toListInterleaved ((+) <$> s1 <*> s2)
833 `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])
835nestTwoAsync :: Expectation
836nestTwoAsync =
837 let s1 = S.foldMapWith (<>) return [1..4]
838 s2 = S.foldMapWith (<>) return [5..8]
839 in sort <$> toListAsync (do
840 x <- s1
841 y <- s2
842 return (x + y))
843 `shouldReturn` sort ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])
845nestTwoAsyncApp :: Expectation
846nestTwoAsyncApp =
847 let s1 = S.foldMapWith (<>) return [1..4]
848 s2 = S.foldMapWith (<>) return [5..8]
849 in sort <$> toListAsync ((+) <$> s1 <*> s2)
850 `shouldReturn` sort ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])
852nestTwoWAsync :: Expectation
853nestTwoWAsync =
854 let s1 = S.foldMapWith (<>) return [1..4]
855 s2 = S.foldMapWith (<>) return [5..8]
856 in sort <$> (S.toList . wAsyncly) (do
857 x <- s1
858 y <- s2
859 return (x + y))
860 `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])
862nestTwoParallel :: Expectation
863nestTwoParallel =
864 let s1 = S.foldMapWith (<>) return [1..4]
865 s2 = S.foldMapWith (<>) return [5..8]
866 in sort <$> (S.toList . parallely) (do
867 x <- s1
868 y <- s2
869 return (x + y))
870 `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])
872nestTwoWAsyncApp :: Expectation
873nestTwoWAsyncApp =
874 let s1 = S.foldMapWith (<>) return [1..4]
875 s2 = S.foldMapWith (<>) return [5..8]
876 in sort <$> (S.toList . wAsyncly) ((+) <$> s1 <*> s2)
877 `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])
879nestTwoParallelApp :: Expectation
880nestTwoParallelApp =
881 let s1 = S.foldMapWith (<>) return [1..4]
882 s2 = S.foldMapWith (<>) return [5..8]
883 in sort <$> (S.toList . parallely) ((+) <$> s1 <*> s2)
884 `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])
886interleaveCheck :: IsStream t
887 => (t IO Int -> SerialT IO Int)
888 -> (t IO Int -> t IO Int -> t IO Int)
889 -> Spec
890interleaveCheck t f =
891 it "Interleave four" $
892 (S.toList . t) ((singleton 0 `f` singleton 1) `f` (singleton 100 `f` singleton 101))
893 `shouldReturn` [0, 100, 1, 101]
895parallelCheck :: (IsStream t, Monad (t IO))
896 => (t IO Int -> SerialT IO Int)
897 -> (t IO Int -> t IO Int -> t IO Int)
898 -> Spec
899parallelCheck t f = do
900 it "Parallel ordering left associated" $
901 (S.toList . t) (((event 4 `f` event 3) `f` event 2) `f` event 1)
902 `shouldReturn` [1..4]
904 it "Parallel ordering right associated" $
905 (S.toList . t) (event 4 `f` (event 3 `f` (event 2 `f` event 1)))
906 `shouldReturn` [1..4]
908 where event n = S.yieldM (threadDelay (n * 200000)) >> return n
910composeAndComposeSimple
911 :: ( IsStream t1, Semigroup (t1 IO Int)
912 , IsStream t2, Monoid (t2 IO Int), Monad (t2 IO)
913#if __GLASGOW_HASKELL__ < 804
914 , Semigroup (t2 IO Int)
915#endif
916 )
917 => (t1 IO Int -> SerialT IO Int)
918 -> (t2 IO Int -> t2 IO Int)
919 -> [[Int]] -> Spec
920composeAndComposeSimple t1 t2 answer = do
921 let rfold = adapt . t2 . S.foldMapWith (<>) return
922 it "Compose right associated outer expr, right folded inner" $
923 (S.toList . t1) (rfold [1,2,3] <> (rfold [4,5,6] <> rfold [7,8,9]))
924 `shouldReturn` head answer
926 it "Compose left associated outer expr, right folded inner" $
927 (S.toList . t1) ((rfold [1,2,3] <> rfold [4,5,6]) <> rfold [7,8,9])
928 `shouldReturn` (answer !! 1)
930 let lfold xs = adapt $ t2 $ foldl (<>) mempty $ fmap return xs
931 it "Compose right associated outer expr, left folded inner" $
932 (S.toList . t1) (lfold [1,2,3] <> (lfold [4,5,6] <> lfold [7,8,9]))
933 `shouldReturn` (answer !! 2)
935 it "Compose left associated outer expr, left folded inner" $
936 (S.toList . t1) ((lfold [1,2,3] <> lfold [4,5,6]) <> lfold [7,8,9])
937 `shouldReturn` (answer !! 3)
939loops
940 :: (IsStream t, Semigroup (t IO Int), Monad (t IO))
941 => (t IO Int -> t IO Int)
942 -> ([Int] -> [Int])
943 -> ([Int] -> [Int])
944 -> Spec
945loops t tsrt hsrt = do
946 it "Tail recursive loop" $ (tsrt <$> (S.toList . adapt) (loopTail 0))
947 `shouldReturn` [0..3]
949 it "Head recursive loop" $ (hsrt <$> (S.toList . adapt) (loopHead 0))
950 `shouldReturn` [0..3]
952 where
953 loopHead x = do
954 -- this print line is important for the test (causes a bind)
955 S.yieldM $ putStrLn "LoopHead..."
956 t $ (if x < 3 then loopHead (x + 1) else nil) <> return x
958 loopTail x = do
959 -- this print line is important for the test (causes a bind)
960 S.yieldM $ putStrLn "LoopTail..."
961 t $ return x <> (if x < 3 then loopTail (x + 1) else nil)
963bindAndComposeSimple
964 :: ( IsStream t1, IsStream t2, Semigroup (t2 IO Int), Monad (t2 IO))
965 => (t1 IO Int -> SerialT IO Int)
966 -> (t2 IO Int -> t2 IO Int)
967 -> Spec
968bindAndComposeSimple t1 t2 = do
969 -- XXX need a bind in the body of forEachWith instead of a simple return
970 it "Compose many (right fold) with bind" $
971 (sort <$> (S.toList . t1)
972 (adapt . t2 $ S.forEachWith (<>) [1..10 :: Int] return))
973 `shouldReturn` [1..10]
975 it "Compose many (left fold) with bind" $
976 let forL xs k = foldl (<>) nil $ fmap k xs
977 in (sort <$> (S.toList . t1) (adapt . t2 $ forL [1..10 :: Int] return))
978 `shouldReturn` [1..10]
980bindAndComposeHierarchy
981 :: ( IsStream t1, Monad (t1 IO)
982 , IsStream t2, Monad (t2 IO))
983 => (t1 IO Int -> SerialT IO Int)
984 -> (t2 IO Int -> t2 IO Int)
985 -> ([t2 IO Int] -> t2 IO Int)
986 -> Spec
987bindAndComposeHierarchy t1 t2 g =
988 it "Bind and compose nested" $
989 (sort <$> (S.toList . t1) bindComposeNested)
990 `shouldReturn` (sort (
991 [12, 18]
992 <> replicate 3 13
993 <> replicate 3 17
994 <> replicate 6 14
995 <> replicate 6 16
996 <> replicate 7 15) :: [Int])
998 where
1000 -- bindComposeNested :: WAsyncT IO Int
1001 bindComposeNested =
1002 let c1 = tripleCompose (return 1) (return 2) (return 3)
1003 c2 = tripleCompose (return 4) (return 5) (return 6)
1004 c3 = tripleCompose (return 7) (return 8) (return 9)
1005 b = tripleBind c1 c2 c3
1006-- it seems to be causing a huge space leak in hspec so disabling this for now
1007-- c = tripleCompose b b b
1008-- m = tripleBind c c c
1009-- in m
1010 in b
1012 tripleCompose a b c = adapt . t2 $ g [a, b, c]
1013 tripleBind mx my mz =
1014 mx >>= \x -> my
1015 >>= \y -> mz
1016 >>= \z -> return (x + y + z)
1018mixedOps :: Spec
1019mixedOps =
1020 it "Compose many ops" $
1021 (sort <$> toListSerial composeMixed)
1022 `shouldReturn` ([8,9,9,9,9,9,10,10,10,10,10,10,10,10,10,10,11,11
1023 ,11,11,11,11,11,11,11,11,12,12,12,12,12,13
1024 ] :: [Int])
1025 where
1027 composeMixed :: SerialT IO Int
1028 composeMixed = do
1029 S.yieldM $ return ()
1030 S.yieldM $ putStr ""
1031 let x = 1
1032 let y = 2
1033 z <- do
1034 x1 <- wAsyncly $ return 1 <> return 2
1035 S.yieldM $ return ()
1036 S.yieldM $ putStr ""
1037 y1 <- asyncly $ return 1 <> return 2
1038 z1 <- do
1039 x11 <- return 1 <> return 2
1040 y11 <- asyncly $ return 1 <> return 2
1041 z11 <- wSerially $ return 1 <> return 2
1042 S.yieldM $ return ()
1043 S.yieldM $ putStr ""
1044 return (x11 + y11 + z11)
1045 return (x1 + y1 + z1)
1046 return (x + y + z)
1048mixedOpsAheadly :: Spec
1049mixedOpsAheadly =
1050 it "Compose many ops" $
1051 (sort <$> toListSerial composeMixed)
1052 `shouldReturn` ([8,9,9,9,9,9,10,10,10,10,10,10,10,10,10,10,11,11
1053 ,11,11,11,11,11,11,11,11,12,12,12,12,12,13
1054 ] :: [Int])
1055 where
1057 composeMixed :: SerialT IO Int
1058 composeMixed = do
1059 S.yieldM $ return ()
1060 S.yieldM $ putStr ""
1061 let x = 1
1062 let y = 2
1063 z <- do
1064 x1 <- wAsyncly $ return 1 <> return 2
1065 S.yieldM $ return ()
1066 S.yieldM $ putStr ""
1067 y1 <- aheadly $ return 1 <> return 2
1068 z1 <- do
1069 x11 <- return 1 <> return 2
1070 y11 <- aheadly $ return 1 <> return 2
1071 z11 <- parallely $ return 1 <> return 2
1072 S.yieldM $ return ()
1073 S.yieldM $ putStr ""
1074 return (x11 + y11 + z11)
1075 return (x1 + y1 + z1)
1076 return (x + y + z)
1078testFromCallback :: IO Int
1079testFromCallback = do
1080 ref <- newIORef Nothing
1081 let stream = S.map Just (IP.fromCallback (setCallback ref))
1082 `Streamly.parallel` runCallback ref
1083 S.sum $ S.map fromJust $ S.takeWhile isJust stream
1085 where
1087 setCallback ref cb = do
1088 writeIORef ref (Just cb)
1090 runCallback ref = S.yieldM $ do
1091 cb <-
1092 S.repeatM (readIORef ref)
1093 & IP.delayPost 0.1
1094 & S.mapMaybe id
1095 & S.head
1097 S.fromList [1..100]
1098 & IP.delayPost 0.001
1099 & S.mapM_ (fromJust cb)
1100 threadDelay 100000
1101 return Nothing