@@ -35,6 +35,7 @@ module Streamly.Internal.Data.Fold.Window
3535 --
3636 lmap
3737 , cumulative
38+ , nub
3839
3940 -- ** Sums
4041 , length
@@ -59,11 +60,13 @@ import Streamly.Internal.Data.Fold.Type (Fold(..), Step(..))
5960import Streamly.Internal.Data.Tuple.Strict
6061 (Tuple'(..), Tuple3Fused' (Tuple3Fused'))
6162
62−import Prelude hiding (length, sum, minimum, maximum)
63−
63+import qualified Data.Map as Map
6464import qualified Streamly.Internal.Data.Fold.Type as Fold
6565import qualified Streamly.Internal.Data.Ring.Foreign as Ring
6666
67+import Prelude hiding (length, sum, minimum, maximum)
68+
69+
6770-- $setup
6871-- >>> import Data.Bifunctor(bimap)
6972-- >>> import qualified Streamly.Data.Fold as Fold
@@ -309,3 +312,39 @@ maximum n = fmap (fmap snd) $ range n
309312{-# INLINE mean #-}
310313mean :: forall m a. (Monad m, Fractional a) => Fold m (a, Maybe a) a
311314mean = Fold.teeWith (/) sum length
315+
316+-- | Drop repeated elements within the specified
317+-- tumbling window in the stream.
318+-- /Pre-release/
319+--
320+-- >>> import qualified Streamly.Prelude as S
321+-- >>> S.fold nub $ S.fromList [(1, Nothing), (1, Nothing), (3, Nothing), (4, Just 1)]
322+-- fromList [(1,1),(3,1),(4,1)]
323+--
324+{-# INLINE nub #-}
325+nub :: (Monad m, Ord a) => Fold m (a, Maybe a) (Map.Map a Int)
326+nub = Fold step initial extract
327+
328+ where
329+
330+ initial =
331+ return $ Partial Map.empty
332+
333+ updateRefCounts new map0 = return $ Map.insertWith (+) new (1::Int) map0
334+
335+ alter a m1 (Just v) =
336+ if v == 1
337+ then do
338+ _ <- return $ Map.delete a m1
339+ return Nothing
340+ else return $ Just (v -1)
341+ alter _ _ Nothing = error "Invalid old element"
342+
343+ step map0 (new, mOld) = do
344+ m1 <- updateRefCounts new map0
345+ Partial <$>
346+ case mOld of
347+ Just a -> Map.alterF (alter a m1) a m1
348+ Nothing -> return m1
349+
350+ extract = return