@@ -14,7 +14,9 @@ import Control.Monad.IO.Class (MonadIO(liftIO))
1414import Control.Monad.State (MonadState, get, modify, runStateT, StateT)
1515import Control.Monad.Trans.Except (runExceptT, ExceptT)
1616import Data.Foldable (forM_, fold)
17+import Data.Function ((&))
1718import Data.List (sort)
19+import Data.Maybe (fromJust, isJust)
1820import System.Mem (performMajorGC)
1921
2022import Data.IORef
@@ -147,6 +149,7 @@ main = hspec $ do
147149
148150 describe "Parallel (<>) time order check" $ parallelCheck parallely (<>)
149151 describe "Parallel mappend time order check" $ parallelCheck parallely mappend
152+ it "fromCallback" $ testFromCallback `shouldReturn` (50*101)
150153
151154checkCleanup :: IsStream t
152155 => Int
@@ -1150,3 +1153,28 @@ mixedOpsAheadly =
11501153 return (x11 + y11 + z11)
11511154 return (x1 + y1 + z1)
11521155 return (x + y + z)
1156+
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
1163+
1164+ where
1165+
1166+ setCallback ref cb = do
1167+ writeIORef ref (Just cb)
1168+
1169+ runCallback ref = S.yieldM $ do
1170+ cb <-
1171+ S.repeatM (readIORef ref)
1172+ & IP.delayPost 0.1
1173+ & S.mapMaybe id
1174+ & S.head
1175+
1176+ S.fromList [1..100]
1177+ & IP.delayPost 0.001
1178+ & S.mapM_ (fromJust cb)
1179+ threadDelay 100000
1180+ return Nothing