HORIZON HASKELLDocslts/ghc-9.14.x45dd5372026-10-11Search names, modules, packages, or :: a typeCtrl K

GHC 9.14.1 · lts/ghc-9.14.x · 45dd537 · 2026-10-11

Modulestreamly-0.11.1Haskell2010

Streamly.Data.Scanl.Prelude

All Scan related combinators including the streamly-core Streamly.Data.Scanl module, concurrency, unordered container operations.

  • 2 types
  • 69 values
  • Packagestreamly-0.11.1
  • Exports71
  • LanguageHaskell2010
  • LicenceBSD-3-Clause
  • SourcePrelude.hs

Setup

0 declarations

To execute the code examples provided in this module in ghci, please run the following commands first.

Example5 expressions
:m:set -XFlexibleContextsimport Control.Concurrent (threadDelay)import qualified Streamly.Data.Stream as Streamimport qualified Streamly.Data.Scanl.Prelude as Scanl

Streamly.Data.Scanl

62 declarations

All Streamly.Data.Scanl combinators are re-exported via this module. For more pre-release combinators also see Streamly.Internal.Data.Scanl module.

valuedeleteBy :: Monad m => (a -> a -> Bool) -> a -> Scanl m a (Maybe a)
#

Returns the latest element omitting the first occurrence that satisfies the given equality predicate.

Example:

Example2 expressions
input = Stream.fromList [1,3,3,5]Stream.toList $ Stream.postscanlMaybe (Scanl.deleteBy (==) 3) input[1,3,5]
valuedistribute :: Monad m => [Scanl m a b] -> Scanl m a [b]
#

Distribute one copy of the stream to each scan and collect the results in a container.


                |-------Scanl m a b--------|
---stream m a---|                          |---m [b]
                |-------Scanl m a b--------|
                |                          |
                           ...
Example1 expression
Stream.toList $ Stream.scanl (Scanl.distribute [Scanl.sum, Scanl.length]) (Stream.enumerateFromTo 1 5)[[0,0],[1,1],[3,2],[6,3],[10,4],[15,5]]
Example1 expression
distribute = Prelude.foldr (Scanl.teeWith (:)) (Scanl.const [])

This is the consumer side dual of the producer side sequence operation.

Stops as soon as any of the scans stop.

valueelemIndices :: (Monad m, Eq a) => a -> Scanl m a (Maybe Int)
#

Returns the index of the latest element if the element matches the given value.

Definition:

Example1 expression
elemIndices a = Scanl.findIndices (== a)
valuefoldMap :: (Monad m, Monoid b) => (a -> b) -> Scanl m a b
#

Definition:

Example1 expression
foldMap f = Scanl.lmap f Scanl.mconcat

Make a scan from a pure function that scans the output of the function using mappend and mempty.

Example2 expressions
sum = Scanl.foldMap Data.Monoid.SumStream.toList $ Stream.scanl sum $ Stream.enumerateFromTo 1 3[Sum {getSum = 0},Sum {getSum = 1},Sum {getSum = 3},Sum {getSum = 6}]
valuefoldMapM :: (Monad m, Monoid b) => (a -> m b) -> Scanl m a b
#

Definition:

Example1 expression
foldMapM f = Scanl.lmapM f Scanl.mconcat

Make a scan from a monadic function that scans the output of the function using mappend and mempty.

Example2 expressions
sum = Scanl.foldMapM (return . Data.Monoid.Sum)Stream.toList $ Stream.scanl sum $ Stream.enumerateFromTo 1 3[Sum {getSum = 0},Sum {getSum = 1},Sum {getSum = 3},Sum {getSum = 6}]
valuemapMaybe :: Monad m => (a -> Maybe b) -> Scanl m b r -> Scanl m a r
#

mapMaybe f scan maps a Maybe returning function f on the input of the scan, filters out Nothing elements, and return the values extracted from Just.

