git.iko.soy/streamly
on 0e73809 help sign in
streamly/src/Streamly/Internal/Data/Fold/SVar.hs
79 lines 2.4 kB history raw
1{-# OPTIONS_GHC -Wno-deprecations #-}
2-- |
3-- Module : Streamly.Internal.Data.Fold.SVar
4-- Copyright : (c) 2017 Composewell Technologies
5-- License : BSD-3-Clause
6-- Maintainer : streamly@composewell.com
7-- Stability : experimental
8-- Portability : GHC
9--
10--
11module Streamly.Internal.Data.Fold.SVar
12 (
13 write
14 , writeLimited
15 )
16where
18#include "inline.hs"
20import Control.Monad (void)
21import Control.Monad.IO.Class (MonadIO(liftIO))
22import Streamly.Internal.Data.Fold (Fold(..))
24import qualified Streamly.Internal.Data.Fold as FL (Step(Done, Partial))
26import Streamly.Internal.Data.SVar
28-- | A fold to write a stream to an SVar. Unlike 'toSVar' this does not allow
29-- for concurrent evaluation of the stream, as the fold receives the input one
30-- element at a time, it just forwards the elements to the SVar. However, we
31-- can safely execute the fold in an independent thread, the SVar can act as a
32-- buffer decoupling the sender from the receiver. Also, we can have multiple
33-- folds running concurrently pusing the streams to the SVar.
34--
35{-# INLINE write #-}
36write :: MonadIO m => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
37write svar winfo = Fold step initial return final
39 where
41 initial = return $ FL.Partial ()
43 -- XXX we can have a separate fold for unlimited buffer case to avoid a
44 -- branch in the step here.
45 step () x =
46 liftIO $ do
47 decrementBufferLimit svar
48 void $ send svar (ChildYield x)
49 return $ FL.Partial ()
51 final () = liftIO $ sendStop svar winfo
53-- | Like write, but applies a yield limit.
54--
55{-# INLINE writeLimited #-}
56writeLimited :: MonadIO m
57 => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
58writeLimited svar winfo = Fold step initial (const (return ())) final
60 where
62 initial = return $ FL.Partial True
64 step True x =
65 liftIO $ do
66 yieldLimitOk <- decrementYieldLimit svar
67 if yieldLimitOk
68 then do
69 decrementBufferLimit svar
70 void $ send svar (ChildYield x)
71 return $ FL.Partial True
72 else do
73 cleanupSVarFromWorker svar
74 sendStop svar winfo
75 return $ FL.Done ()
76 step False _ = return $ FL.Done ()
78 final True = liftIO $ sendStop svar winfo
79 final False = return ()