git.iko.soy/typeable/wai
on master help sign in
Breakdown to Sink, SinkM (incomplete Monad instance)
2ee25d9Michael
conduit/Data/Conduit.hs+28−128
compare with master
unifiedsplit
@@ -5,11 +5,8 @@
5{-# LANGUAGE DeriveDataTypeable #-}
5{-# LANGUAGE DeriveDataTypeable #-}
6module Data.Conduit
6module Data.Conduit
7 ( -- * Helper types
7 ( -- * Helper types
8− SinkResult (..)
9− , ConduitResult (..)
8+ ConduitResult (..)
10 -- * Main types
9 -- * Main types
11− , Sink (..)
12− , SinkInside (..)
13 , Conduit
10 , Conduit
14 -- * Connect pieces together
11 -- * Connect pieces together
15 , ($$)
12 , ($$)
@@ -17,153 +14,55 @@ module Data.Conduit
17 , ($=)
14 , ($=)
18 , (=$)
15 , (=$)
19 -- * Other modules
16 -- * Other modules
17+ -- ** Source
20 , module Data.Conduit.Types.Source
18 , module Data.Conduit.Types.Source
21 , module Data.Conduit.Util.Source
19 , module Data.Conduit.Util.Source
20+ -- ** Sink
21+ , module Data.Conduit.Types.Sink
22+ , module Data.Conduit.Util.Sink
22 ) where
23 ) where
23
24
24import Control.Monad.Trans.Resource
25import Control.Monad.Trans.Resource
25import Control.Monad.Trans.Control (MonadBaseControl)
26import Control.Monad.Trans.Control (MonadBaseControl)
26−import Control.Monad.Trans.Class (MonadTrans (lift))
27−import Control.Monad.Base (MonadBase (liftBase))
28−import Control.Exception.Lifted (throwIO, Exception)
29−import qualified Data.IORef as I
30−import Data.Typeable (Typeable)
31−import Control.Monad (liftM)
32−import Control.Applicative (Applicative (..))
33import Data.Conduit.Types.Source
27import Data.Conduit.Types.Source
34import Data.Conduit.Util.Source
28import Data.Conduit.Util.Source
35−
29+import Data.Conduit.Types.Sink
36−data SinkResult input output = SinkResult [input] (Maybe output)
30+import Data.Conduit.Util.Sink
37−instance Functor (SinkResult input) where
38− fmap f (SinkResult input moutput) = SinkResult input (fmap f moutput)
39
31
40data ConduitResult input output = ConduitResult [input] (Stream output)
32data ConduitResult input output = ConduitResult [input] (Stream output)
41
33
42−type SinkInsideData a m b = Stream a -> ResourceT m (SinkResult a b)
43−data SinkInside a m b =
44− SinkData (SinkInsideData a m b)
45− | SinkNoData b
46−newtype Sink input m output = Sink (ResourceT m (SinkInside input m output))
47−instance Monad m => Functor (Sink input m) where
48− fmap f (Sink msink) = Sink $ do
49− sinkI <- msink
50− return $ case sinkI of
51− SinkData sink -> SinkData $ liftM (fmap f) . sink
52− SinkNoData x -> SinkNoData $ f x
53−instance MonadBaseControl IO m => Applicative (Sink input m) where
54− pure x = Sink $ return $ SinkNoData x
55− Sink mf <*> Sink ma = Sink $ do
56− f <- mf
57− a <- ma
58− case (f, a) of
59− (SinkNoData f', SinkNoData a') -> return $ SinkNoData (f' a')
60− _ -> do
61− let toEither (SinkData x) = Left x
62− toEither (SinkNoData x) = Right x
63− istate <- liftBase $ I.newIORef (toEither f, toEither a)
64− return $ appHelper istate
65−
66−appHelper :: MonadBaseControl IO m
67− => I.IORef
68− ( Either (SinkInsideData input m (a -> b)) (a -> b)
69− , Either (SinkInsideData input m a) a
70− )
71− -> SinkInside input m b
72−appHelper istate = SinkData $ \stream0 -> do
73− state <- liftBase $ I.readIORef istate
74− go state stream0
75− where
76− go (Left f, eb) stream = do
77− SinkResult leftover mres <- f stream
78− case mres of
79− Nothing -> return $ SinkResult leftover Nothing
80− Just res -> do
81− let state' = (Right res, eb)
82− liftBase $ I.writeIORef istate state'
83− go state' $ Chunks leftover
84− go (f@Right{}, Left b) stream = do
85− SinkResult leftover mres <- b stream
86− case mres of
87− Nothing -> return $ SinkResult leftover Nothing
88− Just res -> do
89− let state' = (f, Right res)
90− liftBase $ I.writeIORef istate state'
91− go state' $ Chunks leftover
92− go (Right f, Right b) EOF = return $ SinkResult [] (Just $ f b)
93− go (Right f, Right b) (Chunks s) = return $ SinkResult s (Just $ f b)
94−
95−instance MonadBaseControl IO m => Monad (Sink input m) where
96− return = pure
97− x >>= f = sinkJoin (fmap f x)
98−
99−instance MonadBaseControl IO m => MonadBase IO (Sink input m) where
100− liftBase = lift . liftBase
101−
102−instance MonadTrans (Sink input) where
103− lift f = Sink (lift (liftM SinkNoData f))
104−
105−sinkJoin :: MonadBaseControl IO m => Sink a m (Sink a m b) -> Sink a m b
106−sinkJoin (Sink msinkI) = Sink $ do
107− sinkI <- msinkI
108− case sinkI of
109− SinkNoData (Sink inner) -> inner
110− SinkData msink -> do
111− istate <- liftBase $ I.newIORef Nothing
112− return $ SinkData $ go istate msink
113− where
114− go :: MonadBaseControl IO m
115− => I.IORef (Maybe (SinkInsideData a m b))
116− -> (Stream a -> ResourceT m (SinkResult a (Sink a m b)))
117− -> SinkInsideData a m b
118− go istate outer stream = do
119− state <- liftBase $ I.readIORef istate
120− case state of
121− Nothing -> do
122− SinkResult leftover minner' <- outer stream
123− case minner' of
124− Just (Sink msink) -> do
125− sink <- msink
126− case sink of
127− SinkData inner -> do
128− liftBase $ I.writeIORef istate $ Just inner
129− if null leftover
130− then return $ SinkResult [] Nothing
131− else go istate outer $ Chunks leftover
132− SinkNoData x -> return $ SinkResult leftover $ Just x
133− Nothing -> return $ SinkResult leftover Nothing
134− Just inner -> inner stream
135−
136type Conduit a m b = Stream a -> ResourceT m (ConduitResult a b)
34type Conduit a m b = Stream a -> ResourceT m (ConduitResult a b)
137
35
138infixr 0 $$
36infixr 0 $$
139
37
140−($$) :: MonadBaseControl IO m => Source m a -> Sink a m b -> ResourceT m b
38+($$) :: MonadBaseControl IO m => Source m a -> SinkM a m b -> ResourceT m b
141−source $$ Sink msink = do
39+source $$ SinkM msink = do
142 sinkI <- msink
40 sinkI <- msink
143 case sinkI of
41 case sinkI of
144 SinkNoData output -> return output
42 SinkNoData output -> return output
145− SinkData sink -> do
43+ SinkData push close -> do
146 bs <- bsource source
44 bs <- bsource source
147− connect' bs sink
45+ connect' bs push close
148 where
46 where
149− connect' bs sink = do
47+ connect' bs push close = do
150 stream <- bsourcePull bs
48 stream <- bsourcePull bs
49+ case stream of
50+ EOF -> do
151− SinkResult leftover mres <- sink stream
51+ SinkResult leftover res <- close
52+ bsourceUnpull bs leftover
53+ return res
54+ Chunks a -> do
55+ SinkResult leftover mres <- push a
152− bsourceUnpull bs leftover
56+ bsourceUnpull bs leftover
153− case (stream, mres) of
57+ case mres of
154− (_, Just res) -> return res
58+ Just res -> return res
155− (Chunks _, Nothing) -> connect' bs sink
59+ Nothing -> connect' bs push close
156− (EOF, Nothing) -> liftBase (throwIO NothingAfterEOF)
157
60
158infixr 0 <$$>
61infixr 0 <$$>
159
62
160−(<$$>) :: MonadBaseControl IO m => SourceM m a -> Sink a m b -> ResourceT m b
63+(<$$>) :: MonadBaseControl IO m => SourceM m a -> SinkM a m b -> ResourceT m b
161SourceM msrc <$$> sink = msrc >>= ($$ sink)
64SourceM msrc <$$> sink = msrc >>= ($$ sink)
162
65
163−data NothingAfterEOF = NothingAfterEOF
164− deriving (Show, Typeable)
165−instance Exception NothingAfterEOF
166−
167infixl 1 $=
66infixl 1 $=
168
67
169($=) :: MonadBaseControl IO m
68($=) :: MonadBaseControl IO m
@@ -182,12 +81,13 @@ _ $= _ =
182
81
183infixr 0 =$
82infixr 0 =$
184
83
185−(=$) :: MonadBaseControl IO m => Conduit a m b -> Sink b m c -> Sink a m c
84+(=$) :: MonadBaseControl IO m => Conduit a m b -> SinkM b m c -> SinkM a m c
186−pipe =$ Sink msink = Sink $ do
85+_ =$ _ = SinkM $ do
187− sinkI <- msink
86+ undefined {-
188 case sinkI of
87 case sinkI of
189 SinkData sink -> return $ SinkData $ \stream -> do
88 SinkData sink -> return $ SinkData $ \stream -> do
190 ConduitResult leftover stream' <- pipe stream
89 ConduitResult leftover stream' <- pipe stream
191 SinkResult _thisislost mres <- sink stream'
90 SinkResult _thisislost mres <- sink stream'
192 return $ SinkResult leftover mres
91 return $ SinkResult leftover mres
193 SinkNoData mres -> return $ SinkNoData mres
92 SinkNoData mres -> return $ SinkNoData mres
93+ -}
conduit/Data/Conduit/Binary.hs+7−7
@@ -8,6 +8,7 @@ import Prelude hiding (FilePath)
8import System.IO (hClose)
8import System.IO (hClose)
9import Filesystem.Path.CurrentOS (FilePath)
9import Filesystem.Path.CurrentOS (FilePath)
10import qualified Data.ByteString as S
10import qualified Data.ByteString as S
11+import qualified Data.ByteString.Lazy as L
11import Filesystem (openFile, IOMode (ReadMode, WriteMode))
12import Filesystem (openFile, IOMode (ReadMode, WriteMode))
12import Data.Conduit
13import Data.Conduit
13import Control.Monad.Trans.Control (MonadBaseControl)
14import Control.Monad.Trans.Control (MonadBaseControl)
@@ -28,10 +29,9 @@ sourceFile fp = sourceM
28
29
29sinkFile :: MonadBaseControl IO m
30sinkFile :: MonadBaseControl IO m
30 => FilePath
31 => FilePath
31− -> Sink S.ByteString m ()
32+ -> SinkM S.ByteString m ()
32−sinkFile fp = Sink $ do
33+sinkFile fp = sinkM
33− (key, handle) <- with (liftBase $ openFile fp WriteMode) (liftBase . hClose)
34+ (with (liftBase $ openFile fp WriteMode) (liftBase . hClose))
34− return $ SinkData $ \stream -> do
35− case stream of
36− EOF -> release key >> return (SinkResult [] (Just ()))
35+ (\(key, _) -> release key)
37− Chunks bss -> mapM_ (liftBase . S.hPut handle) bss >> return (SinkResult [] Nothing)
36+ (\(_, handle) bss -> liftBase (L.hPut handle $ L.fromChunks bss) >> return (SinkResult [] Nothing))
37+ (const $ return $ SinkResult [] ())
conduit/Data/Conduit/List.hs+31−24
@@ -21,13 +21,15 @@ import Data.List (foldl')
21fold :: MonadBaseControl IO m
21fold :: MonadBaseControl IO m
22 => (b -> a -> b)
22 => (b -> a -> b)
23 -> b
23 -> b
24− -> Sink a m b
24+ -> SinkM a m b
25−fold f accum0 = Sink $ do
25+fold f accum0 = sinkM
26− iaccum <- liftBase $ I.newIORef accum0
26+ (liftBase $ I.newIORef accum0)
27− return $ SinkData $ go iaccum
27+ (const $ return ())
28+ push
29+ close
28 where
30 where
29− go iaccum EOF = SinkResult [] . Just <$> liftBase (I.readIORef iaccum)
31+ close iaccum = SinkResult [] <$> liftBase (I.readIORef iaccum)
30− go iaccum (Chunks cs) = do
32+ push iaccum cs = do
31 liftBase $ I.atomicModifyIORef iaccum $
33 liftBase $ I.atomicModifyIORef iaccum $
32 \accum -> (foldl' f accum cs, ())
34 \accum -> (foldl' f accum cs, ())
33 return $ SinkResult [] Nothing
35 return $ SinkResult [] Nothing
@@ -42,13 +44,15 @@ fromList l = sourceM
42
44
43take :: MonadBaseControl IO m
45take :: MonadBaseControl IO m
44 => Int
46 => Int
45− -> Sink a m [a]
47+ -> SinkM a m [a]
46−take count0 = Sink $ do
48+take count0 = sinkM
47− istate <- liftBase $ I.newIORef (count0, [])
49+ (liftBase $ I.newIORef (count0, []))
48− return $ SinkData $ go istate
50+ (const $ return ())
51+ push
52+ close
49 where
53 where
50− go istate EOF = SinkResult [] . Just . snd <$> liftBase (I.readIORef istate)
54+ close istate = SinkResult [] . snd <$> liftBase (I.readIORef istate)
51− go istate (Chunks cs) = do
55+ push istate cs = do
52 (count, rest', b) <- liftBase $ I.atomicModifyIORef istate $ \(count, rest) ->
56 (count, rest', b) <- liftBase $ I.atomicModifyIORef istate $ \(count, rest) ->
53 let (a, b) = splitAt count cs
57 let (a, b) = splitAt count cs
54 count' = count - length a
58 count' = count - length a
@@ -56,13 +60,14 @@ take count0 = Sink $ do
56 in ((count', rest'), (count', rest', b))
60 in ((count', rest'), (count', rest', b))
57 return $ SinkResult b $ if count == 0 then Just rest' else Nothing
61 return $ SinkResult b $ if count == 0 then Just rest' else Nothing
58
62
59−head :: MonadBaseControl IO m => Sink a m (Maybe a)
63+head :: MonadBaseControl IO m => SinkM a m (Maybe a)
60−head =
61− Sink $ return $ SinkData $ return . go
64+head = SinkM $ return $ SinkData
65+ { sinkPush = return . push
66+ , sinkClose = return $ SinkResult [] Nothing
67+ }
62 where
68 where
63− go EOF = SinkResult [] $ Just Nothing
64− go (Chunks []) = SinkResult [] Nothing
69+ push [] = SinkResult [] Nothing
65− go (Chunks (a:as)) = SinkResult as (Just (Just a))
70+ push (a:as) = SinkResult as (Just (Just a))
66
71
67map :: Monad m => (a -> b) -> Conduit a m b
72map :: Monad m => (a -> b) -> Conduit a m b
68map f a = return $ ConduitResult [] $ fmap f a
73map f a = return $ ConduitResult [] $ fmap f a
@@ -71,12 +76,14 @@ concatMap :: Monad m => (a -> [b]) -> Conduit a m b
71concatMap _ EOF = return $ ConduitResult [] EOF
76concatMap _ EOF = return $ ConduitResult [] EOF
72concatMap f (Chunks l) = return $ ConduitResult [] $ Chunks $ Prelude.concatMap f l
77concatMap f (Chunks l) = return $ ConduitResult [] $ Chunks $ Prelude.concatMap f l
73
78
74−consume :: MonadBaseControl IO m => Sink a m [a]
79+consume :: MonadBaseControl IO m => SinkM a m [a]
75−consume = Sink $ do
80+consume = sinkM
76− ifront <- liftBase $ I.newIORef id
81+ (liftBase $ I.newIORef id)
77− return $ SinkData $ go ifront
82+ (const $ return ())
83+ push
84+ close
78 where
85 where
79− go ifront EOF = SinkResult [] . Just . ($ []) <$> liftBase (I.readIORef ifront)
86+ close ifront = SinkResult [] . ($ []) <$> liftBase (I.readIORef ifront)
80− go ifront (Chunks cs) = do
87+ push ifront cs = do
81 liftBase $ I.atomicModifyIORef ifront $ \front -> (front . (cs ++), ())
88 liftBase $ I.atomicModifyIORef ifront $ \front -> (front . (cs ++), ())
82 return $ SinkResult [] Nothing
89 return $ SinkResult [] Nothing
conduit/Data/Conduit/Types/Sink.hsnew file+185
@@ -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+ -}
conduit/Data/Conduit/Util/Sink.hsnew file+29
@@ -0,0 +1,29 @@
1+-- | Utilities for constructing 'SinkM's. Please see "Data.Conduit.Types.Sink"
2+-- for more information on the base types.
3+module Data.Conduit.Util.Sink
4+ ( sinkM
5+ ) where
6+
7+import Control.Monad.Trans.Resource (ResourceT)
8+import Data.Conduit.Types.Sink
9+
10+-- | Construct a 'SinkM'. Note that your push and close functions need not
11+-- explicitly perform any cleanup.
12+sinkM :: Monad m
13+ => ResourceT m state -- ^ resource and/or state allocation
14+ -> (state -> ResourceT m ()) -- ^ resource and/or state cleanup
15+ -> (state -> [input] -> ResourceT m (SinkResult input (Maybe output))) -- ^ push
16+ -> (state -> ResourceT m (SinkResult input output)) -- ^ close
17+ -> SinkM input m output
18+sinkM alloc cleanup push close = SinkM $ do
19+ state <- alloc
20+ return SinkData
21+ { sinkPush = \input -> do
22+ res@(SinkResult _ mout) <- push state input
23+ maybe (return ()) (const $ cleanup state) mout
24+ return res
25+ , sinkClose = do
26+ res <- close state
27+ cleanup state
28+ return res
29+ }
conduit/test/main.hs+6
@@ -42,3 +42,9 @@ main = hspecX $ do
42 bs1 <- S.readFile "conduit.cabal"
42 bs1 <- S.readFile "conduit.cabal"
43 bs2 <- S.readFile "tmp"
43 bs2 <- S.readFile "tmp"
44 bs1 @=? bs2
44 bs1 @=? bs2
45+ describe "Monad instance for Sink" $ do
46+ it "binding" $ do
47+ x <- runResourceT $ CL.fromList [1..10] C.<$$> do
48+ _ <- CL.take 5
49+ CL.fold (+) (0 :: Int)
50+ x @?= sum [6..10]