@@ -221,6 +221,9 @@ module Streamly.Fold
221221 , chunksOf
222222 , sessionsOf
223223
224+ , lchunksOf
225+ , lsessionsOf
226+
224227 -- *** By Elements
225228 , splitBy
226229 , splitSuffixBy
@@ -346,9 +349,13 @@ module Streamly.Fold
346349 )
347350where
348351
349−import Control.Concurrent (threadDelay)
352+import Control.Concurrent (threadDelay, forkIO, killThread)
353+import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)
354+import Control.Exception (SomeException(..), catch, mask)
350355import Control.Monad (void)
356+import Control.Monad.Catch (throwM)
351357import Control.Monad.IO.Class (MonadIO(..))
358+import Control.Monad.Trans.Control (control)
352359import Data.Functor.Identity (Identity)
353360import Data.Heap (Entry(..))
354361import Data.Map.Strict (Map)
@@ -2167,26 +2174,104 @@ lconcatMap ::(IsStream t, Monad m) => (a -> t m b)
21672174 -> Fold m b c
21682175 -> Fold m a c
21692176lconcatMap s f1 f2 = undefined
2177+-}
21702178
21712179-- All the grouping transformation that we apply to a stream can also be
21722180-- applied to a fold input stream.
21732181
2174−-- | Group the input elements of a fold input stream by some criterion and fold
2175−-- each group using a given fold before its fed to the fold.
2182+-- | Group the input stream into groups of @n@ elements each and then fold each
2183+-- group using the provided fold function.
21762184--
2177−-- For example, we can copy and distribute a stream to multiple folds and then
2178−-- in each fold we can group the input differently e.g. by one second, one
2179−-- minute and one hour windows respectively and fold each resulting stream of
2180−-- folds.
2185+-- @
2186+--
2187+-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
2188+--
2189+-- @
2190+--
2191+lchunksOf :: Monad m => Int -> Fold m a b -> Fold m b c -> Fold m a c
2192+lchunksOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
2193+ Fold step' initial' extract'
2194+
2195+ where
2196+
2197+ initial' = (Tuple3' 0) <$> initial1 <*> initial2
2198+ step' (Tuple3' i r1 r2) a = do
2199+ if i < n
2200+ then do
2201+ res <- step1 r1 a
2202+ return $ Tuple3' (i + 1) res r2
2203+ else do
2204+ i1 <- initial1
2205+ res1 <- extract1 r1
2206+ res <- step2 r2 res1
2207+ return $ Tuple3' 0 i1 res
2208+ extract' (Tuple3' _ _ r) = extract2 r
2209+
2210+-- | Group the input stream into windows of n second each and then fold each
2211+-- group using the provided fold function.
2212+--
2213+-- For example, we can copy and distribute a stream to multiple folds where
2214+-- each fold can group the input differently e.g. by one second, one minute and
2215+-- one hour windows respectively and fold each resulting stream of folds.
21812216--
21822217-- @
21832218--
21842219-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
21852220--
21862221-- @
2187−lchunksOf :: Int -> Fold m a b -> Fold m b c -> Fold m a c
2188−lchunksOf n f1 f2 = undefined
2189−-}
2222+lsessionsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c
2223+lsessionsOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
2224+ Fold step' initial' extract'
2225+
2226+ where
2227+
2228+ -- XXX MVar may be expensive we need a cheaper synch mechanism here
2229+ initial' = do
2230+ i1 <- initial1
2231+ i2 <- initial2
2232+ mv1 <- liftIO $ newMVar i1
2233+ mv2 <- liftIO $ newMVar (Right i2)
2234+ t <- control $ \run ->
2235+ mask $ \restore -> do
2236+ tid <- forkIO $ catch (restore $ void $ run (timerThread mv1 mv2))
2237+ (handleChildException mv2)
2238+ run (return tid)
2239+ return $ Tuple3' t mv1 mv2
2240+ step' acc@(Tuple3' _ mv1 _) a = do
2241+ r1 <- liftIO $ takeMVar mv1
2242+ res <- step1 r1 a
2243+ liftIO $ putMVar mv1 res
2244+ return acc
2245+ extract' (Tuple3' tid _ mv2) = do
2246+ r2 <- liftIO $ takeMVar mv2
2247+ liftIO $ killThread tid
2248+ case r2 of
2249+ Left e -> throwM e
2250+ Right x -> extract2 x
2251+
2252+ timerThread mv1 mv2 = do
2253+ liftIO $ threadDelay (round $ n * 1000000)
2254+
2255+ r1 <- liftIO $ takeMVar mv1
2256+ i1 <- initial1
2257+ liftIO $ putMVar mv1 i1
2258+
2259+ res1 <- extract1 r1
2260+ r2 <- liftIO $ takeMVar mv2
2261+ res <- case r2 of
2262+ Left _ -> return r2
2263+ Right x -> fmap Right $ step2 x res1
2264+ liftIO $ putMVar mv2 res
2265+ timerThread mv1 mv2
2266+
2267+ handleChildException ::
2268+ MVar (Either SomeException a) -> SomeException -> IO ()
2269+ handleChildException mv2 e = do
2270+ r2 <- takeMVar mv2
2271+ let r = case r2 of
2272+ Left _ -> r2
2273+ Right _ -> Left e
2274+ putMVar mv2 r
21902275
21912276------------------------------------------------------------------------------
21922277-- Windowed classification