git.iko.soy/typeable/wai
on master help sign in
Breakdown to Source, SourceM and BSource
a4da54bMichael
conduit/Data/Conduit.hs+23−71
compare with master
unifiedsplit
@@ -5,20 +5,20 @@
5{-# LANGUAGE DeriveDataTypeable #-}
5{-# LANGUAGE DeriveDataTypeable #-}
6module Data.Conduit
6module Data.Conduit
7 ( -- * Helper types
7 ( -- * Helper types
8− Stream (..)
9− , SinkResult (..)
8+ SinkResult (..)
10 , ConduitResult (..)
9 , ConduitResult (..)
11 -- * Main types
10 -- * Main types
12− , Source (..)
13 , Sink (..)
11 , Sink (..)
14 , SinkInside (..)
12 , SinkInside (..)
15 , Conduit
13 , Conduit
16 -- * Connect pieces together
14 -- * Connect pieces together
17 , ($$)
15 , ($$)
16+ , (<$$>)
18 , ($=)
17 , ($=)
19 , (=$)
18 , (=$)
20− -- * Helper functions
19+ -- * Other modules
21− , mkSource
20+ , module Data.Conduit.Types.Source
21+ , module Data.Conduit.Util.Source
22 ) where
22 ) where
23
23
24import Control.Monad.Trans.Resource
24import Control.Monad.Trans.Resource
@@ -30,13 +30,8 @@ import qualified Data.IORef as I
30import Data.Typeable (Typeable)
30import Data.Typeable (Typeable)
31import Control.Monad (liftM)
31import Control.Monad (liftM)
32import Control.Applicative (Applicative (..))
32import Control.Applicative (Applicative (..))
33−import Data.Monoid (Monoid (mempty, mappend, mconcat))
33+import Data.Conduit.Types.Source
34−
34+import Data.Conduit.Util.Source
35−data Stream a = EOF | Chunks [a]
36− deriving Show
37−instance Functor Stream where
38− fmap _ EOF = EOF
39− fmap f (Chunks l) = Chunks (map f l)
40
35
41data SinkResult input output = SinkResult [input] (Maybe output)
36data SinkResult input output = SinkResult [input] (Maybe output)
42instance Functor (SinkResult input) where
37instance Functor (SinkResult input) where
@@ -44,38 +39,6 @@ instance Functor (SinkResult input) where
44
39
45data ConduitResult input output = ConduitResult [input] (Stream output)
40data ConduitResult input output = ConduitResult [input] (Stream output)
46
41
47−data SourceInside m a = SourceInside
48− { sourcePull :: ResourceT m (Stream a)
49− , sourcePush :: [a] -> ResourceT m ()
50− }
51−newtype Source m a = Source { unSource :: ResourceT m (SourceInside m a) }
52−instance MonadBaseControl IO m => Monoid (Source m a) where
53− mempty = mkSource $ return EOF
54− mappend a b = mconcat [a, b]
55− mconcat [] = mempty
56− mconcat (Source mnext:rest0) = Source $ do
57− next0 <- mnext
58− istate <- liftBase $ I.newIORef $ Just (next0, rest0)
59− unSource $ mkSource $ go istate
60− where
61− go istate = do
62− state <- liftBase $ I.readIORef istate
63− case state of
64− Nothing -> return EOF
65− Just (next, rest) -> do
66− stream <- sourcePull next
67− case stream of
68− EOF ->
69− case rest of
70− Source a:as -> do
71− a' <- a
72− liftBase $ I.writeIORef istate $ Just (a', as)
73− go istate
74− [] -> do
75− liftBase $ I.writeIORef istate Nothing
76− return EOF
77− Chunks _ -> return stream
78−
79type SinkInsideData a m b = Stream a -> ResourceT m (SinkResult a b)
42type SinkInsideData a m b = Stream a -> ResourceT m (SinkResult a b)
80data SinkInside a m b =
43data SinkInside a m b =
81 SinkData (SinkInsideData a m b)
44 SinkData (SinkInsideData a m b)
@@ -172,44 +135,31 @@ sinkJoin (Sink msinkI) = Sink $ do
172
135
173type Conduit a m b = Stream a -> ResourceT m (ConduitResult a b)
136type Conduit a m b = Stream a -> ResourceT m (ConduitResult a b)
174
137
175−mkSource :: MonadBaseControl IO m
176− => ResourceT m (Stream a)
177− -> Source m a
178−mkSource pull = Source $ do
179− buffer <- liftBase $ I.newIORef []
180− return SourceInside
181− { sourcePull = do
182− x <- liftBase $ I.atomicModifyIORef buffer (\a -> ([], a))
183− if null x
184− then pull
185− else return $ Chunks x
186− , sourcePush = \s ->
187− if null s
188− then return ()
189− else liftBase $
190− I.atomicModifyIORef buffer (\a -> (s ++ a, ()))
191− }
192−
193infixr 0 $$
138infixr 0 $$
194
139
195($$) :: MonadBaseControl IO m => Source m a -> Sink a m b -> ResourceT m b
140($$) :: MonadBaseControl IO m => Source m a -> Sink a m b -> ResourceT m b
196−Source msource $$ Sink msink = do
141+source $$ Sink msink = do
197 sinkI <- msink
142 sinkI <- msink
198 case sinkI of
143 case sinkI of
199 SinkNoData output -> return output
144 SinkNoData output -> return output
200 SinkData sink -> do
145 SinkData sink -> do
201− source <- msource
146+ bs <- bsource source
202− connect' source sink
147+ connect' bs sink
203 where
148 where
204− connect' source sink = do
149+ connect' bs sink = do
205− stream <- sourcePull source
150+ stream <- bsourcePull bs
206 SinkResult leftover mres <- sink stream
151 SinkResult leftover mres <- sink stream
207− sourcePush source leftover
152+ bsourceUnpull bs leftover
208 case (stream, mres) of
153 case (stream, mres) of
209 (_, Just res) -> return res
154 (_, Just res) -> return res
210− (Chunks _, Nothing) -> connect' source sink
155+ (Chunks _, Nothing) -> connect' bs sink
211 (EOF, Nothing) -> liftBase (throwIO NothingAfterEOF)
156 (EOF, Nothing) -> liftBase (throwIO NothingAfterEOF)
212
157
158+infixr 0 <$$>
159+
160+(<$$>) :: MonadBaseControl IO m => SourceM m a -> Sink a m b -> ResourceT m b
161+SourceM msrc <$$> sink = msrc >>= ($$ sink)
162+
213data NothingAfterEOF = NothingAfterEOF
163data NothingAfterEOF = NothingAfterEOF
214 deriving (Show, Typeable)
164 deriving (Show, Typeable)
215instance Exception NothingAfterEOF
165instance Exception NothingAfterEOF
@@ -220,13 +170,15 @@ infixl 1 $=
220 => Source m a
170 => Source m a
221 -> Conduit a m b
171 -> Conduit a m b
222 -> Source m b
172 -> Source m b
223−Source msource $= pipe = Source $ do
173+_ $= _ =
224− source <- msource
174+ undefined
175+ {-
225 source' <- unSource $ mkSource $ do
176 source' <- unSource $ mkSource $ do
226 ConduitResult leftover result <- sourcePull source >>= pipe
177 ConduitResult leftover result <- sourcePull source >>= pipe
227 sourcePush source leftover
178 sourcePush source leftover
228 return result
179 return result
229 return source'
180 return source'
181+ -}
230
182
231infixr 0 =$
183infixr 0 =$
232
184
conduit/Data/Conduit/Binary.hs+7−7
@@ -16,15 +16,15 @@ import Control.Monad.Trans.Resource (with, release)
16
16
17sourceFile :: MonadBaseControl IO m
17sourceFile :: MonadBaseControl IO m
18 => FilePath
18 => FilePath
19− -> Source m S.ByteString
19+ -> SourceM m S.ByteString
20−sourceFile fp = Source $ do
20+sourceFile fp = sourceM
21− (key, handle) <- with (liftBase $ openFile fp ReadMode) (liftBase . hClose)
21+ (with (liftBase $ openFile fp ReadMode) (liftBase . hClose))
22− unSource $ mkSource $ do
22+ (\(key, _) -> release key)
23+ (\(_, handle) -> do
23 bs <- liftBase $ S.hGetSome handle 50
24 bs <- liftBase $ S.hGetSome handle 50
24− -- FIXME protection against being called after close?
25 if S.null bs
25 if S.null bs
26− then release key >> return EOF
26+ then return EOF
27− else return $ Chunks [bs]
27+ else return $ Chunks [bs])
28
28
29sinkFile :: MonadBaseControl IO m
29sinkFile :: MonadBaseControl IO m
30 => FilePath
30 => FilePath
conduit/Data/Conduit/List.hs+6−5
@@ -32,12 +32,13 @@ fold f accum0 = Sink $ do
32 \accum -> (foldl' f accum cs, ())
32 \accum -> (foldl' f accum cs, ())
33 return $ SinkResult [] Nothing
33 return $ SinkResult [] Nothing
34
34
35−fromList :: MonadBaseControl IO m => [a] -> Source m a
35+fromList :: MonadBaseControl IO m => [a] -> SourceM m a
36−fromList l = Source $ do
36+fromList l = sourceM
37− il <- liftBase $ I.newIORef l
37+ (liftBase (I.newIORef l))
38− unSource $ mkSource $ do
38+ (const (return ()))
39+ (\il -> do
39 l' <- liftBase $ I.atomicModifyIORef il $ \x -> ([], x)
40 l' <- liftBase $ I.atomicModifyIORef il $ \x -> ([], x)
40− return $ if null l' then EOF else Chunks l'
41+ return $ if null l' then EOF else Chunks l')
41
42
42take :: MonadBaseControl IO m
43take :: MonadBaseControl IO m
43 => Int
44 => Int
conduit/Data/Conduit/Types/Source.hsnew file+138
@@ -0,0 +1,138 @@
1+{-# LANGUAGE FlexibleContexts #-}
2+{-# LANGUAGE DeriveDataTypeable #-}
3+-- | Defines the types for a source, which is a producer of data.
4+module Data.Conduit.Types.Source
5+ ( Stream (..)
6+ , Source (..)
7+ , SourceM (..)
8+ , BSource (..)
9+ , SourceInvariantException (..)
10+ ) where
11+
12+import Control.Monad.Trans.Resource (ResourceT)
13+import Data.Monoid (Monoid (..))
14+import Control.Monad.Base (MonadBase (liftBase))
15+import qualified Data.IORef as I
16+import Control.Monad (liftM)
17+import Data.Typeable (Typeable)
18+import Control.Exception (Exception, throw)
19+
20+-- | In general, a source represents a producer of a stream of data. A source
21+-- can either produce output, or indicate that there is no more data. This type
22+-- represents the output from a 'Source'.
23+data Stream a = EOF -- ^ no more data
24+ | Chunks [a] -- ^ produced some data
25+
26+instance Functor Stream where
27+ fmap _ EOF = EOF
28+ fmap f (Chunks a) = Chunks (map f a)
29+
30+-- | A 'Source' has two operations on it: pull some data, and close the
31+-- 'Source'. Since 'Source' is built on top of 'ResourceT', all acquired
32+-- resources should be automatically released anyway. Closing a 'Source' early
33+-- is merely an optimization to free scarce resources as soon as possible.
34+--
35+-- A 'Source' has two invariants:
36+--
37+-- * It is illegal to call 'sourcePull' after a previous call returns 'EOF'.
38+--
39+-- * It is legal to call 'sourceClose' multiple times.
40+--
41+-- 'Source's are not expected to call 'sourceClose' when they return an 'EOF'
42+-- from 'sourcePull', though based on the second invariant, they are free to do
43+-- so.
44+data Source m a = Source
45+ { sourcePull :: ResourceT m (Stream a)
46+ , sourceClose :: ResourceT m ()
47+ }
48+
49+instance Monad m => Functor (Source m) where
50+ fmap f src = src
51+ { sourcePull = liftM (fmap f) (sourcePull src)
52+ }
53+
54+-- | All but the simplest of 'Source's (e.g., @repeat@ and @cycle@) require
55+-- some type of state to track their current status. This may be in the form of
56+-- a mutable variable (e.g., @IORef@), or via opening a resource like a
57+-- @Handle@. While a 'Source' is given no opportunity to acquire such
58+-- resources, this type is.
59+--
60+-- A 'SourceM' is simply a monadic action that returns a 'Source'. One nice
61+-- consequence of this is the possibility of creating an efficient 'Monoid'
62+-- instance, which will only acquire one resource at a time, instead of bulk
63+-- acquiring all resources at the beginning of running the 'SourceM'.
64+--
65+-- Note that each time you \"call\" a @SourceM@, it is started from scratch. If
66+-- you want a resumable source (e.g., one which can be passed to multiple
67+-- @Sink@s), you likely want to use a 'BSource'.
68+newtype SourceM m a = SourceM { genSource :: ResourceT m (Source m a) }
69+
70+instance Monad m => Functor (SourceM m) where
71+ fmap f (SourceM msrc) = SourceM (liftM (fmap f) msrc)
72+
73+instance MonadBase IO m => Monoid (SourceM m a) where
74+ mempty = SourceM (return Source
75+ { sourcePull = return EOF
76+ , sourceClose = return ()
77+ })
78+ mappend a b = mconcat [a, b]
79+ mconcat [] = mempty
80+ mconcat (SourceM mnext:rest0) = SourceM $ do
81+ -- open up the first SourceM...
82+ next0 <- mnext
83+ -- and place it in a mutable reference along with all of the upcoming
84+ -- SourceMs
85+ istate <- liftBase (I.newIORef (next0, rest0))
86+ return Source
87+ { sourcePull = pull istate
88+ , sourceClose = close istate
89+ }
90+ where
91+ pull istate =
92+ liftBase (I.readIORef istate) >>= pull'
93+ where
94+ pull' (current, rest) = do
95+ stream <- sourcePull current
96+ case stream of
97+ -- end of the current Source
98+ EOF -> do
99+ -- close the current Source
100+ sourceClose current
101+ case rest of
102+ -- ... and open the next one
103+ SourceM ma:as -> do
104+ a <- ma
105+ liftBase (I.writeIORef istate (a, as))
106+ -- continue pulling base on this new state
107+ pull istate
108+ -- no more source, return an EOF
109+ [] -> do
110+ -- give an error message if the first Source
111+ -- invariant is violated (read data after EOF)
112+ liftBase (I.writeIORef istate (throw (PullAfterEOF "SourceM:mconcat")))
113+ return EOF
114+ Chunks _ -> return stream
115+ close istate = do
116+ -- we only need to close the current Source, since they are opened
117+ -- one at a time
118+ (current, _) <- liftBase (I.readIORef istate)
119+ sourceClose current
120+
121+-- | When actually interacting with 'Source's, we usually want to be able to
122+-- buffer the output, in case any intermediate steps return leftover data. A
123+-- 'BSource' allows for such buffering, via the 'bsourceUnpull' function.
124+--
125+-- A 'BSource', unlike a 'SourceM', is resumable, meaning it can be passed to
126+-- multiple 'Sink's without restarting.
127+--
128+-- Finally, a 'BSource' relaxes one of the invariants of a 'Source': calling
129+-- 'bsourcePull' after an 'EOF' will simply return another 'EOF'.
130+data BSource m a = BSource
131+ { bsourcePull :: ResourceT m (Stream a)
132+ , bsourceUnpull :: [a] -> ResourceT m () -- ^ It is the responsibility of the 'BSource' to check if the argument is null.
133+ , bsourceClose :: ResourceT m ()
134+ }
135+
136+data SourceInvariantException = PullAfterEOF String
137+ deriving (Show, Typeable)
138+instance Exception SourceInvariantException
conduit/Data/Conduit/Util/Source.hsnew file+84
@@ -0,0 +1,84 @@
1+{-# LANGUAGE FlexibleContexts #-}
2+-- | Utilities for constructing and converting 'Source', 'SourceM' and
3+-- 'BSource' types. Please see "Data.Conduit.Types.Source" for more information
4+-- on the base types.
5+module Data.Conduit.Util.Source
6+ ( sourceM
7+ , bsource
8+ , bsourceM
9+ ) where
10+
11+import Control.Monad.Trans.Resource (ResourceT)
12+import Data.Conduit.Types.Source
13+import Control.Monad.Base (MonadBase (liftBase))
14+import qualified Data.IORef as I
15+
16+-- | Construct a 'SourceM'.
17+sourceM :: Monad m
18+ => ResourceT m state -- ^ resource and/or state allocation
19+ -> (state -> ResourceT m ()) -- ^ resource and/or state cleanup
20+ -> (state -> ResourceT m (Stream output)) -- ^ Pull function. Note that this need not explicitly perform any cleanup.
21+ -> SourceM m output
22+sourceM alloc cleanup pull = SourceM $ do
23+ state <- alloc
24+ return Source
25+ { sourcePull = pull state
26+ , sourceClose = cleanup state
27+ }
28+
29+-- | State of a 'BSource'
30+data BState a = EmptyOpen -- ^ nothing in buffer, EOF not received yet
31+ | EmptyClosed -- ^ nothing in buffer, EOF has been received
32+ | Open [a] -- ^ something in buffer, EOF not received yet
33+ | Closed [a] -- ^ something in buffer, EOF has been received
34+ deriving Show
35+
36+-- | Convert a 'SourceM' into a 'BSource'.
37+bsource :: MonadBase IO m
38+ => Source m output
39+ -> ResourceT m (BSource m output)
40+bsource src = do
41+ -- state is a boolean indicating whether EOF was reached, and a buffer
42+ istate <- liftBase (I.newIORef EmptyOpen)
43+ return BSource
44+ { bsourcePull = do
45+ mresult <- liftBase $ I.atomicModifyIORef istate $ \state ->
46+ case state of
47+ Open buffer -> (EmptyOpen, Just (Chunks buffer))
48+ Closed buffer -> (EmptyClosed, Just (Chunks buffer))
49+ EmptyOpen -> (EmptyOpen, Nothing)
50+ EmptyClosed -> (EmptyClosed, Just EOF)
51+ case mresult of
52+ Nothing -> do
53+ result <- sourcePull src
54+ case result of
55+ EOF -> do
56+ liftBase $ I.writeIORef istate EmptyClosed
57+ return EOF
58+ _ -> return result
59+ Just result -> return result
60+ , bsourceUnpull =
61+ \x ->
62+ if null x
63+ then return ()
64+ else liftBase $ I.atomicModifyIORef istate $ \state ->
65+ case state of
66+ Open buffer -> (Open (x ++ buffer), ())
67+ Closed buffer -> (Closed (x ++ buffer), ())
68+ EmptyOpen -> (Open x, ())
69+ EmptyClosed -> (Closed x, ())
70+ , bsourceClose = do
71+ action <- liftBase $ I.atomicModifyIORef istate $ \state ->
72+ case state of
73+ Open x -> (Closed x, sourceClose src)
74+ Closed _ -> (state, return ())
75+ EmptyOpen -> (EmptyClosed, sourceClose src)
76+ EmptyClosed -> (state, return ())
77+ action
78+ }
79+
80+-- | Convert a 'SourceM' into a 'BSource'.
81+bsourceM :: MonadBase IO m
82+ => SourceM m output
83+ -> ResourceT m (BSource m output)
84+bsourceM (SourceM msrc) = msrc >>= bsource
conduit/test/main.hs+8−5
@@ -15,27 +15,30 @@ import qualified Data.ByteString as S
15main :: IO ()
15main :: IO ()
16main = hspecX $ do
16main = hspecX $ do
17 describe "sum" $ do
17 describe "sum" $ do
18+ it "works for 1..10" $ do
19+ x <- runResourceT $ CL.fromList [1..10] C.<$$> CL.fold (+) (0 :: Int)
20+ x @?= sum [1..10]
18 prop "is idempotent" $ \list ->
21 prop "is idempotent" $ \list ->
19− (unsafePerformIO $ runResourceT $ CL.fromList list C.$$ CL.fold (+) (0 :: Int))
22+ (unsafePerformIO $ runResourceT $ CL.fromList list C.<$$> CL.fold (+) (0 :: Int))
20 == sum list
23 == sum list
21 describe "Monoid instance for Source" $ do
24 describe "Monoid instance for Source" $ do
22 it "mappend" $ do
25 it "mappend" $ do
23− x <- runResourceT $ (CL.fromList [1..5 :: Int] `mappend` CL.fromList [6..10]) C.$$ CL.fold (+) 0
26+ x <- runResourceT $ (CL.fromList [1..5 :: Int] `mappend` CL.fromList [6..10]) C.<$$> CL.fold (+) 0
24 x @?= sum [1..10]
27 x @?= sum [1..10]
25 it "mconcat" $ do
28 it "mconcat" $ do
26 x <- runResourceT $ mconcat
29 x <- runResourceT $ mconcat
27 [ CL.fromList [1..5 :: Int]
30 [ CL.fromList [1..5 :: Int]
28 , CL.fromList [6..10]
31 , CL.fromList [6..10]
29 , CL.fromList [11..20]
32 , CL.fromList [11..20]
30− ] C.$$ CL.fold (+) 0
33+ ] C.<$$> CL.fold (+) 0
31 x @?= sum [1..20]
34 x @?= sum [1..20]
32 describe "file access" $ do
35 describe "file access" $ do
33 it "read" $ do
36 it "read" $ do
34 bs <- S.readFile "conduit.cabal"
37 bs <- S.readFile "conduit.cabal"
35− bss <- runResourceT $ CB.sourceFile "conduit.cabal" C.$$ CL.consume
38+ bss <- runResourceT $ CB.sourceFile "conduit.cabal" C.<$$> CL.consume
36 bs @=? S.concat bss
39 bs @=? S.concat bss
37 it "write" $ do
40 it "write" $ do
38− runResourceT $ CB.sourceFile "conduit.cabal" C.$$ CB.sinkFile "tmp"
41+ runResourceT $ CB.sourceFile "conduit.cabal" C.<$$> CB.sinkFile "tmp"
39 bs1 <- S.readFile "conduit.cabal"
42 bs1 <- S.readFile "conduit.cabal"
40 bs2 <- S.readFile "tmp"
43 bs2 <- S.readFile "tmp"
41 bs1 @=? bs2
44 bs1 @=? bs2