Dealing with Java quirks while switching to functional programming style

Thus Java constantly gains more features supporting writing code in functional style, starting with 2014’s addition of lambdas and streams of collection elements.

Yet it looks like the language architects are not convinced whether fully embrace that style by not providing basic combinator methods (zip, zipWith, fold) with existing Optional<T> and Stream<T> classes, nor classes that can carry information about a failure like Try<T> or Either<E, T>.

This limits the possibility of writing more complicated logic without a nasty mixture of good ol' try {} s and mutable elements.

In this article, we’ll explore a couple of libraries that make it easier to stay with the FP side.

All code examples and AsciiDoc version of this article are provided in this GitHub repository.

A temptation to use mutable state in a Stream<T> comes when we need to access a previous element that has already been processed. Say we need to summarize changes in Customer objects for auditing purposes.

First, let’s come up with machinery for producing a string describing differences between 2 objects. Here defined is a map of attribute names to their projectors on a customer object.

  @Value
  private static class ComparableAttribute {
    String name;
    Function<Customer, String> getter;
  }

  private static final List<ComparableAttribute> COMPARABLE_ATTRIBUTES = List.of(
      new ComparableAttribute("name", Customer::getName),
      new ComparableAttribute("address", c -> c.getAddress().toString()),
      new ComparableAttribute("born on", c -> c.getBornOn().toString()),
      new ComparableAttribute("is active", c -> c.getActive().toString())
  );

Now let’s try to define the actual builder of a String with changes summary.

  private static Optional<String> valueDiff(String valueName, String v1, String v2) {
    if (v1.equals(v2)) return Optional.empty();
    else return Optional.of(valueName + ": " + v1 + " -> " + v2);
  }

  static String customerDiff(Customer c1, Customer c2) {
    return COMPARABLE_ATTRIBUTES
        .stream()
        .map(attr -> valueDiff(attr.name, attr.getGetter().apply(c1), attr.getter.apply(c2)))
        .filter(Optional::isPresent)
        .map(Optional::get)
        .collect(Collectors.joining(" | "));
  }

(test data at the end of the section):

  @Test
  void customerDiffTest() {
    assertThat(Zipping.customerDiff(c1, c3)).isEqualTo("name: Johny Kovalsky -> Jan Kowalski | born on: 2014-03-18T12:00Z -> 2019-03-18T12:00Z | is active: true -> false");
  }

And then use it:

  static List<String> compareSubsequentChangesWithAtomicRefence(List<Customer> customerStateSnapshots) {
    if (customerStateSnapshots.size() < 2) return Collections.emptyList();
    final var lastValue = new AtomicReference<>(customerStateSnapshots.get(0));
    return customerStateSnapshots
        .stream()
        .skip(1)
        .map(customer -> customerDiff(lastValue.getAndSet(customer), customer))
        .collect(Collectors.toList());
  }

The lastValue constant indicates a strong desire to use Streams and problem of enforcing that used variables must be declared as final. AtomicReference<Customer> is a quick hack for changing a for-each loop to the New Fancy Functional Streams™.

It’s a nasty hack, of course. All the promises of simplicity about reasoning about code are thrown away when one needs to keep track of all the places where a variable can be mutated. One can argue that it’s not a big deal when the mutable state isn’t leaked outside such a method, and it’s a valid claim. However, in this case, it’d be easier to just use a variable and old loops.

Zipping it

  static List<String> compareSubsequentChanges(List<Customer> customerStateSnapshots) {
    if (customerStateSnapshots.size() < 2) return Collections.emptyList();
    final var vavrList = io.vavr.collection.List.ofAll(customerStateSnapshots);
    return vavrList
        .zipWith(vavrList.drop(1), Zipping::customerDiff)
        .asJava();
  }

