on bc0da07 help sign in
262 lines 8.8 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.Exception
3-- Copyright : (c) 2019 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
9module Streamly.Internal.Data.Stream.Exception
10 (
11 before
12 , after_
13 , after
14 , bracket_
15 , bracket
16 , bracket'
17 , onException
18 , finally_
19 , finally
20 , ghandle
21 , handle
22 , retry
23 )
24where
26import Control.Exception (Exception)
27import Control.Monad.Catch (MonadCatch)
28import Data.Map.Strict (Map)
29import Streamly.Internal.Control.Concurrent (MonadAsync)
30import Streamly.Internal.Data.Stream.Type (Stream, fromStreamD, toStreamD)
32import qualified Streamly.Internal.Data.Stream.StreamD as D
34-- $setup
35-- >>> :m
36-- >>> import qualified Streamly.Internal.Data.Stream as Stream
38------------------------------------------------------------------------------
39-- Exceptions
40------------------------------------------------------------------------------
42-- | Run the action @m b@ before the stream yields its first element.
43--
44-- Same as the following but more efficient due to fusion:
45--
46-- >>> before action xs = Stream.nilM action <> xs
47-- >>> before action xs = Stream.concatMap (const xs) (Stream.fromEffect action)
48--
49{-# INLINE before #-}
50before :: Monad m => m b -> Stream m a -> Stream m a
51before action xs = fromStreamD $ D.before action $ toStreamD xs
53-- | Like 'after', with following differences:
54--
55-- * action @m b@ won't run if the stream is garbage collected
56-- after partial evaluation.
57-- * Monad @m@ does not require any other constraints.
58-- * has slightly better performance than 'after'.
59--
60-- Same as the following, but with stream fusion:
61--
62-- >>> after_ action xs = xs <> Stream.nilM action
63--
64-- /Pre-release/
65--
66{-# INLINE after_ #-}
67after_ :: Monad m => m b -> Stream m a -> Stream m a
68after_ action xs = fromStreamD $ D.after_ action $ toStreamD xs
70-- | Run the action @m b@ whenever the stream @Stream m a@ stops normally, or
71-- if it is garbage collected after a partial lazy evaluation.
72--
73-- The semantics of the action @m b@ are similar to the semantics of cleanup
74-- action in 'bracket'.
75--
76-- /See also 'after_'/
77--
78{-# INLINE after #-}
79after :: MonadAsync m
80 => m b -> Stream m a -> Stream m a
81after action xs = fromStreamD $ D.after action $ toStreamD xs
83-- | Run the action @m b@ if the stream aborts due to an exception. The
84-- exception is not caught, simply rethrown.
85--
86-- /Inhibits stream fusion/
87--
88{-# INLINE onException #-}
89onException :: MonadCatch m => m b -> Stream m a -> Stream m a
90onException action xs = fromStreamD $ D.onException action $ toStreamD xs
92-- | Like 'finally' with following differences:
93--
94-- * action @m b@ won't run if the stream is garbage collected
95-- after partial evaluation.
96-- * does not require a 'MonadAsync' constraint.
97-- * has slightly better performance than 'finally'.
98--
99-- /Inhibits stream fusion/
100--
101-- /Pre-release/
102--
103{-# INLINE finally_ #-}
104finally_ :: MonadCatch m => m b -> Stream m a -> Stream m a
105finally_ action xs = fromStreamD $ D.finally_ action $ toStreamD xs
107-- | Run the action @m b@ whenever the stream @Stream m a@ stops normally,
108-- aborts due to an exception or if it is garbage collected after a partial
109-- lazy evaluation.
110--
111-- The semantics of running the action @m b@ are similar to the cleanup action
112-- semantics described in 'bracket'.
113--
114-- >>> finally release = Stream.bracket (return ()) (const release)
115--
116-- /See also 'finally_'/
117--
118-- /Inhibits stream fusion/
119--
120{-# INLINE finally #-}
121finally :: (MonadAsync m, MonadCatch m) => m b -> Stream m a -> Stream m a
122finally action xs = fromStreamD $ D.finally action $ toStreamD xs
124-- | Like 'bracket' but with following differences:
125--
126-- * alloc action @m b@ runs with async exceptions enabled
127-- * cleanup action @b -> m c@ won't run if the stream is garbage collected
128-- after partial evaluation.
129-- * does not require a 'MonadAsync' constraint.
130-- * has slightly better performance than 'bracket'.
131--
132-- /Inhibits stream fusion/
133--
134-- /Pre-release/
135--
136{-# INLINE bracket_ #-}
137bracket_ :: MonadCatch m
138 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
139bracket_ bef aft bet = fromStreamD $
140 D.bracket_ bef aft (toStreamD . bet)
142-- | Run the alloc action @m b@ with async exceptions disabled but keeping
143-- blocking operations interruptible (see 'Control.Exception.mask'). Use the
144-- output @b@ as input to @b -> Stream m a@ to generate an output stream.
145--
146-- @b@ is usually a resource under the state of monad @m@, e.g. a file
147-- handle, that requires a cleanup after use. The cleanup action @b -> m c@,
148-- runs whenever the stream ends normally, due to a sync or async exception or
149-- if it gets garbage collected after a partial lazy evaluation.
150--
151-- 'bracket' only guarantees that the cleanup action runs, and it runs with
152-- async exceptions enabled. The action must ensure that it can successfully
153-- cleanup the resource in the face of sync or async exceptions.
154--
155-- When the stream ends normally or on a sync exception, cleanup action runs
156-- immediately in the current thread context, whereas in other cases it runs in
157-- the GC context, therefore, cleanup may be delayed until the GC gets to run.
158--
159-- /See also: 'bracket_'/
160--
161-- /Inhibits stream fusion/
162--
163{-# INLINE bracket #-}
164bracket :: (MonadAsync m, MonadCatch m)
165 => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
166bracket bef aft = bracket' bef aft aft aft
168-- For a use case of this see the "streamly-process" package. It needs to kill
169-- the process in case of exception or garbage collection, but waits for the
170-- process to terminate in normal cases.
172-- | Like 'bracket' but can use separate cleanup actions depending on the mode
173-- of termination. @bracket' before onStop onGC onException action@ runs
174-- @action@ using the result of @before@. If the stream stops, @onStop@ action
175-- is executed, if the stream is abandoned @onGC@ is executed, if the stream
176-- encounters an exception @onException@ is executed.
177--
178-- /Pre-release/
179{-# INLINE bracket' #-}
180bracket' :: (MonadAsync m, MonadCatch m)
181 => m b
182 -> (b -> m c)
183 -> (b -> m d)
184 -> (b -> m e)
185 -> (b -> Stream m a)
186 -> Stream m a
187bracket' bef aft gc exc bet = fromStreamD $
188 D.bracket' bef aft exc gc (toStreamD . bet)
190-- | Like 'handle' but the exception handler is also provided with the stream
191-- that generated the exception as input. The exception handler can thus
192-- re-evaluate the stream to retry the action that failed. The exception
193-- handler can again call 'ghandle' on it to retry the action multiple times.
194--
195-- This is highly experimental. In a stream of actions we can map the stream
196-- with a retry combinator to retry each action on failure.
197--
198-- /Inhibits stream fusion/
199--
200-- /Pre-release/
201--
202{-# INLINE ghandle #-}
203ghandle :: (MonadCatch m, Exception e)
204 => (e -> Stream m a -> Stream m a) -> Stream m a -> Stream m a
205ghandle handler =
206 fromStreamD
207 . D.ghandle (\e xs -> toStreamD $ handler e (fromStreamD xs))
208 . toStreamD
210-- | When evaluating a stream if an exception occurs, stream evaluation aborts
211-- and the specified exception handler is run with the exception as argument.
212--
213-- /Inhibits stream fusion/
214--
215{-# INLINE handle #-}
216handle :: (MonadCatch m, Exception e)
217 => (e -> Stream m a) -> Stream m a -> Stream m a
218handle handler xs =
219 fromStreamD $ D.handle (toStreamD . handler) $ toStreamD xs
222-- | @retry@ takes 3 arguments
223--
224-- 1. A map @m@ whose keys are exceptions and values are the number of times to
225-- retry the action given that the exception occurs.
226--
227-- 2. A handler @han@ that decides how to handle an exception when the exception
228-- cannot be retried.
229--
230-- 3. The stream itself that we want to run this mechanism on.
231--
232-- When evaluating a stream if an exception occurs,
233--
234-- 1. The stream evaluation aborts
235--
236-- 2. The exception is looked up in @m@
237--
238-- a. If the exception exists and the mapped value is > 0 then,
239--
240-- i. The value is decreased by 1.
241--
242-- ii. The stream is resumed from where the exception was called, retrying
243-- the action.
244--
245-- b. If the exception exists and the mapped value is == 0 then the stream
246-- evaluation stops.
247--
248-- c. If the exception does not exist then we handle the exception using
249-- @han@.
250--
251-- /Internal/
252--
253{-# INLINE retry #-}
254retry :: (MonadCatch m, Exception e, Ord e)
255 => Map e Int
256 -- ^ map from exception to retry count
257 -> (e -> Stream m a)
258 -- ^ default handler for those exceptions that are not in the map
259 -> Stream m a
260 -> Stream m a
261retry emap handler inp =
262 fromStreamD $ D.retry emap (toStreamD . handler) $ toStreamD inp