git.iko.soy/streamly
on master help sign in
Make fromIndicesM concurrent, and add tests for fromIndicesM.
64524bfPranay Sashank
src/Streamly/Prelude.hs+5−3
compare with master
unifiedsplit
@@ -682,7 +682,6 @@ yieldM = K.yieldM
682fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a
682fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a
683fromIndices = fromStreamD . D.fromIndices
683fromIndices = fromStreamD . D.fromIndices
684
684
685−-- XXX this needs to be concurrent
686--
685--
687-- |
686-- |
688-- @
687-- @
@@ -694,8 +693,11 @@ fromIndices = fromStreamD . D.fromIndices
694--
693--
695-- @since 0.6.0
694-- @since 0.6.0
696{-# INLINE fromIndicesM #-}
695{-# INLINE fromIndicesM #-}
697−fromIndicesM :: (IsStream t, Monad m) => (Int -> m a) -> t m a
696+fromIndicesM :: (IsStream t, MonadAsync m) => (Int -> m a) -> t m a
698−fromIndicesM = fromStreamD . D.fromIndicesM
697+fromIndicesM gen = go 0
698+ where
699+ go i = K.mkStream $ \st stp sng yld -> do
700+ K.foldStreamShared st stp sng yld (gen i |: go (i + 1))
699
701
700-- |
702-- |
701-- @
703-- @
test/Prop.hs+27
@@ -173,6 +173,29 @@ constructWithIterateM op len =
173 streamEffect <- run $ readIORef mvl
173 streamEffect <- run $ readIORef mvl
174 listEquals (==) streamEffect list
174 listEquals (==) streamEffect list
175
175
176+constructWithFromIndices ::
177+ IsStream t => (t IO Int -> SerialT IO Int) -> Word8 -> Property
178+constructWithFromIndices op len =
179+ withMaxSuccess maxTestCount $
180+ monadicIO $ do
181+ stream <-
182+ run $ (S.toList . op . S.take (fromIntegral len)) (S.fromIndices id)
183+ let list = take (fromIntegral len) (iterate (+ 1) 0)
184+ listEquals (==) stream list
185+
186+constructWithFromIndicesM ::
187+ IsStream t => (t IO Int -> SerialT IO Int) -> Word8 -> Property
188+constructWithFromIndicesM op len =
189+ withMaxSuccess maxTestCount $
190+ monadicIO $ do
191+ mvl <- run (newIORef [] :: IO (IORef [Int]))
192+ let addIndex mv i = modifyIORef' mv (++ [i]) >> return i
193+ list = take (fromIntegral len) (iterate (+ 1) 0)
194+ run $
195+ S.drain . op $
196+ S.take (fromIntegral len) $ S.fromIndicesM (addIndex mvl)
197+ streamEffect <- run $ readIORef mvl
198+ listEquals (==) streamEffect list
176
199
177-------------------------------------------------------------------------------
200-------------------------------------------------------------------------------
178-- Concurrent generation
201-- Concurrent generation
@@ -1017,6 +1040,10 @@ main = hspec
1017 -- parallelOps $ prop "parallely iterateM" parallely . onstructWithIterate
1040 -- parallelOps $ prop "parallely iterateM" parallely . onstructWithIterate
1018 -- XXX add tests for fromIndices
1041 -- XXX add tests for fromIndices
1019
1042
1043+ serialOps $ prop "serially fromIndices" . constructWithFromIndices
1044+
1045+ serialOps $ prop "serially fromIndicesM" . constructWithFromIndicesM
1046+
1020 describe "Functor operations" $ do
1047 describe "Functor operations" $ do
1021 serialOps $ functorOps S.fromFoldable "serially" (==)
1048 serialOps $ functorOps S.fromFoldable "serially" (==)
1022 serialOps $ functorOps folded "serially folded" (==)
1049 serialOps $ functorOps folded "serially folded" (==)