Generalizing with foldX

  @Value
  private static final class ComparisonState {
    final Customer lastVale;
    final io.vavr.collection.List<String> stateAcc;
  }

  static List<String> compareSubsequentChangesWithFoldLeft(List<Customer> customerStateSnapshots) {
    if (customerStateSnapshots.size() < 2) return Collections.emptyList();
    final var zero = new ComparisonState(customerStateSnapshots.get(0), io.vavr.collection.List.empty());
    final var vavrList = io.vavr.collection.List.ofAll(customerStateSnapshots);
    return vavrList
        .drop(1)
        .foldLeft(zero, (ComparisonState foldAcc, Customer c) ->
            new ComparisonState(c, foldAcc.stateAcc.append(customerDiff(foldAcc.lastVale, c))))
        .stateAcc
        .asJava();
  }

.foldLeft(initialState, (currentState, element) -> newState))

  private Addrees address = Addrees
                .builder()
                .line1(of("Warszawska 1"))
                .line2(empty())
                .zipCode(of("00-000"))
                .city(of("Warsaw"))
                .country(of("Poland"))
                .build();

  private Customer c1 = Customer
            .builder()
            .name("Johny Kovalsky")
            .address(address)
            .active(TRUE)
            .bornOn(ZonedDateTime.of(2014, 3, 18, 12, 0, 0, 0, UTC))
            .build();

  private Customer c2 = Customer
      .builder()
      .name("John Kovalsky")
      .address(address)
      .active(TRUE)
      .bornOn(ZonedDateTime.of(2014, 3, 18, 12, 0, 0, 0, UTC))
      .build();

  private Customer c3 = Customer
            .builder()
            .name("Jan Kowalski")
            .address(address)
            .active(FALSE)
            .bornOn(ZonedDateTime.of(2019, 3, 18, 12, 0, 0, 0, UTC))
            .build();

Checked exceptions don’t blend with Streams. Runtime exceptions don’t blend with predictable methods invocations.

Let’s start with such a service:

  @AllArgsConstructor
  static class CustomerService {
    private io.vavr.collection.List<Customer> customersSource;

    static class ServiceException extends Exception {
      ServiceException(String msg) {
        super(msg);
      }
    }

    Optional<Customer> getByNameOptionalThrowing(String name) throws ServiceException {
      if ("Error-prone Customer".equals(name)) throw new ServiceException("Life is life... Nananana");
      return customersSource.find(c -> c.getName().equals(name)).toJavaOptional();
    }
  }

And a task of getting the average age of a list of customers.

The first attempt…

    try {
      res = names
          .stream()
          .map(name -> cs.getByNameOptionalThrowing(name))
          // further processing
          .collect(Collectors.toList());
    } catch (CustomerService.ServiceException e) {
      log.error("An error obtaining customers", e);
    }

…and a disappointment

Error:(77, 52) java: unreported exception com.kpalka.fpplayground.FailableBehaviour.CustomerService.ServiceException; must be caught or declared to be thrown
Error:(80, 7) java: exception com.kpalka.fpplayground.FailableBehaviour.CustomerService.ServiceException is never thrown in body of corresponding try statement

.map won’t accept something that throws a checked exception. So the second attempt:

  static Integer getAvgAge(CustomerService cs, List<String> names, ZonedDateTime now) {

    // Don't do this at home
    Function<String, Optional<Customer>> aHackYouCanSometimesSpot = name -> {
      try {
        return cs.getByNameOptionalThrowing(name);
      } catch (CustomerService.ServiceException e) {

        // A service tries to inform me in the method signature that something can go wrong.
        // But I cannot use a method that throws an exception inside `.map()`.
        // But I REALLY want to use that fancy Stream feature... Hmm...

        throw new RuntimeException(e);
      }
    };

    // Similar examples are often used to show the possibilities of Stream<T> and method references...

    return names
        .stream()
        .map(aHackYouCanSometimesSpot)

        // ... and when you see such a call to service as a fragment of stream pipeline, you should smell something bad. Things can fail. In a nasty way. And I think such situations make some people, softly said, not very willing to incorporate the newer features of the language to their daily usage


        // the rest of processing
  }

