@@ -174,6 +174,13 @@ module Streamly.Internal.Data.Parser.ParserD
174174 -- , retryMax -- try N times
175175 -- , retryUntil -- try until successful
176176 -- , retryUntilN -- try until successful n times
177+
178+ -- ** Zipping Input
179+ , zipWithM
180+ , zip
181+ , indexed
182+ , makeIndexFilter
183+ , sampleFromthen
177184 )
178185where
179186
@@ -187,9 +194,11 @@ import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
187194
188195import qualified Streamly.Internal.Data.Fold.Type as FL
189196import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
197+import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D
190198
191199import Prelude hiding
192− (any, all, take, takeWhile, sequence, concatMap, maybe, either, span)
200+ (any, all, take, takeWhile, sequence, concatMap, maybe, either, span
201+ , zip, filter)
193202import Streamly.Internal.Data.Parser.ParserD.Tee
194203import Streamly.Internal.Data.Parser.ParserD.Type
195204
@@ -975,6 +984,102 @@ matchBy cmp (D.Stream sstep state) = Parser step initial extract
975984
976985 extract _ = throwM $ ParseError "match: end of input"
977986
987+{-# INLINE zipWithM #-}
988+zipWithM :: MonadThrow m =>
989+ (a -> b -> m c) -> D.Stream m a -> Fold m c x -> Parser m b x
990+zipWithM zf (D.Stream sstep state) (Fold fstep finitial fextract) =
991+ Parser step initial extract
992+
993+ where
994+
995+ initial = do
996+ fres <- finitial
997+ case fres of
998+ FL.Partial fs -> do
999+ r <- sstep defState state
1000+ case r of
1001+ D.Yield x s -> return $ IPartial (Just x, s, fs)
1002+ D.Stop -> do
1003+ x <- fextract fs
1004+ return $ IDone x
1005+ -- Need Skip/Continue in initial to loop right here
1006+ D.Skip s -> return $ IPartial (Nothing, s, fs)
1007+ FL.Done x -> return $ IDone x
1008+
1009+ step (Just a, st, fs) b = do
1010+ c <- zf a b
1011+ fres <- fstep fs c
1012+ case fres of
1013+ FL.Partial fs1 -> do
1014+ r <- sstep defState st
1015+ case r of
1016+ D.Yield x1 s -> return $ Continue 0 (Just x1, s, fs1)
1017+ D.Stop -> do
1018+ x <- fextract fs1
1019+ return $ Done 0 x
1020+ D.Skip s -> return $ Continue 1 (Nothing, s, fs1)
1021+ FL.Done x -> return $ Done 0 x
1022+ step (Nothing, st, fs) b = do
1023+ r <- sstep defState st
1024+ case r of
1025+ D.Yield a s -> do
1026+ c <- zf a b
1027+ fres <- fstep fs c
1028+ case fres of
1029+ FL.Partial fs1 -> return $ Continue 0 (Nothing, s, fs1)
1030+ FL.Done x -> return $ Done 0 x
1031+ D.Stop -> do
1032+ x <- fextract fs
1033+ return $ Done 1 x
1034+ D.Skip s -> return $ Continue 1 (Nothing, s, fs)
1035+
1036+ extract _ = throwM $ ParseError "zipWithM: end of input"
1037+
1038+-- | Zip the input of a fold with a stream.
1039+--
1040+-- /Pre-release/
1041+--
1042+{-# INLINE zip #-}
1043+zip :: MonadThrow m => D.Stream m a -> Fold m (a, b) x -> Parser m b x
1044+zip = zipWithM (curry return)
1045+
1046+-- | Pair each element of a fold input with its index, starting from index 0.
1047+--
1048+-- /Pre-release/
1049+{-# INLINE indexed #-}
1050+indexed :: forall m a b. MonadThrow m => Fold m (Int, a) b -> Parser m a b
1051+indexed = zip (D.enumerateFromIntegral 0 :: D.Stream m Int)
1052+
1053+-- | @makeIndexFilter indexer filter predicate@ generates a fold filtering
1054+-- function using a fold indexing function that attaches an index to each input
1055+-- element and a filtering function that filters using @(index, element) ->
1056+-- Bool) as predicate.
1057+--
1058+-- For example:
1059+--
1060+-- @
1061+-- filterWithIndex = makeIndexFilter indexed filter
1062+-- filterWithAbsTime = makeIndexFilter timestamped filter
1063+-- filterWithRelTime = makeIndexFilter timeIndexed filter
1064+-- @
1065+--
1066+-- /Pre-release/
1067+{-# INLINE makeIndexFilter #-}
1068+makeIndexFilter ::
1069+ (Fold m (s, a) b -> Parser m a b)
1070+ -> (((s, a) -> Bool) -> Fold m (s, a) b -> Fold m (s, a) b)
1071+ -> (((s, a) -> Bool) -> Fold m a b -> Parser m a b)
1072+makeIndexFilter f comb g = f . comb g . FL.lmap snd
1073+
1074+-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
1075+-- then every element at strides of @stride@.
1076+--
1077+-- /Pre-release/
1078+{-# INLINE sampleFromthen #-}
1079+sampleFromthen :: MonadThrow m => Int -> Int -> Fold m a b -> Parser m a b
1080+sampleFromthen offset size =
1081+ makeIndexFilter indexed FL.filter (\(i, _) -> (i + offset) `mod` size == 0)
1082+
9781083--------------------------------------------------------------------------------
9791084--- Spanning
9801085--------------------------------------------------------------------------------