Java 8 Streams and error handling?

Java 8 Streams quickly become your best friend but for serious programs you will need some error handling. Streams are not that designed for that and you can quickly use a library to solve that (functional) need like RxJava but for simple applications or small use cases you can desire to avoid to import an external library.

There are multiple ways to solve it not fully relying on the stream but an external state but this is not that elegant in general.

To solve it I used a small class which is my returned payload and added a few stream helper method in there.

To summarize the use case, it is the processing of N items from their identifier and the output is a list of "succeed" identifiers and a list of "failed" identifiers:

public class Status {
    // out state, + getters
    private Collection<String> succeed = new ArrayList<>();
    private Collection<String> failed = new ArrayList<>();

    public void onItem(final String id, final Consumer<String> delete) {
        try {
            delete.accept(id);
            succeed.add(id);
        } catch (final RuntimeException re) {
            failed.add(id); // in real life store some error message too
        }
    }

    public static Status combine(final Status s1, final Status s2) {
        if (s1 == null) {
            return s2;
        }
        ofNullable(s2.succeed).ifPresent(f -> s1.succeed.addAll(s2.succeed));
        ofNullable(s2.failed).ifPresent(f -> s1.failed.addAll(s2.failed));
        return s1;
    }
}

Nothing crazy for a java developer:

• the state (attributes) is composed of both list of successful item identifiers and failed ones
• we have an onItem() delegating the processing of the current identifier to another "method" (Consumer) and handling the success/failure adding the id to the right list
• and finally combine() is merging together two status in one

Now we have these helpers, the whole trick consists to wrap the lambdas in the onItem() method and if you use parallel streams (or need smething to put in the combiner of collect() then you can rely on the combine() method):

@POST
public Status work(final Input input) {
    return input.getIds().stream()
        .collect(
            Status::new,
            (status, id) -> status.onItem(id, this::doWork),
            Status::combine);
}

Of course this works well for the collect() case and you will likely need to have an onItem() using a Function<> instead of a Consumer<> to handle the map() case but idea is exactly the same: put the state in the Stream and merge it after if needed - that is where collect() is nice since it allows to limit the number of object created in that process.