Example2 expressions
mapMaybe f = Scanl.lmap f . Scanl.catMaybesmapMaybe f = Scanl.mapMaybeM (return . f)
Example3 expressions
f x = if even x then Just x else Nothingscn = Scanl.mapMaybe f Scanl.toListStream.toList $ Stream.scanl scn (Stream.enumerateFromTo 1 10)[[],[],[2],[2],[2,4],[2,4],[2,4,6],[2,4,6],[2,4,6,8],[2,4,6,8],[2,4,6,8,10]]
valuemconcat :: (Monad m, Monoid a) => Scanl m a a
#

Monoid concat. Scan an input stream consisting of monoidal elements using mappend and mempty.

Definition:

Example1 expression
mconcat = Scanl.sconcat mempty
Example2 expressions
monoids = fmap Data.Monoid.Sum $ Stream.enumerateFromTo 1 3Stream.toList $ Stream.scanl Scanl.mconcat monoids[Sum {getSum = 0},Sum {getSum = 1},Sum {getSum = 3},Sum {getSum = 6}]
valuemean :: (Monad m, Fractional a) => Scanl m a a
#

Compute a numerically stable arithmetic mean of all elements in the input stream.

valuepartition :: Monad m => Scanl m b x -> Scanl m c x -> Scanl m (Either b c) x
#

Compose two scans such that the combined scan accepts a stream of Either and routes the Left values to the first scan and Right values to the second scan.

Definition:

Example1 expression
partition = Scanl.partitionBy id
valueproduct :: (Monad m, Num a, Eq a) => Scanl m a a
#

Determine the product of all elements of a stream of numbers. Returns multiplicative identity (1) when the stream is empty. The scan terminates when it encounters (0) in its input.

Same as the following but terminates on multiplication by 0:

Example1 expression
product = fmap Data.Monoid.getProduct $ Scanl.foldMap Data.Monoid.Product
valuerollingHash :: (Monad m, Enum a) => Scanl m a Int64
#

Compute an Int sized polynomial rolling hash of a stream.

Example1 expression
rollingHash = Scanl.rollingHashWithSalt Scanl.defaultSalt
valuescanl :: Monad m => Scanl m a b -> Scanl m b c -> Scanl m a c
#

Scan the input of a Scanl to change it in a stateful manner using another Scanl. The scan stops as soon as any of the scans terminates.

This is basically an append operation.

Pre-release

valuesconcat :: (Monad m, Semigroup a) => a -> Scanl m a a
#

Semigroup concat. Append the elements of an input stream to a provided starting value.

Definition:

Example1 expression
sconcat = Scanl.mkScanl (<>)
Example2 expressions
semigroups = fmap Data.Monoid.Sum $ Stream.enumerateFromTo 1 3Stream.toList $ Stream.scanl (Scanl.sconcat 3) semigroups[Sum {getSum = 3},Sum {getSum = 4},Sum {getSum = 6},Sum {getSum = 9}]
valuesum :: (Monad m, Num a) => Scanl m a a
#

Determine the sum of all elements of a stream of numbers. Returns additive identity (0) when the stream is empty. Note that this is not numerically stable for floating point numbers.

Example1 expression
sum = Scanl.cumulativeScan Scanl.incrSum

Same as following but numerically stable:

Example2 expressions
sum = Scanl.mkScanl (+) 0sum = fmap Data.Monoid.getSum $ Scanl.foldMap Data.Monoid.Sum
valuetee :: Monad m => Scanl m a b -> Scanl m a c -> Scanl m a (b, c)
#

Distribute one copy of the stream to each scan and zip the results.

                |-------Scanl m a b--------|
---stream m a---|                          |---m (b,c)
                |-------Scanl m a c--------|

Definition:

Example1 expression
tee = Scanl.teeWith (,)

Example:

Example2 expressions
t = Scanl.tee Scanl.sum Scanl.lengthStream.toList $ Stream.scanl t (Stream.enumerateFromTo 1.0 10.0)[(0.0,0),(1.0,1),(3.0,2),(6.0,3),(10.0,4),(15.0,5),(21.0,6),(28.0,7),(36.0,8),(45.0,9),(55.0,10)]
valuethe :: (Monad m, Eq a) => Scanl m a (Maybe a)
#

Terminates with Nothing as soon as it finds an element different than the previous one, returns the element if the entire input consists of the same element.

