@@ -1,245 +0,0 @@
2−-- Module : Streamly.Internal.Data.Sink
3−-- Copyright : (c) 2019 Composewell Technologies
5−-- Maintainer : streamly@composewell.com
6−-- Stability : experimental
10−-- The 'Sink' type is a just a special case of 'Fold' and we can do without
11−-- it. However, in some cases 'Sink' is a simpler type and may provide better
12−-- performance than 'Fold' because it does not maintain any state. Folds can
13−-- be used for both pure and monadic computations. Sinks are not applicable to
16−module Streamly.Internal.Data.Sink
44− -- * Input Transformation
45− -- | These are contravariant operations i.e. they apply on the input of the
46− -- 'Sink', for this reason they are prefixed with 'l' for 'left'.
60−import Control.Monad ((>=>), when, void)
61−import Data.Map.Strict (Map)
63− hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
64− foldl, map, mapM, sequence, all, any, sum, product, elem,
65− notElem, maximum, minimum, head, last, tail, length, null,
66− reverse, iterate, init, and, or, lookup, foldr1, (!!),
67− scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip)
69−import Streamly.Internal.Data.Fold.Type (Fold(..), Step(..))
70−import Streamly.Internal.Data.Sink.Type (Sink(..))
72−import qualified Data.Map.Strict as Map
74−------------------------------------------------------------------------------
76−------------------------------------------------------------------------------
78−-- | Convert a 'Sink' to a 'Fold'. When you want to compose sinks and folds
79−-- together, upgrade a sink to a fold before composing.
80−toFold :: Monad m => Sink m a -> Fold m a ()
81−toFold (Sink f) = Fold step begin done
83− begin = return $ Partial ()
84− step _ a = Partial <$> f a
87−------------------------------------------------------------------------------
88−-- Composing with sinks
89−------------------------------------------------------------------------------
91−-- | Distribute one copy each of the input to both the sinks.
99−-- > let pr x = Sink.drainM (putStrLn . ((x ++ " ") ++) . show)
100−-- > sink (Sink.tee (pr \"L") (pr \"R")) (S.enumerateFromTo 1 2)
107−tee :: Monad m => Sink m a -> Sink m a -> Sink m a
108−tee (Sink fL) (Sink fR) = Sink (\a -> fL a >> fR a)
110−-- | Distribute copies of the input to all the sinks in a container.
120−-- > let pr x = Sink.drainM (putStrLn . ((x ++ " ") ++) . show)
121−-- > sink (Sink.distribute [(pr \"L"), (pr \"R")]) (S.enumerateFromTo 1 2)
128−-- This is the consumer side dual of the producer side 'sequence_' operation.
129−{-# INLINE distribute #-}
130−distribute :: Monad m => [Sink m a] -> Sink m a
131−distribute ss = Sink (\a -> Prelude.mapM_ (\(Sink f) -> f a) ss)
133−-- | Demultiplex to multiple consumers without collecting the results. Useful
134−-- to run different effectful computations depending on the value of the stream
135−-- elements, for example handling network packets of different types using
136−-- different handlers.
141−-- -----stream m a-----Map-----|
148−-- > let pr x = Sink.drainM (putStrLn . ((x ++ " ") ++) . show)
149−-- > let table = Data.Map.fromList [(1, pr \"One"), (2, pr \"Two")]
150−-- in Sink.sink (Sink.demux id table) (S.enumerateFromTo 1 100)
155−demux :: (Monad m, Ord k) => (a -> k) -> Map k (Sink m a) -> Sink m a
156−demux f kv = Sink step
161− -- XXX should we raise an exception in Nothing case?
162− -- Ideally we should enforce that it is a total map over k so that look
164− case Map.lookup (f a) kv of
165− Nothing -> return ()
166− Just (Sink g) -> g a
169−demux :: (Monad m, Ord k) => Map k (Sink m a) -> Sink m (a, k)
175− -- XXX should we raise an exception in Nothing case?
176− -- Ideally we should enforce that it is a total map over k so that look
178− case Map.lookup k kv of
179− Nothing -> return ()
180− Just (Sink g) -> g a
182−-- | Split elements in the input stream into two parts using a monadic unzip
183−-- function, direct each part to a different sink.
188−-- -----Stream m a----(b,c)--|
192−-- > let pr x = Sink.drainM (putStrLn . ((x ++ " ") ++) . show)
193−-- in Sink.sink (Sink.unzip return (pr \"L") (pr \"R")) (S.fromPure (1,2))
197−{-# INLINE unzipM #-}
198−unzipM :: Monad m => (a -> m (b,c)) -> Sink m b -> Sink m c -> Sink m a
199−unzipM f (Sink stepB) (Sink stepC) =
200− Sink (f >=> (\(b, c) -> stepB b >> stepC c))
202−-- | Same as 'unzipM' but with a pure unzip function.
204−unzip :: Monad m => (a -> (b,c)) -> Sink m b -> Sink m c -> Sink m a
205−unzip f = unzipM (return . f)
207−------------------------------------------------------------------------------
208−-- Input transformation
209−------------------------------------------------------------------------------
211−-- | Map a pure function on the input of a 'Sink'.
212−{-# INLINABLE lmap #-}
213−lmap :: (a -> b) -> Sink m b -> Sink m a
214−lmap f (Sink step) = Sink (step . f)
216−-- | Map a monadic function on the input of a 'Sink'.
217−{-# INLINABLE lmapM #-}
218−lmapM :: Monad m => (a -> m b) -> Sink m b -> Sink m a
219−lmapM f (Sink step) = Sink (f >=> step)
221−-- | Filter the input of a 'Sink' using a pure predicate function.
222−{-# INLINABLE lfilter #-}
223−lfilter :: Monad m => (a -> Bool) -> Sink m a -> Sink m a
224−lfilter f (Sink step) = Sink (\a -> when (f a) $ step a)
226−-- | Filter the input of a 'Sink' using a monadic predicate function.
227−{-# INLINABLE lfilterM #-}
228−lfilterM :: Monad m => (a -> m Bool) -> Sink m a -> Sink m a
229−lfilterM f (Sink step) = Sink (\a -> f a >>= \use -> when use $ step a)
231−------------------------------------------------------------------------------
233−------------------------------------------------------------------------------
235−-- | Drain all input, running the effects and discarding the results.
236−drain :: Monad m => Sink m a
237−drain = Sink (\_ -> return ())
240−-- > drainM f = lmapM f drain
242−-- Drain all input after passing it through a monadic function.
243−{-# INLINABLE drainM #-}
244−drainM :: Monad m => (a -> m b) -> Sink m a
245−drainM f = Sink (void . f)