fec4a83 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add "handle" to StreamK

fec4a83Harendra Kumar3 years ago

core/src/Streamly/Data/StreamK.hs+12

compare with master

unifiedsplit

@@ -65,6 +65,8 @@ module Streamly.Data.StreamK

65 , fromEffect

67 -- ** From Stream

68+ -- | Please note that 'Stream' type does not observe any exceptions from

69+ -- the consumer of the stream whereas 'StreamK' does.

68 , fromStream

70 , fromStream

69 , toStream

71 , toStream

@@ -122,6 +124,16 @@ module Streamly.Data.StreamK

122 -- * Buffered Operations

124 -- * Buffered Operations

123 , reverse

125 , reverse

124 , sortBy

126 , sortBy

127+

128+ -- * Exceptions

129+ -- | Please note that 'Stream' type does not observe any exceptions from

130+ -- the consumer of the stream whereas 'StreamK' does.

131+ , handle

132+

133+ -- * Resource Management

134+ -- | Please note that 'Stream' type does not observe any exceptions from

135+ -- the consumer of the stream whereas 'StreamK' does.

136+ , bracketIO

125 )

137 )

126where

138where

127

139

core/src/Streamly/Internal/Data/StreamK.hs+51−12

@@ -183,6 +183,9 @@ module Streamly.Internal.Data.StreamK

183 , append

184 , interleave

185

186+ -- * Exceptions

187+ , handle

188+

186 -- * Resource Management

189 -- * Resource Management

187 , bracketIO

190 , bracketIO

188

191

@@ -196,20 +199,19 @@ where

196#include "inline.hs"

199#include "inline.hs"

197#include "assert.hs"

200#include "assert.hs"

198

201

199−import Control.Exception (mask_)

202+import Control.Exception (mask_, Exception)

200import Control.Monad (void, join)

203import Control.Monad (void, join)

201−import Control.Monad.IO.Class (MonadIO(..))

202import Control.Monad.Catch (MonadCatch)

204import Control.Monad.Catch (MonadCatch)

205+import Control.Monad.IO.Class (MonadIO(..))

203import Data.Proxy (Proxy(..))

206import Data.Proxy (Proxy(..))

204import GHC.Types (SPEC(..))

207import GHC.Types (SPEC(..))

205import Streamly.Internal.Data.Array.Type (Array(..))

208import Streamly.Internal.Data.Array.Type (Array(..))

206import Streamly.Internal.Data.Fold.Type (Fold(..))

209import Streamly.Internal.Data.Fold.Type (Fold(..))

210+import Streamly.Internal.Data.IOFinalizer (newIOFinalizer, runIOFinalizer)

211+import Streamly.Internal.Data.Parser.ParserK.Type (ParserK)

207import Streamly.Internal.Data.Producer.Type (Producer(..))

212import Streamly.Internal.Data.Producer.Type (Producer(..))

208import Streamly.Internal.Data.SVar.Type (adaptState, defState)

213import Streamly.Internal.Data.SVar.Type (adaptState, defState)

209import Streamly.Internal.Data.Unbox (sizeOf, Unbox)

214import Streamly.Internal.Data.Unbox (sizeOf, Unbox)

210−import Streamly.Internal.Data.Parser.ParserK.Type (ParserK)

211−import Streamly.Internal.Data.IOFinalizer

212− (newIOFinalizer, runIOFinalizer, clearingIOFinalizer)

213

215

214import qualified Control.Monad.Catch as MC

216import qualified Control.Monad.Catch as MC

215import qualified Streamly.Internal.Data.Array.Type as Array

217import qualified Streamly.Internal.Data.Array.Type as Array

@@ -926,6 +928,38 @@ mapMaybe f = go

926 Nothing -> foldStream (adaptState st) yieldk single stp r

928 Nothing -> foldStream (adaptState st) yieldk single stp r

927 in foldStream (adaptState st) yieldk single stp m1

