ORIGINAL ARTICLE: HUTTON, G. (1999). A tutorial on the universality and expressiveness of fold. Journal of Functional Programming, 9(4), 355-372. DOI: https://doi.org/10.1017/S0956796899003500
We show various forms equivalent to Wolfram’s Fold that work over infinite data streams and asynchronous data items. Exactly the same foldable “aggregator” functions can be applied independently of the manner of data delivery. We show in other submissions how such decoupling can improve the reliability and testability of Kalman filters. However, the technique has very broad applicability. See Graham Hutton’s “Tutorial on the Universality and Expressiveness of Fold”.
We’d like to have the same operations over Lists (or enumerable collections, more generally), lazy Lists (Streams, iterables, or iteractive [sic] collections) and asynchronous tasks (Observables or reactive collections). We call Lists, Streams, and Observables collection domains (or monads, if you like that word). They each maintain ordered collections of values, differing only in the method and timing of delivery. Elements of Lists are distributed in memory, elements of Streams are computed on demand, and elements of Observables arrive asynchronously, on a schedule not known to the observer.
Mathematica Lists have a rich set of built-in operations, like Map, Scan, Take, Fold, FoldList, Drop, Append, and so on. For brevity, we work up examples just with analogues of Take and FoldList for Streams and Observables. It’s easy to build up other operations.
Many of these operations are higher-order. Higher-order operations take query functions as arguments. We may package sophisticated algorithms in queries, from mere counts to incremental statistics to Kalman filters.
The big message here is that we write the algorithms once, and run them over all manner of collections: eager or lazy, in-memory or computed on-the-fly, synchronous or asynchronous. Test them in a friendly environment of Lists; deploy them in an unfriendly environment of Observables, for instance.
Many higher-order operations are composable: they compute Lists from Lists, Streams from Streams, Observables from Observable. That means we may build up more sophisticated algorithms from building blocks, as one does on the Unix command line with the pipe operator.
Cross domains at will, with List as a rock-bottom intermediary. We show how to convert Lists into Streams and vice versa, and Lists into Observables, and vice versa.
Finally, we want “ways out” of the collection domains: operations that return values from Lists, Streams, and Observables. A way-out from Mathematica Lists is Scan. We posit subscribe for Observables and forEach for Streams. These should have the same signatures as Scan, taking a collection and a unary query that operates on each element of the collection. Transformers — ways to and from Lists for Streams and Observables, with Lists as the rock-bottom domain — are reify and disperse for Streams and harvest and dispense for Observables.
The following diagram illustrates:

Typographical Conventions

Types are capitalised, like List, Stream, and Observable.
Honoring the use-mention dichotomy, mentioned or appositive terms are in italics. The word List refers to the List type. To extract an element from a stream, apply extract.
Mathematica built-ins, like List and FoldList, are written in a Mathematica input font, as shown to the left. Literal Mathematica expressions like extract[___] are in the same font.
Some built-ins are types, like List, and some are functions, like FoldList. The word List, when referring to a general notion of the type of lists, is not in Mathematica input font.
A general instance of a type like List is written in lower case: the list [1,2,3] is an instance of type List.
Sometimes List, in font, means an instance of type List, when we wish to mean a list specific to Mathematica. The Mathematica expression {1,2,3} is both a list and a List.
New terms are written at first in bold italics, later in appropriate typography.
Double quote are pejorative: a “way out” of a list is some way of extracting a value, as in Mathematica’s Part built-in.

Lists

In general, any fold takes three arguments: a binary query function, a starting output value, and a collection of input values. The binary query takes two arguments: the current output value and the current input value, and produces a new output value (the queries for right folds and left folds take their arguments in the opposite order, but we don’t consider left folds in this notebook).
Mathematica’s basic Fold produces just one final output value, exiting the collection domain of List. FoldList produces all intermediate output values in a new List. We want FoldList because its results compose easily (and composition is good). Using Haskell-style functional type notation, the type of Mathematica’s FoldList is as follows:

