523f606 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add exception handling combinators

523f606Harendra Kumar7 years ago

src/Streamly/Prelude.hs+46

compare with master

unifiedsplit

@@ -457,6 +457,14 @@ module Streamly.Prelude

457 -- , stripSuffix

458 -- , stripInfix

459

460+ -- * Exceptions

461+ , before

462+ , after

463+ , bracket

464+ , onException

465+ , finally

466+ , handle

467+

460 -- * Deprecated

468 -- * Deprecated

461 , K.once

469 , K.once

462 , each

470 , each

@@ -473,6 +481,8 @@ module Streamly.Prelude

473where

481where

474

482

475import Control.Concurrent (threadDelay)

483import Control.Concurrent (threadDelay)

484+import Control.Exception (Exception)

485+import Control.Monad.Catch (MonadCatch)

476import Control.Monad.IO.Class (MonadIO(..))

486import Control.Monad.IO.Class (MonadIO(..))

477import Control.Monad.Trans (MonadTrans(..))

487import Control.Monad.Trans (MonadTrans(..))

478import Data.Maybe (isJust, fromJust)

488import Data.Maybe (isJust, fromJust)

@@ -2211,3 +2221,39 @@ interposeBy :: (IsStream t, Monad m)

2211-- | A generalization of intersperseM to intersperse sequences.

2221-- | A generalization of intersperseM to intersperse sequences.

2212intercalate :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

2222intercalate :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

2213-}

2223-}

2224+

2225+------------------------------------------------------------------------------

2226+-- Exceptions

2227+------------------------------------------------------------------------------

2228+

2229+-- | Run a side effect before the stream yields its first element.

2230+before :: (IsStream t, Monad m) => m b -> t m a -> t m a

2231+before action xs = D.fromStreamD $ D.before action $ D.toStreamD xs

2232+

2233+-- | Run a side effect whenever the stream stops normally.

2234+after :: (IsStream t, Monad m) => m b -> t m a -> t m a

2235+after action xs = D.fromStreamD $ D.after action $ D.toStreamD xs

2236+

2237+-- | Run a side effect whenever the stream aborts due to an exception.

2238+onException :: (IsStream t, MonadCatch m) => m b -> t m a -> t m a

2239+onException action xs = D.fromStreamD $ D.onException action $ D.toStreamD xs

2240+

2241+-- | Run a side effect whenever the stream stops normally or aborts due to an

2242+-- exception.

2243+finally :: (IsStream t, MonadCatch m) => m b -> t m a -> t m a

2244+finally action xs = D.fromStreamD $ D.finally action $ D.toStreamD xs

2245+

2246+-- | Run the first action before the stream starts and remember its output,

2247+-- generate a stream using the output, run the second action using the

2248+-- remembered value as an argument whenever the stream ends normally or due to

2249+-- an exception.

2250+bracket :: (IsStream t, MonadCatch m)

2251+ => m b -> (b -> m c) -> (b -> t m a) -> t m a

2252+bracket bef aft bet = D.fromStreamD $

2253+ D.bracket bef aft (\x -> toStreamD $ bet x)

2254+

2255+-- | When evaluating a stream if an exception occurs, stream evaluation aborts

2256+-- and the specified exception handler is run with the exception as argument.

2257+handle :: (IsStream t, MonadCatch m, Exception e)

2258+ => (e -> m a) -> t m a -> t m a

2259+handle handler xs = D.fromStreamD $ D.handle handler $ D.toStreamD xs

src/Streamly/Streams/StreamD.hs+107−1

@@ -215,7 +215,7 @@ module Streamly.Streams.StreamD

215 , zipWith

216 , zipWithM

217

218− -- * Comparisions

218+ -- * Comparisons

219 , eqBy

220 , cmpBy

221

@@ -225,9 +225,19 @@ module Streamly.Streams.StreamD

225

226 -- * Transformation comprehensions

227 , the

228+

229+ -- * Exceptions

230+ , before

231+ , after

232+ , bracket

233+ , onException

234+ , finally

235+ , handle

228 )

236 )

229where

237where

230

238

239+import Control.Exception (Exception)

240+import Control.Monad.Catch (MonadCatch)

231import Control.Monad.IO.Class (MonadIO(..))

241import Control.Monad.IO.Class (MonadIO(..))

232import Control.Monad.Trans (MonadTrans(lift))

242import Control.Monad.Trans (MonadTrans(lift))

233import Data.Word (Word32)

243import Data.Word (Word32)

@@ -243,6 +253,8 @@ import Prelude

243 (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat,

253 (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat,

244 reverse)

254 reverse)

245

255

256+import qualified Control.Monad.Catch as MC

257+

246import Streamly.Mem.Array.Types (Array(..))

