@@ -68,7 +68,30 @@ import qualified Streamly.Streams.StreamK as K
6868runOne
6969 :: MonadIO m
7070 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
71−runOne st m0 winfo = go m0
71+runOne st m0 winfo =
72+ case getYieldLimit st of
73+ Nothing -> go m0
74+ Just _ -> runOneLimited st m0 winfo
75+
76+ where
77+
78+ go m = do
79+ liftIO $ decrementBufferLimit sv
80+ foldStreamShared st yieldk single stop m
81+
82+ sv = fromJust $ streamVar st
83+
84+ stop = liftIO $ do
85+ incrementBufferLimit sv
86+ sendStop sv winfo
87+ sendit a = liftIO $ void $ send sv (ChildYield a)
88+ single a = sendit a >> (liftIO $ sendStop sv winfo)
89+ yieldk a r = sendit a >> go r
90+
91+runOneLimited
92+ :: MonadIO m
93+ => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
94+runOneLimited st m0 winfo = go m0
7295
7396 where
7497