FoldList :: binaryQueryFunction ⟶ startingValue ⟶ List ⟶ List

Example: Fibonacci

This version of Fibonacci doesn't actually use the values in the third argument, the collection, it just needs them to be there to “drive” the computation.
FoldList[​​(*binaryqueryfunction*)​​{acc,v}{acc〚2〛,acc〚1〛+acc〚2〛},​​(*startingvalue*)​​{0,1},​​(*collection(List)ofothertenvalues*)​​Range[0,9]]
{{0,1},{1,1},{1,2},{2,3},{3,5},{5,8},{8,13},{13,21},{21,34},{34,55},{55,89}}
This produces a new List from the given List. The new List has one extra element, because FoldList produces the starting accumulation value if the input List is empty. Our request for 10 elements above — via Range[0,9] — produces 11 elements.
FoldList cannot produce an empty List, no matter what the query produces and no matter how many elements are in the input List:
FoldList[{a,b}Null,42,{}]
{42}

Streams

Let's write folds for potentially infinite, lazy, iteractive [sic] collections — streams. Represent a stream as a pair of a value and a thunk (procedure of no arguments); the thunk must produce another stream. An idiomatic way to write a thunk from any expression
E
is E&. It’s elegant and also idiomatic to pick up instances of Stream by matching {v_, thunk_}, so I won’t apologize for the lack of something like a class. Classes are not idiomatic in Mathematica (though see this blog post).
Here's an example stream factory that produces an infinite stream of Integers starting at n:
integersFrom[n_Integer]:={n,integersFrom[n+1]&}
Technically, any such stream is semi-infinite, producing all Integers greater than or equal to
n
. If we extract a value from a slot
<n
, we get Null, which doesn’t print by itself.

extract :: Stream ⟶ value

Extract is a (tail-recursive) way to get the 1-indexed,
n
-th value from any stream. Extract is not Stream-to-Stream, but Stream-to-Value — it’s a “way out.” Other ways-out of Stream are forEach and reify, and we get to them below.
By convention, a finite stream has a Null thunk at the end. Thus, the empty stream, by invoking such a thunk, is Null[]. Extracting anything from such a stream produces Null, which does not print by itself.
The following five, ordered patterns implement extract:
ClearAll[extract];​​(*defaultingn:*)​​extract[{v_,thunk_}]:=extract[{v,thunk},1];​​(*basecase;Null[]istheemptystream*)​​extract[Null[],_]:=Null;​​(*Dothisifyoujustwantthefirstvalue;itisthedefault*)​​extract[{v_,thunk_},1]:=v;(*Somebodyinvokedmythunk!*)​​(*tailrecursion:*)​​extract[{v_,thunk_},n_Integer/;n>0]:=extract[thunk[],n-1];​​(*catch-all:*)extract[___]:=Null;
Get the
630000
-th integer efficiently (if not quickly):
Block[{$IterationLimit=Infinity},​​extract[integersFrom[1],630000]]//AbsoluteTiming
{0.643266,630000}
Extracting 0 values produces Null — nothing in the output cell; in fact, no output cell:
extract[integersFrom[0],0]
Extracting from before the beginning produces Null:
extract[integersFrom[0],-2]
Here is how that works, by invoking the fifth, catch-all pattern, extract[___], in the definition of extract:
ResourceFunction["TraceView"][extract[integersFrom[0],-2]]

take :: Stream ⟶ index ⟶ Stream

