std/iteration/collectors

What to do with the values of a pipeline, and the state of doing it: values arrive one at a time through add, and finish produces the result once they have all arrived. collect(accumulator) is the general terminal operation of Iterate - toList, count, fold and the rest are all written on top of it.

A description and a run are one type, because a value is a copy. collect fills a copy of what it was given, so one accumulator may be collected with twice and the two runs never see each other, and then(downstream) needs no factory: it keeps the downstream and copies it per group.

It stands alone. A collection does not implement it: a container that is merely filled has no result of a run to give and no isDone() to answer, so what gathers into one is a type beside it (ListAccumulator, into).

const numbers = [1, 2, 3, 4, 5]
print numbers.collect(summing { value: Int => value })

• A description and a half-filled run have the same type, so an accumulator that has already been added to and is then handed to collect or to then starts every run from what is in it. Build one where it is used (collect(listing())), the way every collector of this package answers a fresh value.

• ListAccumulator - the one this package ships, and the shape a package writes for its own collection.
• groupingBy - the one accumulator with a then(downstream) for a different one per group.

Two partial results of one accumulator, joined: what lets an accumulator run over pieces of its input - the chunks of parallel(), a divide-and-conquer fold, a map-reduce over the parts of a file - and still answer what one run over the whole input answers.

The contract is three lines. merge is associative: merge(merge(a, b), c) is merge(a, merge(b, c)). It need not be commutative, because it is called in the order of the pieces, left to right. And it is never called with the output of a piece that received no value: such a piece is skipped, and an input without any value takes the accumulator's own finish() instead. That third line is what makes joining(separator:, prefix:, suffix:) mergeable at all - without it every merge would have to decide whether a separator belongs between two pieces one of which is not there.

It joins two outputs and not two accumulators: two trait-typed accumulators need not have the same type, and the output is what comes back from wherever a piece ran anyway. A collector whose output lost what a merge would need - averaging, whose count is gone once it divided - has none; the shape of the workaround is a mergingCollector whose output keeps it.

const counter = counting<Int>()
print counter.merge([1, 2].collect(counter), [3].collect(counter))

• mergingCollector - a fold with a final step and a merge, the way to write one.
• Accumulator - what it adds the merge to.

Collects into anything a pipeline can be turned into (From<Iterate<Item>>), which every collection is. Target can be given on its own, the rest is inferred.

It gathers into a List and calls Target.from once at the end, which is the answer that works for every target. A package that owns a collection may ship an accumulator that writes straight into it - ListAccumulator is the shape - and a caller who wants that one names it (collect(ListAccumulator<Int>())). Nothing picks it up automatically: which run a pipeline makes is written at the call, not inferred from the target.

const numbers = [3, 1, 2, 1]
print numbers.collect(into<Set<Int>>())

Shorthand for ListAccumulator<Item>(), which is what a pipeline ends in most often. Two lists merge by concatenation.

Gathers the values into a List and answers it: the specialised accumulator this package ships, and the shape a package that owns a collection writes for its own (HashSetAccumulator writing straight into a set).

print([3, 1, 2].collect(ListAccumulator<Int>()))

State is the starting point of every run. Every run works on its own copy.

fn counting<Item>(): Accumulator<Item, Int> {
  collector(0, finish: { _ }) { count, _ => count + 1 }
}

collector with a Merge: merge joins the outputs of two pieces of the input, in their order, and is never handed the output of a piece that received no value. It has to be associative, so that the answer does not depend on where the pieces were cut.

An average merges once its output keeps the count:

const averaged = mergingCollector((0.0, 0), finish: { _ }, merge: { first, second =>
  (first.0 + second.0, first.1 + second.1)
}) { state, value: Float => (state.0 + value, state.1 + 1) }
const total = [1.0, 2.0, 6.0].collect(averaged)
print(total.0 / Float.from(total.1))

How many values there are. Two counts merge by +.

Every value, added together, starting from zero. Two sums merge by +.

None for an empty pipeline.

The value with the smallest key, or None for an empty pipeline. Keeps the earlier of two equal keys, and so does its merge: the second piece's value wins only with a smaller key.

The value with the largest key, or None for an empty pipeline. Keeps the earlier of two equal keys, and so does its merge: the second piece's value wins only with a larger key.

Every value, joined into one String, with separator between values and prefix/suffix around all of them.

It buffers the pieces and concatenates once at the end, for the reason concatenated gives: a String is a value, so appending to one copies everything that is already in it.

(matching, rest). Two pairs merge list by list.

groupingBy { _.department }                              // Map<String, List<Employee>>
groupingBy { _.department }.then(counting())             // Map<String, Int>
groupingBy { _.department }.then(maxBy { _.salary }) // Map<String, Employee?>