cff0369 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Refactor ops from Prelude to Prelude.Serial.

cff0369pranaysashank6 years ago

test/Streamly/Test/Prelude.hs+1−114

compare with master

unifiedsplit

@@ -11,127 +11,24 @@ module Streamly.Test.Prelude (main) where

12import Control.Concurrent (myThreadId, threadDelay)

13import Control.Exception (Exception, try)

14−import Control.Monad (when)

15import Control.Monad.Catch (throwM)

14import Control.Monad.Catch (throwM)

16import Control.Monad.Error.Class (throwError, MonadError)

15import Control.Monad.Error.Class (throwError, MonadError)

17import Control.Monad.Trans.Except (runExceptT, ExceptT)

16import Control.Monad.Trans.Except (runExceptT, ExceptT)

18−import Data.Function ((&))

19−import Data.IORef (newIORef, readIORef, writeIORef)

20−import Data.Int (Int64)

21import Data.List (sort)

17import Data.List (sort)

22−import Data.Maybe (fromJust, isJust)

23#if !(MIN_VERSION_base(4,11,0))

18#if !(MIN_VERSION_base(4,11,0))

24import Data.Semigroup (Semigroup(..))

19import Data.Semigroup (Semigroup(..))

25#endif

20#endif

26import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

21import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

27import System.Random (randomIO)

22import System.Random (randomIO)

28import Test.Hspec as H

23import Test.Hspec as H

29−import Test.Hspec.QuickCheck (prop)

30−import Test.QuickCheck (Property, choose)

31−import Test.QuickCheck.Monadic (monadicIO, pick)

33import Streamly.Prelude (SerialT, IsStream)

25import Streamly.Prelude (SerialT, IsStream)

34−import Streamly.Internal.Data.Time.Units

35− (AbsTime, NanoSecond64(..), toRelTime64, diffAbsTime64)

36−import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

38−import qualified Streamly.Internal.Data.Fold as FL

39−import qualified Streamly.Internal.Data.Unfold as UF

40−import qualified Streamly.Internal.Data.Stream.IsStream as IS

41import qualified Streamly.Prelude as S

27import qualified Streamly.Prelude as S

43toListSerial :: SerialT IO a -> IO [a]

29toListSerial :: SerialT IO a -> IO [a]

44toListSerial = S.toList . S.serially

30toListSerial = S.toList . S.serially

46−max_length :: Int

47−max_length = 1000

48−

49−tenPow8 :: Int64

50−tenPow8 = 10^(8 :: Int)

51−

52−tenPow7 :: Int64

53−tenPow7 = 10^(7 :: Int)

54−

55−takeDropTime :: NanoSecond64

56−takeDropTime = NanoSecond64 $ 5 * tenPow8

57−

58−checkTakeDropTime :: (Maybe AbsTime, Maybe AbsTime) -> IO Bool

59−checkTakeDropTime (mt0, mt1) = do

60− let graceTime = NanoSecond64 $ 8 * tenPow7

61− case mt0 of

62− Nothing -> return True

63− Just t0 ->

64− case mt1 of

65− Nothing -> return True

66− Just t1 -> do

67− let tMax = toRelTime64 (takeDropTime + graceTime)

68− let tMin = toRelTime64 (takeDropTime - graceTime)

69− let t = diffAbsTime64 t1 t0

70− let r = t >= tMin && t <= tMax

71− when (not r) $ putStrLn $

72− "t = " ++ show t ++

73− " tMin = " ++ show tMin ++

74− " tMax = " ++ show tMax

75− return r

76−

77−testTakeByTime :: IO Bool

78−testTakeByTime = do

79− r <-

80− S.fold ((,) <$> FL.head <*> FL.last)

81− $ IS.takeByTime takeDropTime

82− $ S.repeatM (threadDelay 1000 >> getTime Monotonic)

83− checkTakeDropTime r

84−

85−testDropByTime :: IO Bool