Now, by calling getAvgAge we aren’t even informed that something can go wrong, so it’s easy to forget to handle an error. The argument of easier reasoning about a program written in the functional style apparently doesn’t apply here.

For a moment let’s try to finish the broken implementation and then fix the error handling part.

We’re going to need a class representing a state that will be used in reducing a stream of customers' age to the average value.

  @AllArgsConstructor
  static class AvgPeriodCounter {
    static final AvgPeriodCounter ZERO = new AvgPeriodCounter(Period.ZERO, 0);
    final Period sum;
    final Integer elementsNumber;

    AvgPeriodCounter plus(Period period) {
      return new AvgPeriodCounter(sum.plus(period), elementsNumber + 1);
    }

    AvgPeriodCounter plus(AvgPeriodCounter avgPeriodCounter) {
      return new AvgPeriodCounter(sum.plus(avgPeriodCounter.sum), elementsNumber + avgPeriodCounter.elementsNumber);
    }

    int getAvgYear(ZonedDateTime relativeTo) {
      return Period.between(relativeTo.minus(sum).toLocalDate(), relativeTo.toLocalDate()).getYears() / elementsNumber;
    }
  }

And again, this time the complete implementation:

  static Integer getAvgAge(CustomerService cs, List<String> names, ZonedDateTime now) {
    Function<String, Optional<Customer>> aHackYouCanSometimesSpot = name -> {
      try {
        return cs.getByNameOptionalThrowing(name);
      } catch (CustomerService.ServiceException e) {
        throw new RuntimeException(e);
      }
    };

    var toAge = periodTo(now);

    return names
        .stream()
        .map(aHackYouCanSometimesSpot)
        .filter(Optional::isPresent)
        .map(Optional::get)
        .map(Customer::getBornOn)
        .map(toAge)
        .reduce(AvgPeriodCounter.ZERO,
            (AvgPeriodCounter acc, Period p) -> acc.plus(p),
            (AvgPeriodCounter acc1, AvgPeriodCounter acc2) -> acc1.plus(acc2)
        )
        .getAvgYear(now);
  }

(test data at the end of the section):

  @Test
  void countAvgForExistingCustomers() {
    assertThat(FailableBehaviour.getAvgAge(cs, existingNames, now))
        .describedAs("Counts the avg of existing customers' age")
        .isEqualTo(avgAge);
  }

  @Test
  void countAvgThrowing() {
    assertThatThrownBy(() -> FailableBehaviour.getAvgAge(cs, List.of(errorProneCustomer), now))
        .describedAs("Method can throw an expected error, but doesn't inform that it can fail in any way. Additionally, meaningful ServiceException is wrapped in a very generic RuntimeException")
        .isExactlyInstanceOf(RuntimeException.class);
  }

It’s worth noticing the (over)complicated reduce available in Java’s Stream. Compared to the foldLeft available in vavr, in reduce, we have two stages. The first is the same as in foldLeft; the second combines the states produced by the first stage.

In the first, the new accumulated state is dependent on the previous value of the stream. In the second, states can be combined independently, which means they can be parallelized. It’s great if you need it. If you don’t you have to deal with the burden of defining additional binary operation, here AvgPeriodCounter plus(AvgPeriodCounter avgPeriodCounter) next to AvgPeriodCounter plus(Period period). So if we already have the whole list in memory, using foldX seams to be a sexier solution than reduce.

Now it’s time to tame the method calls that can fail. vavr offers Try<T> and Either<E, T> classes with which you can inform about a possible failure in the return type. Either is more powerful as you can put in Left (failure) part anything signaling the error. Try can be seen as:

    interface Try<T> extends Either<Throwable, T> {}

