@@ -11,6 +11,7 @@ module Data.Conduit
1111 -- * Main types
1212 , Source (..)
1313 , Sink (..)
14+ , SinkInside (..)
1415 , Conduit
1516 -- * Connect pieces together
1617 , ($$)
@@ -48,42 +49,58 @@ data SourceInside m a = SourceInside
4849 }
4950newtype Source m a = Source (ResourceT m (SourceInside m a))
5051
51−type SinkInside a m b = (Stream a -> ResourceT m (SinkResult a b))
52+type SinkInsideData a m b = Stream a -> ResourceT m (SinkResult a b)
53+data SinkInside a m b =
54+ SinkData (SinkInsideData a m b)
55+ | SinkNoData b
5256newtype Sink input m output = Sink (ResourceT m (SinkInside input m output))
5357instance Monad m => Functor (Sink input m) where
5458 fmap f (Sink msink) = Sink $ do
55− sink <- msink
56− return $ \stream -> liftM (fmap f) (sink stream)
59+ sinkI <- msink
60+ return $ case sinkI of
61+ SinkData sink -> SinkData $ liftM (fmap f) . sink
62+ SinkNoData x -> SinkNoData $ f x
5763instance MonadBaseControl IO m => Applicative (Sink input m) where
58− pure x =
59− Sink (return (\s -> return (SinkResult (toList s) (Just x))))
64+ pure x = Sink $ return $ SinkNoData x
6065 Sink mf <*> Sink ma = Sink $ do
61− istate <- liftBase $ I.newIORef Nothing
6266 f <- mf
6367 a <- ma
64− return $ appHelper istate f a
65−
66−toList :: Stream a -> [a]
67−toList EOF = []
68−toList (Chunks a) = a
68+ case (f, a) of
69+ (SinkNoData f', SinkNoData a') -> return $ SinkNoData (f' a')
70+ _ -> do
71+ let toEither (SinkData x) = Left x
72+ toEither (SinkNoData x) = Right x
73+ istate <- liftBase $ I.newIORef (toEither f, toEither a)
74+ return $ appHelper istate
6975
7076appHelper :: MonadBaseControl IO m
71− => I.IORef (Maybe (Either (b -> c) ()))
72− -> SinkInside a m (b -> c)
73− -> SinkInside a m b
74− -> SinkInside a m c
75−appHelper istate f a stream = do
77+ => I.IORef
78+ ( Either (SinkInsideData input m (a -> b)) (a -> b)
79+ , Either (SinkInsideData input m a) a
80+ )
81+ -> SinkInside input m b
82+appHelper istate = SinkData $ \stream0 -> do
7683 state <- liftBase $ I.readIORef istate
77− go state stream
84+ go state stream0
7885 where
79− go Nothing stream = do
86+ go (Left f, eb) stream = do
8087 SinkResult leftover mres <- f stream
8188 case mres of
8289 Nothing -> return $ SinkResult leftover Nothing
8390 Just res -> do
84− let state' = Just $ Left res
91+ let state' = (Right res, eb)
8592 liftBase $ I.writeIORef istate state'
8693 go state' $ Chunks leftover
94+ go (f@Right{}, Left b) stream = do
95+ SinkResult leftover mres <- b stream
96+ case mres of
97+ Nothing -> return $ SinkResult leftover Nothing
98+ Just res -> do
99+ let state' = (f, Right res)
100+ liftBase $ I.writeIORef istate state'
101+ go state' $ Chunks leftover
102+ go (Right f, Right b) EOF = return $ SinkResult [] (Just $ f b)
103+ go (Right f, Right b) (Chunks s) = return $ SinkResult s (Just $ f b)
87104
88105instance MonadBaseControl IO m => Monad (Sink input m) where
89106 return = pure
@@ -93,24 +110,31 @@ instance MonadBaseControl IO m => MonadBase IO (Sink input m) where
93110 liftBase = lift . liftBase
94111
95112instance MonadTrans (Sink input) where
96− lift f = Sink $ do
97− x <- lift f
98− return $ \s -> return (SinkResult (toList s) (Just x))
113+ lift f = Sink (lift (liftM SinkNoData f))
99114
100115sinkJoin :: MonadBaseControl IO m => Sink a m (Sink a m b) -> Sink a m b
101−sinkJoin (Sink mmsink) = Sink $ do
102− istate <- liftBase $ I.newIORef Nothing
103− msink <- mmsink
104− return $ go istate msink
116+sinkJoin (Sink msinkI) = Sink $ do
117+ sinkI <- msinkI
118+ case sinkI of
119+ SinkNoData (Sink inner) -> inner
120+ SinkData _msink -> do
121+ undefined
122+ {-
123+ istate <- liftBase $ I.newIORef Nothing
124+ return $ go istate msink
105125 where
106− go :: MonadBaseControl IO m => I.IORef (Maybe (SinkInside a m b)) -> SinkInside a m (Sink a m b) -> SinkInside a m b
126+ go :: MonadBaseControl IO m
127+ => I.IORef (Maybe (SinkInside a m b))
128+ -> (Stream a -> ResourceT m (SinkResult a (Sink a m b)))
129+ -> Stream a
130+ -> ResourceT m (SinkResult a b)
107131 go istate outer stream = do
108132 state <- liftBase $ I.readIORef istate
109133 case state of
110134 Nothing -> do
111− SinkResult leftover minner <- outer stream
112− case minner of
113− Just (Sink minner) -> do
135+ SinkResult leftover minner' <- outer stream
136+ case minner' of
137+ Just (SinkData minner) -> do
114138 inner <- minner
115139 liftBase $ I.writeIORef istate $ Just inner
116140 if null leftover
@@ -118,6 +142,7 @@ sinkJoin (Sink mmsink) = Sink $ do
118142 else go istate outer $ Chunks leftover
119143 Nothing -> return $ SinkResult leftover Nothing
120144 Just inner -> inner stream
145+ -}
121146
122147type Conduit a m b = Stream a -> ResourceT m (ConduitResult a b)
123148
@@ -143,9 +168,12 @@ infixr 0 $$
143168
144169($$) :: MonadBaseControl IO m => Source m a -> Sink a m b -> ResourceT m b
145170Source msource $$ Sink msink = do
146− sink <- msink
147− source <- msource
148− connect' source sink
171+ sinkI <- msink
172+ case sinkI of
173+ SinkNoData output -> return output
174+ SinkData sink -> do
175+ source <- msource
176+ connect' source sink
149177 where
150178 connect' source sink = do
151179 stream <- sourcePull source
@@ -180,8 +208,10 @@ infixr 0 =$
180208
181209(=$) :: MonadBaseControl IO m => Conduit a m b -> Sink b m c -> Sink a m c
182210pipe =$ Sink msink = Sink $ do
183− sink <- msink
184− return $ \stream -> do
185− ConduitResult leftover stream' <- pipe stream
186− SinkResult _thisislost mres <- sink stream'
187− return $ SinkResult leftover mres
211+ sinkI <- msink
212+ case sinkI of
213+ SinkData sink -> return $ SinkData $ \stream -> do
214+ ConduitResult leftover stream' <- pipe stream
215+ SinkResult _thisislost mres <- sink stream'
216+ return $ SinkResult leftover mres
217+ SinkNoData mres -> return $ SinkNoData mres