Here’s the first of the big results in this notebook: take is a function that takes a finite number of elements from a possibly infinite stream and produces a finite stream:
ClearAll[take];​​take[stream_]:=take[stream,1];​​(*theemptyStream*)​​take[_,0]:=Null[];​​(*newStreamwithexactlyoneelement:*)​​take[{v_,thunk_},1]:={v,Null};(*treatingNullasathunk!*)​​(*recursivecase*)​​take[{v_,thunk_},n_Integer/;n>0]:={v,take[thunk[],n-1]&};​​(*WhathappensifyoutakemorethantheStreamhas?*)​​take[Null[],_]:=Null[];
Here it is at work on the infinite stream of all Integers, producing a finite stream of three Integers, from which we extract the first value:
extract[take[integersFrom[1],3],1]
... and the last value:
If we extract too far into a finite stream, we get Null, which doesn’t print to the notebook:

Finite Streams

Test extract on a finite stream, produced by integersDownFrom:
Extract all elements:
Go too far — Null in a List self-prints:
Look at the wrong end, and go too far:
Consider extract on take, staying within the known bounds; the first element of every take is the first element of the original stream:
Exceed the bounds; if we take too many, we can still extract the first:
a corner case — integersDownFrom stops at 1, so integersDownFrom[n < 1] must be empty:

reify :: Stream ⟶ List

Reify a stream to a list, rattling out all thunks (this is not tail-recursive; we might have to do better if we overflow the stack; it’s not hard to do better, the function just won’t be so obviously correct):
On a finite take of an infinite stream:
On a finite stream:
On an empty stream:
When you take too many from a finite stream, you get all elements, but no more, in a list:
If you forget to call Null-treated-as-a-thunk, reify does not reduce:

disperse :: List ⟶ Stream

Go the other way, producing a finite stream from a list:
disperse and reify will round-trip. On an empty:
On a finite collection:

Fibonaccis as a Stream

Here’s a direct, semi-infinite, non-tail recursive, fibonacci stream (with memoization, i.e., dynamic programming)
This is efficient, modulo stack overflow: it can find large Fibonacci numbers quickly. Pick up the 200-th element of the semi-infinite stream:
Pick up the first element of a semi-infinite stream that begins at element 200:

foldrStream :: binaryQueryFunction ⟶ initialValue ⟶ Stream ⟶ Stream

Here is a fold for Stream with the same signature and semantics as FoldList for List. The “r” in “foldrStream” means “right.” This is a right-fold, distinct from the (not shown) left-fold:
foldrStream produces a stream from a given stream. It’s first-class, as composable as take.
Just as FoldList can’t produce an empty List, foldrStream can’t produce an empty stream.
This next example is another way — via foldrStream — to produce a semi-infinite stream of fibonaccis. It produces all the fibonaccis, plus the starting value at the beginning. That’s one more than an infinite number, which is exactly the same infinity. Otherwise, it’s just like our way of producing a List of fibonaccis via FoldList in Input 2 of this notebook:
Its first value is the starting value of the accumulation:
The first 10 non-trivial fibonaccis must include one extra accumulation value, the zero accumulation, just as in the original FoldList example. Just take one more, 11, than the desired number, 10:
foldrStream also works on finite streams; here is the count-up variant of integersDownFrom:
Expect the empty stream, Null[], from the following (Null[] prints, unlike Null):
Reify an empty List from an empty stream:
foldrStream does not produce an empty result, but rather the first accumulator value, so we can never reify an empty List from a foldrStream:
As always with any fold, the length of the result is one more than the length of the input stream:

Iterable

forEach :: Stream ⟶ unaryQueryFunction ⟶ Null

An interface is a function that take a collection object as first argument, by convention, just as you'd emulate object-oriented programming in C. There are multiple approaches to object-oriented programming in Mathematica; too many, in fact, so we won’t bother.
An iterable is an interface with one method, forEach, which takes an iterable and a unary, side-effecting query function, and invokes the query on every element of the iterable. It’s side-effecting way to pull values out of an iterable, so it’s a legitimate “way out.” The analogous, non-side-effecting operator is Map, not pursued here for brevity. Streams are the only iterables in this notebook.
It’s bad to invoke forEach on infinite streams, just as with reify. You’ll get a non-terminating computation.
Here is forEach for Stream:
and here it is at work, side-effecting a take of allFibs via Print:

