1{-# OPTIONS_GHC -fno-warn-unused-imports #-}
3-- Module : Streamly.Tutorial
4-- Copyright : (c) 2017 Harendra Kumar
7-- Maintainer : harendra.kumar@gmail.com
9-- Streamly combines streaming, logic programming, concurrency and reactive
10-- programming under a single abstraction with a concise API, giving
11-- programmers a simple yet powerful tool for wide variety of applications.
13-- The 'AsyncT' monad operates on streams. Multiple streams of the same type
14-- can be summed using different sum style operators. For example, '<>' appends
15-- streams serially, '<=>' interleaves streams, '<|' combines streams in
16-- parallel and '<|>' combines streams in a fairly parallel manner. Fold
17-- utilities are provided to choose a fold style and an operator to fold a
18-- container in a desired manner producing an AsyncT stream.
20-- Streams of the same or different types can be zipped or transformed using
21-- applicative or monadic composition. The monadic composition is exactly the
22-- same as a list transformer except that it can be concurrent. It can be
23-- likened to nested loops in the imperative programming parlance. In addition
24-- to the regular monadic bind operator '>>=', other operators are provided to
25-- compose streams in alternate manners. For example, '>->' interleaves the
26-- iterations of a loop, '>>|' runs iterations in parallel and '>|>' runs all
27-- iterations in fairly parallel manner.
29-- With streamly you can write concurrent and streaming applications without
30-- being aware of threads or synchronization. No explicit thread control is
31-- needed, where applicable the concurrency rate is automatically controlled
32-- based on the demand by the consumer. However, combinators are provided to
33-- fine tune the concurrency control. Streaming and concurrency allows
34-- expressing reactive applications conveniently. See 'Streamly.Examples' for a
35-- simple SDL based FRP example.
37-- For examples and other ways to use the library see 'Streamly.Examples' as
40module Streamly.Tutorial
42 -- * AsyncT Monad Transformer
45 -- ** Generating Streams
48 -- ** Eliminating Streams
51 -- * Semigroup Style Composition
54 -- ** Serial composition (Append)
57 -- ** Interleaved composition
60 -- ** Parallel composition
63 -- ** Fairly Parallel composition
66 -- ** Custom composition
69 -- * Monoid Style composition
72 -- * Transforming Streams
84 -- ** Interleaved Iterations
85 -- $interleavedNesting
87 -- ** Concurrent Iterations
90 -- ** Fairly Concurrent Iterations
91 -- $fairlyConcurrentNesting
99 -- * Concurrent Programming Examples
102 -- * Reactive Programming
105 -- * State Machine Model
111 -- * Interworking with Streaming Libraries
114 -- * Comparison with Existing Packages
120import Streamly.Prelude
121-- import Streamly.Examples
123import Control.Applicative
125import Control.Monad.IO.Class (MonadIO(..))
126import Control.Monad.Trans.Class (MonadTrans (lift))
130-- 'AsyncT' is a monad transformer, the type 'AsyncT m a' represents a stream
131-- of values of type 'a' in the underlying monad 'm'. In this section we
132-- describe how to create new values of type 'AsyncT' in the first place and
133-- how to eliminate or run a composed 'AsyncT' computation. Later sections
134-- discuss how to compose existing AsyncT streams or computations to create
135-- composite AsyncT streams.
139-- Pure values can be placed into the stream type using 'return' or 'pure'.
140-- Effects in the IO monad can be lifted to the stream type using the 'liftIO'
141-- combinator. In a transformer stack we can lift actions from the lower monad
142-- using the 'lift' combinator.
148-- liftIO $ putStrLn "Hello world!"
153-- 'runStreaming' runs composed actions in a streaming monad, lowering the type
154-- into the underlying monad and discarding the values generated by the stream;
155-- 'toList' collects the resulting stream in a list. 'uncons' can be used
156-- to deconstruct a stream into head and tail. Before you run a stream you need
157-- to express the type of the stream by using one of the streaming type
158-- combinators ('serially', 'asyncly', 'parallely' etc.).
164-- runStreaming $ serially $ liftIO $ putStrLn "Hello world!"
165-- xs \<- toList $ serially $ return 1 <> return 2
170-- AsyncT streams of the same type can be combined into a composite stream in
171-- many different ways using semigroup style binary composition operators
172-- ('<>', '<=>', '<|', '<|>', 'mplus').
173-- We will use the following utility functions to illustrate these operators in
174-- the following sections. The @traced@ function prints the 'threadId' of the
175-- thread in which the computation runs and the @timer@ function adds a delay
176-- specified by the argument in seconds and then prints the argument.
180-- import Control.Concurrent
182-- traced m = liftIO (myThreadId >>= print) >> m
184-- timer n = liftIO $ do
185-- threadDelay (n * 1000000)
187-- putStrLn (show tid ++ ": Timer " ++ show n)
191-- The '<>' operator composes multiple AsyncT streams in series i.e. the next
192-- event happens after the previous one is completed. The following example
193-- prints the sequence 3, 2, 1 and takes a total of 6 seconds:
196-- main = runStreaming $ serially $ timer 3 <> timer 2 <> timer 1
199-- ThreadId 36: Timer 3
200-- ThreadId 36: Timer 2
201-- ThreadId 36: Timer 1
205-- The '<=>' operator interleaves the two computations i.e. it yields one item
206-- from the first stream and then one item from the second stream. The
207-- following example should print the sequence 1, 3, 2, 4:
210-- main = runStreaming $ serially $ (timer 1 <> timer 2) \<=> (timer 3 <> timer 4)
213-- ThreadId 36: Timer 1
214-- ThreadId 36: Timer 3
215-- ThreadId 36: Timer 2
216-- ThreadId 36: Timer 4
221-- The '<|' operator runs computations in parallel, preferring the first
222-- computation over the second. The second computation is run in parallel with
223-- the first only if the first computation is not producing enough output to
224-- keep the consumer busy otherwise the second computation is run serially
225-- after the first. We maintain a pool of minimum number of threads, for
226-- running the computations, that is big enough to match the pull rate of the
227-- consumer of the stream.
228-- Note that the left bias of the operator '<|' is suggested by its shape.
229-- It is also half cut '<|>' which is fairly parallel in contrast.
231-- The following example runs all the parallel computations in a single thread
232-- one after another, because none of them blocks. Note that if the consumer
233-- were faster than the producer this would start new threads for each
234-- computation to keep up even if none of them blocks:
237-- main = runStreaming $ serially $ traced (sqrt 9) <| traced (sqrt 16) <| traced (sqrt 25)
245-- In the following example since the first computation blocks we start the
246-- next one in a separate thread and so on:
249-- main = runStreaming $ serially $ timer 3 <| timer 2 <| timer 1
252-- ThreadId 42: Timer 1
253-- ThreadId 41: Timer 2
254-- ThreadId 40: Timer 3
257-- When we have a hierarchy of computations this operator follows DFS style:
260-- main = runStreaming $ serially $ (p 1 <| p 2) <| (p 3 <| p 4)
261-- where p = liftIO . print
270-- Note that since this operator is left biased it should not be used when the
271-- computations have timers that are relative to each other because all
272-- computations may not be started at the same time and therefore timers in all
273-- of them will not start at the same time. When relative timing among all
274-- computations is important or when we need to start all computations at once
275-- for some reason use '<|>' instead. However, '<|' is useful in situations
276-- when we want to optimally utilize the resources and we know that the
277-- computations can run in parallel but we do not care if they actually run in
278-- parallel or not, that decision is left to the scheduler. Also, note that
279-- this operator can be used to fold infinite containers in contrast to '<|>',
280-- because it does not require us to run all of them at the same time.
284-- The 'Alternative' composition operator '<|>', like '<|', runs the composed
285-- computations in parallel. However, unlike '<|' it runs all of the
286-- computations in fairly parallel manner using a round robin scheduling. Note
287-- that this should not be used on infinite containers, as it will lead to an
288-- infinite scheduling queue.
290-- The following example sends a query to three search engines and prints the
291-- name of the search engine as a response arrives:
295-- import Network.HTTP.Simple
297-- main = runStreaming $ serially $ google \<|> bing \<|> duckduckgo
299-- google = get "https://www.google.com/search?q=haskell"
300-- bing = get "https://www.bing.com/search?q=haskell"
301-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
302-- get s = liftIO (httpNoBody (parseRequest_ s) >> putStrLn (show s))
307-- The 'async' API can be used to create references to asynchronously running
308-- 'AsyncT' computations. We can then use 'uncons' to explore the
309-- streams arbitrarily and then recompose them to create a new 'AsyncT' stream.
310-- This way we can dynamically decide which stream to explore at any given time.
311-- Take an example of a merge sort of two sorted streams. We need to keep
312-- consuming items from the stream which has the lowest item in the sort order.
313-- This can be achieved using async references to streams. See
314-- 'Streamly.Examples.MergeSortedStreams'.
318-- Containers can be folded using a suitable identity ('mempty',
319-- 'empty', 'mzero') and one of the binary operators mentioned earlier. Some
320-- fold utilities are provided by the library for convenience:
322-- * 'foldWith' folds a 'Foldable' container of AsyncT computations using the
324-- * 'foldMapWith' folds like foldWith but also maps a function before folding.
325-- * 'forEachWith' is like foldMapwith but the container argument comes before
326-- the function argument.
332-- runStreaming $ serially $ liftIO $ putStrLn "Hello world!"
333-- xs \<- toList . serially $ return 1 <> return 2; print xs
334-- xs \<- toList . serially $ foldWith (<>) (map return [1..10]); print xs
335-- xs \<- toList . serially $ foldMapWith (<>) return [1..10]; print xs
336-- xs \<- toList . serially $ forEachWith (<>) [1..10] return; print xs
341-- The previous section discussed ways to compose streams without any
342-- transformation using sum style operators. In this section we will explore
343-- how to transform streams using 'Functor', 'Applicative' or 'Monad'
344-- style compositions. The applicative and monad composition of 'AsyncT' is
345-- exactly the same as a list transformer.
346-- For simplicity of illustration we are using streams of
347-- pure values in the following examples. However, the real application of
348-- AsyncT arises when these streams are generated using monadic actions. In the
349-- pure stream cases AsyncT is equivalent to the list monad.
353-- The simplest way to transform a stream is by mapping a function on
354-- all elements of the stream using 'fmap'.
360-- let nums = foldMapWith (<>) return [1..10]
361-- let strings = fmap show nums
362-- (toList $ serially $ fmap putStrLn strings) >>= sequence
367-- You can compose two or more streams using a function in 'Applicative' manner
368-- as well. In functional programmer's parlance the applicative implements
369-- non-determinism, exploring all possible combination of choices from both the
370-- streams. From an imperative programmer's point of view the applicative
371-- behaves like nested loops i.e. for each element in the first argument and
372-- for each element in the second argument apply the function. Note that the
373-- behavior of applicative instance of 'AsyncT' is exactly the same as that of
374-- a list transformer.
376-- The following example prints all combinations of size, color and shape:
379-- main = (toList . serially $ (,,) \<$> sizes \<*> colors \<*> shapes) >>= print
383-- toStream xs = foldMapWith (<>) return xs
384-- sizes = toStream [1, 2, 3]
385-- colors = toStream ["red", "green", "blue"]
386-- shapes = toStream ["triangle", "square", "circle"]
391-- The monadic composition is similar to Applicative composition and just like
392-- a list transformer. It is like nested loops in the imperative parlance. The
393-- monadic continuation is performed for each element in a stream.
395-- For example the following program repeatedly reads strings from standard
396-- input and echoes them to the standard output. This is like one infinite
401-- import Data.Semigroup (cycle1)
403-- main = runStreaming $ serially $ do
404-- str <- cycle1 (liftIO getLine)
405-- liftIO $ putStrLn str
408-- This example is equivalent to three nested loops, the first one for sizes,
409-- the second one for colors and the third one for shapes. It prints all
410-- combinations of size, color and shape.
413-- main = runStreaming $ serially $ do
417-- liftIO $ putStrLn $ show (sz, cl, sh)
421-- toStream xs = foldMapWith (<>) return xs
422-- sizes = toStream [1, 2, 3]
423-- colors = toStream ["red", "green", "blue"]
424-- shapes = toStream ["triangle", "square", "circle"]
428-- $interleavedNesting
433-- main = runStreaming $ interleaved $
437-- liftIO $ putStrLn $ show (sz, cl, sh)
441-- toStream xs = foldMapWith (<>) return xs
442-- sizes = toStream [1, 2, 3]
443-- colors = toStream ["red", "green", "blue"]
444-- shapes = toStream ["triangle", "square", "circle"]
451-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
452-- import Control.Concurrent
454-- printWithTid s = do
457-- putStrLn (show tid ++ ": " ++ s)
459-- main = runStreaming $ asyncly $ do
460-- liftIO $ hSetBuffering stdout LineBuffering
464-- liftIO $ putStrLn $ show (sz, cl, sh)
468-- toStream xs = foldMapWith (<>) return xs
469-- sizes = toStream [1, 2, 3]
470-- colors = toStream ["red", "green", "blue"]
471-- shapes = toStream ["triangle", "square", "circle"]
474-- $fairlyConcurrentNesting
478-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
479-- import Control.Concurrent
481-- printWithTid s = do
484-- putStrLn (show tid ++ ": " ++ s)
486-- main = runStreaming $ parallely $ do
487-- liftIO $ hSetBuffering stdout LineBuffering
491-- liftIO $ putStrLn $ show (sz, cl, sh)
495-- toStream xs = foldMapWith (<>) return xs
496-- sizes = toStream [1, 2, 3]
497-- colors = toStream ["red", "green", "blue"]
498-- shapes = toStream ["triangle", "square", "circle"]
505-- Applicative composition
509-- There are two ways to achieve concurrency. We can generate individual
510-- elements of a stream concurrently by folding with parallel composition
511-- operators i.e. '<|' or '<|>'. 'forEachWith' can be useful in such cases.
513-- In the following example, we square each number concurrently but then
514-- sum and print them serially:
518-- import Data.List (sum)
521-- squares \<- toList $ serially $ foldMapWith (<|) (\x -\> return $ x * x) [1..100]
522-- print $ sum squares
525-- The following example not just computes the squares concurrently but also
526-- computes the square root of their sums concurrently by using the parallel
531-- import Data.List (sum)
534-- z \<- toList $ asyncly $ foldMapWith (\<|) (\\x -> return $ x * x) [1..100] >>= \\xsq ->
535-- foldMapWith (\<|) (\\x -> return $ x * x) [1..100] >>= \\ysq ->
536-- return $ sqrt (xsq + ysq)
542-- Let us see a reactive programming example:
546-- import Control.Concurrent (threadDelay)
547-- import Control.Monad (when)
548-- import Control.Monad.State
549-- import Data.Semigroup (cycle1)
551-- data Event = Harm Int | Heal Int | Quit deriving (Show)
553-- userAction :: MonadIO m => AsyncT m Event
554-- userAction = cycle1 $ liftIO askUser
559-- "potion" -> return (Heal 10)
560-- "quit" -> return Quit
561-- _ -> putStrLn "What?" >> askUser
563-- acidRain :: MonadIO m => AsyncT m Event
564-- acidRain = cycle1 $ liftIO (threadDelay 1000000) >> return (Harm 1)
566-- game :: (MonadAsync m, MonadState Int m) => AsyncT m ()
568-- event \<- userAction \<|> acidRain
570-- Harm n -> modify $ \h -> h - n
571-- Heal n -> modify $ \h -> h + n
572-- Quit -> fail "quit"
575-- when (h <= 0) $ fail "You die!"
576-- liftIO $ putStrLn $ "Health = " ++ show h
579-- putStrLn "Your health is deteriorating due to acid rain,\
580-- \ type \"potion\" or \"quit\""
581-- runStateT (runStreaming $ serially $ game) 60
585-- State machine stuff
589-- Even though streamly covers all that is provided by the 'async' package or
590-- most of what is provided by the 'streaming', 'pipes' or 'conduit' packages,
591-- I would not say that it renders those useless. Streamly is like monad if
592-- 'async' is applicative and monads and applicatives both have their use
593-- cases. It can completely repalce 'async', the ZipAsync type is equivalent to
594-- the functionality provided by 'async'.
596-- pipes and conduit are like Arrows and streamly is like monad to them. You
597-- would use pipes and conduit when you do not need the product style
598-- composition and the implicit concurrency.
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