So the errors can be only a (sub)instance of Throwable. Say it’s good enough for now. Let’s add the following method to CustomerService:

    Try<Optional<Customer>> getByNameWithTry(String name) {
      return Try.of(() -> getByNameOptionalThrowing(name));

Notice that there’s no throws ServiceException in the method signature, which means it can be used within Stream. Notice that the getByNameOptionalThrowing(name) call is now wrapped in a lambda expression. Without it, the exception would be thrown immediately, before vavr had a chance to wrap call execution into Try.Success or Try.Failure.

Now we’re ready to use the new method of the service.

  static Try<Integer> getAvgAgeWithTry(CustomerService cs, List<String> names, ZonedDateTime now) {

    var toAge = periodTo(now);

    return Try.traverse(names, cs::getByNameWithTry) // Try<Seq<Optional<Customer>>>
        .map(customers -> customers // Seq<Optional<Customer>>
            .filter(Optional::isPresent)
            .map(Optional::get)
            .map(Customer::getBornOn)
            .map(toAge)
            .foldLeft(AvgPeriodCounter.ZERO, AvgPeriodCounter::plus)
            .getAvgYear(now)
        );
  }

in the client code of this method, you can add on the result:

    result
        .onFailure(e -> log.warn("Cannot obtain customers {}", names, e))
        .onSuccess(result -> {
          if (log.isDebugEnabled()) {
            log.debug("For customers {} received the average age {}", names, result);
          }
        });

The first thing here is to use Try.traverse instead of mapping each name in the list like name → cs.getByNameWithTry(name). With the latter, we’d end up with Stream<Try<Optional<Customer>>> and what we’re interested in is a kind of Try<Stream<Optional<Customer>>>, so that we have a Stream<Optional<Customer>> to process. The traverse method does the job of "flipping" a Stream, or actually Seq, with Try.

With a simplification, it can be thought of as:

cs.getByNameWithTry(names.get(0)).flatMap(customer0 ->
  cs.getByNameWithTry(names.get(1)).flatMap(customer1 -> /* and so on */
    cs.getByNameWithTry(names.get(N)).map(customerN -> List.of(
        customer0, customer1, /* and so on */ customerN
    ))));

If any of the flatMap s end with Try.Failure (here if the Try.of(() → …​) caught an exception) the call chain is short-circuited and the end result is the first Try.Failure. Otherwise, it’s Try.Success with the list of processed elements.

Back to the logic. Later, the Try<T> behaves similarly to Option<T> in the way we can call map and flatMap, plus some specific method like onFailure, onSuccess, recover, and recoverWith. So we’re mapping a Seq of customers to their ages and reducing those to an average. The result can be processed further with map s and flatMap s, or we can finally call .get() first ensuring ourselves with .isSuccess() if the call succeeded.

If we’re into asynchronous processing, we can switch Try to Future (the vavr’s one, not the Java one which has method naming inconsistent with Stream/Option part). However, then we must be careful not to call .get or we’d block the execution. What we can do is use .onSuccess and .onFailure to complete the Promise of whatever framework or library expects.

Test data used in this section:

  io.vavr.collection.List<Customer> testCustomers = io.vavr.collection.List.of(
      ZonedDateTime.of(1970, 1, 1, 1, 0, 0, 0, UTC),
      ZonedDateTime.of(1990, 1, 1, 1, 0, 0, 0, UTC))
      .zipWithIndex()
      .map(bornOnIdx -> new Customer("Test John " + bornOnIdx._2,
          new Customer.Addrees(empty(), empty(), empty(), empty(), empty()),
          bornOnIdx._1,
          TRUE)
      );

  ZonedDateTime now = ZonedDateTime.of(2020, 12, 31, 1, 0, 0, 0, UTC);
  Integer avgAge = 41;
  List<String> existingNames = List.of("Test John 0", "Test John 1");
  List<String> existingAndNonexistingNames = List.of("Test John 0", "Non-existing John");

  String errorProneCustomer = "Error-prone Customer";

  FailableBehaviour.CustomerService cs = new FailableBehaviour.CustomerService(testCustomers);

Summary