@@ -10,7 +10,7 @@ module Data.Conduit
1010 , Sink (..)
1111 , Conduit
1212 -- * Connect pieces together
13− , connect
13+ , ($$)
1414 , connectLeft
1515 , connectRight
1616 -- * Helper functions
@@ -24,6 +24,7 @@ import Control.Exception.Lifted (throwIO, Exception)
2424import qualified Data.IORef as I
2525import Data.Typeable (Typeable)
2626import Control.Monad (liftM)
27+import Control.Applicative (Applicative (..))
2728
2829data Stream a = EOF | Chunks [a]
2930 deriving Show
@@ -49,6 +50,63 @@ instance Monad m => Functor (Sink input m) where
4950 fmap f (Sink msink) = Sink $ do
5051 sink <- msink
5152 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
52110
53111type Conduit a m b = Stream a -> ResourceT m (ConduitResult a b)
54112
@@ -70,8 +128,8 @@ mkSource pull = Source $ do
70128 I.atomicModifyIORef buffer (\a -> (s ++ a, ()))
71129 }
72130
73−connect :: MonadBaseControl IO m => Source m a -> Sink a m b -> ResourceT m b
74−connect (Source msource) (Sink msink) = do
131+($$) :: MonadBaseControl IO m => Source m a -> Sink a m b -> ResourceT m b
132+Source msource $$ Sink msink = do
75133 sink <- msink
76134 source <- msource
77135 connect' source sink