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 is a general computing framework based on streaming IO. The IO
10-- monad and pure lists are a special case of streamly. On one hand, streamly
11-- extends the lists of pure values to lists of monadic actions, on the other
12-- hand it extends the IO monad with concurrrent non-determinism. In simple
13-- imperative terms we can say that streamly extends the IO monad with @for@
14-- loops and nested @for@ loops with concurrency support. You can understand
15-- this analogy better once you can go through this tutorial.
17-- Streaming in general enables writing modular, composable and scalable
18-- applications with ease, and concurrency allows you to make them scale and
19-- perform well. Streamly enables writing scalable concurrent applications
20-- without being aware of threads or synchronization. No explicit thread
21-- control is needed, where applicable the concurrency rate is automatically
22-- controlled based on the demand by the consumer. However, combinators can be
23-- used to fine tune the concurrency control.
25-- Streaming and concurrency together enable expressing reactive applications
26-- conveniently. See the @CirclingSquare@ example in the examples directory for
27-- a simple SDL based FRP example. To summarize, streamly provides a unified
28-- computing framework for streaming, non-determinism and functional reactive
29-- programming in an elegant and simple API that is a natural extension of pure
30-- lists to monadic streams.
32-- In this tutorial we will go over the basic concepts and how to
33-- use the library. See the last section for further reading resources.
35module Streamly.Tutorial
40 -- * Concurrent Streams
43 -- * Combining Streams
46 -- * Imports and Supporting Code
49 -- * Generating Streams
52 -- * Generating Streams Concurrently
53 -- $generatingConcurrently
55 -- * Eliminating Streams
58 -- * Concurrent Pipeline Stages
59 -- $concurrentApplication
61 -- * Transforming Streams
64 -- * Mapping Concurrently
65 -- $concurrentTransformation
72 -- *** Deep Serial Composition ('Serial')
75 -- *** Wide Serial Composition ('WSerial')
78 -- *** Deep Ahead Composition ('Ahead')
81 -- *** Deep Asynchronous Composition ('Async')
84 -- *** Wide Asynchronous Composition ('WAsync')
87 -- *** Parallel Asynchronous Composition ('Parallel')
90 -- *** Custom composition
102 -- *** Deep Serial Nesting ('Serial')
105 -- *** Wide Serial Nesting ('WSerial')
106 -- $interleavedNesting
108 -- *** Deep Ahead Nesting ('Ahead')
111 -- *** Deep Asynchronous Nesting ('Async')
112 -- $concurrentNesting
114 -- *** Wide Asynchronous Nesting ('WAsync')
117 -- *** Parallel Asynchronous Nesting ('Parallel')
135 -- ** Parallel Zipping
138 -- * Monad transformers
139 -- $monadtransformers
141 -- * Concurrent Programming
144 -- * Reactive Programming
147 -- * Writing Concurrent Programs
153 -- * Interoperation with Streaming Libraries
156 -- * Comparison with Existing Packages
159 -- * Where to go next?
165import Streamly.Prelude
167import Control.Applicative
169import Control.Monad.IO.Class (MonadIO(..))
170import Control.Monad.Trans.Class (MonadTrans (lift))
174-- The way a list represents a sequence of pure values, a stream represents a
175-- sequence of monadic actions. The monadic stream API offered by Streamly is
176-- very close to the Haskell "Prelude" pure lists' API, it can be considered as a
177-- natural extension of lists to monadic actions. Streamly streams provide
178-- concurrent composition and merging of streams. It can be considered as a
179-- concurrent list transformer. In contrast to the "Prelude" lists, merging or
180-- appending streams of arbitrary length is scalable and inexpensive.
182-- The basic stream type is 'Serial', it represents a sequence of IO actions,
183-- and is a 'Monad'. The 'Serial' monad is almost a drop in replacement for
184-- the 'IO' monad, IO monad is a special case of the 'Serial' monad; IO monad
185-- represents a single IO action whereas the 'Serial' monad represents a series
186-- of IO actions. The only change you need to make to go from 'IO' to 'Serial'
187-- is to use 'runStream' to run the monad and to prefix the IO actions with
188-- either 'yieldM' or 'liftIO'. If you use liftIO you can switch from 'Serial'
189-- to IO monad by simply removing the 'runStream' function; no other changes
190-- are needed unless you have used some stream specific composition or
193-- Similarly, the 'Serial' type is almost a drop in replacement for pure lists,
194-- pure lists are a special case of monadic streams. If you use 'nil' in place
195-- of '[]' and '|:' in place ':' you can replace a list with a 'Serial' stream.
196-- The only difference is that the elements must be monadic type and to operate
197-- on the streams we must use the corresponding functions from
198-- "Streamly.Prelude" instead of using the base "Prelude".
202-- Streams can be generated concurrently, even infnite streams can be generated
203-- concurrently using controlled concurrency, streams can be merged
204-- concurrently, multiple stages in a streaming pipeline can run concurrently,
205-- streams can be mapped concurrently, they can be zipped concurrently, and in
206-- a monadic composition they combine like a list transformer providing
207-- concurrent non-determinism.
209-- There are three basic concurrent stream styles, 'Ahead', 'Async', and
210-- 'Parallel'. The 'Ahead' style streams are similar to 'Serial' except that
211-- they can speculatively execute multiple stream actions concurrently in
212-- advance. 'Ahead' would return exactly the same stream as 'Serial' except
213-- that it may execute the actions concurrently. The 'Async' style streams,
214-- like 'Ahead', speculatively execute multiple stream actions in advance but
215-- return the results in their finishing order rather than in the stream
216-- traversal order. 'Parallel' is like 'Async' except that it provides
217-- unbounded parallelism instead of controlled parallelism.
219-- For easy reference, we can classify the stream types based on /execution order/,
220-- /consumption order/, and /bounded or unbounded/ concurrency.
221-- Execution could be serial (i.e. synchronous) or asynchronous. In serial
222-- execution we execute the next action in the stream only after the previous
223-- one has finished executing. In asynchronous execution multiple actions in
224-- the stream can be executed asynchronously i.e. the next action can start
225-- executing even before the first one has finished. Consumption order
226-- determines the order in which the outputs generated by the composition are
227-- consumed. Consumption could be serial or asynchronous. In serial
228-- consumption, the outputs are consumed in the traversal order, in
229-- asynchronous consumption the outputs are consumed as they arrive i.e. first
230-- come first serve order.
233-- +------------+--------------+--------------+--------------+
234-- | Type | Execution | Consumption | Concurrency |
235-- +============+==============+==============+==============+
236-- | 'Serial' | Serial | Serial | None |
237-- +------------+--------------+--------------+--------------+
238-- | 'Ahead' | Asynchronous | Serial | bounded |
239-- +------------+--------------+--------------+--------------+
240-- | 'Async' | Asynchronous | Asynchronous | bounded |
241-- +------------+--------------+--------------+--------------+
242-- | 'Parallel' | Asynchronous | Asynchronous | unbounded |
243-- +------------+--------------+--------------+--------------+
246-- All these types can be freely inter-converted using type conversion
247-- combinators or type annotations without any cost, to acheive the desired
248-- composition style. To force a particular type of composition we coerce the
249-- stream type using the corresponding type adapting combinator from
250-- 'serially', 'aheadly', 'asyncly', or 'parallely'. The default stream type
251-- is inferred as 'Serial' unless you change it by using one of the combinators
252-- or using a type annotation.
256-- Streams can be combined using semigroup or monoid composition to form a
257-- composite stream. Traversal of a composition of streams could be @deep@ or
258-- @wide@. Deep goes depth first i.e. each stream is traversed fully before
259-- we traverse the next stream. Wide goes breadth first i.e. one element from
260-- each stream is traversed before coming back to the first stream again.
262-- Each stream type has a wide traversal variant prefixed by 'W'. The wide
263-- variant differs only in the Semigroup\/Monoid, Applicative\/Monad
264-- compositions of the streams.
265-- The following table summarizes the basic types and the corresponding wide
269-- +------------+-----------+
271-- +============+===========+
272-- | 'Serial' | 'WSerial' |
273-- +------------+-----------+
274-- | 'Ahead' | 'WAhead' |
275-- +------------+-----------+
276-- | 'Async' | 'WAsync' |
277-- +------------+-----------+
280-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that
281-- zip streams serially or concurrently using 'Applicative' operation. These
282-- types are not monads they are only applicatives and they do not differ in
283-- 'Semigroup' composition.
288-- When writing concurrent programs it is advised to not use the concurrent
289-- style stream combinators blindly at the top level. That might create too
290-- much concurrency where it is not even required, and can even degrade
291-- performance in some cases. In some cases it can also lead to surprising
292-- behavior because of some code that is supposed to be serial becoming
293-- concurrent. Please be aware that all concurrency capable APIs that you may
294-- have used under the scope of a concurrent stream combinator will become
295-- concurrent. For example if you have a 'repeatM' somewhere in your program
296-- and you use 'parallely' on top, the 'repeatM' becomes fully parallel,
297-- resulting into an infinite parallel execution . Instead, use the
298-- /Keep It Serial and Stupid/ principle, start with the default serial
299-- composition and enable concurrent combinators only when and where necessary.
300-- When you use a concurrent combinator you can use an explicit 'serially'
301-- combinator to suppress any unnecessary concurrency under the scope of that
306-- To represent streams in an arbitrary monad use the more general monad
307-- transformer types for example the monad transformer type corresponding to
308-- the 'Serial' type is 'SerialT'. @SerialT m a@ represents a stream of values
309-- of type 'a' in some underlying monad 'm'. For example, @SerialT IO Int@ is a
310-- stream of 'Int' in 'IO' monad. In fact, the type 'Serial' is a synonym for
313-- Similarly we have monad transformer types for other stream types as well viz.
314-- 'WSerialT', 'AsyncT', 'WAsyncT' and 'ParallelT'.
316-- To lift a value from an underlying monad in a monad transformer stack into a
317-- singleton stream use 'lift' and to lift from an IO action use 'liftIO'.
320-- > 'runStream' $ liftIO $ putStrLn "Hello world!"
322-- > 'runStream' $ lift $ putStrLn "Hello world!"
329-- We will assume the following imports in this tutorial. Go ahead, fire up a
330-- GHCi session and import these lines to start playing.
333-- > import "Streamly"
334-- > import "Streamly.Prelude" ((|:))
335-- > import qualified "Streamly.Prelude" as S
338-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
339-- a monadic action at the head of the stream.
342-- > S.'toList' S.'nil'
344-- > S.'toList' $ 'getLine' |: 'getLine' |: S.'nil'
350-- To create a singleton stream from a pure value use 'yield' or 'pure' and to
351-- create a singleton stream from a monadic action use 'yieldM'. Note that in
352-- case of Zip applicative streams "pure" repeats the value to generate an
356-- > S.'toList' $ 'pure' 1
358-- > S.'toList' $ 'yield' 1
360-- > S.'toList' $ S.'yieldM' 'getLine'
365-- To create a stream from pure values in a 'Foldable' container use
366-- 'fromFoldable' which is equivalent to a fold using 'cons' and 'nil':
369-- > S.'toList' $ S.'fromFoldable' [1..3]
371-- > S.'toList' $ 'Prelude.foldr' S.'cons' S.'nil' [1..3]
375-- To create a stream from monadic actions in a 'Foldable' container just use a
376-- right fold using 'consM' and 'nil':
379-- > 'runStream' $ 'Prelude.foldr' ('|:') S.'nil' ['putStr' "Hello ", 'putStrLn' "world!"]
383-- For more ways to construct a stream see the module "Streamly.Prelude".
385-- $generatingConcurrently
387-- Monadic construction and generation functions e.g. 'consM', 'unfoldrM',
388-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' etc work
389-- concurrently when used with appropriate stream type combinator. The pure
390-- versions of these APIs are not concurrent, however you can use the monadic
391-- versions even for pure computations by wrapping the pure value in a monad to
392-- get the concurrent generation capability where required.
394-- The following code finishes in 3 seconds (6 seconds when serial):
397-- > let p n = threadDelay (n * 1000000) >> return n
398-- > S.'toList' $ 'parallely' $ p 3 |: p 2 |: p 1 |: S.'nil'
400-- > S.'toList' $ 'aheadly' $ p 3 |: p 2 |: p 1 |: S.'nil'
403-- The following finishes in 10 seconds (100 seconds when serial):
406-- > 'runStream' $ 'asyncly' $ S.'replicateM' 10 $ p 10
412-- We have already seen 'runStream' and 'toList' to eliminate a stream in the
413-- examples above. 'runStream' runs a stream discarding the results i.e. only
414-- for effects. 'toList' runs the stream and collects the results in a list.
416-- For other ways to eliminate a stream see the @Folding@ section in
417-- "Streamly.Prelude" module.
421-- Transformation over a stream is the equivalent of a @for@ loop construct in
422-- imperative paradigm. We iterate over every element in the stream and perform
423-- certain transformations for each element. Transformations may involve
424-- mapping functions over the elements, filtering elements from the stream or
425-- folding all the elements in the stream into a single value. Streamly streams
426-- are exactly like lists and you can perform all the transformations in the
427-- same way as you would on lists.
429-- Here is a simple console echo program that just echoes every input line,
433-- > 'runStream' $ S.'repeatM' getLine & S.'mapM' putStrLn
436-- The following code snippet reads lines from standard input, filters blank
437-- lines, drops the first non-blank line, takes the next two, up cases them,
438-- numbers them and prints them:
442-- import qualified "Streamly.Prelude" as S
443-- import Data.Char (toUpper)
444-- import Data.Function ((&))
446-- main = 'runStream' $
447-- S.'repeatM' getLine
448-- & S.'filter' (not . null)
451-- & fmap (map toUpper)
452-- & S.'zipWith' (\\n s -> show n ++ " " ++ s) (S.'fromFoldable' [1..])
453-- & S.'mapM' putStrLn
456-- $concurrentTransformation
458-- Monadic transformation functions 'mapM' and 'sequence' work concurrently
459-- when used with appropriate stream type combinators. The pure versions do not
460-- work concurrently, however you can use the monadic versions even for pure
461-- computations to get the concurrent transformation capability where required.
463-- This would print a value every second (2 seconds when serial):
466-- > let p n = threadDelay (n * 1000000) >> return n
467-- > runStream $ aheadly $ S.'mapM' (\\x -> p 1 >> print x) (serially $ repeatM (p 1))
471-- $concurrentApplication
473-- The concurrent function application operators '|$' and '|&' apply a stream
474-- argument to a stream function concurrently to compose a concurrent pipeline
475-- of stream processing functions:
477-- Because both the stages run concurrently, we would see a delay of only 1
478-- second instead of 2 seconds in the following:
481-- > let p n = threadDelay (n * 1000000) >> return n
482-- > runStream $ S.'repeatM' (p 1) '|&' S.'mapM' (\\x -> p 1 >> print x)
487-- We can combine two streams into a single stream using semigroup composition
488-- operation '<>'. Streams can be combined in many different ways as described
489-- in the following sections, the '<>' operation behaves differently depending
490-- on the stream type in effect. The stream type and therefore the composition
491-- style can be changed at any point using one of the type combinators as
496-- In most of example snippets we do not repeat the imports. Where imports are
497-- not explicitly specified use the imports shown below.
501-- import "Streamly.Prelude" ((|:), nil)
502-- import qualified "Streamly.Prelude" as S
504-- import Control.Concurrent
505-- import Control.Monad (forever)
508-- To illustrate concurrent vs serial composition aspects, we will use the
509-- following @delay@ function to introduce a sleep or delay specified in
510-- seconds. After the delay it prints the number of seconds it slept.
513-- delay n = S.'yieldM' $ do
514-- threadDelay (n * 1000000)
516-- putStrLn (show tid ++ ": Delay " ++ show n)
521-- The 'Semigroup' operation '<>' of the 'Serial' type combines the two streams
522-- in a /serial depth first/ manner. We use the 'serially' type combinator to
523-- effect 'Serial' style of composition. We can also use an explicit 'Serial'
524-- type annotation for the stream to acheive the same effect. However, since
525-- 'Serial' is the default type unless explicitly specified by using a
526-- combinator, we can omit using an explicit combinator or type annotation for
527-- this style of composition.
529-- When two streams with multiple elements are combined in this manner, the
530-- monadic actions in the two streams are performed sequentially i.e. first all
531-- actions in the first stream are performed sequentially and then all actions
532-- in the second stream are performed sequentially. We call it
533-- /serial depth first/ as the full depth of one stream is fully traversed
534-- before we move to the next. The following example prints the sequence 1, 2,
538-- main = 'runStream' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
547-- All actions in both the streams are performed serially in the same thread.
548-- In the following example we can see that all actions are performed in the
549-- same thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
552-- main = 'runStream' $ delay 3 <> delay 2 <> delay 1
555-- ThreadId 36: Delay 3
556-- ThreadId 36: Delay 2
557-- ThreadId 36: Delay 1
560-- The polymorphic version of the binary operation '<>' of the 'Serial' type is
561-- 'serial'. We can use 'serial' to join streams in a sequential manner
562-- irrespective of the type of stream:
565-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`serial` (print 3 |: print 4 |: nil)
570-- The 'Semigroup' operation '<>' of the 'WSerial' type combines the two
571-- streams in a /serial breadth first/ manner. We use the 'wSerially' type
572-- combinator to effect 'WSerial' style of composition. We can also use the
573-- 'WSerial' type annotation for the stream to acheive the same effect.
575-- When two streams with multiple elements are combined in this manner, we
576-- traverse all the streams in a breadth first manner i.e. one action from each
577-- stream is peformed and yielded to the resulting stream before we come back
578-- to the first stream again and so on.
579-- The following example prints the sequence 1, 3, 2, 4
582-- main = 'runStream' . 'wSerially' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
591-- Even though the monadic actions of the two streams are performed in an
592-- interleaved manner they are all performed serially in the same thread. In
593-- the following example we can see that all actions are performed in the same
594-- thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
597-- main = 'runStream' . 'wSerially' $ delay 3 <> delay 2 <> delay 1
600-- ThreadId 36: Delay 3
601-- ThreadId 36: Delay 2
602-- ThreadId 36: Delay 1
605-- The polymorphic version of the 'WSerial' binary operation '<>' is called
606-- 'wSerial'. We can use 'wSerial' to join streams in an interleaved manner
607-- irrespective of the type, notice that we have not used the 'wSerially'
608-- combinator in the following example:
611-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`wSerial` (print 3 |: print 4 |: nil)
620-- Note that this composition cannot be used to fold infinite number of streams
621-- since it requires preserving the state until a stream is finished.
625-- The 'Semigroup' operation '<>' of the 'Ahead' type combines two streams in a
626-- /serial depth first/ manner with concurrent lookahead. We use the 'aheadly'
627-- type combinator to effect 'Ahead' style of composition. We can also use an
628-- explicit 'Ahead' type annotation for the stream to acheive the same effect.
630-- When two streams are combined in this manner, the streams are traversed in
631-- depth first manner just like 'Serial', however it can execute the next
632-- stream concurrently and keep the results ready when its turn arrives.
633-- Concurrent execution of the next stream(s) is performed if the first stream
634-- blocks or if it cannot produce output at the rate that is enough to meet the
635-- consumer demand. Multiple streams can be executed concurrently to meet the
636-- demand. The following example would print the result in a second even
637-- though each action in each stream takes one second:
641-- xs \<- 'toList' . 'aheadly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
643-- where p n = threadDelay 1000000 >> return n
649-- Each stream is constructed 'aheadly' and then both the streams are merged
650-- 'aheadly', therefore, all the actions can run concurrently but the result is
651-- presented in serial order.
653-- You can also use the polymorphic combinator 'ahead' in place of '<>' to
654-- compose any type of streams in this manner.
658-- The 'Semigroup' operation '<>' of the 'Async' type combines the two
659-- streams in a depth first manner with parallel look ahead. We use the
660-- 'asyncly' type combinator to effect 'Async' style of composition. We
661-- can also use the 'Async' type annotation for the stream type to acheive
664-- When two streams with multiple elements are combined in this manner, the
665-- streams are traversed in depth first manner just like 'Serial', however it
666-- can execute the next stream concurrently and return the results from it
667-- as they arrive i.e. the results from the next stream may be yielded even
668-- before the results from the first stream. Concurrent execution of the next
669-- stream(s) is performed if the first stream blocks or if it cannot produce
670-- output at the rate that is enough to meet the consumer demand. Multiple
671-- streams can be executed concurrently to meet the demand.
672-- In the example below each element in the stream introduces a constant delay
673-- of 1 second, however, it takes just one second to produce all the results.
674-- The results are not guaranteed to be in any particular order:
678-- xs \<- 'runStream' . 'asyncly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
680-- where p n = threadDelay 1000000 >> return n
686-- The constituent streams are also composed in 'Async' manner and the
687-- composition of streams too. We can compose the constituent streams to run
688-- serially, in that case it would take 2 seconds to produce all the results.
689-- The elements in the serial streams would be in serial order in the results:
693-- xs \<- 'runStream' . 'asyncly' $ (serially $ p 1 |: p 2 |: nil) <> (serially $ p 3 |: p 4 |: nil)
695-- where p n = threadDelay 1000000 >> return n
701-- In the following example we can see that new threads are started when a
702-- computation blocks. Notice that the output from the stream with the
703-- shortest delay is printed first. The whole computation takes @maximum of
704-- (3, 2, 1) = 3@ seconds:
707-- main = 'runStream' . 'asyncly' $ delay 3 '<>' delay 2 '<>' delay 1
710-- ThreadId 42: Delay 1
711-- ThreadId 41: Delay 2
712-- ThreadId 40: Delay 3
715-- When we have a tree of computations composed using this style, the tree is
716-- traversed in DFS style just like the 'Serial' style, the only difference is
717-- that here we can move on to executing the next stream if a stream blocks.
718-- However, we will not start new threads if we have sufficient output to
719-- saturate the consumer. This is why we call it left-biased demand driven or
720-- adaptive concurrency style, the concurrency tends to stay on the left side
721-- of the composition as long as possible. More threads are started based on
722-- the pull rate of the consumer. The following example prints an output every
723-- second as all of the actions are concurrent.
726-- main = 'runStream' . 'asyncly' $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)
735-- All the computations may even run in a single thread when more threads are
736-- not needed. As you can see, in the following example the computations are
737-- run in a single thread one after another, because none of them blocks.
738-- However, if the thread consuming the stream were faster than the producer
739-- then it would have started parallel threads for each computation to keep up
740-- even if none of them blocks:
743-- main = 'runStream' . 'asyncly' $ traced (sqrt 9) '<>' traced (sqrt 16) '<>' traced (sqrt 25)
744-- where traced m = S.'yieldM' (myThreadId >>= print) >> return m
752-- Note that the order of printing in the above examples may change due to
753-- variations in scheduling latencies for concurrent threads.
755-- The polymorphic version of the 'Async' binary operation '<>' is called
756-- 'async'. We can use 'async' to join streams in a left biased
757-- adaptively concurrent manner irrespective of the type, notice that we have
758-- not used the 'asyncly' combinator in the following example:
761-- main = 'runStream' $ delay 3 \`async` delay 2 \`async` delay 1
764-- ThreadId 42: Delay 1
765-- ThreadId 41: Delay 2
766-- ThreadId 40: Delay 3
769-- Since the concurrency provided by this operator is demand driven it cannot
770-- be used when the composed computations start timers that are relative to
771-- each other because all computations may not be started at the same time and
772-- therefore timers in all of them may not start at the same time. When
773-- relative timing among all computations is important or when we need to start
774-- all computations at once for any reason 'Parallel' style must be used
777-- 'Async' style should be preferred over 'Parallel' or 'WAsync' unless you
778-- really need those. It utilizes the resources optimally. It should be used
779-- when we know that the computations can run in parallel but we do not care if
780-- they actually run in parallel or not, that decision can be left to the
781-- scheduler based on demand. Also, note that this operator can be used to fold
782-- infinite number of streams in contrast to the 'Parallel' or 'WAsync' styles,
783-- because it does not require us to run all of them at the same time in a fair
788-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in
789-- a concurrent manner using /breadth first traversal/. We use the 'wAsyncly'
790-- type combinator to effect 'WAsync' style of composition. We can also use the
791-- 'WAsync' type annotation for the stream to achieve the same effect.
793-- When streams with multiple elements are combined in this manner, we traverse
794-- all the streams concurrently in a breadth first manner i.e. one action from
795-- each stream is performed and yielded to the resulting stream before we come
796-- back to the first stream again and so on. Even though we execute the actions
797-- in a breadth first order the outputs are consumed on a first come first
800-- In the following example we can see that outputs are produced in the breadth
801-- first traversal order but this is not guaranteed.
804-- main = 'runStream' . 'wAsyncly' $ (serially $ print 1 |: print 2 |: nil) <> (serially $ print 3 |: print 4 |: nil)
813-- The polymorphic version of the binary operation '<>' of the 'WAsync' type is
814-- 'wAsync'. We can use 'wAsync' to join streams using a breadth first
815-- concurrent traversal irrespective of the type, notice that we have not used
816-- the 'wAsyncly' combinator in the following example:
819-- main = 'runStream' $ delay 3 \`wAsync` delay 2 \`wAsync` delay 1
822-- ThreadId 42: Delay 1
823-- ThreadId 41: Delay 2
824-- ThreadId 40: Delay 3
827-- Since the concurrency provided by this style is demand driven it may not
828-- be used when the composed computations start timers that are relative to
829-- each other because all computations may not be started at the same time and
830-- therefore timers in all of them may not start at the same time. When
831-- relative timing among all computations is important or when we need to start
832-- all computations at once for any reason 'Parallel' style must be used
838-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two
839-- streams in a fairly concurrent manner with round robin scheduling. We use
840-- the 'parallely' type combinator to effect 'Parallel' style of composition.
841-- We can also use the 'Parallel' type annotation for the stream type to
842-- acheive the same effect.
844-- When two streams with multiple elements are combined in this manner, the
845-- monadic actions in both the streams are performed concurrently with a fair
846-- round robin scheduling. The outputs are yielded in the order in which the
847-- actions complete. This is pretty similar to the 'WAsync' type, the
848-- difference is that 'WAsync' is adaptive to the consumer demand and may or
849-- may not execute all actions in parallel depending on the demand, whereas
850-- 'Parallel' runs all the streams in parallel irrespective of the demand.
852-- The following example sends a query to all the three search engines in
853-- parallel and prints the name of the search engines in the order in which the
858-- import qualified Streamly.Prelude as S
859-- import Network.HTTP.Simple
861-- main = 'runStream' . 'parallely' $ google \<> bing \<> duckduckgo
863-- google = get "https://www.google.com/search?q=haskell"
864-- bing = get "https://www.bing.com/search?q=haskell"
865-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
866-- get s = S.'yieldM' (httpNoBody (parseRequest_ s) >> putStrLn (show s))
869-- The polymorphic version of the binary operation '<>' of the 'Parallel' type
870-- is 'parallel'. We can use 'parallel' to join streams in a fairly concurrent
871-- manner irrespective of the type, notice that we have not used the
872-- 'parallely' combinator in the following example:
875-- main = 'runStream' $ delay 3 \`parallel` delay 2 \`wAsync` delay 1
878-- ThreadId 42: Delay 1
879-- ThreadId 41: Delay 2
880-- ThreadId 40: Delay 3
883-- Note that this style of composition cannot be used to combine infinite
884-- number of streams, as it will lead to an infinite sized scheduling queue.
889-- The 'mkAsync' API can be used to create references to asynchronously running
890-- stream computations. We can then use 'uncons' to explore the streams
891-- arbitrarily and then recompose individual elements to create a new stream.
892-- This way we can dynamically decide which stream to explore at any given
893-- time. Take an example of a merge sort of two sorted streams. We need to
894-- keep consuming items from the stream which has the lowest item in the sort
895-- order. This can be achieved using async references to streams. See
896-- "MergeSort.hs" in the examples directory.
900-- We can use 'Monoid' instances to fold a container of streams in the desired
901-- style using 'fold' or 'foldMap'. We have also provided some fold utilities
902-- to fold streams using the polymorphic combine operations:
904-- * 'foldWith' is like 'fold', it folds a 'Foldable' container of streams
905-- using the given composition operator.
906-- * 'foldMapWith' is like 'foldMap', it folds like @foldWith@ but also maps a
907-- function before folding.
908-- * 'forEachWith' is like @foldMapwith@ but the container argument comes before
909-- the function argument.
911-- All of the following are equivalent and start ten concurrent tasks each with
912-- a delay from 1 to 10 seconds, resulting in the printing of each number every
917-- import qualified "Streamly.Prelude" as S
918-- import Control.Concurrent
921-- 'runStream' $ 'asyncly' $ foldMap delay [1..10]
922-- 'runStream' $ 'foldWith' 'async' (map delay [1..10])
923-- 'runStream' $ 'foldMapWith' 'async' delay [1..10]
924-- 'runStream' $ 'forEachWith' 'async' [1..10] delay
925-- where delay n = S.'yieldM' $ threadDelay (n * 1000000) >> print n
930-- Till now we discussed ways to apply transformations on a stream or to merge
931-- streams together to create another stream. We mentioned earlier that
932-- transforming a stream is similar to a @for@ loop in the imperative paradigm.
933-- We will now discuss the concept of a nested composition of streams which is
934-- analogous to nested @for@ loops in the imperative paradigm. Functional
935-- programmers call this style of composition a list transformer or @ListT@.
936-- Logic programmers call it a logic monad or non-deterministic composition,
937-- but for ordinary imperative minded people like me it is easier to think in
938-- terms of good old nested @for@ loops.
942-- In functional programmer's parlance the 'Monad' instances of different
943-- 'IsStream' types implement non-determinism, exploring all possible
944-- combination of choices from both the streams. From an imperative
945-- programmer's point of view it behaves like nested loops i.e. for each
946-- element in the first stream and for each element in the second stream
947-- execute the body of the loop.
949-- The 'Monad' instances of 'Serial', 'WSerial', 'Async' and 'WAsync'
950-- stream types support different flavors of nested looping. In other words,
951-- they are all variants of list transformer. The nesting behavior of these
952-- types correspond exactly to the way they merge streams as we discussed in
953-- the previous section.
958-- The 'Monad' composition of the 'Serial' type behaves like a standard list
959-- transformer. This is the default when we do not use an explicit type
960-- combinator. However, the 'serially' type combinator can be used to switch to
961-- this style of composition. We will see how this style of composition works
962-- in the following examples.
964-- Let's start with an example with a simple @for@ loop without any nesting.
965-- For simplicity of illustration we are using streams of pure values in all
966-- the examples. However, the streams could also be made of monadic actions
971-- import qualified "Streamly.Prelude" as S
973-- main = 'runStream' $ do
974-- x <- S.'fromFoldable' [3,2,1]
978-- ThreadId 30: Delay 3
979-- ThreadId 30: Delay 2
980-- ThreadId 30: Delay 1
983-- As we can see, the code after the @fromFoldable@ statement is run three
984-- times, once for each value of @x@ drawn from the stream. All the three
985-- iterations are serial and run in the same thread one after another. In
986-- imperative terms this is equivalent to a @for@ loop with three iterations.
988-- A console echo loop copying standard input to standard output can simply be
993-- import qualified "Streamly.Prelude" as S
995-- main = 'runStream' $ forever $ S.yieldM getLine >>= S.yieldM . putStrLn
998-- When multiple streams are composed using this style they nest in a DFS
999-- manner i.e. nested iterations of a loop are executed before we proceed to
1000-- the next iteration of the parent loop. This behaves just like nested @for@
1001-- loops in imperative programming.
1005-- import qualified "Streamly.Prelude" as S
1007-- main = 'runStream' $ do
1008-- x <- S.'fromFoldable' [1,2]
1009-- y <- S.'fromFoldable' [3,4]
1010-- S.'yieldM' $ putStrLn $ show (x, y)
1019-- Notice that this is analogous to merging streams of type 'Serial' or merging
1020-- streams using 'serial'.
1024-- The 'Monad' composition of 'Ahead' type behaves just like 'Serial' except
1025-- that it can speculatively perform a bounded number of next iterations of a
1026-- loop concurrently.
1028-- The 'aheadly' type combinator can be used to switch to this style of
1029-- composition. Alternatively, a type annotation can be used to specify the
1030-- type of the stream as 'Ahead'.
1034-- import "Streamly.Prelude" as S
1036-- comp = 'toList' . 'aheadly' $ do
1037-- x <- S.'fromFoldable' [3,2,1]
1038-- delay x >> return x
1040-- main = comp >> print
1043-- ThreadId 40: Delay 1
1044-- ThreadId 39: Delay 2
1045-- ThreadId 38: Delay 3
1049-- This code finishes in 3 seconds, 'Serial' would take 6 seconds. As we can
1050-- see all the three iterations are concurrent and run in different threads,
1051-- however, the results are returned in the serial order.
1053-- Concurrency is demand driven, when multiple streams are composed using this
1054-- style, the iterations are executed in a depth first manner just like
1055-- 'Serial' i.e. nested iterations are executed before we proceed to the next
1056-- outer iteration. The only difference is that we may execute multiple future
1057-- iterations concurrently and keep the results ready.
1060-- $concurrentNesting
1062-- The 'Monad' composition of 'Async' type can perform the iterations of a
1063-- loop concurrently. Concurrency is demand driven i.e. more concurrent
1064-- iterations are started only if the previous iterations are not able to
1065-- produce enough output for the consumer of the output stream. This works
1066-- exactly the same way as the merging of two streams 'asyncly' works.
1067-- This is the concurrent analogue of 'Serial' style monadic composition.
1069-- The 'asyncly' type combinator can be used to switch to this style of
1070-- composition. Alternatively, a type annotation can be used to specify the
1071-- type of the stream as 'Async'.
1075-- import "Streamly.Prelude"
1077-- main = 'runStream' . 'asyncly' $ do
1078-- x <- S.'fromFoldable' [3,2,1]
1082-- ThreadId 40: Delay 1
1083-- ThreadId 39: Delay 2
1084-- ThreadId 38: Delay 3
1087-- As we can see the code after the @fromFoldable@ statement is run three
1088-- times, once for each value of @x@. All the three iterations are concurrent
1089-- and run in different threads. The iteration with least delay finishes first.
1090-- When compared to imperative programming, this can be viewed as a @for@ loop
1091-- with three concurrent iterations.
1093-- Concurrency is demand driven just as in the case of 'async' merging.
1094-- When multiple streams are composed using this style, the iterations are
1095-- triggered in a depth first manner just like 'Serial' i.e. nested iterations are
1096-- executed before we proceed to the next iteration at higher level. However,
1097-- unlike 'Serial' more than one iterations may be started concurrently based
1098-- on the demand from the consumer of the stream.
1102-- import qualified "Streamly.Prelude" as S
1104-- main = 'runStream' . 'asyncly' $ do
1105-- x <- S.'fromFoldable' [1,2]
1106-- y <- S.'fromFoldable' [3,4]
1107-- S.'yieldM' $ putStrLn $ show (x, y)
1116-- $interleavedNesting
1118-- The 'Monad' composition of 'WSerial' type interleaves the iterations of
1119-- outer and inner loops in a nested loop composition. This works exactly the
1120-- same way as the merging of two streams in 'wSerially' fashion works. The
1121-- 'wSerially' type combinator can be used to switch to this style of
1122-- composition. Alternatively, a type annotation can be used to specify the
1123-- type of the stream as 'WSerial'.
1127-- import qualified "Streamly.Prelude" as S
1129-- main = 'runStream' . 'wSerially' $ do
1130-- x <- S.'fromFoldable' [1,2]
1131-- y <- S.'fromFoldable' [3,4]
1132-- S.yieldM $ putStrLn $ show (x, y)
1144-- Just like 'Async' the 'Monad' composition of 'WAsync' runs the iterations of
1145-- a loop concurrently. The difference is in the nested loop behavior. The
1146-- nested loops in this type are traversed and executed in a breadth first
1147-- manner rather than the depth first manner of 'Async' style.
1148-- The loop nesting works exactly the same way as the merging of streams
1149-- 'wAsyncly' works. The 'wAsyncly' type combinator can be used to switch to
1150-- this style of composition. Alternatively, a type annotation can be used to
1151-- specify the type of the stream as 'WAsync'.
1155-- import qualified "Streamly.Prelude" as S
1157-- main = 'runStream' . 'wAsyncly' $ do
1158-- x <- S.'fromFoldable' [1,2]
1159-- y <- S.'fromFoldable' [3,4]
1160-- S.'yieldM' $ putStrLn $ show (x, y)
1171-- Just like 'Async' or 'WAsync' the 'Monad' composition of 'Parallel' runs the
1172-- iterations of a loop concurrently. The difference is in the nested loop
1173-- behavior. The streams at each nest level is run fully concurrently
1174-- irrespective of the demand. The loop nesting works exactly the same way as
1175-- the merging of streams 'parallely' works. The 'parallely' type combinator
1176-- can be used to switch to this style of composition. Alternatively, a type
1177-- annotation can be used to specify the type of the stream as 'Parallel'.
1181-- import qualified "Streamly.Prelude" as S
1183-- main = 'runStream' . 'parallely' $ do
1184-- x <- S.'fromFoldable' [3,2,1]
1188-- ThreadId 40: Delay 1
1189-- ThreadId 39: Delay 2
1190-- ThreadId 38: Delay 3
1195-- Streamly code is usually written in a way that is agnostic of the
1196-- specific monadic composition type. We use a polymorphic type with a
1197-- 'IsStream' type class constraint. When running the stream we can choose the
1198-- specific mode of composition. For example take a look at the following code.
1202-- import qualified "Streamly.Prelude" as S
1204-- composed :: (IsStream t, Monad (t IO)) => t IO ()
1209-- S.'yieldM' $ putStrLn $ show (sz, cl, sh)
1213-- sizes = S.'fromFoldable' [1, 2, 3]
1214-- colors = S.'fromFoldable' ["red", "green", "blue"]
1215-- shapes = S.'fromFoldable' ["triangle", "square", "circle"]
1218-- Now we can interpret this in whatever way we want:
1221-- main = 'runStream' . 'serially' $ composed
1222-- main = 'runStream' . 'wSerially' $ composed
1223-- main = 'runStream' . 'asyncly' $ composed
1224-- main = 'runStream' . 'wAsyncly' $ composed
1225-- main = 'runStream' . 'parallely' $ composed
1228-- As an exercise try to figure out the output of this code for each mode of
1233-- 'fmap' transforms a stream by mapping a function on all elements of the
1234-- stream. 'fmap' behaves in the same way for all stream types, it is always
1239-- import qualified "Streamly.Prelude" as S
1241-- main = (S.'toList' $ fmap show $ S.'fromFoldable' [1..10]) >>= print
1244-- Also see the 'mapM' and 'sequence' functions for mapping actions, in the
1245-- "Streamly.Prelude" module.
1249-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For
1250-- zipping applicatives separate types 'ZipSerial' and 'ZipAsync' are
1253-- The following example uses the 'Serial' applicative, it runs all iterations
1254-- serially and takes a total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):
1258-- import qualified "Streamly.Prelude" as S
1259-- import Control.Concurrent
1263-- d n = delay n >> return n
1265-- main = (S.'toList' . 'serially' $ (,) \<$> s1 \<*> s2) >>= print
1268-- ThreadId 36: Delay 1
1269-- ThreadId 36: Delay 3
1270-- ThreadId 36: Delay 4
1271-- ThreadId 36: Delay 2
1272-- ThreadId 36: Delay 3
1273-- ThreadId 36: Delay 4
1274-- [(1,3),(1,4),(2,3),(2,4)]
1277-- Similalrly 'WSerial' applicative runs the iterations in an interleaved
1278-- order but since it is serial it takes a total of 17 seconds:
1281-- main = (S.'toList' . 'wSerially' $ (,) \<$> s1 \<*> s2) >>= print
1284-- ThreadId 36: Delay 1
1285-- ThreadId 36: Delay 3
1286-- ThreadId 36: Delay 2
1287-- ThreadId 36: Delay 3
1288-- ThreadId 36: Delay 4
1289-- ThreadId 36: Delay 4
1290-- [(1,3),(2,3),(1,4),(2,4)]
1293-- 'Async' can run the iterations concurrently and therefore takes a total
1294-- of 10 seconds (1 + 2 + 3 + 4):
1297-- main = (S.'toList' . 'asyncly' $ (,) \<$> s1 \<*> s2) >>= print
1300-- ThreadId 34: Delay 1
1301-- ThreadId 36: Delay 2
1302-- ThreadId 35: Delay 3
1303-- ThreadId 36: Delay 3
1304-- ThreadId 35: Delay 4
1305-- ThreadId 36: Delay 4
1306-- [(1,3),(2,3),(1,4),(2,4)]
1309-- Similalrly 'WAsync' as well can run the iterations concurrently and
1310-- therefore takes a total of 10 seconds (1 + 2 + 3 + 4):
1313-- main = (S.'toList' . 'wAsyncly' $ (,) \<$> s1 \<*> s2) >>= print
1316-- ThreadId 34: Delay 1
1317-- ThreadId 36: Delay 2
1318-- ThreadId 35: Delay 3
1319-- ThreadId 36: Delay 3
1320-- ThreadId 35: Delay 4
1321-- ThreadId 36: Delay 4
1322-- [(1,3),(2,3),(1,4),(2,4)]
1327-- Zipping is a special transformation where the corresponding elements of two
1328-- streams are combined together using a zip function producing a new stream of
1329-- outputs. Two different types are provided for serial and concurrent zipping.
1330-- These types provide an applicative instance that can be used to lift
1331-- functions to zip the argument streams.
1332-- Also see the zipping functions in the "Streamly.Prelude" module.
1336-- The applicative instance of 'ZipSerial' type zips streams serially.
1337-- 'zipSerially' type combinator can be used to switch to serial applicative
1342-- import qualified "Streamly.Prelude" as S
1343-- import Control.Concurrent
1345-- d n = delay n >> return n
1346-- s1 = 'serially' $ d 1 <> d 2
1347-- s2 = 'serially' $ d 3 <> d 4
1349-- main = (S.'toList' . 'zipSerially' $ (,) \<$> s1 \<*> s2) >>= print
1352-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since
1353-- everything runs serially:
1356-- ThreadId 29: Delay 1
1357-- ThreadId 29: Delay 3
1358-- ThreadId 29: Delay 2
1359-- ThreadId 29: Delay 4
1365-- The applicative instance of 'ZipAsync' type zips streams concurrently.
1366-- 'zipAsyncly' type combinator can be used to switch to parallel applicative
1372-- import qualified "Streamly.Prelude" as S
1373-- import Control.Concurrent
1374-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
1376-- d n = delay n >> return n
1377-- s1 = 'serially' $ d 1 <> d 2
1378-- s2 = 'serially' $ d 3 <> d 4
1381-- hSetBuffering stdout LineBuffering
1382-- (S.'toList' . 'zipAsyncly' $ (,) \<$> s1 \<*> s2) >>= print
1385-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
1386-- are produced concurrently, and 2 and 4 are produced concurrently:
1389-- ThreadId 32: Delay 1
1390-- ThreadId 32: Delay 2
1391-- ThreadId 33: Delay 3
1392-- ThreadId 33: Delay 4
1398-- When writing concurrent programs there are two distinct places where the
1399-- programmer can control the concurrency. First, when /composing/ a stream by
1400-- merging multiple streams we can choose an appropriate sum style operators to
1401-- combine them concurrently or serially. Second, when /processing/ a stream in
1402-- a monadic composition we can choose one of the monad composition types to
1403-- choose the desired type of concurrency.
1405-- In the following example the squares of @x@ and @y@ are computed
1406-- concurrently using the 'async' operation and the square roots of their
1407-- sum are computed serially because of the 'streamly' combinator. We can
1408-- choose different combinators for the monadic processing and the stream
1409-- generation, to control the concurrency. We can also use the 'asyncly'
1410-- combinator instead of explicitly folding with 'async'.
1414-- import qualified "Streamly.Prelude" as S
1415-- import Data.List (sum)
1419-- $ 'serially' -- Serial monadic processing (sqrt below)
1421-- x2 \<- 'forEachWith' 'async' [1..100] $ -- Concurrent @"for"@ loop
1422-- \\x -> return $ x * x -- body of the loop
1423-- y2 \<- 'forEachWith' 'async' [1..100] $
1424-- \\y -> return $ y * y
1425-- return $ sqrt (x2 + y2)
1429-- We can see how this directly maps to the imperative style
1430-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators
1431-- and operators instead of the ugly pragmas.
1433-- For more concurrent programming examples see,
1434-- "ListDir.hs", "MergeSort.hs" and "SearchQuery.hs" in the examples directory.
1438-- Reactive programming is nothing but concurrent streaming which is what
1439-- streamly is all about. With streamly we can generate streams of events,
1440-- merge streams that are generated concurrently and process events
1441-- concurrently. We can do all this without any knowledge about the specifics
1442-- of the implementation of concurrency. In the following example you will see
1443-- that the code is just regular Haskell code without much streamly APIs used
1444-- (active hyperlinks are the streamly APIs) and yet it is a reactive
1447-- This application has two independent and concurrent sources of event
1448-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates
1449-- events that deteriorate the health of the character in the game.
1450-- @userAction@ can be "potion" or "quit". When the user types "potion" the
1451-- health improves and the game continues.
1454-- {-\# LANGUAGE FlexibleContexts #-}
1457-- import Streamly.Prelude as S
1458-- import Control.Monad (when)
1459-- import Control.Monad.IO.Class (MonadIO(..))
1460-- import Control.Monad.State (MonadState, get, modify, runStateT)
1462-- data Event = Harm Int | Heal Int | Quit deriving (Show)
1464-- userAction :: MonadAsync m => 'SerialT' m Event
1465-- userAction = S.repeatM $ liftIO askUser
1468-- command <- getLine
1470-- "potion" -> return (Heal 10)
1471-- "quit" -> return Quit
1472-- _ -> putStrLn "What?" >> askUser
1474-- acidRain :: MonadAsync m => SerialT m Event
1475-- acidRain = asyncly $ constRate 1 $ S.repeatM $ liftIO $ return $ Harm 1
1477-- game :: ('MonadAsync' m, MonadState Int m) => 'SerialT' m ()
1479-- event \<- userAction \`parallel` acidRain
1481-- Harm n -> modify $ \\h -> h - n
1482-- Heal n -> modify $ \\h -> h + n
1483-- Quit -> fail "quit"
1486-- when (h <= 0) $ fail "You die!"
1487-- liftIO $ putStrLn $ "Health = " ++ show h
1490-- putStrLn "Your health is deteriorating due to acid rain,\\
1491-- \\ type \\"potion\\" or \\"quit\\""
1492-- _ <- runStateT ('runStream' game) 60
1496-- You can also find the source of this example in the examples directory as
1497-- "AcidRain.hs". It has been adapted from Gabriel's
1498-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>
1500-- This is much simpler compared to the pipes version because of the builtin
1501-- concurrency in streamly. You can also find a SDL based reactive programming
1502-- example adapted from Yampa in "Streamly.Examples.CirclingSquare".
1506-- Streamly is highly optimized for performance, it is designed for serious
1507-- high performing, concurrent and scalable applications. We have created the
1508-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>
1509-- package which is specifically and carefully designed to measure the
1510-- performance of Haskell streaming libraries fairly and squarely in the right
1511-- way. Streamly performs at par or even better than most streaming libraries
1512-- for serial operations even though it needs to deal with the concurrency
1517-- We can use @unfoldr@ and @uncons@ to convert one streaming type to another.
1519-- Interop with @vector@:
1523-- import qualified Streamly.Prelude as S
1524-- import qualified Data.Vector.Fusion.Stream.Monadic as V
1526-- -- | vector to streamly
1527-- fromVector :: (IsStream t, Monad m) => V.Stream m a -> t m a
1528-- fromVector = S.unfoldrM unconsV
1533-- then return Nothing
1536-- return $ Just (h, V.tail v)
1538-- -- | streamly to vector
1539-- toVector :: Monad m => SerialT m a -> V.Stream m a
1540-- toVector = V.unfoldrM (S.uncons . adapt)
1543-- S.toList (fromVector (V.fromList [1..3])) >>= print
1544-- V.toList (toVector (S.fromFoldable [1..3])) >>= print
1547-- Interop with @pipes@:
1551-- import qualified "Streamly.Prelude" as S
1552-- import qualified Pipes as P
1553-- import qualified Pipes.Prelude as P
1555-- -- | pipes to streamly
1556-- fromPipes :: (IsStream t, Monad m) => P.Producer a m r -> t m a
1557-- fromPipes = S.'unfoldrM' unconsP
1559-- -- Adapt P.next to return a Maybe instead of Either
1560-- unconsP p = P.next p >>= either (\\_ -> return Nothing) (return . Just)
1562-- -- | streamly to pipes
1563-- toPipes :: Monad m => SerialT m a -> P.Producer a m ()
1564-- toPipes = P.unfoldr unconsS
1566-- -- Adapt S.uncons to return an Either instead of Maybe
1567-- unconsS s = S.'uncons' s >>= maybe (return $ Left ()) (return . Right)
1570-- S.'toList' (fromPipes (P.each [1..3])) >>= print
1571-- P.toListM (toPipes (S.'fromFoldable' [1..3])) >>= print
1574-- Interop with @streaming@:
1578-- import qualified "Streamly.Prelude" as S
1579-- import qualified Streaming as SG
1580-- import qualified Streaming.Prelude as SG
1582-- -- | streaming to streamly
1583-- fromStreaming :: (IsStream t, Monad m) => SG.Stream (SG.Of a) m r -> t m a
1584-- fromStreaming = S.unfoldrM SG.uncons
1586-- -- | streamly to streaming
1587-- toStreaming :: Monad m => SerialT m a -> SG.Stream (SG.Of a) m ()
1588-- toStreaming = SG.unfoldr unconsS
1590-- -- Adapt S.uncons to return an Either instead of Maybe
1591-- unconsS s = S.'uncons' s >>= maybe (return $ Left ()) (return . Right)
1594-- S.toList (fromStreaming (SG.each [1..3])) >>= print
1595-- SG.toList (toStreaming (S.fromFoldable [1..3])) >>= print
1598-- Interop with @conduit@:
1602-- import qualified "Streamly.Prelude" as S
1603-- import qualified Data.Conduit as C
1604-- import qualified Data.Conduit.List as C
1605-- import qualified Data.Conduit.Combinators as C
1607-- -- It seems there is no way out of a conduit as it does not provide an
1608-- -- uncons or a tail function. We can convert streamly to conduit though.
1610-- -- | streamly to conduit
1611-- toConduit :: Monad m => SerialT m a -> C.ConduitT i a m ()
1612-- toConduit s = C.unfoldM S.'uncons' s
1615-- C.runConduit (toConduit (S.'fromFoldable' [1..3]) C..| C.sinkList) >>= print
1620-- List transformers and logic programming monads also provide a product style
1621-- composition similar to streamly, however streamly generalizes it with the
1622-- time dimension; allowing streams to be composed in an asynchronous and
1623-- concurrent fashion in many different ways. It also provides multiple
1624-- alternative ways of composing streams e.g. serial, interleaved or
1627-- This seemingly simple addition of asynchronicity and concurrency to product
1628-- style streaming composition unifies a number of disparate abstractions into
1629-- one powerful, concise and elegant abstraction. A wide variety of
1630-- programming problems can be solved elegantly with this abstraction. In
1631-- particular, it unifies three major programming domains namely
1632-- non-deterministic (logic) programming, concurrent programming and functional
1633-- reactive programming. In other words, you can do everything with this one
1634-- abstraction that you could do with the popular libraries listed under these
1635-- categories in the list below.
1638-- +-----------------+----------------+
1639-- | Non-determinism | <https://hackage.haskell.org/package/pipes pipes> |
1640-- | +----------------+
1641-- | | <https://hackage.haskell.org/package/list-t list-t> |
1642-- | +----------------+
1643-- | | <https://hackage.haskell.org/package/logict logict> |
1644-- +-----------------+----------------+
1645-- | Streaming | <https://hackage.haskell.org/package/vector vector> |
1646-- | +----------------+
1647-- | | <https://hackage.haskell.org/package/streaming streaming> |
1648-- | +----------------+
1649-- | | <https://hackage.haskell.org/package/pipes pipes> |
1650-- | +----------------+
1651-- | | <https://hackage.haskell.org/package/conduit conduit> |
1652-- +-----------------+----------------+
1653-- | Concurrency | <https://hackage.haskell.org/package/async async> |
1654-- | +----------------+
1655-- | | <https://hackage.haskell.org/package/transient transient> |
1656-- +-----------------+----------------+
1657-- | FRP | <https://hackage.haskell.org/package/Yampa Yampa> |
1658-- | +----------------+
1659-- | | <https://hackage.haskell.org/package/dunai dunai> |
1660-- | +----------------+
1661-- | | <https://hackage.haskell.org/package/reflex reflex> |
1662-- +-----------------+----------------+
1665-- Streamly is a list-transformer. It provides all the functionality provided
1666-- by any of the list transformer and logic programming packages listed above.
1667-- In addition, Streamly naturally integrates the concurrency dimension to the
1668-- basic list transformer functionality.
1670-- When it comes to streaming, in terms of the streaming API streamly is almost
1671-- identical to the vector package. Streamly, vector and streaming packages all
1672-- represent a stream as data and are therefore similar in the fundamental
1673-- approach to streaming. The fundamental difference is that streamly adds
1674-- concurrency support and the monad instance provides concurrent looping.
1675-- Other streaming libraries like pipes, conduit and machines represent and
1676-- compose stream processors rather than the stream data and therefore fall in
1677-- another class of streaming libraries and have comparatively more complicated
1680-- When it comes to concurrency, streamly can do everything that the @async@
1681-- package can do and more. async provides applicative concurrency whereas
1682-- streamly provides both applicative and monadic concurrency. The
1683-- 'ZipAsync' type behaves like the applicative instance of async. In
1684-- comparison to transient streamly has a first class streaming interface and
1685-- is a monad transformer that can be used universally in any Haskell monad
1686-- transformer stack. Streamly was in fact originally inspired by the
1687-- concurrency implementation in @transient@ though it has no resemblence with
1688-- that and takes a lazy pull approach versus transient's strict push approach.
1690-- The non-determinism, concurrency and streaming combination make streamly a
1691-- strong FRP capable library as well. FRP is fundamentally stream of events
1692-- that can be processed concurrently. The example in this tutorial as well as
1693-- the "Streamly.Examples.CirclingSquare" example from Yampa demonstrate the
1694-- basic FRP capability of streamly. In core concepts streamly is strikingly
1695-- similar to @dunai@. dunai was designed from a FRP perspective and streamly
1696-- was originally designed from a concurrency perspective. However, both have
1697-- similarity at the core.
1701-- * Read the documentation of "Streamly" module
1702-- * Read the documentation of "Streamly.Prelude" module
1703-- * See the examples in the "examples" directory of the package
1704-- * See the tests in the "test" directory of the package
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