valuetoListRev :: Monad m => Scanl m a [a]
#

Buffers the input stream to a list in the reverse order of the input.

Definition:

Example1 expression
toListRev = Scanl.mkScanl (flip (:)) []

Warning! working on large lists accumulated as buffers in memory could be very inefficient, consider using Streamly.Array instead.

valuetopBy
  1. :: (MonadIO m, Unbox a)
  2. => a -> a -> Ordering
  3. -> Int
  4. -> Scanl m a (MutArray a)
#

Get the top n elements using the supplied comparison function.

To get bottom n elements instead:

Example1 expression
bottomBy cmp = Scanl.topBy (flip cmp)

Example:

Example2 expressions
stream = Stream.fromList [2::Int,7,9,3,1,5,6,11,17]Stream.toList (Stream.scanl (Scanl.topBy compare 3) stream) >>= mapM MutArray.toList[[],[17],[17,11],[17,11,9],[17,11,9],[17,11,9],[17,11,9],[17,11,9],[17,11,9],[17,11,9]]

Pre-release

valueuniqBy :: Monad m => (a -> a -> Bool) -> Scanl m a (Maybe a)
#

Return the latest unique element using the supplied comparison function. Returns Nothing if the current element is same as the last element otherwise returns Just.

Example, strip duplicate path separators:

Example3 expressions
input = Stream.fromList "//a//b"f x y = x == '/' && y == '/'Stream.toList $ Stream.postscanlMaybe (Scanl.uniqBy f) input"/a/b"

Space: O(1)

Pre-release

valueunzip :: Monad m => Scanl m a x -> Scanl m b y -> Scanl m (a, b) (x, y)
#

Send the elements of tuples in a stream of tuples through two different scans.


                          |-------Scanl m a x--------|
---------stream of (a,b)--|                          |----m (x,y)
                          |-------Scanl m b y--------|

Definition:

Example1 expression
unzip = Scanl.unzipWith id

This is the consumer side dual of the producer side zip operation.

valueclassify
  1. :: (MonadIO m, Ord k)
  2. => a -> k
  3. -> Scanl m a b
  4. -> Scanl m a (Maybe (k, b))
#

Scans the values for each key using the supplied scan.

Once the scan for a key terminates, any future values of the key are ignored.

Equivalent to the following except that the scan is not restarted:

Example1 expression
classify f fld = Scanl.demux f (const fld)
valueclassifyIO
  1. :: (MonadIO m, Ord k)
  2. => a -> k
  3. -> Scanl m a b
  4. -> Scanl m a (Maybe (k, b))
#

Same as classify except that it uses mutable IORef cells in the Map, providing better performance.

Equivalent to the following except that the scan is not restarted:

Example1 expression
classifyIO f fld = Scanl.demuxIO f (const fld)
valuecountDistinct :: (Monad m, Ord a) => Scanl m a Int
#

Count non-duplicate elements in the stream.

Definition:

Example2 expressions
countDistinct = fmap Set.size Scanl.toSetcountDistinct = Scanl.postscanl Scanl.nub $ Scanl.catMaybes $ Scanl.length

The memory used is proportional to the number of distinct elements in the stream, to guard against using too much memory use it as a scan and terminate if the count reaches more than a threshold.

Space: \mathcal{O}(n)

Pre-release

valuecountDistinctInt :: Monad m => Scanl m Int Int
#

Like countDistinct but specialized to a stream of Int, for better performance.

Definition:

Example2 expressions
countDistinctInt = fmap IntSet.size Scanl.toIntSetcountDistinctInt = Scanl.postscanl Scanl.nubInt $ Scanl.catMaybes $ Scanl.length

Pre-release

valuedemux
  1. :: (Monad m, Ord k)
  2. => a -> k
  3. -> k -> m (Maybe (Scanl m a b))
  4. -> Scanl m a (Maybe (k, b))
#

demux getKey getScan: In a key value stream, scan values corresponding to each key using a key specific scan. getScan is invoked to generate a key specific scan when a key is encountered for the first time in the stream. If a scan does not exist corresponding to the key then Nothing is returned otherwise the result of the scan is returned.

