git.iko.soy/streamly
on 626e355 help sign in
streamly/src/Streamly/Internal/Data/Fold/Async.hs
133 lines 4.7 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 final) =
67 Fold step' initial' done' final'
69 where
71 initial' = do
72 res <- initial
73 case res of
74 Partial s -> do
75 mv <- liftIO $ newMVar False
76 t <-
77 withRunInIO $ \run ->
78 mask $ \restore -> do
79 tid <-
80 forkIO
81 $ catch
82 (restore $ void $ run (timerThread mv))
83 (handleChildException mv)
84 run (return tid)
85 return $ Partial $ Tuple3' s mv t
86 Done b -> return $ Done b
88 step' (Tuple3' s mv t) a = do
89 val <- liftIO $ readMVar mv
90 if val
91 then do
92 res <- step s a
93 case res of
94 Partial sres -> Done <$> final sres
95 Done bres -> return $ Done bres
96 else do
97 res <- step s a
98 case res of
99 Partial fs -> return $ Partial $ Tuple3' fs mv t
100 Done b -> liftIO (killThread t) >> return (Done b)
102 done' (Tuple3' s _ _) = done s
104 final' (Tuple3' s _ _) = final s
106 timerThread mv = do
107 liftIO $ threadDelay (round $ n * 1000000)
108 -- Use IORef + CAS? instead of MVar since its a Bool?
109 liftIO $ void $ swapMVar mv True
111 handleChildException :: MVar Bool -> SomeException -> IO ()
112 handleChildException mv _ = void $ swapMVar mv True
114-- For example, we can copy and distribute a stream to multiple folds where
115-- each fold can group the input differently e.g. by one second, one minute and
116-- one hour windows respectively and fold each resulting stream of folds.
118-- XXX This needs to be fixed like intervalsOf in Data.Stream.Time.
120-- | Group the input stream into windows of n second each using the first fold
121-- and then fold the resulting groups using the second fold.
122--
123-- >>> intervals = Fold.intervalsOf 0.5 Fold.toList Fold.toList
124-- >>> Stream.fold intervals $ Stream.delay 0.2 $ Stream.fromList [1..10]
125-- [[1,2,3,4],[5,6,7],[8,9,10]]
126--
127-- > intervalsOf n split = many (takeInterval n split)
128--
129-- /Pre-release/
130--
131{-# INLINE intervalsOf #-}
132intervalsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c
133intervalsOf n split = many (takeInterval n split)