1{-# OPTIONS_GHC -fno-warn-unused-imports #-}
4-- Copyright : (c) 2017 Composewell Technologies
7-- Maintainer : streamly@composewell.com
9-- In this tutorial we will show how Streamly can be used for idiomatic
10-- dataflow programming. Before you go through this tutorial we recommend that
11-- you take a look at the Streamly Getting Started guide so that you are ready
12-- to try out the examples.
19 -- * Imports and Supporting Code
22 -- * Generating Streams
25 -- * Eliminating Streams
28 -- * Transforming Streams
36 -- *** Deep Serial Composition ('Serial')
39 -- *** Wide Serial Composition ('WSerial')
51 -- *** Deep Serial Nesting ('Serial')
54 -- *** Wide Serial Nesting ('WSerial')
55 -- $interleavedNesting
72 -- * Where to go next?
77import Streamly.Prelude
79import Control.Applicative
81import Control.Monad.IO.Class (MonadIO(..))
82import Control.Monad.Trans.Class (MonadTrans (lift))
84-- CAUTION: please keep setup and imports sections in sync
88-- >>> import Data.Function ((&))
89-- >>> import Streamly.Prelude ((|:), (|&))
90-- >>> import qualified Streamly.Prelude as Stream
91-- >>> import qualified Streamly.Data.Fold as Fold
96-- In most of example snippets we do not repeat the imports. Where imports are
97-- not explicitly specified use the imports shown below.
100-- >>> import Data.Function ((&))
101-- >>> import Streamly.Prelude ((|:), (|&))
102-- >>> import qualified Streamly.Prelude as Stream
103-- >>> import qualified Streamly.Data.Fold as Fold
108-- The monadic stream API offered by Streamly is very close to the Haskell
109-- "Prelude" pure lists' API, it can be considered as a natural extension of
110-- lists to monadic actions. Streamly streams provide concurrent composition
111-- and merging of streams. It can be considered as a concurrent list
114-- The basic stream type is 'Serial', it represents a sequence of IO actions,
115-- and is a 'Monad'. The 'Serial' monad is almost a drop in replacement for
116-- the 'IO' monad, IO monad is a special case of the 'Serial' monad; IO monad
117-- represents a single IO action whereas the 'Serial' monad represents a series
118-- of IO actions. The only change you need to make to go from 'IO' to 'Serial'
119-- is to use 'drain' to run the monad and to prefix the IO actions with
120-- either 'fromEffect' or 'liftIO'. If you use liftIO you can switch from 'Serial'
121-- to IO monad by simply removing the 'drain' function; no other changes
122-- are needed unless you have used some stream specific composition or
125-- Similarly, the 'Serial' type is almost a drop in replacement for pure lists,
126-- pure lists are a special case of monadic streams. If you use 'nil' in place
127-- of '[]' and '|:' in place ':' you can replace a list with a 'Serial' stream.
128-- The only difference is that the elements must be monadic type and to operate
129-- on the streams we must use the corresponding functions from
130-- "Streamly.Prelude" instead of using the base "Prelude".
134-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
135-- a monadic action at the head of the stream.
137-- >>> Stream.toList Stream.nil
140-- Stream.toList $ getLine |: getLine |: Stream.nil
145-- To create a singleton stream from a pure value use 'fromPure' or 'pure' and to
146-- create a singleton stream from a monadic action use 'fromEffect'. Note that in
147-- case of Zip applicative streams "pure" repeats the value to generate an
150-- >>> Stream.toList $ pure 1
153-- >>> Stream.toList $ Stream.fromPure 1
156-- Stream.toList $ Stream.fromEffect getLine
160-- To create a stream from pure values in a 'Foldable' container use
161-- 'fromFoldable' which is equivalent to a fold using 'cons' and 'nil':
163-- >>> Stream.toList $ Stream.fromFoldable [1..3]
166-- >>> Stream.toList $ Prelude.foldr Stream.cons Stream.nil [1..3]
169-- To create a stream from monadic actions in a 'Foldable' container just use a
170-- right fold using 'consM' and 'nil':
172-- >>> Stream.drain $ Prelude.foldr (|:) Stream.nil [putStr "Hello ", putStrLn "world!"]
175-- For more ways to construct a stream see the module "Streamly.Prelude".
179-- We have already seen 'drain' and toList to eliminate a stream in the
180-- examples above. 'drain' runs a stream discarding the results i.e. only
181-- for effects. 'toList' runs the stream and collects the results in a list.
183-- For other ways to eliminate a stream see the @Folding@ section in
184-- "Streamly.Prelude" module.
188-- Transformation over a stream is the equivalent of a @for@ loop construct in
189-- imperative paradigm. We iterate over every element in the stream and perform
190-- certain transformations for each element. Transformations may involve
191-- mapping functions over the elements, filtering elements from the stream or
192-- folding all the elements in the stream into a single value. Streamly streams
193-- are exactly like lists and you can perform all the transformations in the
194-- same way as you would on lists.
196-- Here is a simple console echo program that just echoes every input line,
201-- Stream.repeatM getLine
202-- & Stream.mapM putStrLn
206-- The following code snippet reads lines from standard input, filters blank
207-- lines, drops the first non-blank line, takes the next two, up cases them,
208-- numbers them and prints them:
210-- >>> import Data.Char (toUpper)
213-- Stream.repeatM getLine
214-- & Stream.filter (not . null)
217-- & fmap (map toUpper)
218-- & Stream.zipWith (\n s -> show n ++ " " ++ s) (Stream.fromFoldable [1..])
219-- & Stream.mapM putStrLn
226-- We can combine two streams into a single stream using semigroup composition
227-- operation '<>'. Streams can be combined in many different ways as described
228-- in the following sections, the '<>' operation behaves differently depending
229-- on the stream type in effect. The stream type and therefore the composition
230-- style can be changed at any point using one of the type combinators as
235-- The 'Semigroup' operation '<>' of the 'Serial' type combines the two streams
236-- in a /serial depth first/ manner. We use the 'fromSerial' type combinator to
237-- effect 'Serial' style of composition. We can also use an explicit 'Serial'
238-- type annotation for the stream to achieve the same effect. However, since
239-- 'Serial' is the default type unless explicitly specified by using a
240-- combinator, we can omit using an explicit combinator or type annotation for
241-- this style of composition.
243-- When two streams with multiple elements are combined in this manner, the
244-- monadic actions in the two streams are performed sequentially i.e. first all
245-- actions in the first stream are performed sequentially and then all actions
246-- in the second stream are performed sequentially. We call it
247-- /serial depth first/ as the full depth of one stream is fully traversed
248-- before we move to the next. The following example prints the sequence 1, 2,
251-- >>> stream1 = print 1 |: print 2 |: Stream.nil
252-- >>> stream2 = print 3 |: print 4 |: Stream.nil
253-- >>> Stream.drain $ stream1 <> stream2
259-- All actions in both the streams are performed serially in the same thread.
261-- The polymorphic version of the binary operation '<>' of the 'Serial' type is
262-- 'serial'. We can use 'serial' to join streams in a sequential manner
263-- irrespective of the type of stream:
265-- >>> Stream.drain $ stream1 `Stream.serial` stream2
274-- The 'Semigroup' operation '<>' of the 'WSerial' type combines the two
275-- streams in a /serial breadth first/ manner. We use the fromWSerial type
276-- combinator to effect 'WSerial' style of composition. We can also use the
277-- 'WSerial' type annotation for the stream to achieve the same effect.
279-- When two streams with multiple elements are combined in this manner, we
280-- traverse all the streams in a breadth first manner i.e. one action from each
281-- stream is performed and yielded to the resulting stream before we come back
282-- to the first stream again and so on.
283-- The following example prints the sequence 1, 3, 2, 4
285-- >>> stream1 = print 1 |: print 2 |: Stream.nil
286-- >>> stream2 = print 3 |: print 4 |: Stream.nil
287-- >>> Stream.drain $ Stream.fromWSerial $ stream1 <> stream2
293-- Even though the monadic actions of the two streams are performed in an
294-- interleaved manner they are all performed serially in the same thread.
296-- The polymorphic version of the 'WSerial' binary operation '<>' is called
297-- 'wSerial'. We can use 'wSerial' to join streams in an interleaved manner
298-- irrespective of the type, notice that we have not used the fromWSerial
299-- combinator in the following example:
301-- >>> Stream.drain $ stream1 `Stream.wSerial` stream2
307-- Note that this composition cannot be used to fold infinite number of streams
308-- since it requires preserving the state until a stream is finished.
312-- We can use 'Monoid' instances to fold a container of streams in the desired
313-- style using 'fold' or 'foldMap'. We have also provided some fold utilities
314-- to fold streams using the polymorphic combine operations:
316-- * 'concatFoldableWith' is like 'fold', it folds a 'Foldable' container of
317-- streams using the given composition operator.
318-- * 'concatMapFoldableWith' is like 'foldMap', it folds like
319-- @concatFoldableWith@ but also maps a function before folding.
320-- * 'concatForFoldableWith' is like @concatMapFoldableWith@ but the container
321-- argument comes before the function argument.
323-- All of the following are equivalent:
326-- traced = Stream.fromEffect . print
328-- Stream.drain $ foldMap traced [1..10]
329-- Stream.drain $ Stream.concatFoldableWith Stream.serial (map traced [1..10])
330-- Stream.drain $ Stream.concatMapFoldableWith Stream.serial traced [1..10]
331-- Stream.drain $ Stream.concatForFoldableWith Stream.serial [1..10] traced
337-- Till now we discussed ways to apply transformations on a stream or to merge
338-- streams together to create another stream. We mentioned earlier that
339-- transforming a stream is similar to a @for@ loop in the imperative paradigm.
340-- We will now discuss the concept of a nested composition of streams which is
341-- analogous to nested @for@ loops in the imperative paradigm. Functional
342-- programmers call this style of composition a list transformer or @ListT@.
343-- Logic programmers call it a logic monad or non-deterministic composition,
344-- but for ordinary imperative minded people like me it is easier to think in
345-- terms of good old nested @for@ loops.
349-- In functional programmer's parlance the 'Monad' instances of different
350-- 'IsStream' types implement non-determinism, exploring all possible
351-- combination of choices from both the streams. From an imperative
352-- programmer's point of view it behaves like nested loops i.e. for each
353-- element in the first stream and for each element in the second stream
354-- execute the body of the loop.
356-- The 'Monad' instances of 'Serial', 'WSerial', 'Async' and 'WAsync'
357-- stream types support different flavors of nested looping. In other words,
358-- they are all variants of list transformer. The nesting behavior of these
359-- types correspond exactly to the way they merge streams as we discussed in
360-- the previous section.
365-- The 'Monad' composition of the 'Serial' type behaves like a standard list
366-- transformer. This is the default when we do not use an explicit type
367-- combinator. However, the 'fromSerial' type combinator can be used to switch to
368-- this style of composition. We will see how this style of composition works
369-- in the following examples.
371-- Let's start with an example with a simple @for@ loop without any nesting.
372-- For simplicity of illustration we are using streams of pure values in all
373-- the examples. However, the streams could also be made of monadic actions
378-- x <- Stream.fromFoldable [3,2,1]
379-- Stream.fromEffect $ print x
385-- As we can see, the code after the @fromFoldable@ statement is run three
386-- times, once for each value of @x@ drawn from the stream. All the three
387-- iterations are serial and run in the same thread one after another. In
388-- imperative terms this is equivalent to a @for@ loop with three iterations.
390-- We can write the console echo program that we wrote earlier using the monad
396-- x <- Stream.repeatM getLine
397-- Stream.fromEffect $ putStrLn x
400-- When multiple streams are composed using this style they nest in a DFS
405-- x <- Stream.fromFoldable [1,2]
406-- y <- Stream.fromFoldable [3,4]
407-- Stream.fromEffect $ print (x, y)
414-- i.e. inner loop iterations ((1,3), (1,4)) are executed before we proceed to
415-- the next iteration of the outer loop ((2,3), (2,4)). This behaves just like
416-- nested @for@ loops in imperative programming.
418-- Notice that this is analogous to merging streams of type 'Serial' or merging
419-- streams using 'serial'.
421-- $interleavedNesting
423-- The 'Monad' composition of 'WSerial' type interleaves the iterations of
424-- outer and inner loops in a nested loop composition.
427-- Stream.drain $ Stream.fromWSerial $ do
428-- x <- Stream.fromFoldable [1,2]
429-- y <- Stream.fromFoldable [3,4]
430-- Stream.fromEffect $ print (x, y)
437-- Note that (2,3) is preferred to (1,4). This works exactly the same way as
438-- the merging of two streams using 'wSerial' works.
440-- The fromWSerial type combinator can be used to switch to this style of
441-- composition. Alternatively, a type annotation can be used to specify the
442-- type of the stream as 'WSerial'.
447-- Streamly code is usually written in a way that is agnostic of the
448-- specific monadic composition type. We use a polymorphic type with a
449-- 'IsStream' type class constraint. When running the stream we can choose the
450-- specific mode of composition. For example take a look at the following code.
453-- composed :: (Stream.IsStream t, Monad (t IO)) => t IO ()
458-- Stream.fromEffect $ print (sz, cl, sh)
460-- sizes = Stream.fromFoldable [1, 2, 3]
461-- colors = Stream.fromFoldable ["red", "green", "blue"]
462-- shapes = Stream.fromFoldable ["triangle", "square", "circle"]
465-- Now we can interpret this in whatever way we want:
468-- main = Stream.'drain' $ Stream.'fromSerial' $ composed
469-- main = Stream.'drain' $ Stream.'fromWSerial' $ composed
470-- main = Stream.'drain' $ Stream.'fromAsync' $ composed
471-- main = Stream.'drain' $ Stream.'fromWAsync' $ composed
472-- main = Stream.'drain' $ Stream.'fromParallel' $ composed
475-- As an exercise try to figure out the output of this code for each mode of
480-- 'fmap' transforms a stream by mapping a function on all elements of the
481-- stream. 'fmap' behaves in the same way for all stream types, it is always
484-- >>> (Stream.toList $ fmap show $ Stream.fromFoldable [1..10]) >>= print
485-- ["1","2","3","4","5","6","7","8","9","10"]
487-- Also see functions 'mapM' and 'sequence' from "Streamly.Prelude" module
488-- which can map actions concurrently depending on the type of the input stream.
492-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For
493-- zipping applicatives separate types 'ZipSerial' and 'ZipAsync' are
496-- The following is an example of 'Serial' applicative, it runs all iterations
499-- >>> p n = Stream.fromEffect (print n) >> return n
500-- >>> s1 = p 1 <> p 2
501-- >>> s2 = p 3 <> p 4
503-- >>> (Stream.toList $ Stream.fromSerial $ (,) <$> s1 <*> s2) >>= print
510-- [(1,3),(1,4),(2,3),(2,4)]
512-- Similarly, 'WSerial' applicative runs the iterations in an interleaved
513-- order but being serial it too takes a total of 17 seconds:
515-- >>> (Stream.toList $ Stream.fromWSerial $ (,) <$> s1 <*> s2) >>= print
522-- [(1,3),(2,3),(1,4),(2,4)]
526-- Zipping is a special transformation where the corresponding elements of two
527-- streams are combined together using a zip function producing a new stream of
528-- outputs. Two different types are provided for serial and concurrent zipping.
529-- These types provide an applicative instance that can be used to lift
530-- functions to zip the argument streams.
531-- Also see the zipping functions in the "Streamly.Prelude" module.
535-- The applicative instance of 'ZipSerial' type zips streams serially.
536-- 'fromZipSerial' type combinator can be used to switch to serial applicative
539-- >>> p n = Stream.fromEffect (print n) >> return n
540-- >>> s1 = Stream.fromSerial $ p 1 <> p 2
541-- >>> s2 = Stream.fromSerial $ p 3 <> p 4
542-- >>> (Stream.toList $ Stream.fromZipSerial $ (,) <$> s1 <*> s2) >>= print
552-- * Read the concurrent streams tutorial
553-- * See the examples in <https://github.com/composewell/streamly-examples streamly-examples> repo.
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list