If a scan terminates, another instance of the scan is started upon receiving an input with that key, getScan is invoked again whenever the key is encountered again.

This can be used to scan a stream, splitting it based on different keys.

Since the scan generator function is monadic we can add scans dynamically. For example, we can maintain a Map of keys to scans in an IORef and lookup the scan from that corresponding to a key. This Map can be changed dynamically, scans for new keys can be added or scans for old keys can be deleted or modified.

Compare with classify, the scan in classify is a static scan.

Pre-release

valuedemuxIO
  1. :: (MonadIO m, Ord k)
  2. => a -> k
  3. -> k -> m (Maybe (Scanl m a b))
  4. -> Scanl m a (Maybe (k, b))
#

This is specialized version of demux that uses mutable IO cells as scan accumulators for better performance.

valuenub :: (Monad m, Ord a) => Scanl m a (Maybe a)
#

Returns Just for the first occurrence of an element, returns Nothing for any other occurrences.

Example:

Example2 expressions
stream = Stream.fromList [1::Int,1,2,3,4,4,5,1,5,7]Stream.toList $ Stream.postscanlMaybe Scanl.nub stream[1,2,3,4,5,7]

Pre-release

valuetoIntSet :: Monad m => Scanl m Int IntSet
#

Scan the input adding it to an int set. For integer inputs this performs better than toSet.

Definition:

Example1 expression
toIntSet = Scanl.mkScanl (flip IntSet.insert) IntSet.empty
valuetoSet :: (Monad m, Ord a) => Scanl m a (Set a)
#

Scan the input adding it to a set.

Definition:

Example1 expression
toSet = Scanl.mkScanl (flip Set.insert) Set.empty
valuecatEithers :: Scanl m a b -> Scanl m (Either a a) b
#

Remove the either wrapper and flatten both lefts and as well as rights in the output stream.

Definition:

Example1 expression
catEithers = Scanl.lmap (either id id)

Pre-release

valuecatMaybes :: Monad m => Scanl m a b -> Scanl m (Maybe a) b
#

Modify a scan to receive a Maybe input, the Just values are unwrapped and sent to the original scan, Nothing values are discarded.

Example2 expressions
catMaybes = Scanl.mapMaybe idcatMaybes = Scanl.filter isJust . Scanl.lmap fromJust
valuedrain :: Monad m => Scanl m a ()
#

A scan that drains all its input, running the effects and discarding the results.

Example2 expressions
drain = Scanl.drainMapM (const (return ()))drain = Scanl.mkScanl (\_ _ -> ()) ()
valuefilter :: Monad m => (a -> Bool) -> Scanl m a r -> Scanl m a r
#

Include only those elements that pass a predicate.

Example1 expression
Stream.toList $ Stream.scanl (Scanl.filter (> 5) Scanl.sum) $ Stream.fromList [1..10][0,0,0,0,0,0,6,13,21,30,40]
Example3 expressions
filter p = Scanl.postscanlMaybe (Scanl.filtering p)filter p = Scanl.filterM (return . p)filter p = Scanl.mapMaybe (\x -> if p x then Just x else Nothing)
valuefilterM :: Monad m => (a -> m Bool) -> Scanl m a r -> Scanl m a r
#

Like filter but with a monadic predicate.

Example2 expressions
f p x = p x >>= \r -> return $ if r then Just x else NothingfilterM p = Scanl.mapMaybeM (f p)
valuelatest :: Monad m => Scanl m a (Maybe a)
#

Returns the latest element of the input stream, if any.

Example2 expressions
latest = Scanl.mkScanl1 (\_ x -> x)latest = fmap getLast $ Scanl.foldMap (Last . Just)
valuelength :: Monad m => Scanl m a Int
#

Determine the length of the input stream.

Definition:

Example2 expressions
length = Scanl.genericLengthlength = fmap getSum $ Scanl.foldMap (Sum . const  1)
valuelmap :: (a -> b) -> Scanl m b r -> Scanl m a r
#

lmap f scan maps the function f on the input of the scan.

Definition:

