git.iko.soy/streamly
on a79caed help sign in
streamly/src/Streamly/Internal/Data/Stream/Concurrent/Channel.hs
109 lines 3.4 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 module Streamly.Internal.Data.Stream.Concurrent.Channel.Type
12 , module Streamly.Internal.Data.Stream.Concurrent.Channel.Operations
13 , module Streamly.Internal.Data.Stream.Concurrent.Channel.Append
14 , module Streamly.Internal.Data.Stream.Concurrent.Channel.Interleave
15 , module Streamly.Internal.Data.Stream.Concurrent.Channel.Dispatcher
16 , module Streamly.Internal.Data.Stream.Concurrent.Channel.Consumer
18 -- * Channel
19 , Channel (..)
20 , newChannel
21 , withChannel
22 , withChannelK
23 -- quiesceChannel -- wait for running tasks but do not schedule any more.
25 -- * Configuration
26 , Config
27 , defaultConfig
29 -- ** Limits
30 , maxThreads
31 , maxBuffer
33 -- ** Rate Control
34 , Rate(..)
35 , rate
36 , avgRate
37 , minRate
38 , maxRate
39 , constRate
41 -- ** Stop behavior
42 , StopWhen (..)
43 , stopWhen
44 , getStopWhen
46 -- ** Scheduling behavior
47 , eager
48 , ordered
49 , interleaved
51 -- ** Diagnostics
52 , inspect
53 )
54where
56import Streamly.Internal.Control.Concurrent (MonadAsync)
57import Streamly.Internal.Data.Stream (Stream)
59import qualified Streamly.Internal.Data.StreamK as K
61import Streamly.Internal.Data.Channel.Types
63import Streamly.Internal.Data.Stream.Concurrent.Channel.Type
64import Streamly.Internal.Data.Stream.Concurrent.Channel.Operations
65import Streamly.Internal.Data.Stream.Concurrent.Channel.Append
66import Streamly.Internal.Data.Stream.Concurrent.Channel.Interleave
67import Streamly.Internal.Data.Stream.Concurrent.Channel.Dispatcher
68import Streamly.Internal.Data.Stream.Concurrent.Channel.Consumer
70-- | Create a new concurrent stream evaluation channel. The monad
71-- state used to run the stream actions is captured from the call site of
72-- newChannel.
73{-# INLINE newChannel #-}
74newChannel :: MonadAsync m =>
75 (Config -> Config) -> m (Channel m a)
76newChannel modifier =
77 let cfg = modifier defaultConfig
78 in if getInterleaved cfg
79 then newInterleaveChannel modifier
80 else newAppendChannel modifier
82-- | Allocate a channel and evaluate the stream using the channel and the
83-- supplied evaluator function. The evaluator is run in a worker thread.
84{-# INLINE withChannelK #-}
85withChannelK :: MonadAsync m =>
86 (Config -> Config)
87 -> K.StreamK m a
88 -> (Channel m b -> K.StreamK m a -> K.StreamK m b)
89 -> K.StreamK m b
90withChannelK modifier input evaluator = K.concatEffect action
92 where
94 action = do
95 chan <- newChannel modifier
96 toChannelK chan (evaluator chan input)
97 return $ fromChannelK chan
99-- | Allocate a channel and evaluate the stream using the channel and the
100-- supplied evaluator function. The evaluator is run in a worker thread.
101{-# INLINE withChannel #-}
102withChannel :: MonadAsync m =>
103 (Config -> Config)
104 -> Stream m a
105 -> (Channel m b -> Stream m a -> Stream m b)
106 -> Stream m b
107withChannel modifier input evaluator =
108 let f chan stream = K.fromStream $ evaluator chan (K.toStream stream)
109 in K.toStream $ withChannelK modifier (K.fromStream input) f