git.iko.soy/streamly
on eecc006 help sign in
streamly/src/Streamly/Internal/Data/Fold/Async.hs
121 lines 4.3 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Fold.Async
3-- Copyright : (c) 2019 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9module Streamly.Internal.Data.Fold.Async
10 (
11 -- * Trimming
12 takeInterval
14 -- * Splitting
15 , intervalsOf
16 )
17where
19import Control.Concurrent (threadDelay, forkIO, killThread)
20import Control.Concurrent.MVar (MVar, newMVar, swapMVar, readMVar)
21import Control.Exception (SomeException(..), catch, mask)
22import Control.Monad (void)
23import Control.Monad.IO.Class (MonadIO(..))
24import Streamly.Data.Fold (many)
25import Streamly.Internal.Data.Fold.Type (Fold(..), Step (..))
26import Streamly.Internal.Control.Concurrent (MonadAsync, withRunInIO)
27import Streamly.Internal.Data.Tuple.Strict (Tuple3'(..))
29-- $setup
30-- >>> :m
31-- >>> :set -fno-warn-deprecations
32-- >>> :set -XFlexibleContexts
33-- >>> import qualified Streamly.Prelude as Stream
34-- >>> import qualified Streamly.Data.Fold as Fold
35-- >>> import qualified Streamly.Internal.Data.Fold.Async as Fold
37-- XXX We can use asyncClock here. A parser can be used to return an input that
38-- arrives after the timeout.
39-- XXX If n is 0 return immediately in initial.
40-- XXX we should probably discard the input received after the timeout like
41-- takeEndBy_.
42--
43-- | @takeInterval n fold@ uses @fold@ to fold the input items arriving within
44-- a window of first @n@ seconds.
45--
46-- >>> input = Stream.delay 0.1 $ Stream.fromList [1..]
47-- >>> Stream.fold (Fold.takeInterval 1.0 Fold.toList) input
48-- [1,2,3,4,5,6,7,8,9,10,11]
49--
50-- Stops when @fold@ stops or when the timeout occurs. Note that the fold needs
51-- an input after the timeout to stop. For example, if no input is pushed to
52-- the fold until one hour after the timeout had occurred, then the fold will
53-- be done only after consuming that input.
54--
55-- /Pre-release/
56--
57{-# INLINE takeInterval #-}
58takeInterval :: MonadAsync m => Double -> Fold m a b -> Fold m a b
59takeInterval n (Fold step initial done) = Fold step' initial' done'
61 where
63 initial' = do
64 res <- initial
65 case res of
66 Partial s -> do
67 mv <- liftIO $ newMVar False
68 t <-
69 withRunInIO $ \run ->
70 mask $ \restore -> do
71 tid <-
72 forkIO
73 $ catch
74 (restore $ void $ run (timerThread mv))
75 (handleChildException mv)
76 run (return tid)
77 return $ Partial $ Tuple3' s mv t
78 Done b -> return $ Done b
80 step' (Tuple3' s mv t) a = do
81 val <- liftIO $ readMVar mv
82 if val
83 then do
84 res <- step s a
85 case res of
86 Partial sres -> Done <$> done sres
87 Done bres -> return $ Done bres
88 else do
89 res <- step s a
90 case res of
91 Partial fs -> return $ Partial $ Tuple3' fs mv t
92 Done b -> liftIO (killThread t) >> return (Done b)
94 done' (Tuple3' s _ _) = done s
96 timerThread mv = do
97 liftIO $ threadDelay (round $ n * 1000000)
98 -- Use IORef + CAS? instead of MVar since its a Bool?
99 liftIO $ void $ swapMVar mv True
101 handleChildException :: MVar Bool -> SomeException -> IO ()
102 handleChildException mv _ = void $ swapMVar mv True
104-- For example, we can copy and distribute a stream to multiple folds where
105-- each fold can group the input differently e.g. by one second, one minute and
106-- one hour windows respectively and fold each resulting stream of folds.
108-- | Group the input stream into windows of n second each using the first fold
109-- and then fold the resulting groups using the second fold.
110--
111-- >>> intervals = Fold.intervalsOf 0.5 Fold.toList Fold.toList
112-- >>> Stream.fold intervals $ Stream.delay 0.2 $ Stream.fromList [1..10]
113-- [[1,2,3,4],[5,6,7],[8,9,10]]
114--
115-- > intervalsOf n split = many (takeInterval n split)
116--
117-- /Pre-release/
118--
119{-# INLINE intervalsOf #-}
120intervalsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c
121intervalsOf n split = many (takeInterval n split)