929 in foldStream (adaptState st) yieldk single stp m1

928

930

931+-------------------------------------------------------------------------------

932+-- Exception Handling

933+-------------------------------------------------------------------------------

934+

935+-- | Like 'Streamly.Data.Stream.handle' but with one significant difference,

936+-- this function observes exceptions from the consumer of the stream as well.

937+--

938+-- You can also convert 'StreamK' to 'Stream' and use exception handling from

939+-- 'Stream' module:

940+--

941+-- >>> handle f s = StreamK.fromStream $ Stream.handle (\e -> StreamK.toStream (f e)) (StreamK.toStream s)

942+--

943+{-# INLINABLE handle #-}

944+handle :: (MonadCatch m, Exception e)

945+ => (e -> m (StreamK m a)) -> StreamK m a -> StreamK m a

946+handle f stream = go stream

947+

948+ where

949+

950+ go m1 = mkStream $ \st yld sng stp ->

951+ let yieldk a r = yld a $ go r

952+ in do

953+ res <- MC.try (foldStream (adaptState st) yieldk sng stp m1)

954+ case res of

955+ Right r -> return r

956+ Left e -> do

957+ r <- f e

958+ foldStream (adaptState st) yld sng stp r

959+

960+-------------------------------------------------------------------------------

961+-- Resource Management

962+-------------------------------------------------------------------------------

929

963

930-- If we are folding the stream and we do not drain the entire stream (e.g. if

964-- If we are folding the stream and we do not drain the entire stream (e.g. if

931-- the fold terminates before the stream) then the finalizer will run on GC.

965-- the fold terminates before the stream) then the finalizer will run on GC.

@@ -935,13 +969,16 @@ mapMaybe f = go

935-- the entire chain can be invoked when the stream ends voluntarily or if

969-- the entire chain can be invoked when the stream ends voluntarily or if

936-- someone decides to abandon the stream.

970-- someone decides to abandon the stream.

937

971

938−{-

972+-- | Like 'Streamly.Data.Stream.bracketIO' but with one significant difference,

939−bracketIO :: (MonadIO m, MonadCatch m)

973+-- this function observes exceptions from the consumer of the stream as well.

940− => IO b -> (b -> IO c) -> (b -> StreamK m a) -> StreamK m a

974+-- Therefore, it cleans up the resource promptly when the consumer encounters

941−bracketIO bef aft bet =

975+-- an exception.

976+--

977+-- You can also convert 'StreamK' to 'Stream' and use resource handling from

978+-- 'Stream' module:

979+--

942− fromStream $ Stream.bracketIO bef aft (\x -> toStream $ bet x)

980+-- >>> bracketIO bef aft bet = StreamK.fromStream $ Stream.bracketIO bef aft (StreamK.toStream . bet)

943−-}

981+--

944−

945{-# INLINABLE bracketIO #-}

982{-# INLINABLE bracketIO #-}

946bracketIO :: (MonadIO m, MonadCatch m)

983bracketIO :: (MonadIO m, MonadCatch m)

947 => IO b -> (b -> IO c) -> (b -> StreamK m a) -> StreamK m a

984 => IO b -> (b -> IO c) -> (b -> StreamK m a) -> StreamK m a

@@ -957,6 +994,8 @@ bracketIO bef aft bet =

957

994

958 go ref m1 = mkStream $ \st yld sng stp ->

995 go ref m1 = mkStream $ \st yld sng stp ->

959 let

996 let

997+ -- We can discard exceptions on continuations to make it equivalent

998+ -- to StreamD, but it seems like a desirable behavior.

960 stop = liftIO (runIOFinalizer ref) >> stp

999 stop = liftIO (runIOFinalizer ref) >> stp

961 single a = liftIO (runIOFinalizer ref) >> sng a

1000 single a = liftIO (runIOFinalizer ref) >> sng a

962 yieldk a r = yld a $ go ref r

1001 yieldk a r = yld a $ go ref r