git.iko.soy/streamly
on 4e34693 help sign in
streamly/src/Streamly/Internal/Data/Fold/Async.hs
130 lines 4.6 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 (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_.
43-- XXX The foldMany doctest is important for consistency with the "many" fold,
44-- and intervalsOf. We should put it in tests.
46-- | @takeInterval n fold@ uses @fold@ to fold the input items arriving within
47-- a window of first @n@ seconds.
48--
49-- >>> input = Stream.delay 0.2 $ Stream.fromList [1..10]
50-- >>> Stream.fold (Fold.takeInterval 1.0 Fold.toList) input
51-- [1,2,3,4,5,6]
52--
53-- >>> f = Fold.takeInterval 0.5 Fold.toList
54-- >>> Stream.fold Fold.toList $ Stream.foldMany f input
55-- [[1,2,3,4],[5,6,7],[8,9,10]]
56--
57-- Stops when @fold@ stops or when the timeout occurs. Note that the fold needs
58-- an input after the timeout to stop. For example, if no input is pushed to
59-- the fold until one hour after the timeout had occurred, then the fold will
60-- be done only after consuming that input.
61--
62-- /Pre-release/
63--
64{-# INLINE takeInterval #-}
65takeInterval :: MonadAsync m => Double -> Fold m a b -> Fold m a b
66takeInterval n (Fold step initial done) = Fold step' initial' done'
68 where
70 initial' = do
71 res <- initial
72 case res of
73 Partial s -> do
74 mv <- liftIO $ newMVar False
75 t <-
76 withRunInIO $ \run ->
77 mask $ \restore -> do
78 tid <-
79 forkIO
80 $ catch
81 (restore $ void $ run (timerThread mv))
82 (handleChildException mv)
83 run (return tid)
84 return $ Partial $ Tuple3' s mv t
85 Done b -> return $ Done b
87 step' (Tuple3' s mv t) a = do
88 val <- liftIO $ readMVar mv
89 if val
90 then do
91 res <- step s a
92 case res of
93 Partial sres -> Done <$> done sres
94 Done bres -> return $ Done bres
95 else do
96 res <- step s a
97 case res of
98 Partial fs -> return $ Partial $ Tuple3' fs mv t
99 Done b -> liftIO (killThread t) >> return (Done b)
101 done' (Tuple3' s _ _) = done s
103 timerThread mv = do
104 liftIO $ threadDelay (round $ n * 1000000)
105 -- Use IORef + CAS? instead of MVar since its a Bool?
106 liftIO $ void $ swapMVar mv True
108 handleChildException :: MVar Bool -> SomeException -> IO ()
109 handleChildException mv _ = void $ swapMVar mv True
111-- For example, we can copy and distribute a stream to multiple folds where
112-- each fold can group the input differently e.g. by one second, one minute and
113-- one hour windows respectively and fold each resulting stream of folds.
115-- XXX This needs to be fixed like intervalsOf in Data.Stream.Time.
117-- | Group the input stream into windows of n second each using the first fold
118-- and then fold the resulting groups using the second fold.
119--
120-- >>> intervals = Fold.intervalsOf 0.5 Fold.toList Fold.toList
121-- >>> Stream.fold intervals $ Stream.delay 0.2 $ Stream.fromList [1..10]
122-- [[1,2,3,4],[5,6,7],[8,9,10]]
123--
124-- > intervalsOf n split = many (takeInterval n split)
125--
126-- /Pre-release/
127--
128{-# INLINE intervalsOf #-}
129intervalsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c
130intervalsOf n split = many (takeInterval n split)