@@ -0,0 +1,185 @@
1+{-# LANGUAGE FlexibleContexts #-}
2+{-# LANGUAGE MultiParamTypeClasses #-}
3+-- | Defines the types for a sink, which is a consumer of data.
4+module Data.Conduit.Types.Sink
5+ ( SinkResult (..)
6+ , Sink (..)
7+ , SinkM (..)
8+ ) where
9+
10+import Control.Monad.Trans.Resource (ResourceT)
11+import Control.Monad.Trans.Class (MonadTrans (lift))
12+import Control.Monad (liftM)
13+import Control.Applicative (Applicative (..))
14+import Control.Monad.Base (MonadBase (liftBase))
15+import qualified Data.IORef as I
16+
17+-- | When a sink is given data, it can return both leftover data and a result.
18+data SinkResult input output = SinkResult [input] output
19+instance Functor (SinkResult input) where
20+ fmap f (SinkResult input output) = SinkResult input (f output)
21+
22+-- | In general, a sink will consume data and eventually produce an output when
23+-- it has consumed \"enough\" data. There are two caveats to that statement:
24+--
25+-- * Some sinks do not actually require any data to produce an output. This is
26+-- included with a sink in order to allow for a 'Monad' instance.
27+--
28+-- * Some sinks will consume all available data and only produce a result at
29+-- the \"end\" of a data stream (e.g., @sum@).
30+--
31+-- To allow for the first caveat, we have the 'SinkNoData' constructor. For the
32+-- second, the 'SinkData' constructor has two records: one for receiving more
33+-- input, and the other to indicate the end of a stream. Note that, at the end
34+-- of a stream, some output is required. If a specific 'Sink' implementation
35+-- cannot always produce output, this should be indicated in its return value,
36+-- using something like a 'Maybe' or 'Either'.
37+--
38+-- Invariants:
39+--
40+-- * After a 'Sink' produces a result (either via 'sinkPush' or 'sinkClose'),
41+-- neither of those two functions may be called on the @Sink@ again.
42+--
43+-- * If a @Sink@ needs to clean up any resources (e.g., close a file handle),
44+-- it must do so whenever it returns a result, either via @sinkPush@ or
45+-- @sinkClose@.
46+data Sink input m output =
47+ SinkNoData output
48+ | SinkData
49+ { sinkPush :: [input] -> ResourceT m (SinkResult input (Maybe output))
50+ , sinkClose :: ResourceT m (SinkResult input output)
51+ }
52+
53+instance Monad m => Functor (Sink input m) where
54+ fmap f (SinkNoData x) = SinkNoData (f x)
55+ fmap f (SinkData p c) = SinkData
56+ { sinkPush = liftM (fmap (fmap f)) . p
57+ , sinkClose = liftM (fmap f) c
58+ }
59+
60+-- | Most 'Sink's require some type of state, similar to 'Source's. Like a
61+-- @SourceM@ for a @Source@, a @SinkM@ is a simple monadic wrapper around a
62+-- @Sink@ which allows initialization of such state. See @SourceM@ for further
63+-- caveats.
64+--
65+-- Note that this type provides a 'Monad' instance, allowing you to easily
66+-- compose @SinkM@s together.
67+newtype SinkM input m output = SinkM { genSink :: ResourceT m (Sink input m output) }
68+
69+instance Monad m => Functor (SinkM input m) where
70+ fmap f (SinkM msink) = SinkM (liftM (fmap f) msink)
71+
72+instance MonadBase IO m => Applicative (SinkM input m) where
73+ pure x = SinkM (return (SinkNoData x))
74+ SinkM mf <*> SinkM ma = SinkM $ do
75+ f <- mf
76+ a <- ma
77+ case (f, a) of
78+ (SinkNoData f', SinkNoData a') -> return (SinkNoData (f' a'))
79+ _ -> do
80+ let toEither (SinkData x y) = SinkPair x y
81+ toEither (SinkNoData x) = SinkOutput x
82+ istate <- liftBase (I.newIORef (toEither f, toEither a))
83+ return $ appHelper istate
84+
85+type SinkPush input m output = [input] -> ResourceT m (SinkResult input (Maybe output))
86+type SinkClose input m output = ResourceT m (SinkResult input output)
87+data SinkEither input m output = SinkPair (SinkPush input m output) (SinkClose input m output) | SinkOutput output
88+type SinkState input m a b = I.IORef (SinkEither input m (a -> b), SinkEither input m a)
89+
90+appHelper :: MonadBase IO m => SinkState input m a b -> Sink input m b
91+appHelper istate = SinkData (pushHelper istate) (closeHelper istate)
92+
93+pushHelper :: MonadBase IO m
94+ => SinkState input m a b
95+ -> [input]
96+ -> ResourceT m (SinkResult input (Maybe b))
97+pushHelper istate stream0 = do
98+ state <- liftBase $ I.readIORef istate
99+ go state stream0
100+ where
101+ go (SinkPair f _, eb) stream = do
102+ SinkResult leftover mres <- f stream
103+ case mres of
104+ Nothing -> return $ SinkResult leftover Nothing
105+ Just res -> do
106+ let state' = (SinkOutput res, eb)
107+ liftBase $ I.writeIORef istate state'
108+ go state' leftover
109+ go (f@SinkOutput{}, SinkPair b _) stream = do
110+ SinkResult leftover mres <- b stream
111+ case mres of
112+ Nothing -> return $ SinkResult leftover Nothing
113+ Just res -> do
114+ let state' = (f, SinkOutput res)
115+ liftBase $ I.writeIORef istate state'
116+ go state' leftover
117+ go (SinkOutput f, SinkOutput b) leftover = return $ SinkResult leftover (Just $ f b)
118+
119+closeHelper :: MonadBase IO m
120+ => SinkState input m a b
121+ -> ResourceT m (SinkResult input b)
122+closeHelper istate = do
123+ (sf, sa) <- liftBase $ I.readIORef istate
124+ case sf of
125+ SinkOutput f -> go' f sa
126+ SinkPair _ close -> do
127+ SinkResult leftover f <- close
128+ go f leftover sa
129+ where
130+ go f leftover sa
131+ | null leftover = go' f sa
132+ go f leftover (SinkOutput a) = return $ SinkResult leftover (f a)
133+ go f leftover sp@(SinkPair push _) = do
134+ SinkResult leftover' mres <- push leftover
135+ case mres of
136+ Just a -> return $ SinkResult leftover' (f a)
137+ Nothing -> go f leftover' sp
138+ go' f (SinkPair _ close) = do
139+ SinkResult leftover a <- close
140+ return $ SinkResult leftover (f a)
141+ go' f (SinkOutput a) = return $ SinkResult [] (f a)
142+
143+instance MonadBase IO m => Monad (SinkM input m) where
144+ return = pure
145+ x >>= f = sinkJoin (fmap f x)
146+
147+instance MonadBase IO m => MonadBase IO (SinkM input m) where
148+ liftBase = lift . liftBase
149+
150+instance MonadTrans (SinkM input) where
151+ lift f = SinkM (lift (liftM SinkNoData f))
152+
153+sinkJoin :: MonadBase IO m => SinkM a m (SinkM a m b) -> SinkM a m b
154+sinkJoin = error "sinkJoin"
155+{-
156+sinkJoin (Sink msinkI) = Sink $ do
157+ sinkI <- msinkI
158+ case sinkI of
159+ SinkNoData (Sink inner) -> inner
160+ SinkData msink -> do
161+ istate <- liftBase $ I.newIORef Nothing
162+ return $ SinkData $ go istate msink
163+ where
164+ go :: MonadBaseControl IO m
165+ => I.IORef (Maybe (SinkInsideData a m b))
166+ -> (Stream a -> ResourceT m (SinkResult a (Sink a m b)))
167+ -> SinkInsideData a m b
168+ go istate outer stream = do
169+ state <- liftBase $ I.readIORef istate
170+ case state of
171+ Nothing -> do
172+ SinkResult leftover minner' <- outer stream
173+ case minner' of
174+ Just (Sink msink) -> do
175+ sink <- msink
176+ case sink of
177+ SinkData inner -> do
178+ liftBase $ I.writeIORef istate $ Just inner
179+ if null leftover
180+ then return $ SinkResult [] Nothing
181+ else go istate outer $ Chunks leftover
182+ SinkNoData x -> return $ SinkResult leftover $ Just x
183+ Nothing -> return $ SinkResult leftover Nothing
184+ Just inner -> inner stream
185+ -}