Example1 expression
lmap = Scanl.lmapM return

Example:

Example2 expressions
sumSquared = Scanl.lmap (\x -> x * x) Scanl.sumStream.toList $ Stream.scanl sumSquared (Stream.enumerateFromTo 1 10)[0,1,5,14,30,55,91,140,204,285,385]
valuelmapM :: Monad m => (a -> m b) -> Scanl m b r -> Scanl m a r
#

lmapM f scan maps the monadic function f on the input of the scan.

valuemaximum :: (Monad m, Ord a) => Scanl m a (Maybe a)
#

Determine the maximum element in a stream.

Definitions:

Example2 expressions
maximum = Scanl.maximumBy comparemaximum = Scanl.mkScanl1 max

Same as the following but without a default maximum. The Max Monoid uses the minBound as the default maximum:

Example1 expression
maximum = fmap Data.Semigroup.getMax $ Scanl.foldMap Data.Semigroup.Max
valueminimum :: (Monad m, Ord a) => Scanl m a (Maybe a)
#

Determine the minimum element in a stream using the supplied comparison function.

Definitions:

Example2 expressions
minimum = Scanl.minimumBy compareminimum = Scanl.mkScanl1 min

Same as the following but without a default minimum. The Min Monoid uses the maxBound as the default maximum:

Example1 expression
maximum = fmap Data.Semigroup.getMin $ Scanl.foldMap Data.Semigroup.Min
valuemkScanl :: Monad m => (b -> a -> b) -> b -> Scanl m a b
#

Make a scan from a left fold style pure step function and initial value of the accumulator.

If your Scanl returns only Partial (i.e. never returns a Done) then you can use mkScanl* constructors.

valuemkScanl1 :: Monad m => (a -> a -> a) -> Scanl m a (Maybe a)
#

Make a strict left scan, for non-empty streams, using first element as the starting value. Returns Nothing if the stream is empty.

Pre-release

valuemkScanlM :: Monad m => (b -> a -> m b) -> m b -> Scanl m a b
#

Make a scan from a left fold style monadic step function and initial value of the accumulator.

valuemkScanr :: Monad m => (a -> b -> b) -> b -> Scanl m a b
#

Make a scan using a right fold style step function and a terminal value. It performs a strict right fold via a left fold using function composition. Note that a strict right fold can only be useful for constructing strict structures in memory. For reductions this will be very inefficient.

Definitions:

Example2 expressions
mkScanr f z = fmap (flip appEndo z) $ Scanl.foldMap (Endo . f)mkScanr f z = fmap ($ z) $ Scanl.mkScanl (\g x -> g . f x) id

Example:

Example1 expression
Stream.toList $ Stream.scanl (Scanl.mkScanr (:) []) $ Stream.enumerateFromTo 1 5[[],[1],[1,2],[1,2,3],[1,2,3,4],[1,2,3,4,5]]
valuemorphInner :: (forall x. m x -> n x) -> Scanl m a b -> Scanl n a b
#

Change the underlying monad of a scan. Also known as hoist.

Pre-release

valuepostscanl :: Monad m => Scanl m a b -> Scanl m b c -> Scanl m a c
#

Postscan the input of a Scanl to change it in a stateful manner using another Scanl.

This is basically an append operation.

Pre-release

valuermapM :: Monad m => (b -> m c) -> Scanl m a b -> Scanl m a c
#

Map a monadic function on the output of a scan.

valuetake :: Monad m => Int -> Scanl m a b -> Scanl m a b
#

Take at most n input elements and scan them using the supplied scan. A negative count is treated as 0.

Example1 expression
Stream.toList $ Stream.scanl (Scanl.take 2 Scanl.toList) $ Stream.fromList [1..10][[],[1],[1,2]]
valuetakeEndBy :: Monad m => (a -> Bool) -> Scanl m a b -> Scanl m a b
#

Take the input, stop when the predicate succeeds taking the succeeding element as well.

Example:

Example3 expressions
input = Stream.fromList "hello\nthere\n"line = Scanl.takeEndBy (== '\n') Scanl.toListStream.toList $ Stream.scanl line input["","h","he","hel","hell","hello","hello\n"]
valuetakeEndBy_ :: Monad m => (a -> Bool) -> Scanl m a b -> Scanl m a b
#

