@@ -9,38 +9,126 @@
99
1010module Streamly.Test.Prelude (main) where
1111
12+import Control.Concurrent (threadDelay)
1213import Control.Exception (Exception, try)
14+import Control.Monad (when)
1315import Control.Monad.Catch (throwM)
1416import Control.Monad.Error.Class (throwError, MonadError)
1517import Control.Monad.Trans.Except (runExceptT, ExceptT)
18+import Data.Function ((&))
19+import Data.IORef (newIORef, readIORef, writeIORef)
20+import Data.Int (Int64)
1621import Data.List (sort)
22+import Data.Maybe (fromJust, isJust)
1723#if !(MIN_VERSION_base(4,11,0))
1824import Data.Semigroup (Semigroup(..))
19−import Data.IORef
2025#endif
21−
2226import Test.Hspec as H
23−
24−import Streamly.Prelude
25− ( SerialT, IsStream)
27+import Test.Hspec.QuickCheck (prop)
28+import Test.QuickCheck (Property, choose)
29+import Test.QuickCheck.Monadic (monadicIO, pick)
30+
31+import Streamly.Prelude (SerialT, IsStream)
32+import Streamly.Internal.Data.Time.Units
33+ (AbsTime, NanoSecond64(..), toRelTime64, diffAbsTime64)
34+import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
35+
36+import qualified Streamly.Internal.Data.Fold as FL
37+import qualified Streamly.Internal.Data.Unfold as UF
38+import qualified Streamly.Internal.Data.Stream.IsStream as IS
2639import qualified Streamly.Prelude as S
2740
2841toListSerial :: SerialT IO a -> IO [a]
2942toListSerial = S.toList . S.serially
3043
31−main :: IO ()
32−main = hspec $ H.parallel $ do
44+max_length :: Int
45+max_length = 1000
46+
47+tenPow8 :: Int64
48+tenPow8 = 10^(8 :: Int)
49+
50+tenPow7 :: Int64
51+tenPow7 = 10^(7 :: Int)
52+
53+takeDropTime :: NanoSecond64
54+takeDropTime = NanoSecond64 $ 5 * tenPow8
55+
56+checkTakeDropTime :: (Maybe AbsTime, Maybe AbsTime) -> IO Bool
57+checkTakeDropTime (mt0, mt1) = do
58+ let graceTime = NanoSecond64 $ 8 * tenPow7
59+ case mt0 of
60+ Nothing -> return True
61+ Just t0 ->
62+ case mt1 of
63+ Nothing -> return True
64+ Just t1 -> do
65+ let tMax = toRelTime64 (takeDropTime + graceTime)
66+ let tMin = toRelTime64 (takeDropTime - graceTime)
67+ let t = diffAbsTime64 t1 t0
68+ let r = t >= tMin && t <= tMax
69+ when (not r) $ putStrLn $
70+ "t = " ++ show t ++
71+ " tMin = " ++ show tMin ++
72+ " tMax = " ++ show tMax
73+ return r
74+
75+testTakeByTime :: IO Bool
76+testTakeByTime = do
77+ r <-
78+ S.fold ((,) <$> FL.head <*> FL.last)
79+ $ IS.takeByTime takeDropTime
80+ $ S.repeatM (threadDelay 1000 >> getTime Monotonic)
81+ checkTakeDropTime r
82+
83+testDropByTime :: IO Bool
84+testDropByTime = do
85+ t0 <- getTime Monotonic
86+ mt1 <-
87+ S.head
88+ $ IS.dropByTime takeDropTime
89+ $ S.repeatM (threadDelay 1000 >> getTime Monotonic)
90+ checkTakeDropTime (Just t0, mt1)
91+
92+unfold :: Property
93+unfold = monadicIO $ do
94+ a <- pick $ choose (0, max_length `div` 2)
95+ b <- pick $ choose (0, max_length)
96+ let unf = UF.enumerateFromToIntegral b
97+ ls <- S.toList $ S.unfold unf a
98+ return $ ls == [a..b]
99+
100+unfold0 :: Property
101+unfold0 = monadicIO $ do
102+ a <- pick $ choose (0, max_length `div` 2)
103+ b <- pick $ choose (0, max_length)
104+ let unf = UF.supply (UF.enumerateFromToIntegral b) a
105+ ls <- S.toList $ IS.unfold0 unf
106+ return $ ls == [a..b]
107+
108+testFromCallback :: IO Int
109+testFromCallback = do
110+ ref <- newIORef Nothing
111+ let stream = S.map Just (IS.fromCallback (setCallback ref))
112+ `S.parallel` runCallback ref
113+ S.sum $ S.map fromJust $ S.takeWhile isJust stream
33114
34− describe "Miscellaneous combined examples" mixedOps
35− describe "Miscellaneous combined examples aheadly" mixedOpsAheadly
36− describe "Simple MonadError and MonadThrow" simpleMonadError
115+ where
37116
38− {-
39− describe "Composed MonadError serially" $ composeWithMonadError serially
40− describe "Composed MonadError wSerially" $ composeWithMonadError wSerially
41− describe "Composed MonadError asyncly" $ composeWithMonadError asyncly
42− describe "Composed MonadError wAsyncly" $ composeWithMonadError wAsyncly
43− -}
117+ setCallback ref cb = do
118+ writeIORef ref (Just cb)
119+
120+ runCallback ref = S.yieldM $ do
121+ cb <-
122+ S.repeatM (readIORef ref)
123+ & IS.delayPost 0.1
124+ & S.mapMaybe id
125+ & S.head
126+
127+ S.fromList [1..100]
128+ & IS.delayPost 0.001
129+ & S.mapM_ (fromJust cb)
130+ threadDelay 100000
131+ return Nothing
44132
45133-- XXX need to test that we have promptly cleaned up everything after the error
46134-- XXX We can also check the output that we are expected to get before the
@@ -138,3 +226,26 @@ mixedOpsAheadly =
138226 return (x11 + y11 + z11)
139227 return (x1 + y1 + z1)
140228 return (x + y + z)
229+
230+main :: IO ()
231+main = hspec $ H.parallel $ do
232+ describe "Filtering" $ do
233+ it "takeByTime" (testTakeByTime `shouldReturn` True)
234+ it "dropByTime" (testDropByTime `shouldReturn` True)
235+
236+ describe "From Generators" $ do
237+ prop "unfold" unfold
238+ prop "unfold0" unfold0
239+
240+ it "fromCallback" $ testFromCallback `shouldReturn` (50*101)
241+
242+ describe "Miscellaneous combined examples" mixedOps
243+ describe "Miscellaneous combined examples aheadly" mixedOpsAheadly
244+ describe "Simple MonadError and MonadThrow" simpleMonadError
245+
246+ {-
247+ describe "Composed MonadError serially" $ composeWithMonadError serially
248+ describe "Composed MonadError wSerially" $ composeWithMonadError wSerially
249+ describe "Composed MonadError asyncly" $ composeWithMonadError asyncly
250+ describe "Composed MonadError wAsyncly" $ composeWithMonadError wAsyncly
251+ -}