86−testDropByTime = do

87− t0 <- getTime Monotonic

88− mt1 <-

89− S.head

90− $ IS.dropByTime takeDropTime

91− $ S.repeatM (threadDelay 1000 >> getTime Monotonic)

92− checkTakeDropTime (Just t0, mt1)

93−

94−unfold :: Property

95−unfold = monadicIO $ do

96− a <- pick $ choose (0, max_length `div` 2)

97− b <- pick $ choose (0, max_length)

98− let unf = UF.enumerateFromToIntegral b

99− ls <- S.toList $ S.unfold unf a

100− return $ ls == [a..b]

101−

102−unfold0 :: Property

103−unfold0 = monadicIO $ do

104− a <- pick $ choose (0, max_length `div` 2)

105− b <- pick $ choose (0, max_length)

106− let unf = UF.supply (UF.enumerateFromToIntegral b) a

107− ls <- S.toList $ IS.unfold0 unf

108− return $ ls == [a..b]

109−

110−testFromCallback :: IO Int

111−testFromCallback = do

112− ref <- newIORef Nothing

113− let stream = S.map Just (IS.fromCallback (setCallback ref))

114− `S.parallel` runCallback ref

115− S.sum $ S.map fromJust $ S.takeWhile isJust stream

116−

117− where

118−

119− setCallback ref cb = do

120− writeIORef ref (Just cb)

121−

122− runCallback ref = S.yieldM $ do

123− cb <-

124− S.repeatM (readIORef ref)

125− & IS.delayPost 0.1

126− & S.mapMaybe id

127− & S.head

128−

129− S.fromList [1..100]

130− & IS.delayPost 0.001

131− & S.mapM_ (fromJust cb)

132− threadDelay 100000

133− return Nothing

134−

135-- XXX need to test that we have promptly cleaned up everything after the error

32-- XXX need to test that we have promptly cleaned up everything after the error

136-- XXX We can also check the output that we are expected to get before the

33-- XXX We can also check the output that we are expected to get before the

137-- error occurs.

34-- error occurs.

@@ -275,16 +172,6 @@ parallelLoops = do

275

172

276main :: IO ()

173main :: IO ()

277main = hspec $ H.parallel $ do

174main = hspec $ H.parallel $ do

278− describe "Filtering" $ do

279− it "takeByTime" (testTakeByTime `shouldReturn` True)

280− it "dropByTime" (testDropByTime `shouldReturn` True)

281−

282− describe "From Generators" $ do

283− prop "unfold" unfold

284− prop "unfold0" unfold0

285−

286− it "fromCallback" $ testFromCallback `shouldReturn` (50*101)

287−

288 describe "Miscellaneous combined examples" mixedOps

175 describe "Miscellaneous combined examples" mixedOps

289 describe "Miscellaneous combined examples aheadly" mixedOpsAheadly

176 describe "Miscellaneous combined examples aheadly" mixedOpsAheadly

290 describe "Simple MonadError and MonadThrow" simpleMonadError

177 describe "Simple MonadError and MonadThrow" simpleMonadError

@@ -296,4 +183,4 @@ main = hspec $ H.parallel $ do

296 describe "Composed MonadError wSerially" $ composeWithMonadError wSerially

183 describe "Composed MonadError wSerially" $ composeWithMonadError wSerially

297 describe "Composed MonadError asyncly" $ composeWithMonadError asyncly

184 describe "Composed MonadError asyncly" $ composeWithMonadError asyncly

298 describe "Composed MonadError wAsyncly" $ composeWithMonadError wAsyncly

185 describe "Composed MonadError wAsyncly" $ composeWithMonadError wAsyncly

299− -}

186+ -}

test/Streamly/Test/Prelude/Serial.hs+111−1

@@ -9,7 +9,13 @@

10module Streamly.Test.Prelude.Serial where

12+import Control.Concurrent ( threadDelay )

