git.iko.soy/streamly
on efe0be3 help sign in
streamly/core/src/Streamly/Internal/Data/Pipe.hs
276 lines 8.3 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Pipe
3-- Copyright : (c) 2019 Composewell Technologies
4-- License : BSD3
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9-- There are three fundamental types in streamly. They are streams
10-- ("Streamly.Data.Stream"), pipes ("Streamly.Internal.Data.Pipe") and folds ("Streamly.Data.Fold").
11-- Streams are sources or producers of values, multiple sources can be merged
12-- into a single source but a source cannot be split into multiple stream
13-- sources. Folds are sinks or consumers, a stream can be split and
14-- distributed to multiple folds but the results cannot be merged back into a
15-- stream source again. Pipes are transformations, a stream source can be split
16-- and distributed to multiple pipes each pipe can apply its own transform on
17-- the stream and the results can be merged back into a single pipe. Pipes can
18-- be attached to a source to produce a source or they can be attached to a
19-- fold to produce a fold, or multiple pipes can be merged or zipped into a
20-- single pipe.
21--
22-- > import qualified Streamly.Internal.Data.Pipe as Pipe
24module Streamly.Internal.Data.Pipe
25 (
26 -- * Pipe Type
27 Pipe
29 -- * Pipes
30 -- ** Mapping
31 , map
32 , mapM
34 {-
35 -- ** Filtering
36 , lfilter
37 , lfilterM
38 -- , ldeleteBy
39 -- , luniq
41 {-
42 -- ** Mapping Filters
43 , lmapMaybe
44 , lmapMaybeM
46 -- ** Scanning Filters
47 , lfindIndices
48 , lelemIndices
50 -- ** Insertion
51 -- | Insertion adds more elements to the stream.
53 , linsertBy
54 , lintersperseM
56 -- ** Reordering
57 , lreverse
58 -}
60 -- * Parsing
61 -- ** Trimming
62 , ltake
63 -- , lrunFor -- time
64 , ltakeWhile
65 {-
66 , ltakeWhileM
67 , ldrop
68 , ldropWhile
69 , ldropWhileM
70 -}
72 -- ** Splitting
73 -- | Streams can be split into segments in space or in time. We use the
74 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
75 -- and the term @session@ to refer to a length in time (time window).
77 -- In imperative terms, grouped folding can be considered as a nested loop
78 -- where we loop over the stream to group elements and then loop over
79 -- individual groups to fold them to a single value that is yielded in the
80 -- output stream.
82 -- *** By Chunks
83 , chunksOf
84 , sessionsOf
86 -- *** By Elements
87 , splitBy
88 , splitSuffixBy
89 , splitSuffixBy'
90 -- , splitPrefixBy
91 , wordsBy
93 -- *** By Sequences
94 , splitOn
95 , splitSuffixOn
96 -- , splitPrefixOn
97 -- , wordsOn
99 -- Keeping the delimiters
100 , splitOn'
101 , splitSuffixOn'
102 -- , splitPrefixOn'
104 -- Splitting by multiple sequences
105 -- , splitOnAny
106 -- , splitSuffixOnAny
107 -- , splitPrefixOnAny
109 -- ** Grouping
110 , groups
111 , groupsBy
112 , groupsRollingBy
113 -}
115 -- * Composing Pipes
116 , tee
117 , zipWith
118 , compose
120 {-
121 -- * Distributing
122 -- |
123 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy
124 -- of the stream to each fold and combines the results using a function.
125 --
126 -- @
127 --
128 -- |-------Fold m a b--------|
129 -- ---stream m a---| |---m (b,c,...)
130 -- |-------Fold m a c--------|
131 -- | |
132 -- ...
133 -- @
134 --
135 -- To compute the average of numbers in a stream without going through the
136 -- stream twice:
137 --
138 -- >>> let avg = (/) <$> FL.sum <*> fmap fromIntegral FL.length
139 -- >>> FL.foldl' avg (S.enumerateFromTo 1.0 100.0)
140 -- 50.5
141 --
142 -- The 'Semigroup' and 'Monoid' instances of a distributing fold distribute
143 -- the input to both the folds and combines the outputs using Monoid or
144 -- Semigroup instances of the output types:
145 --
146 -- >>> import Data.Monoid (Sum)
147 -- >>> FL.foldl' (FL.head <> FL.last) (fmap Sum $ S.enumerateFromTo 1.0 100.0)
148 -- Just (Sum {getSum = 101.0})
149 --
150 -- The 'Num', 'Floating', and 'Fractional' instances work in the same way.
152 , tee
153 , distribute
155 -- * Partitioning
156 -- |
157 -- Direct items in the input stream to different folds using a function to
158 -- select the fold. This is useful to demultiplex the input stream.
159 -- , partitionByM
160 -- , partitionBy
161 , partition
163 -- * Demultiplexing
164 , demux
165 -- , demuxWith
166 , demux_
167 -- , demuxWith_
169 -- * Classifying
170 , classify
171 -- , classifyWith
173 -- * Unzipping
174 , unzip
175 -- These can be expressed using lmap/lmapM and unzip
176 -- , unzipWith
177 -- , unzipWithM
179 -- * Nested Folds
180 -- , concatMap
181 -- , chunksOf
182 , duplicate -- experimental
184 -- * Windowed Classification
185 -- | Split the stream into windows or chunks in space or time. Each window
186 -- can be associated with a key, all events associated with a particular
187 -- key in the window can be folded to a single result. The stream is split
188 -- into windows of specified size, the window can be terminated early if
189 -- the closing flag is specified in the input stream.
190 --
191 -- The term "chunk" is used for a space window and the term "session" is
192 -- used for a time window.
194 -- ** Tumbling Windows
195 -- | A new window starts after the previous window is finished.
196 -- , classifyChunksOf
197 , classifySessionsOf
199 -- ** Keep Alive Windows
200 -- | The window size is extended if an event arrives within the specified
201 -- window size. This can represent sessions with idle or inactive timeout.
202 -- , classifyKeepAliveChunks
203 , classifyKeepAliveSessions
205 {-
206 -- ** Sliding Windows
207 -- | A new window starts after the specified slide from the previous
208 -- window. Therefore windows can overlap.
209 , classifySlidingChunks
210 , classifySlidingSessions
211 -}
212 -- ** Sliding Window Buffers
213 -- , slidingChunkBuffer
214 -- , slidingSessionBuffer
215-}
216 )
217where
219-- import Control.Concurrent (threadDelay, forkIO, killThread)
220-- import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)
221-- import Control.Exception (SomeException(..), catch, mask)
222-- import Control.Monad (void)
223-- import Control.Monad.Catch (throwM)
224-- import Control.Monad.IO.Class (MonadIO(..))
225-- import Control.Monad.Trans (lift)
226-- import Control.Monad.Trans.Control (control)
227-- import Data.Functor.Identity (Identity)
228-- import Data.Heap (Entry(..))
229-- import Data.Map.Strict (Map)
230-- import Data.Maybe (fromJust, isJust, isNothing)
232-- import Foreign.Storable (Storable(..))
233import Prelude
234 hiding (id, filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
235 foldl, map, mapM_, sequence, all, any, sum, product, elem,
236 notElem, maximum, minimum, head, last, tail, length, null,
237 reverse, iterate, init, and, or, lookup, foldr1, (!!),
238 scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip,
239 span, splitAt, break, mapM)
241-- import qualified Data.Heap as H
242-- import qualified Data.Map.Strict as Map
243-- import qualified Prelude
245-- import Streamly.Data.Fold.Types (Fold(..))
246import Streamly.Internal.Data.Pipe.Type
247 (Pipe(..), PipeState(..), Step(..), zipWith, tee, map, compose)
248-- import Streamly.Internal.Data.Array.Type (Array)
249-- import Streamly.Internal.Data.Ring.Unboxed (Ring)
250-- import Streamly.Internal.Data.Stream (Stream)
251-- import Streamly.Internal.Data.Time.Units
252-- (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
253-- toAbsTime)
255-- import Streamly.Internal.Data.Strict
257-- import qualified Streamly.Internal.Data.Array.Type as A
258-- import qualified Streamly.Data.Stream as S
259-- import qualified Streamly.Internal.Data.Stream as D
260-- import qualified Streamly.Internal.Data.StreamK as K
261-- import qualified Streamly.Internal.Data.Stream.Common as P
263------------------------------------------------------------------------------
264-- Pipes
265------------------------------------------------------------------------------
267-- | Lift a monadic function to a 'Pipe'.
268--
269-- @since 0.7.0
270{-# INLINE mapM #-}
271mapM :: Monad m => (a -> m b) -> Pipe m a b
272mapM f = Pipe consume undefined ()
273 where
274 consume _ a = do
275 r <- f a
276 return $ Yield r (Consume ())