git.iko.soy/typeable/wai
on master help sign in
Monad instance for Sink
cd6501fMichael
conduit/Data/Conduit.hs+61−3
compare with master
unifiedsplit
@@ -10,7 +10,7 @@ module Data.Conduit
10 , Sink (..)
10 , Sink (..)
11 , Conduit
11 , Conduit
12 -- * Connect pieces together
12 -- * Connect pieces together
13− , connect
13+ , ($$)
14 , connectLeft
14 , connectLeft
15 , connectRight
15 , connectRight
16 -- * Helper functions
16 -- * Helper functions
@@ -24,6 +24,7 @@ import Control.Exception.Lifted (throwIO, Exception)
24import qualified Data.IORef as I
24import qualified Data.IORef as I
25import Data.Typeable (Typeable)
25import Data.Typeable (Typeable)
26import Control.Monad (liftM)
26import Control.Monad (liftM)
27+import Control.Applicative (Applicative (..))
27
28
28data Stream a = EOF | Chunks [a]
29data Stream a = EOF | Chunks [a]
29 deriving Show
30 deriving Show
@@ -49,6 +50,63 @@ instance Monad m => Functor (Sink input m) where
49 fmap f (Sink msink) = Sink $ do
50 fmap f (Sink msink) = Sink $ do
50 sink <- msink
51 sink <- msink
51 return $ \stream -> liftM (fmap f) (sink stream)
52 return $ \stream -> liftM (fmap f) (sink stream)
53+instance MonadBaseControl IO m => Applicative (Sink input m) where
54+ pure x =
55+ Sink (return (\s -> return (SinkResult (toList s) (Just x))))
56+ where
57+ toList EOF = []
58+ toList (Chunks a) = a
59+ Sink mf <*> Sink ma = Sink $ do
60+ istate <- liftBase $ I.newIORef Nothing
61+ f <- mf
62+ a <- ma
63+ return $ appHelper istate f a
64+
65+appHelper :: MonadBaseControl IO m
66+ => I.IORef (Maybe (Either (b -> c) ()))
67+ -> SinkInside a m (b -> c)
68+ -> SinkInside a m b
69+ -> SinkInside a m c
70+appHelper istate f a stream = do
71+ state <- liftBase $ I.readIORef istate
72+ go state stream
73+ where
74+ go Nothing stream = do
75+ SinkResult leftover mres <- f stream
76+ case mres of
77+ Nothing -> return $ SinkResult leftover Nothing
78+ Just res -> do
79+ let state' = Just $ Left res
80+ liftBase $ I.writeIORef istate state'
81+ if null leftover
82+ then return $ SinkResult [] Nothing
83+ else go state' $ Chunks leftover
84+
85+instance MonadBaseControl IO m => Monad (Sink input m) where
86+ return = pure
87+ x >>= f = sinkJoin (fmap f x)
88+
89+sinkJoin :: MonadBaseControl IO m => Sink a m (Sink a m b) -> Sink a m b
90+sinkJoin (Sink mmsink) = Sink $ do
91+ istate <- liftBase $ I.newIORef Nothing
92+ msink <- mmsink
93+ return $ go istate msink
94+ where
95+ go :: MonadBaseControl IO m => I.IORef (Maybe (SinkInside a m b)) -> SinkInside a m (Sink a m b) -> SinkInside a m b
96+ go istate outer stream = do
97+ state <- liftBase $ I.readIORef istate
98+ case state of
99+ Nothing -> do
100+ SinkResult leftover minner <- outer stream
101+ case minner of
102+ Just (Sink minner) -> do
103+ inner <- minner
104+ liftBase $ I.writeIORef istate $ Just inner
105+ if null leftover
106+ then return $ SinkResult [] Nothing
107+ else go istate outer $ Chunks leftover
108+ Nothing -> return $ SinkResult leftover Nothing
109+ Just inner -> inner stream
52
110
53type Conduit a m b = Stream a -> ResourceT m (ConduitResult a b)
111type Conduit a m b = Stream a -> ResourceT m (ConduitResult a b)
54
112
@@ -70,8 +128,8 @@ mkSource pull = Source $ do
70 I.atomicModifyIORef buffer (\a -> (s ++ a, ()))
128 I.atomicModifyIORef buffer (\a -> (s ++ a, ()))
71 }
129 }
72
130
73−connect :: MonadBaseControl IO m => Source m a -> Sink a m b -> ResourceT m b
131+($$) :: MonadBaseControl IO m => Source m a -> Sink a m b -> ResourceT m b
74−connect (Source msource) (Sink msink) = do
132+Source msource $$ Sink msink = do
75 sink <- msink
133 sink <- msink
76 source <- msource
134 source <- msource
77 connect' source sink
135 connect' source sink
conduit/Data/Conduit/List.hs+8−2
@@ -2,8 +2,10 @@
2module Data.Conduit.List
2module Data.Conduit.List
3 ( fold
3 ( fold
4 , fromList
4 , fromList
5+ , take
5 ) where
6 ) where
6
7
8+import Prelude hiding (take)
7import Data.Conduit
9import Data.Conduit
8import Control.Monad.Trans.Control (MonadBaseControl)
10import Control.Monad.Trans.Control (MonadBaseControl)
9import Control.Monad.Base (liftBase)
11import Control.Monad.Base (liftBase)
@@ -43,5 +45,9 @@ take count0 = Sink $ do
43 where
45 where
44 go istate EOF = SinkResult [] . Just . snd <$> liftBase (I.readIORef istate)
46 go istate EOF = SinkResult [] . Just . snd <$> liftBase (I.readIORef istate)
45 go istate (Chunks cs) = do
47 go istate (Chunks cs) = do
46− (count, rest) <- liftBase $ I.readIORef
48+ (count, rest', b) <- liftBase $ I.atomicModifyIORef istate $ \(count, rest) ->
47− undefined
49+ let (a, b) = splitAt count cs
50+ count' = count - length a
51+ rest' = rest ++ a
52+ in ((count', rest'), (count', rest', b))
53+ return $ SinkResult b $ if count == 0 then Just rest' else Nothing
conduit/sum.hs+5−1
@@ -1,9 +1,13 @@
1+import Prelude hiding (take)
1import Data.Conduit
2import Data.Conduit
2import Data.Conduit.List
3import Data.Conduit.List
3import Control.Monad.Trans.Resource
4import Control.Monad.Trans.Resource
4
5
5main :: IO ()
6main :: IO ()
7+main = do
6−main = runResourceT (fromList [1..10] `connect` sum') >>= putStrLn
8+ runResourceT (fromList [1..10] $$ do
9+ _ <- take 5
10+ sum') >>= putStrLn
7
11
8sum' :: Sink Int IO String
12sum' :: Sink Int IO String
9sum' = fmap show $ fold (+) 0
13sum' = fmap show $ fold (+) 0