Chapter 6. Reactive Data Streams

In particular circumstances, our application may be required to handle huge file uploads. This can be done by putting all of these in the memory, by creating a temporary file, or by acting directly on the stream. Out of these three, the last option works the best for us, as it removes I/O stream limitations (such as blocking, memory, and threads) and also eliminates the need to buffer (that is, acting on input at the rate needed).

Handling huge file uploads belongs to the set of unavoidable operations that can be heavy on resources. Some other tasks that belong to the same category are processing real-time data for monitoring, analysis, bulk data transfers, and processing large datasets. In this chapter, we will discuss the Iteratee approach used to handle such situations. This chapter covers the basics of handling data streams with a brief explanation of the following topics:

• Iteratees

• Enumerators

• Enumeratees

This chapter may seem intense at times but the topics...

Basics of handling data streams

Consider that we connected a mobile device (such as a tablet, phone, MP3 player, and so on) to its charger and plugged it in. The consequences of this can be as follows:

• The device's battery starts charging and continues to do so until the occurrence of one of the other options

• The device's battery is completely charged and minimal power is drawn by the device to continue running

• The device's battery can not be charged due to malfunctioning of the device

Here, the power supply is the source, the device is the sink, while the charger is the channel that enables transfer of energy from the source to the sink. The processing or task performed by the device is that of charging its battery.

Well, this covers most of the Iteratee approach without any of the usual jargon. Simply put, the power supply represents a data source, the charger acts as the Enumerator, and the device as the Iteratee.

Oops, we missed the Enumeratee! Suppose that the energy from a regular power...

Iteratees

Iteratee is defined as a trait, Iteratee[E, +A], where E is the input type and A is the result type. The state of an Iteratee is represented by an instance of Step, which is defined as follows:

sealed trait Step[E, +A] {

  def it: Iteratee[E, A] = this match {
    case Step.Done(a, e) => Done(a, e)
    case Step.Cont(k) => Cont(k)
    case Step.Error(msg, e) => Error(msg, e)
  }

}

object Step {

  //done state of an iteratee
  case class Done[+A, E](a: A, remaining: Input[E]) extends Step[E, A]

  //continuing state of an iteratee.
  case class Cont[E, +A](k: Input[E] => Iteratee[E, A]) extends Step[E, A]

  //error state of an iteratee
  case class Error[E](msg: String, input: Input[E]) extends Step[E, Nothing]
}

The input used here represents an element of the data stream, which can be empty, an element, or an end of file indicator. Therefore, Input is defined as follows:

sealed trait Input[+E] {
  def map[U](f: (E => U)): Input[U] = this match {
    case Input...

Enumerator

Similar to the Iteratee, an Enumerator is also defined through a trait and backed by an object of the same name:

trait Enumerator[E] {
  parent =>
  def apply[A](i: Iteratee[E, A]): Future[Iteratee[E, A]]
  ...
}
object Enumerator{
def apply[E](in: E*): Enumerator[E] = in.length match {
    case 0 => Enumerator.empty
    case 1 => new Enumerator[E] {
      def apply[A](i: Iteratee[E, A]): Future[Iteratee[E, A]] = i.pureFoldNoEC {
        case Step.Cont(k) => k(Input.El(in.head))
        case _ => i
      }
    }
    case _ => new Enumerator[E] {
      def apply[A](i: Iteratee[E, A]): Future[Iteratee[E, A]] = enumerateSeq(in, i)
    }
  }
...
}

Observe that the apply method of the trait and its companion object are different. The apply method of the trait accepts Iteratee[E, A] and returns Future[Iteratee[E, A]], while that of the companion object accepts a sequence of type E and returns an Enumerator[E].

Now, let's define a simple data flow using the companion object...

Enumeratees

Enumeratee is also defined using a trait and its companion object with the same Enumeratee name.

It is defined as follows:

trait Enumeratee[From, To] {
...
def applyOn[A](inner: Iteratee[To, A]): Iteratee[From, Iteratee[To, A]]

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

An Enumeratee transforms the Iteratee given to it as input and returns a new Iteratee. Let's look at a method that defines an Enumeratee by implementing the applyOn method. An Enumeratee's flatten method accepts Future[Enumeratee] and returns an another Enumeratee, which is defined as follows:

  def flatten[From, To](futureOfEnumeratee: Future[Enumeratee[From, To]]) = new Enumeratee[From, To] {
    def applyOn[A](it: Iteratee[To, A]): Iteratee[From, Iteratee[To, A]] =
      Iteratee.flatten(futureOfEnumeratee.map(_.applyOn[A](it))(dec))
  }

In the preceding snippet, applyOn is called on the Enumeratee whose future is passed and dec is defaultExecutionContext.

Defining...

Summary

In this chapter, we discussed the concept of Iteratees, Enumerators, and Enumeratees. We also saw how they were implemented in Play Framework and used internally. This chapter also walked you through a simple example to illustrate how data flow can be defined using the API exposed by Play Framework.

In the next chapter, we will explore the features provided in a Play application through a global plugin.