13+import Control.Monad ( when )

14+import Data.Function ( (&) )

15+import Data.IORef ( newIORef, readIORef, writeIORef )

16+import Data.Int (Int64)

12import Data.List (intercalate)

17import Data.List (intercalate)

18+import Data.Maybe ( isJust, fromJust )

13#if __GLASGOW_HASKELL__ < 808

19#if __GLASGOW_HASKELL__ < 808

14import Data.Semigroup ((<>))

20import Data.Semigroup ((<>))

15#endif

21#endif

@@ -26,13 +32,18 @@ import Test.QuickCheck

26 , vectorOf

32 , vectorOf

27 , withMaxSuccess

33 , withMaxSuccess

28 )

34 )

29−import Test.QuickCheck.Monadic (assert, monadicIO, run)

35+import Test.QuickCheck.Monadic (assert, monadicIO, pick, run)

30import Test.Hspec as H

36import Test.Hspec as H

32import Streamly.Prelude

38import Streamly.Prelude

33 ( SerialT, IsStream, avgRate, maxBuffer, serial, serially)

39 ( SerialT, IsStream, avgRate, maxBuffer, serial, serially)

34import qualified Streamly.Prelude as S

40import qualified Streamly.Prelude as S

35import qualified Streamly.Data.Fold as FL

41import qualified Streamly.Data.Fold as FL

42+import qualified Streamly.Internal.Data.Unfold as UF

43+import qualified Streamly.Internal.Data.Stream.IsStream as IS

44+import Streamly.Internal.Data.Time.Units

45+ (AbsTime, NanoSecond64(..), toRelTime64, diffAbsTime64)

46+import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

37import Streamly.Test.Common

48import Streamly.Test.Common

38import Streamly.Test.Prelude.Common

49import Streamly.Test.Prelude.Common

@@ -163,6 +174,95 @@ associativityCheck desc t = prop desc assocCheckProp

163 assocStream <- run $ S.toList $ t $ xStream <> yStream <> zStream

174 assocStream <- run $ S.toList $ t $ xStream <> yStream <> zStream

164 listEquals (==) infixAssocstream assocStream

175 listEquals (==) infixAssocstream assocStream

165

176

177+max_length :: Int

178+max_length = 1000

179+

180+tenPow8 :: Int64

181+tenPow8 = 10^(8 :: Int)

182+

183+tenPow7 :: Int64

184+tenPow7 = 10^(7 :: Int)

185+

186+takeDropTime :: NanoSecond64

187+takeDropTime = NanoSecond64 $ 5 * tenPow8

188+

189+checkTakeDropTime :: (Maybe AbsTime, Maybe AbsTime) -> IO Bool

190+checkTakeDropTime (mt0, mt1) = do

191+ let graceTime = NanoSecond64 $ 8 * tenPow7

192+ case mt0 of

193+ Nothing -> return True

194+ Just t0 ->

195+ case mt1 of

196+ Nothing -> return True

197+ Just t1 -> do

198+ let tMax = toRelTime64 (takeDropTime + graceTime)

199+ let tMin = toRelTime64 (takeDropTime - graceTime)

200+ let t = diffAbsTime64 t1 t0

201+ let r = t >= tMin && t <= tMax

202+ when (not r) $ putStrLn $

203+ "t = " ++ show t ++

204+ " tMin = " ++ show tMin ++

205+ " tMax = " ++ show tMax

206+ return r

207+

208+testTakeByTime :: IO Bool

209+testTakeByTime = do

210+ r <-

211+ S.fold ((,) <$> FL.head <*> FL.last)

212+ $ IS.takeByTime takeDropTime

213+ $ S.repeatM (threadDelay 1000 >> getTime Monotonic)

214+ checkTakeDropTime r

215+

216+testDropByTime :: IO Bool

217+testDropByTime = do

218+ t0 <- getTime Monotonic

219+ mt1 <-

220+ S.head

