@@ -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