@@ -73,6 +73,7 @@ module Streamly.Internal.Data.Stream.StreamD.Transform
73 , foldFilter
73 , foldFilter
74 , deleteBy
74 , deleteBy
77 -- * Trimming
78 -- * Trimming
78 -- | Produce a subset of the stream trimmed at ends.
79 -- | Produce a subset of the stream trimmed at ends.
@@ -135,6 +136,7 @@ import Data.Maybe (fromJust, isJust)
135import Fusion.Plugin.Types (Fuse(..))
136import Fusion.Plugin.Types (Fuse(..))
136import GHC.Types (SPEC(..))
137import GHC.Types (SPEC(..))
137import qualified Control.Monad.Catch as MC
138import qualified Control.Monad.Catch as MC
139+import qualified Data.Set as Set
139import Streamly.Internal.Control.Concurrent (MonadAsync)
141import Streamly.Internal.Control.Concurrent (MonadAsync)
140import Streamly.Internal.Control.ForkLifted (fork, forkManaged)
142import Streamly.Internal.Control.ForkLifted (fork, forkManaged)
@@ -796,6 +798,26 @@ uniq (Stream step state) = Stream step' (Nothing, state)
796 Skip s -> return $ Skip (Just x, s)
798 Skip s -> return $ Skip (Just x, s)
797 Stop -> return Stop
799 Stop -> return Stop
801+-- | The memory used is proportional to the number of unique elements in the
802+-- stream. If we want to limit the memory we can just use "take" to limit the
803+-- uniq elements in the stream.
804+{-# INLINE_NORMAL nub #-}
805+nub :: (Monad m, Ord a) => Stream m a -> Stream m a
806+nub (Stream step1 state1) = Stream step (Set.empty, state1)
810+ step gst (set, st) = do
817+ else Yield x (Set.insert x set, s)
818+ Skip s -> Skip (set, s)
799{-# INLINE_NORMAL deleteBy #-}
821{-# INLINE_NORMAL deleteBy #-}
800deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
822deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
801deleteBy eq x (Stream step state) = Stream step' (state, False)
823deleteBy eq x (Stream step state) = Stream step' (state, False)