Observable

An observable is an interface with one method: subscribe, which has the same signature and semantics as forEach. The difference is that an observable produces its results asynchronously. The big lesson of this whole notebook is that the query function can’t tell whether it’s invoked on values in a list, values from a stream, or values (observations) from an observable. Put your algorithm in the query, and you’ve separated the concerns of algorithms from the concern of synchrony. This strategy works for very big algorithms, like Kalman filters.
Model the asynchronous behavior of observables with Mathematica Tasks; model the observables themselves differently because we need the subscribe interface for them.
Here's an finite Task. You MUST interact with this, say via Dynamic, to appreciate it. Evaluate the following input cell and watch the values tick-over five times in the output cell:
That task cleans itself up. If you evaluate the following cell quickly enough, while the Task is running, you’ll see the task in some kind of waiting area inside Mathematica. If you evaluate the following cell after the Task is finished, you’ll see an empty List.
Observables must produce values every tick so we can subscribe query functions to them. Model observables as stateful thunks that produce new values every time they’re invoked, then invoke the thunks inside asynchronous Tasks that start when you subscribe.
Such tasks implement cold observables, which don't produce values until subscribed. By contrast, hot observables produce values whether subscribed or not, and you can subscribe and unsubscribe on-the-fly. We'll finesse those now because the purpose here is to show that subscribe is almost the same as forEach. Our purpose here is illustrative, not to produce a complete Observable library (see this book by Lee Campbell for much more).

Fibonacci as an Observable

Here’s a factory for fibonacci observables. Call it, and you’ll get a function that you can apply to a List to get a number of results equal to the Length of the List:
Here we show 11 calls of oldFibs, resulting in statefully updated values:
Of course, we’ve left garbage in the environment (e.g., local variables created by the Module). But we’ll finesse that now, too. All we did was call oldFibs 11 times. We only want the last 10 values, but we must also produce the original accumulation value, just as was automatically done with FoldList and foldrStream. We want all our folds to behave the same.
Watch this until it stops changing:
Catch it in time, or not; this shows tasks-in-progress:
Just as we forEach a unary query over a synchronous List or stream, we subscribe a unary query to an observable. Such queries are also called observers, but they’re just ordinary unary functions.

subscribe :: Observable ⟶ unaryQueryFunction ⟶ Null

We emphasize: the syntax and semantics of forEach and subscribe are virtually identical:
Watch this until you get tired (it’s infinite), then clean it up:
Here’s an infinite observable of all the integers, just like our iterable integersFrom:
Watch the following for a while, then clean it up:
Here are our ways to get back and forth between Lists and Observables (doing this right requires another aspect of observers, namely OnCompleted, not covered here [see the book by Campbell for more]. We’ll joke it with Null for now).

dispense :: List ⟶ Observable

Here’s a finite instance. Clean it up once it starts to produce Null.

harvest :: Observable ⟶ List

The task does not clean itself up. We can capture the output list by evaluating r$ non-dynamically before doing the cleanup:

takeObservable :: Observable ⟶ Integer ⟶ Observable

Here’s our version of take for observables:
and the usual test:
All that’s left is to produce a new observable by folding a binary query function over another observable.

foldObservable :: binaryQueryFunction ⟶ initialValue ⟶ Observable ⟶ Observable

There are lots of better ways to do this, but we’ve captured the gist.

Conclusion

We’ve established that folding over Lists produces Lists, folding over iterable streams produces iterable streams, and that folding over observables produces observables. All our algorithms, expressed as query functions, can run over finite, infinite, synchronous, and asynchronous collections.

CITE THIS NOTEBOOK

Folds over Lists, Streams, and Asynchronous Observables​
by Brian Beckman​
Wolfram Community, STAFF PICKS, November 29, 2023
​https://community.wolfram.com/groups/-/m/t/3074231