Like takeEndBy but drops the element on which the predicate succeeds.

Example:

Example3 expressions
input = Stream.fromList "hello\nthere\n"line = Scanl.takeEndBy_ (== '\n') Scanl.toListStream.toList $ Stream.scanl line input["","h","he","hel","hell","hello","hello"]
valueteeWith
  1. :: Monad m
  2. => b -> c -> d
  3. -> Scanl m a b
  4. -> Scanl m a c
  5. -> Scanl m a d
#

teeWith k f1 f2 distributes its input to both f1 and f2 until any one of them terminates. The outputs of the two scans are combined using the function k.

Definition:

Example1 expression
teeWith k f1 f2 = fmap (uncurry k) (Scanl.tee f1 f2)

Example:

Example2 expressions
avg = Scanl.teeWith (/) Scanl.sum (fmap fromIntegral Scanl.length)Stream.toList $ Stream.postscanl avg $ Stream.fromList [1.0..10.0][1.0,1.5,2.0,2.5,3.0,3.5,4.0,4.5,5.0,5.5]

Note that nested applications of teeWith do not fuse.

Pre-release

valuetoList :: Monad m => Scanl m a [a]
#

Scans the input stream building a list.

Warning! working on large lists accumulated as buffers in memory could be very inefficient, consider using Streamly.Data.Array instead.

Example1 expression
toList = Scanl.mkScanr (:) []
datadata Scanl (m :: Type -> Type) a b
#

The type Scanl m a b represents a consumer of an input stream of values of type a and returning a final value of type b in Monad m. The constructor of a scan is Scanl step initial extract final.

The scan uses an internal state of type s. The initial value of the state s is created by initial. This function is called once and only once before the scan starts consuming input. Any resource allocation can be done in this function.

The step function is called on each input, it consumes an input and returns the next intermediate state (see Step) or the final result b if the scan terminates.

The extract function is used by the scan driver to map the current state s of the scan to the scan result. Thus extract can be called multiple times.

Before a scan terminates, final is called once and only once (unless the scan terminated in initial itself). Any resources allocated by initial can be released in final. In scan that do not require any cleanup extract and final are typically the same.

When implementing scan combinators, care should be taken to cleanup any state of the argument folds held by the fold by calling the respective final at all exit points of the scan. Also, final should not be called more than once. Note that if a scan terminates by Done constructor, there is no state to cleanup.

NOTE: The constructor is not yet released, smart constructors are provided to create scans.

Instances2Functor, Applicative
  • Functor m => Functor (Scanl m a)Defined in streamly-core-0.3.1 · Streamly.Internal.Data.Scanl.Type

    Maps a function on the output of the scan (the type b).

  • Monad m => Applicative (Scanl m a)Defined in streamly-core-0.3.1 · Streamly.Internal.Data.Scanl.Type

Concurrent Operations

0 declarations

Configuration

datadata Config
#

An abstract type for specifying the configuration parameters of a Channel. Use Config -> Config modifier functions to modify the default configuration. See the individual modifier documentation for default values.

valuemaxBuffer :: Int -> Config -> Config
#

Specify the maximum size of the buffer for storing the results from concurrent computations. If the buffer becomes full we stop spawning more concurrent tasks until there is space in the buffer. A value of 0 resets the buffer size to default, a negative value means there is no limit. The default value is 1500.

CAUTION! using an unbounded maxBuffer value (i.e. a negative value) coupled with an unbounded maxThreads value is a recipe for disaster in presence of infinite streams, or very large streams. Especially, it must not be used when pure is used in ZipAsyncM streams as pure in applicative zip streams generates an infinite stream causing unbounded concurrent generation with no limit on the buffer or threads.

valueboundThreads :: Bool -> Config -> Config
#

Spawn bound threads (i.e., spawn threads using forkOS instead of forkIO). The default value is False.

Currently, this only takes effect only for concurrent folds.

valueinspect :: Bool -> Config -> Config
#

