git.iko.soy/typeable/wai
on master help sign in
Sinks can avoid taking data
6b54882Michael Snoyman
conduit/Data/Conduit.hs+68−38
compare with master
unifiedsplit
@@ -11,6 +11,7 @@ module Data.Conduit
11 -- * Main types
11 -- * Main types
12 , Source (..)
12 , Source (..)
13 , Sink (..)
13 , Sink (..)
14+ , SinkInside (..)
14 , Conduit
15 , Conduit
15 -- * Connect pieces together
16 -- * Connect pieces together
16 , ($$)
17 , ($$)
@@ -48,42 +49,58 @@ data SourceInside m a = SourceInside
48 }
49 }
49newtype Source m a = Source (ResourceT m (SourceInside m a))
50newtype Source m a = Source (ResourceT m (SourceInside m a))
50
51
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
52newtype Sink input m output = Sink (ResourceT m (SinkInside input m output))
56newtype Sink input m output = Sink (ResourceT m (SinkInside input m output))
53instance Monad m => Functor (Sink input m) where
57instance Monad m => Functor (Sink input m) where
54 fmap f (Sink msink) = Sink $ do
58 fmap f (Sink msink) = Sink $ do
55− sink <- msink
59+ sinkI <- msink
60+ return $ case sinkI of
56− return $ \stream -> liftM (fmap f) (sink stream)
61+ SinkData sink -> SinkData $ liftM (fmap f) . sink
62+ SinkNoData x -> SinkNoData $ f x
57instance MonadBaseControl IO m => Applicative (Sink input m) where
63instance MonadBaseControl IO m => Applicative (Sink input m) where
58− pure x =
64+ pure x = Sink $ return $ SinkNoData x
59− Sink (return (\s -> return (SinkResult (toList s) (Just x))))
60 Sink mf <*> Sink ma = Sink $ do
65 Sink mf <*> Sink ma = Sink $ do
61− istate <- liftBase $ I.newIORef Nothing
62 f <- mf
66 f <- mf
63 a <- ma
67 a <- ma
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)
64− return $ appHelper istate f a
74+ return $ appHelper istate
65−
66−toList :: Stream a -> [a]
67−toList EOF = []
68−toList (Chunks a) = a
69
75
70appHelper :: MonadBaseControl IO m
76appHelper :: MonadBaseControl IO m
71− => I.IORef (Maybe (Either (b -> c) ()))
77+ => I.IORef
72− -> SinkInside a m (b -> c)
78+ ( Either (SinkInsideData input m (a -> b)) (a -> b)
73− -> SinkInside a m b
79+ , Either (SinkInsideData input m a) a
80+ )
74− -> SinkInside a m c
81+ -> SinkInside input m b
75−appHelper istate f a stream = do
82+appHelper istate = SinkData $ \stream0 -> do
76 state <- liftBase $ I.readIORef istate
83 state <- liftBase $ I.readIORef istate
77− go state stream
84+ go state stream0
78 where
85 where
79− go Nothing stream = do
86+ go (Left f, eb) stream = do
80 SinkResult leftover mres <- f stream
87 SinkResult leftover mres <- f stream
81 case mres of
88 case mres of
82 Nothing -> return $ SinkResult leftover Nothing
89 Nothing -> return $ SinkResult leftover Nothing
83 Just res -> do
90 Just res -> do
84− let state' = Just $ Left res
91+ let state' = (Right res, eb)
85 liftBase $ I.writeIORef istate state'
92 liftBase $ I.writeIORef istate state'
86 go state' $ Chunks leftover
93 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)
87
104
88instance MonadBaseControl IO m => Monad (Sink input m) where
105instance MonadBaseControl IO m => Monad (Sink input m) where
89 return = pure
106 return = pure
@@ -93,24 +110,31 @@ instance MonadBaseControl IO m => MonadBase IO (Sink input m) where
93 liftBase = lift . liftBase
110 liftBase = lift . liftBase
94
111
95instance MonadTrans (Sink input) where
112instance MonadTrans (Sink input) where
96− lift f = Sink $ do
113+ lift f = Sink (lift (liftM SinkNoData f))
97− x <- lift f
98− return $ \s -> return (SinkResult (toList s) (Just x))
99
114
100sinkJoin :: MonadBaseControl IO m => Sink a m (Sink a m b) -> Sink a m b
115sinkJoin :: MonadBaseControl IO m => Sink a m (Sink a m b) -> Sink a m b
101−sinkJoin (Sink mmsink) = Sink $ do
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+ {-
102− istate <- liftBase $ I.newIORef Nothing
123+ istate <- liftBase $ I.newIORef Nothing
103− msink <- mmsink
104− return $ go istate msink
124+ return $ go istate msink
105 where
125 where
126+ go :: MonadBaseControl IO m
106− go :: MonadBaseControl IO m => I.IORef (Maybe (SinkInside a m b)) -> SinkInside a m (Sink a m b) -> SinkInside a m b
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)
107 go istate outer stream = do
131 go istate outer stream = do
108 state <- liftBase $ I.readIORef istate
132 state <- liftBase $ I.readIORef istate
109 case state of
133 case state of
110 Nothing -> do
134 Nothing -> do
111− SinkResult leftover minner <- outer stream
135+ SinkResult leftover minner' <- outer stream
112− case minner of
136+ case minner' of
113− Just (Sink minner) -> do
137+ Just (SinkData minner) -> do
114 inner <- minner
138 inner <- minner
115 liftBase $ I.writeIORef istate $ Just inner
139 liftBase $ I.writeIORef istate $ Just inner
116 if null leftover
140 if null leftover
@@ -118,6 +142,7 @@ sinkJoin (Sink mmsink) = Sink $ do
118 else go istate outer $ Chunks leftover
142 else go istate outer $ Chunks leftover
119 Nothing -> return $ SinkResult leftover Nothing
143 Nothing -> return $ SinkResult leftover Nothing
120 Just inner -> inner stream
144 Just inner -> inner stream
145+ -}
121
146
122type Conduit a m b = Stream a -> ResourceT m (ConduitResult a b)
147type Conduit a m b = Stream a -> ResourceT m (ConduitResult a b)
123
148
@@ -143,9 +168,12 @@ infixr 0 $$
143
168
144($$) :: MonadBaseControl IO m => Source m a -> Sink a m b -> ResourceT m b
169($$) :: MonadBaseControl IO m => Source m a -> Sink a m b -> ResourceT m b
145Source msource $$ Sink msink = do
170Source msource $$ Sink msink = do
146− sink <- msink
171+ sinkI <- msink
172+ case sinkI of
173+ SinkNoData output -> return output
174+ SinkData sink -> do
147− source <- msource
175+ source <- msource
148− connect' source sink
176+ connect' source sink
149 where
177 where
150 connect' source sink = do
178 connect' source sink = do
151 stream <- sourcePull source
179 stream <- sourcePull source
@@ -180,8 +208,10 @@ infixr 0 =$
180
208
181(=$) :: MonadBaseControl IO m => Conduit a m b -> Sink b m c -> Sink a m c
209(=$) :: MonadBaseControl IO m => Conduit a m b -> Sink b m c -> Sink a m c
182pipe =$ Sink msink = Sink $ do
210pipe =$ Sink msink = Sink $ do
183− sink <- msink
211+ sinkI <- msink
212+ case sinkI of
184− return $ \stream -> do
213+ SinkData sink -> return $ SinkData $ \stream -> do
185− ConduitResult leftover stream' <- pipe stream
214+ ConduitResult leftover stream' <- pipe stream
186− SinkResult _thisislost mres <- sink stream'
215+ SinkResult _thisislost mres <- sink stream'
187− return $ SinkResult leftover mres
216+ return $ SinkResult leftover mres
217+ SinkNoData mres -> return $ SinkNoData mres
conduit/Data/Conduit/List.hs+3−3
@@ -23,7 +23,7 @@ fold :: MonadBaseControl IO m
23 -> Sink a m b
23 -> Sink a m b
24fold f accum0 = Sink $ do
24fold f accum0 = Sink $ do
25 iaccum <- liftBase $ I.newIORef accum0
25 iaccum <- liftBase $ I.newIORef accum0
26− return $ go iaccum
26+ return $ SinkData $ go iaccum
27 where
27 where
28 go iaccum EOF = SinkResult [] . Just <$> liftBase (I.readIORef iaccum)
28 go iaccum EOF = SinkResult [] . Just <$> liftBase (I.readIORef iaccum)
29 go iaccum (Chunks cs) = do
29 go iaccum (Chunks cs) = do
@@ -45,7 +45,7 @@ take :: MonadBaseControl IO m
45 -> Sink a m [a]
45 -> Sink a m [a]
46take count0 = Sink $ do
46take count0 = Sink $ do
47 istate <- liftBase $ I.newIORef (count0, [])
47 istate <- liftBase $ I.newIORef (count0, [])
48− return $ go istate
48+ return $ SinkData $ go istate
49 where
49 where
50 go istate EOF = SinkResult [] . Just . snd <$> liftBase (I.readIORef istate)
50 go istate EOF = SinkResult [] . Just . snd <$> liftBase (I.readIORef istate)
51 go istate (Chunks cs) = do
51 go istate (Chunks cs) = do
@@ -58,7 +58,7 @@ take count0 = Sink $ do
58
58
59head :: MonadBaseControl IO m => Sink a m (Maybe a)
59head :: MonadBaseControl IO m => Sink a m (Maybe a)
60head =
60head =
61− Sink $ return $ return . go
61+ Sink $ return $ SinkData $ return . go
62 where
62 where
63 go EOF = SinkResult [] $ Just Nothing
63 go EOF = SinkResult [] $ Just Nothing
64 go (Chunks []) = SinkResult [] Nothing
64 go (Chunks []) = SinkResult [] Nothing
warp/Network/Wai/Handler/Warp.hs+1−1
@@ -474,7 +474,7 @@ sinkSocket :: T.Handle
474 -> Socket
474 -> Socket
475 -> C.Sink B.ByteString IO ()
475 -> C.Sink B.ByteString IO ()
476sinkSocket th sock =
476sinkSocket th sock =
477− C.Sink $ return go
477+ C.Sink $ return $ C.SinkData go
478 where
478 where
479 go C.EOF = liftIO (T.resume th) >> return (C.SinkResult [] (Just ()))
479 go C.EOF = liftIO (T.resume th) >> return (C.SinkResult [] (Just ()))
480 go (C.Chunks []) = return $ C.SinkResult [] Nothing
480 go (C.Chunks []) = return $ C.SinkResult [] Nothing