twitter/algebird — pull request #346 by ejconlon (2014-09-15 23:41:50+00:00)
Title: Add Fold trait for composable incremental processing

Problem

(Little-f) folds are a standard part of a Scala-programmer's toolkit. Most sequence-like data
structures support them directly (Traversable, com.twitter.concurrent.Spool), and some suggest a natural implementation (com.twitter.io.Reader). Consider calculating a message digest over these structures
(simplified):

import java.security.MessageDigest

def digestSeq(bs: Seq[Buf]): Array[Byte] =
(bs.foldLeft(new MessageDigest) { case (md, buf) => md.update(buf) }).digest

def digestSpool(bs: Spool[Buf]): Future[Array[Byte]] =
spool.foldLeft(new MessageDigest) { case (md, buf) => md.update(buf) } map { _.digest }

def digestReader(reader: Reader): Future[Array[Byte]] =
// ... some recursive flatMapping and mapping ...

This is noisy - we can imagine separate mapping, accumulating, and iterating concerns and express
it this way:

val digest: Fold[Buf, Array[Byte]] =
Fold.foldMutable(
case (md, buf) => md.update(buf)
{ _ => new MessageDigest }
{ _.digest }
)

def digestSeq(bs: Seq[Buf]): Array[Byte] =
digest.overTraversable(bs)

def digestSpool(bs: Spool[Buf]): Future[Array[Byte]] =
digest.overSpool(bs)

def digestReader(reader: Reader): Future[Array[Byte]] =
digest.overReader(bs)

Now what if we wanted to perform another operation over the stream, say, "Aggregation":

import java.security.MessageDigest

type Aggregation
def emptyAgg: Aggregation
def updateAgg(agg: Aggregation, buf: Buf): Aggregation

def digestAndAggregateSeq(bs: Seq[Buf]): (Array[Byte], Aggregation) = {
val (md, agg) =
(bs.foldLeft((new MessageDigest, emptyAgg)) { case ((md, agg), buf) =>
(md.update(buf), updateAgg(agg, buf)
})
(md.digest, agg)
}

// Spool impl is equally complicated

We have to explicitly fuse the second fold. Instead we could do this:

val aggregate =
Fold.fold(
updateAgg _,
emptyAgg,
id
)

val digestAndAggregate = digest join aggregate

def digestAndAggregateSeq(bs: Seq[Buf]): (Digest, Aggregation) =
digestAndAggregate.overTraverable(bs)

def digestAndAggregateSpool(bs: Spool[Buf]): (Digest, Aggregation) =
digestAndAggregate.overSpool(bs)

Much simpler!

Solution

This RB introduces the Fold[I, O] type, following libraries like
https://hackage.haskell.org/package/foldl . (We emphasize that the ideas here are not novel to the
world, and not even novel to algebird - see Aggregation.)

Simplified:

trait Fold[I, O] {
// Produces a new Fold that transforms the output of this Fold
def map[P](f: O => P): Fold[I, P]

// Produces a new Fold that tees inputs to this and another Fold, returning both results
def join[P](other: Fold[I, P]): Fold[I, (O, P)]

// Runs this fold over the elements of a Traversable
def overTraversable(is: Traversable[I]): O

// Runs this fold over the elements of a Spool, flatMapping along the way
def overSpool(spool: Spool[I]): Future[O]

// "Build" this so we can run it with overTraversable/overSpool.
type X
def build: FoldState[X, I, O]
}

case class FoldState[X, I, O](
add: (X, I) => X, // Adds I to our state
start: X, // State
end: X => O // Transform state to final result
)

Folds do not have internal state themselves; they instead "build()" a FoldState we can use to fold.
Because of this, Folds can be run many times, and they can be run over datatypes not defined
in algebird.

Fold is so simple and fundamental that algebird-core seems like an appropriate place. It makes
consuming Seqs, Spools, and Readers easy and consistent. A number of teams have expressed interest
in using and sharing this common type.

This type also admits an effectful variant that can sequence Futures before, during, and after
folding.

Result

We will have a basic vocabulary to express Folds simply, succinctly, and reusably.