So this time, our main piece of code will be quite similar to the previous one:

ConnectableObservable<String> input = CreateObservable.from(System.in);

Observable<Double> a = varStream("a", input);
Observable<Double> b = varStream("b", input);

reactiveSum(a, b); // The difference

input.connect();

The only difference is that we are going to take a more functional approach in calculating our sum and not to keep the same state. We won't be implementing the Observer interface; instead, we are going to pass lambdas to subscribe. This solution is much cleaner.

The CreateObservable.from(InputStream) method is a lot like we used previously. We will skip it and look at the Observable<Double> varStream(String, Observable<String>) method, which creates the Observable instances representing the collectors:

public static Observable<Double> varStream(
  final String name, Observable<String> input) {
    final Pattern pattern =     Pattern.compile(
      "\s*" + name + "\s*[:|=]\s*(-?\d+\.?\d*)$"
    );
    return input
    .map(pattern::matcher) // (1)
    .filter(m -> m.matches() && m.group(1) != null) // (2)
    .map(matcher -> matcher.group(1)) // (3)
    .map(Double::parseDouble); // (4)
  }
)

This method is much shorter than used previously and looks simpler. But semantically, it is the same. It creates an Observable instance connected to a source observable producing arbitrary strings, and if a string is in the format it expects, it extracts a double number from it and emits this number. The logic responsible for checking the format of the input and extracting the number is only four lines and is represented by simple lambdas. Let's examine it:

And as for the new void reactiveSum(Observable<Double>, Observable<Double>) method's implementation, use the following code:

public static void reactiveSum(
  Observable<Double> a,
  Observable<Double> b) {
    Observable
      .combineLatest(a, b, (x, y) -> x + y) // (1)
      .subscribe( // (2)
        sum -> System.out.println("update : a + b = " + sum),
    error -> {
      System.out.println("Got an error!");
      error.printStackTrace();
    },
    () -> System.out.println("Exiting...")
  );
}

Let's take a look at the following code:

Everything becomes simpler using lambdas. Looking at the preceding program, we can see that most of the logic is composed of small, independent functions, chained using other functions. This is what we mean by being functional, to express our programs using such small reusable functions that take other functions and return functions and data abstractions, which transform input data using chains of functions in order to produce the wanted result. But let's look at these functions in depth.