@@ -14,7 +14,9 @@ import Control.Monad.IO.Class (MonadIO(liftIO))
14import Control.Monad.State (MonadState, get, modify, runStateT, StateT)
14import Control.Monad.State (MonadState, get, modify, runStateT, StateT)
15import Control.Monad.Trans.Except (runExceptT, ExceptT)
15import Control.Monad.Trans.Except (runExceptT, ExceptT)
16import Data.Foldable (forM_, fold)
16import Data.Foldable (forM_, fold)
17+import Data.Function ((&))
17import Data.List (sort)
18import Data.List (sort)
19+import Data.Maybe (fromJust, isJust)
18import System.Mem (performMajorGC)
20import System.Mem (performMajorGC)
20import Data.IORef
22import Data.IORef
@@ -147,6 +149,7 @@ main = hspec $ do
148 describe "Parallel (<>) time order check" $ parallelCheck parallely (<>)
150 describe "Parallel (<>) time order check" $ parallelCheck parallely (<>)
149 describe "Parallel mappend time order check" $ parallelCheck parallely mappend
151 describe "Parallel mappend time order check" $ parallelCheck parallely mappend
152+ it "fromCallback" $ testFromCallback `shouldReturn` (50*101)
151checkCleanup :: IsStream t
154checkCleanup :: IsStream t
@@ -1150,3 +1153,28 @@ mixedOpsAheadly =
1150 return (x11 + y11 + z11)
1153 return (x11 + y11 + z11)
1151 return (x1 + y1 + z1)
1154 return (x1 + y1 + z1)
1152 return (x + y + z)
1155 return (x + y + z)
1157+testFromCallback :: IO Int
1158+testFromCallback = do
1159+ ref <- newIORef Nothing
1160+ let stream = S.map Just (IP.fromCallback (setCallback ref))
1161+ `Streamly.parallel` runCallback ref
1162+ S.sum $ S.map fromJust $ S.takeWhile isJust stream
1166+ setCallback ref cb = do
1167+ writeIORef ref (Just cb)
1169+ runCallback ref = S.yieldM $ do
1171+ S.repeatM (readIORef ref)
1176+ S.fromList [1..100]
1177+ & IP.delayPost 0.001
1178+ & S.mapM_ (fromJust cb)