composewell/streamly
1
1
Fork 0
mirror of https://github.com/composewell/streamly.git synced 2024-10-05 23:37:43 +03:00
Code Issues Packages Projects Releases Wiki Activity

Change Scan Step to align it with Stream Step

Browse Source
This commit is contained in:
Harendra Kumar 2024-07-23 10:00:52 +05:30
parent 73272a235a
commit de38ff7bd1
5 changed files with 42 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
core/src/Streamly/Internal/Data/Fold/Combinators.hs
View File

@ -492,7 +492,7 @@ runScanWith isMany
step (sL, sR) x = do
rL <- stepL sL x
case rL of
Scan.Yield sL1 b -> do
Scan.Yield b sL1 -> do
rR <- stepR sR b
case rR of
Partial sR1 -> return $ Partial (sL1, sR1)

4
core/src/Streamly/Internal/Data/Fold/Type.hs
View File

@ -655,7 +655,7 @@ fromScan (Scan consume initial) =
fstep (FromScanInit ss) a = do
r <- consume ss a
return $ case r of
Scan.Yield s b -> Partial (FromScanGo s b)
Scan.Yield b s -> Partial (FromScanGo s b)
Scan.Skip s -> Partial (FromScanInit s)
-- XXX We have lost the input here.
-- XXX Need to change folds to always return Done on the next input
@ -663,7 +663,7 @@ fromScan (Scan consume initial) =
fstep (FromScanGo ss acc) a = do
r <- consume ss a
return $ case r of
Scan.Yield s b -> Partial (FromScanGo s b)
Scan.Yield b s -> Partial (FromScanGo s b)
Scan.Skip s -> Partial (FromScanGo s acc)
-- XXX We have lost the input here.
Scan.Stop -> Done (Just acc)

2
core/src/Streamly/Internal/Data/Pipe/Type.hs
View File

@ -701,6 +701,6 @@ fromScan (Scan step initial) = Pipe consume undefined initial
consume st a = do
r <- step st a
return $ case r of
Scan.Yield s b -> YieldC s b
Scan.Yield b s -> YieldC s b
Scan.Skip s -> SkipC s
Scan.Stop -> Stop

59
core/src/Streamly/Internal/Data/Scan.hs
View File

@ -71,7 +71,6 @@ where
#include "inline.hs"
import Control.Arrow (Arrow(..))
import Control.Category (Category(..))
import Data.Functor ((<&>))
import Data.Maybe (isJust, fromJust)
import Fusion.Plugin.Types (Fuse(..))
import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
@ -101,14 +100,14 @@ import Prelude hiding (filter, zipWith, map, mapM, id, unzip, null)
-- | The result of a scan step.
{-# ANN type Step Fuse #-}
data Step s b =
Yield s b -- ^ Yield output and keep consuming
Yield b s -- ^ Yield output and keep consuming
| Skip s -- ^ No output, keep consuming
| Stop -- ^ Stop consuming, last input is unsed
-- | 'fmap' maps a function on the step output.
instance Functor (Step s) where
{-# INLINE_NORMAL fmap #-}
fmap f (Yield s b) = Yield s (f b)
fmap f (Yield b s) = Yield (f b) s
fmap _ (Skip s) = Skip s
fmap _ Stop = Stop
@ -169,11 +168,11 @@ compose
step (sL, sR) x = do
rL <- stepL sL x
case rL of
Yield sL1 bL -> do
Yield bL sL1 -> do
rR <- stepR sR bL
return
$ case rR of
Yield sR1 br -> Yield (sL1, sR1) br
Yield br sR1 -> Yield br (sL1, sR1)
Skip sR1 -> Skip (sL1, sR1)
Stop -> Stop
Skip sL1 -> return $ Skip (sL1, sR)
@ -191,7 +190,7 @@ compose
--
{-# INLINE functionM #-}
functionM :: Monad m => (a -> m b) -> Scan m a b
functionM f = Scan (\() a -> f a <&> Yield ()) ()
functionM f = Scan (\() a -> fmap (\x -> Yield x ()) (f a)) ()
-- | A scan representing mapping of a pure function.
--
@ -251,19 +250,27 @@ teeWithMay f (Scan stepL initialL) (Scan stepR initialR) =
resR <- stepR sR a
return
$ case resL of
Yield sL1 bL ->
Yield bL sL1 ->
case resR of
Yield sR1 bR ->
Yield (TeeWith sL1 sR1) (f (Just bL) (Just bR))
Yield bR sR1 ->
Yield
(f (Just bL) (Just bR))
(TeeWith sL1 sR1)
Skip sR1 ->
Yield (TeeWith sL1 sR1) (f (Just bL) Nothing)
Yield
(f (Just bL) Nothing)
(TeeWith sL1 sR1)
Stop -> Stop
Skip sL1 ->
case resR of
Yield sR1 bR ->
Yield (TeeWith sL1 sR1) (f Nothing (Just bR))
Yield bR sR1 ->
Yield
(f Nothing (Just bR))
(TeeWith sL1 sR1)
Skip sR1 ->
Yield (TeeWith sL1 sR1) (f Nothing Nothing)
Yield
(f Nothing Nothing)
(TeeWith sL1 sR1)
Stop -> Stop
Stop -> Stop
@ -285,7 +292,7 @@ teeWith f s1 s2 =
-- | Zips the outputs only when both scans produce outputs, discards otherwise.
instance Monad m => Applicative (Scan m a) where
{-# INLINE pure #-}
pure b = Scan (\_ _ -> pure $ Yield () b) ()
pure b = Scan (\_ _ -> pure $ Yield b ()) ()
(<*>) = teeWith id
@ -311,19 +318,27 @@ unzipMay (Scan stepL initialL) (Scan stepR initialR) =
resR <- stepR sR b
return
$ case resL of
Yield sL1 bL ->
Yield bL sL1 ->
case resR of
Yield sR1 bR ->
Yield (Tuple' sL1 sR1) (Just bL, Just bR)
Yield bR sR1 ->
Yield
(Just bL, Just bR)
(Tuple' sL1 sR1)
Skip sR1 ->
Yield (Tuple' sL1 sR1) (Just bL, Nothing)
Yield
(Just bL, Nothing)
(Tuple' sL1 sR1)
Stop -> Stop
Skip sL1 ->
case resR of
Yield sR1 bR ->
Yield (Tuple' sL1 sR1) (Nothing, Just bR)
Yield bR sR1 ->
Yield
(Nothing, Just bR)
(Tuple' sL1 sR1)
Skip sR1 ->
Yield (Tuple' sL1 sR1) (Nothing, Nothing)
Yield
(Nothing, Nothing)
(Tuple' sL1 sR1)
Stop -> Stop
Stop -> Stop
@ -368,7 +383,7 @@ filterM f = Scan (\() a -> f a >>= g a) ()
g a b =
return
$ if b
then Yield () a
then Yield a ()
else Skip ()
-- | A filtering scan using a pure predicate.

2
core/src/Streamly/Internal/Data/Stream/Transform.hs
View File

@ -245,7 +245,7 @@ runScan (Scan consume initial) (Stream stream_step state) =
res <- consume cs x
return
$ case res of
Scan.Yield cs1 b -> Yield b (ScanConsume s cs1)
Scan.Yield b cs1 -> Yield b (ScanConsume s cs1)
Scan.Skip cs1 -> Skip (ScanConsume s cs1)
Scan.Stop -> Stop
Skip s -> return $ Skip (ScanConsume s cs)