{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE CPP #-}
{-# LANGUAGE Trustworthy #-}
module Data.Conduit.List
(
sourceList
, sourceNull
, unfold
, unfoldEither
, unfoldM
, unfoldEitherM
, enumFromTo
, iterate
, replicate
, replicateM
, fold
, foldMap
, take
, drop
, head
, peek
, consume
, sinkNull
, foldMapM
, foldM
, mapM_
, map
, mapMaybe
, mapFoldable
, catMaybes
, concat
, concatMap
, concatMapAccum
, scanl
, scan
, mapAccum
, chunksOf
, groupBy
, groupOn1
, isolate
, filter
, mapM
, iterM
, scanlM
, scanM
, mapAccumM
, mapMaybeM
, mapFoldableM
, concatMapM
, concatMapAccumM
, sequence
) where
import qualified Prelude
import Prelude
( ($), return, (==), (-), Int
, (.), id, Maybe (..), Monad
, Either (..)
, Bool (..)
, (>>)
, (>>=)
, seq
, otherwise
, Enum, Eq
, maybe
, (<=)
, (>)
, error
, (++)
, show
)
import Data.Monoid (Monoid, mempty, mappend)
import qualified Data.Foldable as F
import Data.Conduit
import Data.Conduit.Internal.Fusion
import Data.Conduit.Internal.List.Stream
import qualified Data.Conduit.Internal as CI
import Control.Monad (when, (<=<), liftM, void)
import Control.Monad.Trans.Class (lift)
#include "fusion-macros.h"
unfold, unfoldC :: Monad m
=> (b -> Maybe (a, b))
-> b
-> ConduitT i a m ()
unfoldC f =
go
where
go seed =
case f seed of
Just (a, seed') -> yield a >> go seed'
Nothing -> return ()
{-# INLINE unfoldC #-}
STREAMING(unfold, unfoldC, unfoldS, f x)
unfoldEither, unfoldEitherC :: Monad m
=> (b -> Either r (a, b))
-> b
-> ConduitT i a m r
unfoldEitherC f =
go
where
go seed =
case f seed of
Right (a, seed') -> yield a >> go seed'
Left r -> return r
{-# INLINE unfoldEitherC #-}
STREAMING(unfoldEither, unfoldEitherC, unfoldEitherS, f x)
unfoldM, unfoldMC :: Monad m
=> (b -> m (Maybe (a, b)))
-> b
-> ConduitT i a m ()
unfoldMC f =
go
where
go seed = do
mres <- lift $ f seed
case mres of
Just (a, seed') -> yield a >> go seed'
Nothing -> return ()
STREAMING(unfoldM, unfoldMC, unfoldMS, f seed)
unfoldEitherM, unfoldEitherMC :: Monad m
=> (b -> m (Either r (a, b)))
-> b
-> ConduitT i a m r
unfoldEitherMC f =
go
where
go seed = do
mres <- lift $ f seed
case mres of
Right (a, seed') -> yield a >> go seed'
Left r -> return r
STREAMING(unfoldEitherM, unfoldEitherMC, unfoldEitherMS, f seed)
sourceList, sourceListC :: Monad m => [a] -> ConduitT i a m ()
sourceListC = Prelude.mapM_ yield
{-# INLINE sourceListC #-}
STREAMING(sourceList, sourceListC, sourceListS, xs)
enumFromTo, enumFromToC :: (Enum a, Prelude.Ord a, Monad m)
=> a
-> a
-> ConduitT i a m ()
enumFromToC x0 y =
loop x0
where
loop x
| x Prelude.> y = return ()
| otherwise = yield x >> loop (Prelude.succ x)
{-# INLINE enumFromToC #-}
STREAMING(enumFromTo, enumFromToC, enumFromToS, x0 y)
iterate, iterateC :: Monad m => (a -> a) -> a -> ConduitT i a m ()
iterateC f =
go
where
go a = yield a >> go (f a)
{-# INLINE iterateC #-}
STREAMING(iterate, iterateC, iterateS, f a)
replicate, replicateC :: Monad m => Int -> a -> ConduitT i a m ()
replicateC cnt0 a =
loop cnt0
where
loop i
| i <= 0 = return ()
| otherwise = yield a >> loop (i - 1)
{-# INLINE replicateC #-}
STREAMING(replicate, replicateC, replicateS, cnt0 a)
replicateM, replicateMC :: Monad m => Int -> m a -> ConduitT i a m ()
replicateMC cnt0 ma =
loop cnt0
where
loop i
| i <= 0 = return ()
| otherwise = lift ma >>= yield >> loop (i - 1)
{-# INLINE replicateMC #-}
STREAMING(replicateM, replicateMC, replicateMS, cnt0 ma)
fold, foldC :: Monad m
=> (b -> a -> b)
-> b
-> ConduitT a o m b
foldC f =
loop
where
loop !accum = await >>= maybe (return accum) (loop . f accum)
{-# INLINE foldC #-}
STREAMING(fold, foldC, foldS, f accum)
foldM, foldMC :: Monad m
=> (b -> a -> m b)
-> b
-> ConduitT a o m b
foldMC f =
loop
where
loop accum = do
await >>= maybe (return accum) go
where
go a = do
accum' <- lift $ f accum a
accum' `seq` loop accum'
{-# INLINE foldMC #-}
STREAMING(foldM, foldMC, foldMS, f accum)
connectFold :: Monad m => ConduitT () a m () -> (b -> a -> b) -> b -> m b
connectFold (CI.ConduitT src0) f =
go (src0 CI.Done)
where
go (CI.Done ()) b = return b
go (CI.HaveOutput src a) b = go src Prelude.$! f b a
go (CI.NeedInput _ c) b = go (c ()) b
go (CI.Leftover src ()) b = go src b
go (CI.PipeM msrc) b = do
src <- msrc
go src b
{-# INLINE connectFold #-}
{-# RULES "conduit: $$ fold" forall src f b. runConduit (src .| fold f b) = connectFold src f b #-}
connectFoldM :: Monad m => ConduitT () a m () -> (b -> a -> m b) -> b -> m b
connectFoldM (CI.ConduitT src0) f =
go (src0 CI.Done)
where
go (CI.Done ()) b = return b
go (CI.HaveOutput src a) b = do
!b' <- f b a
go src b'
go (CI.NeedInput _ c) b = go (c ()) b
go (CI.Leftover src ()) b = go src b
go (CI.PipeM msrc) b = do
src <- msrc
go src b
{-# INLINE connectFoldM #-}
{-# RULES "conduit: $$ foldM" forall src f b. runConduit (src .| foldM f b) = connectFoldM src f b #-}
foldMap :: (Monad m, Monoid b)
=> (a -> b)
-> ConduitT a o m b
INLINE_RULE(foldMap, f, let combiner accum = mappend accum . f in fold combiner mempty)
foldMapM :: (Monad m, Monoid b)
=> (a -> m b)
-> ConduitT a o m b
INLINE_RULE(foldMapM, f, let combiner accum = liftM (mappend accum) . f in foldM combiner mempty)
mapM_, mapM_C :: Monad m
=> (a -> m ())
-> ConduitT a o m ()
mapM_C f = awaitForever $ lift . f
{-# INLINE mapM_C #-}
STREAMING(mapM_, mapM_C, mapM_S, f)
srcMapM_ :: Monad m => ConduitT () a m () -> (a -> m ()) -> m ()
srcMapM_ (CI.ConduitT src) f =
go (src CI.Done)
where
go (CI.Done ()) = return ()
go (CI.PipeM mp) = mp >>= go
go (CI.Leftover p ()) = go p
go (CI.HaveOutput p o) = f o >> go p
go (CI.NeedInput _ c) = go (c ())
{-# INLINE srcMapM_ #-}
{-# RULES "conduit: connect to mapM_" [2] forall f src. runConduit (src .| mapM_ f) = srcMapM_ src f #-}
drop, dropC :: Monad m
=> Int
-> ConduitT a o m ()
dropC =
loop
where
loop i | i <= 0 = return ()
loop count = await >>= maybe (return ()) (\_ -> loop (count - 1))
{-# INLINE dropC #-}
STREAMING(drop, dropC, dropS, i)
take, takeC :: Monad m
=> Int
-> ConduitT a o m [a]
takeC =
loop id
where
loop front count | count <= 0 = return $ front []
loop front count = await >>= maybe
(return $ front [])
(\x -> loop (front . (x:)) (count - 1))
{-# INLINE takeC #-}
STREAMING(take, takeC, takeS, i)
head, headC :: Monad m => ConduitT a o m (Maybe a)
headC = await
{-# INLINE headC #-}
STREAMING0(head, headC, headS)
peek :: Monad m => ConduitT a o m (Maybe a)
peek = await >>= maybe (return Nothing) (\x -> leftover x >> return (Just x))
map, mapC :: Monad m => (a -> b) -> ConduitT a b m ()
mapC f = awaitForever $ yield . f
{-# INLINE mapC #-}
STREAMING(map, mapC, mapS, f)
mapM, mapMC :: Monad m => (a -> m b) -> ConduitT a b m ()
mapMC f = awaitForever $ \a -> lift (f a) >>= yield
{-# INLINE mapMC #-}
STREAMING(mapM, mapMC, mapMS, f)
iterM, iterMC :: Monad m => (a -> m ()) -> ConduitT a a m ()
iterMC f = awaitForever $ \a -> lift (f a) >> yield a
{-# INLINE iterMC #-}
STREAMING(iterM, iterMC, iterMS, f)
mapMaybe, mapMaybeC :: Monad m => (a -> Maybe b) -> ConduitT a b m ()
mapMaybeC f = awaitForever $ maybe (return ()) yield . f
{-# INLINE mapMaybeC #-}
STREAMING(mapMaybe, mapMaybeC, mapMaybeS, f)
mapMaybeM, mapMaybeMC :: Monad m => (a -> m (Maybe b)) -> ConduitT a b m ()
mapMaybeMC f = awaitForever $ maybe (return ()) yield <=< lift . f
{-# INLINE mapMaybeMC #-}
STREAMING(mapMaybeM, mapMaybeMC, mapMaybeMS, f)
catMaybes, catMaybesC :: Monad m => ConduitT (Maybe a) a m ()
catMaybesC = awaitForever $ maybe (return ()) yield
{-# INLINE catMaybesC #-}
STREAMING0(catMaybes, catMaybesC, catMaybesS)
concat, concatC :: (Monad m, F.Foldable f) => ConduitT (f a) a m ()
concatC = awaitForever $ F.mapM_ yield
{-# INLINE concatC #-}
STREAMING0(concat, concatC, concatS)
concatMap, concatMapC :: Monad m => (a -> [b]) -> ConduitT a b m ()
concatMapC f = awaitForever $ sourceList . f
{-# INLINE concatMapC #-}
STREAMING(concatMap, concatMapC, concatMapS, f)
concatMapM, concatMapMC :: Monad m => (a -> m [b]) -> ConduitT a b m ()
concatMapMC f = awaitForever $ sourceList <=< lift . f
{-# INLINE concatMapMC #-}
STREAMING(concatMapM, concatMapMC, concatMapMS, f)
concatMapAccum, concatMapAccumC :: Monad m => (a -> accum -> (accum, [b])) -> accum -> ConduitT a b m ()
concatMapAccumC f x0 = void (mapAccum f x0) .| concat
{-# INLINE concatMapAccumC #-}
STREAMING(concatMapAccum, concatMapAccumC, concatMapAccumS, f x0)
scanl :: Monad m => (a -> s -> (s, b)) -> s -> ConduitT a b m ()
scanl f s = void $ mapAccum f s
{-# DEPRECATED scanl "Use mapAccum instead" #-}
scanlM :: Monad m => (a -> s -> m (s, b)) -> s -> ConduitT a b m ()
scanlM f s = void $ mapAccumM f s
{-# DEPRECATED scanlM "Use mapAccumM instead" #-}
mapAccum, mapAccumC :: Monad m => (a -> s -> (s, b)) -> s -> ConduitT a b m s
mapAccumC f =
loop
where
loop !s = await >>= maybe (return s) go
where
go a = case f a s of
(s', b) -> yield b >> loop s'
STREAMING(mapAccum, mapAccumC, mapAccumS, f s)
mapAccumM, mapAccumMC :: Monad m => (a -> s -> m (s, b)) -> s -> ConduitT a b m s
mapAccumMC f =
loop
where
loop !s = await >>= maybe (return s) go
where
go a = do (s', b) <- lift $ f a s
yield b
loop s'
{-# INLINE mapAccumMC #-}
STREAMING(mapAccumM, mapAccumMC, mapAccumMS, f s)
scan :: Monad m => (a -> b -> b) -> b -> ConduitT a b m b
INLINE_RULE(scan, f, mapAccum (\a b -> let r = f a b in (r, r)))
scanM :: Monad m => (a -> b -> m b) -> b -> ConduitT a b m b
INLINE_RULE(scanM, f, mapAccumM (\a b -> f a b >>= \r -> return (r, r)))
concatMapAccumM, concatMapAccumMC :: Monad m => (a -> accum -> m (accum, [b])) -> accum -> ConduitT a b m ()
concatMapAccumMC f x0 = void (mapAccumM f x0) .| concat
{-# INLINE concatMapAccumMC #-}
STREAMING(concatMapAccumM, concatMapAccumMC, concatMapAccumMS, f x0)
mapFoldable, mapFoldableC :: (Monad m, F.Foldable f) => (a -> f b) -> ConduitT a b m ()
mapFoldableC f = awaitForever $ F.mapM_ yield . f
{-# INLINE mapFoldableC #-}
STREAMING(mapFoldable, mapFoldableC, mapFoldableS, f)
mapFoldableM, mapFoldableMC :: (Monad m, F.Foldable f) => (a -> m (f b)) -> ConduitT a b m ()
mapFoldableMC f = awaitForever $ F.mapM_ yield <=< lift . f
{-# INLINE mapFoldableMC #-}
STREAMING(mapFoldableM, mapFoldableMC, mapFoldableMS, f)
consume, consumeC :: Monad m => ConduitT a o m [a]
consumeC =
loop id
where
loop front = await >>= maybe (return $ front []) (\x -> loop $ front . (x:))
{-# INLINE consumeC #-}
STREAMING0(consume, consumeC, consumeS)
chunksOf :: Monad m => Int -> ConduitT a [a] m ()
chunksOf n = if n > 0 then loop n id else error $ "chunksOf size must be positive (given " ++ show n ++ ")"
where
loop 0 rest = yield (rest []) >> loop n id
loop count rest = await >>= \ma -> case ma of
Nothing -> case rest [] of
[] -> return ()
nonempty -> yield nonempty
Just a -> loop (count - 1) (rest . (a :))
groupBy, groupByC :: Monad m => (a -> a -> Bool) -> ConduitT a [a] m ()
groupByC f =
start
where
start = await >>= maybe (return ()) (loop id)
loop rest x =
await >>= maybe (yield (x : rest [])) go
where
go y
| f x y = loop (rest . (y:)) x
| otherwise = yield (x : rest []) >> loop id y
STREAMING(groupBy, groupByC, groupByS, f)
groupOn1, groupOn1C :: (Monad m, Eq b)
=> (a -> b)
-> ConduitT a (a, [a]) m ()
groupOn1C f =
start
where
start = await >>= maybe (return ()) (loop id)
loop rest x =
await >>= maybe (yield (x, rest [])) go
where
go y
| f x == f y = loop (rest . (y:)) x
| otherwise = yield (x, rest []) >> loop id y
STREAMING(groupOn1, groupOn1C, groupOn1S, f)
isolate, isolateC :: Monad m => Int -> ConduitT a a m ()
isolateC =
loop
where
loop count | count <= 0 = return ()
loop count = await >>= maybe (return ()) (\x -> yield x >> loop (count - 1))
STREAMING(isolate, isolateC, isolateS, count)
filter, filterC :: Monad m => (a -> Bool) -> ConduitT a a m ()
filterC f = awaitForever $ \i -> when (f i) (yield i)
STREAMING(filter, filterC, filterS, f)
filterFuseRight
:: Monad m
=> ConduitT i o m ()
-> (o -> Bool)
-> ConduitT i o m ()
filterFuseRight (CI.ConduitT src) f = CI.ConduitT $ \rest -> let
go (CI.Done ()) = rest ()
go (CI.PipeM mp) = CI.PipeM (liftM go mp)
go (CI.Leftover p i) = CI.Leftover (go p) i
go (CI.HaveOutput p o)
| f o = CI.HaveOutput (go p) o
| otherwise = go p
go (CI.NeedInput p c) = CI.NeedInput (go . p) (go . c)
in go (src CI.Done)
{-# RULES "conduit: source/filter fusion .|" forall f src. src .| filter f = filterFuseRight src f #-}
{-# INLINE filterFuseRight #-}
sinkNull, sinkNullC :: Monad m => ConduitT i o m ()
sinkNullC = awaitForever $ \_ -> return ()
{-# INLINE sinkNullC #-}
STREAMING0(sinkNull, sinkNullC, sinkNullS)
srcSinkNull :: Monad m => ConduitT () o m () -> m ()
srcSinkNull (CI.ConduitT src) =
go (src CI.Done)
where
go (CI.Done ()) = return ()
go (CI.PipeM mp) = mp >>= go
go (CI.Leftover p ()) = go p
go (CI.HaveOutput p _) = go p
go (CI.NeedInput _ c) = go (c ())
{-# INLINE srcSinkNull #-}
{-# RULES "conduit: connect to sinkNull" forall src. runConduit (src .| sinkNull) = srcSinkNull src #-}
sourceNull, sourceNullC :: Monad m => ConduitT i o m ()
sourceNullC = return ()
{-# INLINE sourceNullC #-}
STREAMING0(sourceNull, sourceNullC, sourceNullS)
sequence :: Monad m
=> ConduitT i o m o
-> ConduitT i o m ()
sequence sink =
self
where
self = awaitForever $ \i -> leftover i >> sink >>= yield