258import Streamly.Mem.Array.Types (Array(..))

247import Streamly.Fold.Types (Fold(..))

259import Streamly.Fold.Types (Fold(..))

248import Streamly.SVar (MonadAsync, defState, adaptState)

260import Streamly.SVar (MonadAsync, defState, adaptState)

@@ -1663,6 +1675,100 @@ mapM_ m = drain . mapM m

1663fromStreamD :: (K.IsStream t, Monad m) => Stream m a -> t m a

1675fromStreamD :: (K.IsStream t, Monad m) => Stream m a -> t m a

1664fromStreamD = K.fromStream . toStreamK

1676fromStreamD = K.fromStream . toStreamK

1665

1677

1678+------------------------------------------------------------------------------

1679+-- Exceptions

1680+------------------------------------------------------------------------------

1681+

1682+-- | Run a side effect before the stream yields its first element.

1683+before :: Monad m => m b -> Stream m a -> Stream m a

1684+before action (Stream step state) = Stream step' Nothing

1685+

1686+ where

1687+

1688+ {-# INLINE_LATE step' #-}

1689+ step' _ Nothing = action >> return (Skip (Just state))

1690+

1691+ step' gst (Just st) = do

1692+ res <- step gst st

1693+ case res of

1694+ Yield x s -> return $ Yield x (Just s)

1695+ Skip s -> return $ Skip (Just s)

1696+ Stop -> return Stop

1697+

1698+-- | Run a side effect whenever the stream stops normally.

1699+after :: Monad m => m b -> Stream m a -> Stream m a

1700+after action (Stream step state) = Stream step' state

1701+

1702+ where

1703+

1704+ {-# INLINE_LATE step' #-}

1705+ step' gst st = do

1706+ res <- step gst st

1707+ case res of

1708+ Yield x s -> return $ Yield x s

1709+ Skip s -> return $ Skip s

1710+ Stop -> action >> return Stop

1711+

1712+-- | Run a side effect whenever the stream aborts due to an exception. The

1713+-- exception is not caught simply rethrown.

1714+onException :: MonadCatch m => m b -> Stream m a -> Stream m a

1715+onException action (Stream step state) = Stream step' state

1716+

1717+ where

1718+

1719+ {-# INLINE_LATE step' #-}

1720+ step' gst st = do

1721+ res <- step gst st `MC.onException` action

1722+ case res of

1723+ Yield x s -> return $ Yield x s

1724+ Skip s -> return $ Skip s

1725+ Stop -> action >> return Stop

1726+

1727+-- XXX check if a monolithic implementation would be more performant.

1728+--

1729+-- | Run a side effect whenever the stream stops normally or aborts due to an

1730+-- exception.

1731+finally :: MonadCatch m => m b -> Stream m a -> Stream m a

1732+finally action xs = after action $ onException action xs

1733+

1734+-- | Run the first action before the stream starts and remember its output,

1735+-- generate a stream using the output, run the second action using the

1736+-- remembered value as an argument whenever the stream ends normally or due to

1737+-- an exception.

1738+bracket :: MonadCatch m => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a

1739+bracket bef aft bet = Stream step' Nothing

1740+

1741+ where

1742+

1743+ {-# INLINE_LATE step' #-}

1744+ step' _ Nothing = bef >>= \x -> return (Skip (Just (bet x, x)))

1745+

1746+ step' gst (Just (Stream step state, v)) = do

1747+ res <- step gst state `MC.onException` aft v

1748+ case res of

1749+ Yield x s -> return $ Yield x (Just (Stream step s, v))

1750+ Skip s -> return $ Skip (Just (Stream step s, v))

1751+ Stop -> aft v >> return Stop

1752+

1753+-- | When evaluating a stream if an exception occurs, stream evaluation aborts

1754+-- and the specified exception handler is run with the exception as argument.

1755+handle :: (MonadCatch m, Exception e) => (e -> m a) -> Stream m a -> Stream m a

1756+handle f (Stream step state) = Stream step' (Just state)

1757+

1758+ where

1759+

1760+ {-# INLINE_LATE step' #-}

1761+ step' gst (Just st) = do

1762+ res <- MC.try $ step gst st

1763+ case res of

1764+ Left e -> f e >>= \x -> return (Yield x Nothing)

1765+ Right r -> case r of

1766+ Yield x s -> return $ Yield x (Just s)

1767+ Skip s -> return $ Skip (Just s)

1768+ Stop -> return Stop

1769+

1770+ step' _ Nothing = return Stop

1771+

1666------------------------------------------------------------------------------

1772------------------------------------------------------------------------------

1667-- Transformation by Folding (Scans)

1773-- Transformation by Folding (Scans)

1668------------------------------------------------------------------------------

1774------------------------------------------------------------------------------