Enumeratees

Handling data streams reactively

The realm of Enumeratees

Enumeratee is a very important component of the iteratees api. It provides the means for adapting and tranforming streams of data. An Enumeratee that might sound familiar is the Enumeratee.map.

Starting with a simple problems, consider the following Iteratee:

  val sum:Iteratee[Int,Int] = Iteratee.fold[Int,Int](0){ (s,e) => s + e }

This Iteratee takes Ints as input and computed their sum. Now if we have an Enumerator like the following:

  val strings:Enumerator[String] = Enumerator("1","2","3","4")

Then obviously we can not apply the strings:Enumerator[String] to an Iteratee[Int,Int]. What we need is transform each String to the corresponding Int so that the source and the consumer can be fit together. This means we either have to adapt the Iteratee[Int,Int] to be Iteratee[String,Int], or adapt the Enumerator[String] to be rather an Enumerator[Int]. Enumeratees are the right tool for doing that. We can create an Enumeratee[String,Int] and adapt Iteratee using it:

  //create am Enumeratee using the map method on Enumeratee
  val toInt: Enumeratee[String,Int] = Enumeratee.map[String]{ s => s.toInt } 

  val adaptedIteratee: Iteratee[String,Int] = toInt.transform(sum)

  //this works!
  strings >>> adaptedIteratee

There exist a symbolic alternative to the transform method, &>> which we can use in our previous example:

  strings >>> toInt &>> sum 

The map method will create an 'Enumeratee' that uses a function to map the input from the From type to the To type. We can also adapt the Enumerator:

  val adaptedEnumerator: Enumerator[Int] = strings.through(toInt)

  //this works!
  adaptedEnumerator >>> sum

Here too, we can use a symbolic version of the through method:

  strings &> toInt >>> sum

Let's have a look at the transform signature defined in the 'Enumeratee' trait:

  trait Enumeratee[From, To] {

    def transform[A](inner: Iteratee[To, A]): Iteratee[From, A] = ...

  }

Fairly simple signature, same for the through defined on an Enumerator :

  trait Enumerator[E]{

    def through[To](enumeratee: Enumeratee[E, To]): Enumerator[To] 

transform and through methods on an Enumeratee and Enumerator respectively both use the apply method on Enumeratee that has a slightly more sophisticated signature:

  trait Enumeratee[From, To] {

    def apply[A](inner: Iteratee[To, A]): Iteratee[From, Iteratee[To, A]] = ...

Indeed, an Enumeratee is more powerful than just transforming an Iteratee type. It really acts like an adapter in that you can get back your original Iteratee after pushing some different input through an Enumeratee. So in the previous example, we can get back the original Iteratee[Int,Int] to continue pushing some Ints in:

    val sum:Iteratee[Int,Int] = Iteratee.fold[Int,Int](0){ (s,e) => s + e }
  

    //create am Enumeratee using the map method on Enumeratee
    val toInt: Enumeratee[String,Int] = Enumeratee.map[String]{ s => s.toInt } 

    val adaptedIteratee: Iteratee[String,Iteratee[Int,Int]] = toInt(sum)

    // pushing some strings
    val afterPushingStrings:  Iteratee[String,Iteratee[Int,Int]] = Enumerator("1","2","3","4") >>> adaptedIteratee

    val originalIteratee: Iteratee[Int,Int] = flatten(afterPushingString.run)

    val moreInts : Iteratee[Int,Int] =  Enumerator(5,6,7) >>> originalIteratee

    moreInts.run.onRedeem(sum => println(sum) ) // eventually prints 28

That's why we call the adapted (original) Iteratee inner and the resulting Iteratee outer. Now that the Enumeratee full picture is clear, it is very important to know that transform drops the left input of the inner Iteratee when it is Done. This means that if we use Enumeratee.map to transform input, if the inner Iteratee is Done with some left transformed input, the transform method will just ignore it.

That might have seemed like a bit too much of details but it is useful for grasping the model. Back to our example on Enumeratee.map, there is a more general method Enumeratee.mapInput which gives the opportunity to return an EOF on some signal:

  val toIntOrEnd: Enumeratee[String,Int ] = Enumeratee.mapInput[String] {
     case Input.El("end") => Input.EOF
     case other => other.map(e => e.toInt)
  }

Enmeratee.map and Enumeratee.mapImput are pretty straight forward, they operate on a per chunk bases and they convert them. Another useful Enumeratee is the Enumeratee.filter :

    def filter[E](predicate: E => Boolean): Enumeratee[E, E]

The signature is pretty obvious, Enumeratee.filter creates an Enumeratee[E,E] and it will test each chunk of input using the given predicate: E => Boolean and it passes it along to the inner (adapted) iteratee if it statisfied that predicate:

  val numbers = Enumerator(1,2,3,4,5,6,7,8,9,10)

  val onlyOdds = Enumeratee.filter[Int](i => i % 2 != 0)

  numbers.through(onlyOdds) >>> sum

Other methods exist like Enumeratee.collect, Enumeratee.drop, Enumeratee.dropWhile, Enumeratee.take, Enumeratee.takeWhile that work on the same principle. Let try to use the Enumeratee.take on an Input of chunks of bytes:

  // complutes the size in bytes
  val fillInMemory: Iteratee[Array[Byte],Int] = Iteratee.consume[Array[Byte]]()

  val limitTo100: Enumeratee[Array[Byte],[Array[Byte]]] = Enumeratee.take[Array[Byte]](100)

  val limitedFillInMemory: Iteratee[Array[Byte,Int]] = limit &>> fillInMemory
  

It looks good, but how many bytes are we taking? What should be done ideally is to limit the size, in bytes, of loaded input. What is done above instead is limit the number of chunks; whatever the size of each chunk is. It seems that the Enumeratee.take is not enough here since it has no information about the type of input (in our case an Array[Byte]) and this is why it can't count what's inside.

Luckily there is the Traversable object that offers a set of methods for creating Enumeratees for Input types that are TraversableLike. An Array[Byte] is TraversableLike and so we can useTraversable.take:

  // complutes the size in bytes
  val fillInMemory: Iteratee[Array[Byte],Int] = Iteratee.consume[Array[Byte]]()

  val limitTo100: Enumeratee[Array[Byte],[Array[Byte]]] = Traversable.take[Array[Byte]](100)

  // We are sure not to get more than 100 bytes loaded into memory
  val limitedFillInMemory: Iteratee[Array[Byte,Int]] = limit &>> fillInMemory

Other Traversable methods exist including Traversable.takeUpTo, Traversable.drop.

Finally you can compose different Enumeratees using the compose method that has the symbolic equivelant ><>. Note that any left input on the Done of the composed Enumeratees will be dropped. If you use composeConcat aliased >+>, any left input will be concatened.