Print debug information about the Channel when the stream ends. When the stream does not end normally, the channel debug information is printed when the channel is garbage collected. If you are expecting but not seeing the debug info try adding a performMajorGC before the program ends.

Combinators

valueparTeeWith
  1. :: MonadAsync m
  2. => Config -> Config
  3. -> a -> b -> c
  4. -> Scanl m x a
  5. -> Scanl m x b
  6. -> Scanl m x c
#

Execute both the scans in a tee concurrently.

Example:

Example6 expressions
src = Stream.delay 1 (Stream.enumerateFromTo 1 3)delay x = threadDelay 1000000 >> print x >> return xc1 = Scanl.lmapM delay Scanl.sumc2 = Scanl.lmapM delay Scanl.lengthdst = Scanl.parTeeWith id (,) c1 c2Stream.toList $ Stream.scanl dst src...
valueparDistributeScanM
  1. :: MonadAsync m
  2. => Config -> Config
  3. -> m [Scanl m a b]
  4. -> Stream m a
  5. -> Stream m [b]
#

Evaluate a stream by distributing its inputs across zero or more concurrently running scans. New scans can be generated dynamically. Use parDistributeScan for an eaiser to use interface, if you do not need the power of parDistributeScanM.

Before processing each input element, the supplied action is executed to produce additional scans. These scans are appended to the set of currently active scans. If you do not want the same scan to be added repeatedly, ensure that the action only generates it once (see the example below).

If there are no scans currently active, the input element is discarded. The results from all active scans are collected and lattened into the the output stream.

Concurrency and buffering:

If the input buffer (see maxBuffer) is bounded, a scan may block until space becomes available. If any scan is blocked on buffer, all scans are blocked. Processing continues only when all scans have buffer space available.

Example:

Example4 expressions
import Data.IORefref <- newIORef [Scanl.take 5 Scanl.sum, Scanl.take 5 Scanl.length :: Scanl.Scanl IO Int Int]gen = atomicModifyIORef ref (\xs -> ([], xs))Stream.toList $ Scanl.parDistributeScanM id gen (Stream.enumerateFromTo 1 10)...
valueparDistributeScan
  1. :: MonadAsync m
  2. => Config -> Config
  3. -> [Scanl m a b]
  4. -> Stream m a
  5. -> Stream m [b]
#

A pure variant of parDistributeScanM that uses a fixed list of scans.

The provided scans are started once and run concurrently for the duration of the stream. Each input element is distributed to all active scans, and their outputs are collected and emitted together.

Example:

Example2 expressions
xs = [Scanl.take 5 Scanl.sum, Scanl.take 5 Scanl.length :: Scanl.Scanl IO Int Int]Stream.toList $ Scanl.parDistributeScan id xs (Stream.enumerateFromTo 1 10)...
valueparDemuxScanM
  1. :: (MonadAsync m, Ord k)
  2. => Config -> Config
  3. -> a -> k
  4. -> k -> m (Scanl m a b)
  5. -> Stream m a
  6. -> Stream m [(k, b)]
#

Evaluate a stream by routing each input to a scan determined by a key.

For each distinct key, the first input encountered triggers the creation of a new scan (via the supplied key-to-scan function). This scan is then added to the set of currently active scans. Subsequent inputs with the same key are directed to the same scan.

If no scan can be created for a key, the input element is discarded.

When a constituent scan completes, its final output is emitted as part of the composed output stream. The output of parDemuxScanM is a stream of key–value pairs, where each value is the output produced by the scan corresponding to that key.

For a simpler interface, use parDemuxScan if you do not need the full flexibility of parDemuxScanM.

Example:

Example9 expressions
import qualified Data.Map.Strict as Mapimport Data.Maybe (fromJust)f1 = ("even", Scanl.take 5 Scanl.sum)f2 = ("odd",  Scanl.take 5 Scanl.sum)kv = Map.fromList [f1, f2]getScan k = return (fromJust $ Map.lookup k kv)getKey x = if even x then "even" else "odd"input = Stream.enumerateFromTo 1 10Stream.toList $ Scanl.parDemuxScanM id getKey getScan input...