221+ $ IS.dropByTime takeDropTime

222+ $ S.repeatM (threadDelay 1000 >> getTime Monotonic)

223+ checkTakeDropTime (Just t0, mt1)

224+

225+unfold :: Property

226+unfold = monadicIO $ do

227+ a <- pick $ choose (0, max_length `div` 2)

228+ b <- pick $ choose (0, max_length)

229+ let unf = UF.enumerateFromToIntegral b

230+ ls <- S.toList $ S.unfold unf a

231+ return $ ls == [a..b]

232+

233+unfold0 :: Property

234+unfold0 = monadicIO $ do

235+ a <- pick $ choose (0, max_length `div` 2)

236+ b <- pick $ choose (0, max_length)

237+ let unf = UF.supply (UF.enumerateFromToIntegral b) a

238+ ls <- S.toList $ IS.unfold0 unf

239+ return $ ls == [a..b]

240+

241+testFromCallback :: IO Int

242+testFromCallback = do

243+ ref <- newIORef Nothing

244+ let stream = S.map Just (IS.fromCallback (setCallback ref))

245+ `S.parallel` runCallback ref

246+ S.sum $ S.map fromJust $ S.takeWhile isJust stream

247+

248+ where

249+

250+ setCallback ref cb = do

251+ writeIORef ref (Just cb)

252+

253+ runCallback ref = S.yieldM $ do

254+ cb <-

255+ S.repeatM (readIORef ref)

256+ & IS.delayPost 0.1

257+ & S.mapMaybe id

258+ & S.head

259+

260+ S.fromList [1..100]

261+ & IS.delayPost 0.001

262+ & S.mapM_ (fromJust cb)

263+ threadDelay 100000

264+ return Nothing

265+

166main :: IO ()

266main :: IO ()

167main = hspec

267main = hspec

168 $ H.parallel

268 $ H.parallel

@@ -209,6 +309,10 @@ main = hspec

209 serialOps $ prop "serially fromIndices" . constructWithFromIndices

309 serialOps $ prop "serially fromIndices" . constructWithFromIndices

210 serialOps $ prop "serially fromIndicesM" . constructWithFromIndicesM

310 serialOps $ prop "serially fromIndicesM" . constructWithFromIndicesM

211

311

312+ describe "From Generators" $ do

313+ prop "unfold" unfold

314+ prop "unfold0" unfold0

315+

212 describe "Functor operations" $ do

316 describe "Functor operations" $ do

213 serialOps $ functorOps S.fromFoldable "serially" (==)

317 serialOps $ functorOps S.fromFoldable "serially" (==)

214 serialOps $ functorOps folded "serially folded" (==)

318 serialOps $ functorOps folded "serially folded" (==)

@@ -269,6 +373,10 @@ main = hspec

269 serialOps $ transformCombineOpsOrdered S.fromFoldable "serially" (==)

373 serialOps $ transformCombineOpsOrdered S.fromFoldable "serially" (==)

270 serialOps $ transformCombineOpsOrdered folded "serially" (==)

374 serialOps $ transformCombineOpsOrdered folded "serially" (==)

271

375

376+ describe "Filtering" $ do

377+ it "takeByTime" (testTakeByTime `shouldReturn` True)

378+ it "dropByTime" (testDropByTime `shouldReturn` True)

379+

272 describe "Stream group and split operations" $ do

380 describe "Stream group and split operations" $ do

273 groupSplitOps "serially"

381 groupSplitOps "serially"

274

382

@@ -289,3 +397,5 @@ main = hspec

289 prop "testGroupsBySep" testGroupsBySep

397 prop "testGroupsBySep" testGroupsBySep

290

398

291 describe "Composed MonadThrow serially" $ composeWithMonadThrow serially

399 describe "Composed MonadThrow serially" $ composeWithMonadThrow serially

400+

401+ it "fromCallback" $ testFromCallback `shouldReturn` (50*101)