stream.map(obj -> doMap(obj)).collect(Collectors.toList());
private String doMap(Object obj) {
if (objectIsInvalid) {
throw new ParseException("Object could not be parsed");
}
}
Problem: how can I throw the exception and make the stream iteration know that it should not break the whole iteration, but continue with the next element (and eventually log failed objects)?
Here's one weird trick you can use to improve your exception handling.
Let's say your mapper function is like this:
String doMap(Object obj) {
if (isInvalid(obj)) {
throw new IllegalArgumentException("info about obj");
} else {
return obj.toString();
}
}
This returns a result if the object is valid, but it throws an exception if the object is invalid. Unfortunately if you stick this directly into a pipeline, any error will stop the pipeline execution. What you want is something like an "either" type that can hold either a value or an error indicator (which would be a exception in Java).
Turns out that CompletableFuture can hold either a value or an exception. Although it's intended for asynchronous processing -- which isn't occurring here -- we only have to contort it a little bit to use for our purposes.
First, given a stream of objects to process, we call the mapping function wrapped in a call to supplyAsync:
CompletableFuture<String>[] cfArray =
stream.map(obj -> CompletableFuture.supplyAsync(() -> doMap(obj), Runnable::run))
.toArray(n -> (CompletableFuture<String>[])new CompletableFuture<?>[n]);
(Unfortunately, the generic array creation gives an unchecked warning, which will have to be suppressed.)
The odd construct
CompletableFuture.supplyAsync(supplier, Runnable::run)
runs the supplier "asynchronously" on the provided Executor Runnable::run, which simply runs the task immediately in this thread. In other words, it runs the supplier synchronously.
The trick is that the CompletableFuture instance returned from this call contains either the value from the supplier, if it returned normally, or it contains an exception, if the supplier threw one. (I'm disregarding cancellation here.) We then gather the CompletableFuture instances into an array. Why an array? It's setup for the next part:
CompletableFuture.allOf(cfArray).join();
This normally waits for the array of CFs to complete. Since they've been run synchronously, they should already all be complete. What's important for this case is that join() will throw a CompletionException if any of the CFs in the array has completed exceptionally. If the join completes normally, we can simply gather up the return values. If the join throws an exception, we can either propagate it, or we can catch it and process the exceptions stored in the CFs in the array. For example,
try {
CompletableFuture.allOf(cfArray).join();
// no errors
return Arrays.stream(cfArray)
.map(CompletableFuture::join)
.collect(toList());
} catch (CompletionException ce) {
long errcount =
Arrays.stream(cfArray)
.filter(CompletableFuture::isCompletedExceptionally)
.count();
System.out.println("errcount = " + errcount);
return Collections.emptyList();
}
If all are successful, this returns a list of the values. If there are any exceptions, this counts the number of exceptions and returns an empty list. Of course, you could easily do something else, like log the details of the exceptions, filter out the exceptions and return a list of valid values, etc.
Without an exception you can work with Optionals:
stream.map(obj -> doMap(obj))
.filter(obj -> obj.isPresent())
.collect(Collectors.toList());
private Optional<String> doMap(Object obj) {
if (objectIsInvalid) {
return Optional.empty();
}
}
If you want a means to quietly ignore those elements that throw an exception when mapped, you can define a helper method that either returns a success or nothing, as a stream of 0 or 1 elements, and then flatMap your stream with that method.
import static java.util.stream.Collectors.toList;
import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Stream;
public class Fish {
private int weight;
public Fish(int weight) { this.weight = weight; }
public String size() {
if (weight < 10) {
return "small";
} else if (weight < 50) {
return "medium";
} else if (weight < 100) {
return "large";
} else {
throw new RuntimeException("I don't believe you");
}
}
public static <T> Stream<T> successOrNothing(Supplier<T> supplier) {
try {
return Stream.of(supplier.get());
} catch (Exception e) {
return Stream.empty();
}
}
public static void main(String[] args) {
Stream<Fish> fishes = Stream.of(new Fish(1000), new Fish(2));
List<String> sizes = fishes
.flatMap(fish -> successOrNothing(fish::size))
.collect(toList());
System.out.println(sizes);
}
}
Each element in the stream gets mapped to 1 element (if the call returned successfully) or 0 (if it did not). This might be easier or more straightforward then the "change the method to return null or Optional.empty() instead of throwing" pattern, especially when dealing with a method whose contract already is defined and you don't want to change.
Caveat: this solution is still silently ignoring exceptions; that is inappropriate in many situations.
I would do following...
stream.map(doMapLenient())
.filter(Objects::nonNull)
.collect(Collectors.toList());
private String doMap(Object obj) {
if (objectIsInvalid(obj)) {
throw new ParseException("Object could not be parsed");
}
return "Streams are cool";
}
private Function<Object, String> doMapLenient() {
return obj -> {
try {
return doMap(obj);
} catch (ParseExcepion ex) {
return null;
}
}
here you can add nicely also some logging in catch part
You can just explicitly iterate over the stream elements using an iterator:
Stream<Object> stream = ... // throws unchecked exceptions;
Iterator<Object> iterator = stream.iterator();
while (iterator.hasNext()) {
try {
Object obj = it.next();
// process
} catch (ParseException ex) {
// handle
}
}
