git.iko.soy/streamly
on 63407dc help sign in
streamly/src/Streamly/Internal/Data/Stream/Concurrent/Channel.hs
106 lines 3.0 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.Concurrent.Channel
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
9module Streamly.Internal.Data.Stream.Concurrent.Channel
10 (
11 -- * Channel
12 Channel (..)
13 , newChannel
14 , withChannel
15 , withChannelK
16 , fromChannel
17 , toChannel
18 , toChannelK
19 , stopChannel
20 -- quiesceChannel -- wait for running tasks but do not schedule any more.
22 -- * Configuration
23 , Config
24 , defaultConfig
26 -- ** Limits
27 , maxThreads
28 , maxBuffer
30 -- ** Rate Control
31 , Rate(..)
32 , rate
33 , avgRate
34 , minRate
35 , maxRate
36 , constRate
38 -- ** Stop behavior
39 , StopWhen (..)
40 , stopWhen
41 , getStopWhen
43 -- ** Scheduling behavior
44 , eager
45 , ordered
46 , interleaved
48 -- ** Diagnostics
49 , inspect
50 )
51where
53import Streamly.Internal.Control.Concurrent (MonadAsync)
54import Streamly.Internal.Data.Stream (Stream)
55import Streamly.Internal.Data.Stream.Concurrent.Channel.Operations
56 (fromChannel, fromChannelK, toChannel, toChannelK)
58import qualified Streamly.Internal.Data.Stream.Concurrent.Channel.Append
59 as Append
60import qualified Streamly.Internal.Data.Stream.Concurrent.Channel.Interleave
61 as Interleave
62import qualified Streamly.Internal.Data.StreamK as K
64import Streamly.Internal.Data.Stream.Concurrent.Channel.Type
65import Streamly.Internal.Data.Stream.Channel.Types
67-- | Create a new concurrent stream evaluation channel. The monad
68-- state used to run the stream actions is captured from the call site of
69-- newChannel.
70{-# INLINE newChannel #-}
71newChannel :: MonadAsync m =>
72 (Config -> Config) -> m (Channel m a)
73newChannel modifier =
74 let cfg = modifier defaultConfig
75 in if getInterleaved cfg
76 then Interleave.newChannel modifier
77 else Append.newChannel modifier
79-- | Allocate a channel and evaluate the stream using the channel and the
80-- supplied evaluator function. The evaluator is run in a worker thread.
81{-# INLINE withChannelK #-}
82withChannelK :: MonadAsync m =>
83 (Config -> Config)
84 -> K.StreamK m a
85 -> (Channel m b -> K.StreamK m a -> K.StreamK m b)
86 -> K.StreamK m b
87withChannelK modifier input evaluator = K.concatEffect action
89 where
91 action = do
92 chan <- newChannel modifier
93 toChannelK chan (evaluator chan input)
94 return $ fromChannelK chan
96-- | Allocate a channel and evaluate the stream using the channel and the
97-- supplied evaluator function. The evaluator is run in a worker thread.
98{-# INLINE withChannel #-}
99withChannel :: MonadAsync m =>
100 (Config -> Config)
101 -> Stream m a
102 -> (Channel m b -> Stream m a -> Stream m b)
103 -> Stream m b
104withChannel modifier input evaluator =
105 let f chan stream = K.fromStream $ evaluator chan (K.toStream stream)
106 in K.toStream $ withChannelK modifier (K.fromStream input) f