Java 8: лямбда-потоки, фильтрация по методам с исключением

У меня возникла проблема при проверке лямбда-выражений Java 8. Обычно это работает нормально, но теперь у меня есть методы, которые бросают IOException. Лучше всего, если вы посмотрите на следующий код:

class Bank{
    ....
    public Set<String> getActiveAccountNumbers() throws IOException {
        Stream<Account> s =  accounts.values().stream();
        s = s.filter(a -> a.isActive());
        Stream<String> ss = s.map(a -> a.getNumber());
        return ss.collect(Collectors.toSet());
    }
    ....
}

interface Account{
    ....
    boolean isActive() throws IOException;
    String getNumber() throws IOException;
    ....
}

Проблема в том, что он не компилируется, потому что я должен перехватить возможные исключения из isActive- и getNumber-Methods. Но даже если я явно использую блок try-catch-Block, как показано ниже, он все равно не скомпилируется, потому что я не улавливаю исключение. Так что либо есть ошибка в JDK, либо я не знаю, как перехватить эти исключения.

class Bank{
    ....
    //Doesn't compile either
    public Set<String> getActiveAccountNumbers() throws IOException {
        try{
            Stream<Account> s =  accounts.values().stream();
            s = s.filter(a -> a.isActive());
            Stream<String> ss = s.map(a -> a.getNumber());
            return ss.collect(Collectors.toSet());
        }catch(IOException ex){
        }
    }
    ....
}

Как я могу заставить это работать? Может кто-нибудь подсказать мне правильное решение?

Вы должны поймать исключение, прежде чем оно выйдет из лямбды:

s = s.filter(a -> { try { return a.isActive(); } 
                    catch (IOException e) { throw new UncheckedIOException(e); }}});

Примите во внимание тот факт, что лямбда не оценивается в том месте, где вы ее пишете, а в каком-то совершенно не связанном месте, в классе JDK. Так что это будет точка, где будет выброшено это проверенное исключение, и в этом месте оно не объявлено.

Вы можете справиться с этим, используя оболочку вашей лямбды, которая переводит проверенные исключения в непроверенные:

public static <T> T uncheckCall(Callable<T> callable) {
  try { return callable.call(); }
  catch (RuntimeException e) { throw e; }
  catch (Exception e) { throw new RuntimeException(e); }
}

Ваш пример будет написан как

return s.filter(a -> uncheckCall(a::isActive))
        .map(Account::getNumber)
        .collect(toSet());

В своих проектах я занимаюсь этим вопросом без обертывания; вместо этого я использую метод, который эффективно предотвращает проверку исключений компилятором. Излишне говорить, что с этим следует обращаться осторожно, и все участники проекта должны знать, что проверенное исключение может появиться там, где оно не объявлено. Это код слесарного дела:

public static <T> T uncheckCall(Callable<T> callable) {
  try { return callable.call(); }
  catch (Exception e) { return sneakyThrow(e); }
}
public static void uncheckRun(RunnableExc r) {
  try { r.run(); } catch (Exception e) { sneakyThrow(e); }
}
public interface RunnableExc { void run() throws Exception; }


@SuppressWarnings("unchecked")
private static <T extends Throwable> void sneakyThrow(Throwable t) throws T {
  throw (T) t;
}

и вы можете ожидать получить IOExceptionбросок в лицо, даже если collectне объявить об этом. В большинстве случаев , но не во всех реальных, вы все равно захотите просто сбросить исключение и обработать его как общий сбой. Во всех этих случаях ничего не теряется в ясности или правильности. Просто остерегайтесь тех других случаев, когда вы действительно захотите отреагировать на исключение на месте. Разработчик не будет предупрежден компилятором о том, что его IOExceptionнужно перехватить, и компилятор на самом деле будет жаловаться, если вы попытаетесь его перехватить, потому что мы обманули его, полагая, что такое исключение не может быть сгенерировано.

Вы также можете распространять свою статическую боль с помощью лямбд, так что все это выглядит читабельно:

s.filter(a -> propagate(a::isActive))

propagateздесь получает java.util.concurrent.Callableв качестве параметра и преобразует любое исключение, пойманное во время вызова RuntimeException. В Гуаве есть похожий метод преобразования Throwables # пропагандировать (Throwable) .

Этот метод, по-видимому, необходим для цепочки лямбда-методов, поэтому я надеюсь, что однажды он будет добавлен к одной из популярных библиотек, или это распространяющееся поведение будет по умолчанию.

public class PropagateExceptionsSample {
    // a simplified version of Throwables#propagate
    public static RuntimeException runtime(Throwable e) {
        if (e instanceof RuntimeException) {
            return (RuntimeException)e;
        }

        return new RuntimeException(e);
    }

    // this is a new one, n/a in public libs
    // Callable just suits as a functional interface in JDK throwing Exception 
    public static <V> V propagate(Callable<V> callable){
        try {
            return callable.call();
        } catch (Exception e) {
            throw runtime(e);
        }
    }

    public static void main(String[] args) {
        class Account{
            String name;    
            Account(String name) { this.name = name;}

            public boolean isActive() throws IOException {
                return name.startsWith("a");
            }
        }


        List<Account> accounts = new ArrayList<>(Arrays.asList(new Account("andrey"), new Account("angela"), new Account("pamela")));

        Stream<Account> s = accounts.stream();

        s
          .filter(a -> propagate(a::isActive))
          .map(a -> a.name)
          .forEach(System.out::println);
    }
}

Этот UtilExceptionвспомогательный класс позволяет вам использовать любые проверенные исключения в потоках Java, например:

Stream.of("java.lang.Object", "java.lang.Integer", "java.lang.String")
      .map(rethrowFunction(Class::forName))
      .collect(Collectors.toList());

Примечание Class::forNameкидает ClassNotFoundException, что проверено . Сам поток также выдает ClassNotFoundException, а НЕ какое-то неконтролируемое исключение обтекания.

public final class UtilException {

@FunctionalInterface
public interface Consumer_WithExceptions<T, E extends Exception> {
    void accept(T t) throws E;
    }

@FunctionalInterface
public interface BiConsumer_WithExceptions<T, U, E extends Exception> {
    void accept(T t, U u) throws E;
    }

@FunctionalInterface
public interface Function_WithExceptions<T, R, E extends Exception> {
    R apply(T t) throws E;
    }

@FunctionalInterface
public interface Supplier_WithExceptions<T, E extends Exception> {
    T get() throws E;
    }

@FunctionalInterface
public interface Runnable_WithExceptions<E extends Exception> {
    void run() throws E;
    }

/** .forEach(rethrowConsumer(name -> System.out.println(Class.forName(name)))); or .forEach(rethrowConsumer(ClassNameUtil::println)); */
public static <T, E extends Exception> Consumer<T> rethrowConsumer(Consumer_WithExceptions<T, E> consumer) throws E {
    return t -> {
        try { consumer.accept(t); }
        catch (Exception exception) { throwAsUnchecked(exception); }
        };
    }

public static <T, U, E extends Exception> BiConsumer<T, U> rethrowBiConsumer(BiConsumer_WithExceptions<T, U, E> biConsumer) throws E {
    return (t, u) -> {
        try { biConsumer.accept(t, u); }
        catch (Exception exception) { throwAsUnchecked(exception); }
        };
    }

/** .map(rethrowFunction(name -> Class.forName(name))) or .map(rethrowFunction(Class::forName)) */
public static <T, R, E extends Exception> Function<T, R> rethrowFunction(Function_WithExceptions<T, R, E> function) throws E {
    return t -> {
        try { return function.apply(t); }
        catch (Exception exception) { throwAsUnchecked(exception); return null; }
        };
    }

/** rethrowSupplier(() -> new StringJoiner(new String(new byte[]{77, 97, 114, 107}, "UTF-8"))), */
public static <T, E extends Exception> Supplier<T> rethrowSupplier(Supplier_WithExceptions<T, E> function) throws E {
    return () -> {
        try { return function.get(); }
        catch (Exception exception) { throwAsUnchecked(exception); return null; }
        };
    }

/** uncheck(() -> Class.forName("xxx")); */
public static void uncheck(Runnable_WithExceptions t)
    {
    try { t.run(); }
    catch (Exception exception) { throwAsUnchecked(exception); }
    }

/** uncheck(() -> Class.forName("xxx")); */
public static <R, E extends Exception> R uncheck(Supplier_WithExceptions<R, E> supplier)
    {
    try { return supplier.get(); }
    catch (Exception exception) { throwAsUnchecked(exception); return null; }
    }

/** uncheck(Class::forName, "xxx"); */
public static <T, R, E extends Exception> R uncheck(Function_WithExceptions<T, R, E> function, T t) {
    try { return function.apply(t); }
    catch (Exception exception) { throwAsUnchecked(exception); return null; }
    }

@SuppressWarnings ("unchecked")
private static <E extends Throwable> void throwAsUnchecked(Exception exception) throws E { throw (E)exception; }

}

Много других примеров того, как его использовать (после статического импорта UtilException):

@Test
public void test_Consumer_with_checked_exceptions() throws IllegalAccessException {
    Stream.of("java.lang.Object", "java.lang.Integer", "java.lang.String")
          .forEach(rethrowConsumer(className -> System.out.println(Class.forName(className))));

    Stream.of("java.lang.Object", "java.lang.Integer", "java.lang.String")
          .forEach(rethrowConsumer(System.out::println));
    }

@Test
public void test_Function_with_checked_exceptions() throws ClassNotFoundException {
    List<Class> classes1
          = Stream.of("Object", "Integer", "String")
                  .map(rethrowFunction(className -> Class.forName("java.lang." + className)))
                  .collect(Collectors.toList());

    List<Class> classes2
          = Stream.of("java.lang.Object", "java.lang.Integer", "java.lang.String")
                  .map(rethrowFunction(Class::forName))
                  .collect(Collectors.toList());
    }

@Test
public void test_Supplier_with_checked_exceptions() throws ClassNotFoundException {
    Collector.of(
          rethrowSupplier(() -> new StringJoiner(new String(new byte[]{77, 97, 114, 107}, "UTF-8"))),
          StringJoiner::add, StringJoiner::merge, StringJoiner::toString);
    }

@Test    
public void test_uncheck_exception_thrown_by_method() {
    Class clazz1 = uncheck(() -> Class.forName("java.lang.String"));

    Class clazz2 = uncheck(Class::forName, "java.lang.String");
    }

@Test (expected = ClassNotFoundException.class)
public void test_if_correct_exception_is_still_thrown_by_method() {
    Class clazz3 = uncheck(Class::forName, "INVALID");
    }

Но не используйте его до понимания следующих преимуществ, недостатков и ограничений :

• Если вызывающий код должен обрабатывать проверенное исключение, вы ДОЛЖНЫ добавить его в предложение throws метода, который содержит поток. Компилятор больше не будет заставлять вас добавлять его, поэтому его легче забыть.

• Если вызывающий код уже обрабатывает проверенное исключение, компилятор напомнит вам добавить предложение throws к объявлению метода, содержащему поток (если вы этого не сделаете, он скажет: исключение никогда не выдается в теле соответствующего оператора try ).

• В любом случае вы не сможете окружить сам поток, чтобы поймать проверенное исключение ВНУТРИ метода, который содержит этот поток (если вы попытаетесь, компилятор скажет: исключение никогда не выдается в теле соответствующего оператора try).

• Если вы вызываете метод, который буквально никогда не может выбросить исключение, которое он объявляет, вам не следует включать предложение throws. Например: new String (byteArr, "UTF-8") выбрасывает UnsupportedEncodingException, но UTF-8 гарантируется, что спецификация Java всегда присутствует. Здесь, декларация бросков - неприятность, и любое решение заставить ее замолчать с минимальным шаблоном приветствуется.

• Если вы ненавидите проверенные исключения и чувствуете, что их никогда не следует добавлять в язык Java для начала (все больше людей думают так, а я НЕ один из них), то просто не добавляйте проверенное исключение в бросает предложение метода, который содержит поток. Таким образом, проверенное исключение будет вести себя так же, как исключение UNchecked.

• Если вы реализуете строгий интерфейс, в котором у вас нет возможности добавить объявление throws, и, тем не менее, выбрасывать исключение вполне уместно, тогда обертывание исключения только для того, чтобы получить привилегию для его выброса, приводит к трассировке стека с ложными исключениями которые не дают никакой информации о том, что на самом деле пошло не так. Хорошим примером является Runnable.run (), который не выдает никаких проверенных исключений. В этом случае вы можете решить не добавлять проверенное исключение в предложение throws метода, содержащего поток.

• В любом случае, если вы решите НЕ добавлять (или забыть добавить) проверенное исключение к предложению throws метода, содержащего поток, учтите следующие 2 последствия выброса исключений CHECKED:

1) вызывающий код не сможет поймать его по имени (если вы попытаетесь, компилятор скажет: исключение никогда не выдается в теле соответствующего оператора try). Он будет пузыриться и, вероятно, будет перехвачен в цикле основной программы каким-то «catch Exception» или «catch Throwable», что может быть тем, что вы хотите в любом случае.

2) Это нарушает принцип наименьшего удивления: вам больше не будет достаточно перехватить RuntimeException, чтобы иметь возможность гарантировать перехват всех возможных исключений. По этой причине я считаю, что это следует делать не в программном коде, а только в бизнес-коде, который вы полностью контролируете.

В заключение: я считаю, что ограничения здесь несерьезны, и UtilExceptionкласс можно использовать без страха. Тем не менее, это зависит от вас!

• Ссылки:
  • http://www.philandstuff.com/2012/04/28/sneakily-throwing-checked-exceptions.html
  • http://www.mail-archive.com/javaposse@googlegroups.com/msg05984.html
  • Проект Ломбок аннотация: @SneakyThrows
  • Мнение Брайана Гетца (против) здесь: Как я могу выбросить CHECKED исключения из потоков Java 8?
  • /software/225931/workaround-for-java-checked-exceptions?newreg=ddf0dd15e8174af8ba52e091cf85688e *

Вы можете потенциально свернуть свой собственный Streamвариант, обернув лямбду, чтобы создать исключение без проверки, а затем развернуть это исключение при операциях терминала:

@FunctionalInterface
public interface ThrowingPredicate<T, X extends Throwable> {
    public boolean test(T t) throws X;
}

@FunctionalInterface
public interface ThrowingFunction<T, R, X extends Throwable> {
    public R apply(T t) throws X;
}

@FunctionalInterface
public interface ThrowingSupplier<R, X extends Throwable> {
    public R get() throws X;
}

public interface ThrowingStream<T, X extends Throwable> {
    public ThrowingStream<T, X> filter(
            ThrowingPredicate<? super T, ? extends X> predicate);

    public <R> ThrowingStream<T, R> map(
            ThrowingFunction<? super T, ? extends R, ? extends X> mapper);

    public <A, R> R collect(Collector<? super T, A, R> collector) throws X;

    // etc
}

class StreamAdapter<T, X extends Throwable> implements ThrowingStream<T, X> {
    private static class AdapterException extends RuntimeException {
        public AdapterException(Throwable cause) {
            super(cause);
        }
    }

    private final Stream<T> delegate;
    private final Class<X> x;

    StreamAdapter(Stream<T> delegate, Class<X> x) {
        this.delegate = delegate;
        this.x = x;
    }

    private <R> R maskException(ThrowingSupplier<R, X> method) {
        try {
            return method.get();
        } catch (Throwable t) {
            if (x.isInstance(t)) {
                throw new AdapterException(t);
            } else {
                throw t;
            }
        }
    }

    @Override
    public ThrowingStream<T, X> filter(ThrowingPredicate<T, X> predicate) {
        return new StreamAdapter<>(
                delegate.filter(t -> maskException(() -> predicate.test(t))), x);
    }

    @Override
    public <R> ThrowingStream<R, X> map(ThrowingFunction<T, R, X> mapper) {
        return new StreamAdapter<>(
                delegate.map(t -> maskException(() -> mapper.apply(t))), x);
    }

    private <R> R unmaskException(Supplier<R> method) throws X {
        try {
            return method.get();
        } catch (AdapterException e) {
            throw x.cast(e.getCause());
        }
    }

    @Override
    public <A, R> R collect(Collector<T, A, R> collector) throws X {
        return unmaskException(() -> delegate.collect(collector));
    }
}

Тогда вы можете использовать это точно так же, как Stream:

Stream<Account> s = accounts.values().stream();
ThrowingStream<Account, IOException> ts = new StreamAdapter<>(s, IOException.class);
return ts.filter(Account::isActive).map(Account::getNumber).collect(toSet());

Это решение потребует довольно много шаблонного, поэтому я предлагаю вам взглянуть на библиотеку, которую я уже создал, которая в точности соответствует тому, что я описал здесь для всего Streamкласса (и даже больше!).

Используйте метод #propagate (). Пример не-гуавской реализации из блога Java 8 Сэма Берана :

public class Throwables {
    public interface ExceptionWrapper<E> {
        E wrap(Exception e);
    }

    public static <T> T propagate(Callable<T> callable) throws RuntimeException {
        return propagate(callable, RuntimeException::new);
    }

    public static <T, E extends Throwable> T propagate(Callable<T> callable, ExceptionWrapper<E> wrapper) throws E {
        try {
            return callable.call();
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e) {
            throw wrapper.wrap(e);
        }
    }
}

Это не дает прямого ответа на вопрос (есть много других ответов), но в первую очередь пытается избежать проблемы:

По моему опыту, необходимость обрабатывать исключения в Stream(или другом лямбда-выражении) часто вытекает из того факта, что исключения объявляются как методы, в которых они не должны создаваться. Это часто происходит от смешивания бизнес-логики с входом и выходом. Ваш Accountинтерфейс является прекрасным примером:

interface Account {
    boolean isActive() throws IOException;
    String getNumber() throws IOException;
}

Вместо того, чтобы бросать IOExceptionна каждый получатель, рассмотрите этот дизайн:

interface AccountReader {
    Account readAccount(…) throws IOException;
}

interface Account {
    boolean isActive();
    String getNumber();
}

Метод AccountReader.readAccount(…)может считывать учетную запись из базы данных, файла или чего-либо еще и выдавать исключение, если это не удается. Он создает Accountобъект, который уже содержит все значения, готовые к использованию. Поскольку значения уже были загружены readAccount(…), получатели не будут генерировать исключение. Таким образом, вы можете свободно использовать их в лямбдах без необходимости оборачивать, маскировать или скрывать исключения.

Конечно, это не всегда возможно сделать так, как я описал, но часто это так и приводит к более чистому коду (ИМХО):

• Лучшее разделение интересов и следование принципу единой ответственности
• Меньше шаблонного: вам не нужно загромождать свой код throws IOExceptionни для чего, кроме как для удовлетворения компилятора
• Обработка ошибок: вы обрабатываете ошибки там, где они возникают - при чтении из файла или базы данных - вместо того, чтобы где-то в середине вашей бизнес-логики, только потому, что вы хотите получить значение поля
• Вы можете быть в состоянии сделать Account неизменным и извлечь выгоду из его преимуществ (например, безопасность потока)
• Вам не нужны «грязные трюки» или обходные пути для использования Accountв лямбдах (например, в Stream)

Это можно решить с помощью простого кода ниже с Stream и Try в AbacusUtil :

Stream.of(accounts).filter(a -> Try.call(a::isActive)).map(a -> Try.call(a::getNumber)).toSet();

Раскрытие информации: я разработчик AbacusUtil.

Расширяя решение @marcg, вы обычно можете бросить и поймать проверенное исключение в Streams; то есть компилятор попросит вас перехватить / перебросить, как если бы вы были вне потоков!

@FunctionalInterface
public interface Predicate_WithExceptions<T, E extends Exception> {
    boolean test(T t) throws E;
}

/**
 * .filter(rethrowPredicate(t -> t.isActive()))
 */
public static <T, E extends Exception> Predicate<T> rethrowPredicate(Predicate_WithExceptions<T, E> predicate) throws E {
    return t -> {
        try {
            return predicate.test(t);
        } catch (Exception exception) {
            return throwActualException(exception);
        }
    };
}

@SuppressWarnings("unchecked")
private static <T, E extends Exception> T throwActualException(Exception exception) throws E {
    throw (E) exception;
}

Тогда ваш пример будет записан следующим образом (добавление тестов, чтобы показать это более четко):

@Test
public void testPredicate() throws MyTestException {
    List<String> nonEmptyStrings = Stream.of("ciao", "")
            .filter(rethrowPredicate(s -> notEmpty(s)))
            .collect(toList());
    assertEquals(1, nonEmptyStrings.size());
    assertEquals("ciao", nonEmptyStrings.get(0));
}

private class MyTestException extends Exception { }

private boolean notEmpty(String value) throws MyTestException {
    if(value==null) {
        throw new MyTestException();
    }
    return !value.isEmpty();
}

@Test
public void testPredicateRaisingException() throws MyTestException {
    try {
        Stream.of("ciao", null)
                .filter(rethrowPredicate(s -> notEmpty(s)))
                .collect(toList());
        fail();
    } catch (MyTestException e) {
        //OK
    }
}

Чтобы правильно добавить код обработки IOException (в RuntimeException), ваш метод будет выглядеть следующим образом:

Stream<Account> s =  accounts.values().stream();

s = s.filter(a -> { try { return a.isActive(); } 
  catch (IOException e) { throw new RuntimeException(e); }});

Stream<String> ss = s.map(a -> { try { return a.getNumber() }
  catch (IOException e) { throw new RuntimeException(e); }});

return ss.collect(Collectors.toSet());

Проблема в том, что IOException нужно будет записать как RuntimeExceptionи преобразовать обратно вIOException - и это добавит еще больше кода к вышеуказанному методу.

Зачем использовать, Streamкогда это можно сделать именно так - и метод выбрасывает, IOExceptionпоэтому для этого тоже не требуется никакого дополнительного кода:

Set<String> set = new HashSet<>();
for(Account a: accounts.values()){
  if(a.isActive()){
     set.add(a.getNumber());
  } 
}
return set;

Помня об этой проблеме, я разработал небольшую библиотеку для работы с проверенными исключениями и лямбдами. Пользовательские адаптеры позволяют интегрироваться с существующими функциональными типами:

stream().map(unchecked(URI::new)) //with a static import

https://github.com/TouK/ThrowingFunction/

Ваш пример может быть записан как:

import utils.stream.Unthrow;

class Bank{
   ....
   public Set<String> getActiveAccountNumbers() {
       return accounts.values().stream()
           .filter(a -> Unthrow.wrap(() -> a.isActive()))
           .map(a -> Unthrow.wrap(() -> a.getNumber()))
           .collect(Collectors.toSet());
   }
   ....
}

Класс Unthrow можно взять здесь https://github.com/SeregaLBN/StreamUnthrower

Если вы не возражаете против использования сторонних библиотек, в библиотеке AOL cyclops-реагировать , disclosure :: I contributor есть класс ExceptionSoftener, который может помочь здесь.

 s.filter(softenPredicate(a->a.isActive()));

Функциональные интерфейсы в Java не объявляют никаких проверенных или непроверенных исключений. Нам нужно изменить сигнатуру методов с:

boolean isActive() throws IOException; 
String getNumber() throwsIOException;

Для того, чтобы:

boolean isActive();
String getNumber();

Или обработайте его с помощью блока try-catch:

public Set<String> getActiveAccountNumbers() {
  Stream<Account> s =  accounts.values().stream();
  s = s.filter(a -> 
    try{
      a.isActive();
    }catch(IOException e){
      throw new RuntimeException(e);
    }
  );
  Stream<String> ss = s.map(a -> 
    try{
      a.getNumber();
    }catch(IOException e){
      throw new RuntimeException(e);
    }
  );
  return ss.collect(Collectors.toSet());
}

Другой вариант - написать пользовательскую оболочку или использовать библиотеку, например ThrowingFunction. С библиотекой нам нужно только добавить зависимость в наш pom.xml:

<dependency>
    <groupId>pl.touk</groupId>
    <artifactId>throwing-function</artifactId>
    <version>1.3</version>
</dependency>

И использовать определенные классы, такие как ThrowingFunction, ThrowingConsumer, ThrowingPredicate, ThrowingRunnable, ThrowingSupplier.

В конце код выглядит так:

public Set<String> getActiveAccountNumbers() {
  return accounts.values().stream()
    .filter(ThrowingPredicate.unchecked(Account::isActive))
    .map(ThrowingFunction.unchecked(Account::getNumber))
    .collect(Collectors.toSet());
}

Я не вижу какого-либо способа обработки проверенного исключения в потоке (Java-8), единственный способ, которым я применил, - это перехватить проверенное исключение в потоке и повторно выбросить их как непроверенное исключение.

        Arrays.stream(VERSIONS)
        .map(version -> TemplateStore.class
                .getClassLoader().getResourceAsStream(String.format(TEMPLATE_FILE_MASK, version)))
        .map(inputStream -> {
            try {
                return ((EdiTemplates) JAXBContext.newInstance(EdiTemplates.class).createUnmarshaller()
                        .unmarshal(inputStream)).getMessageTemplate();
            } catch (JAXBException e) {
                throw new IllegalArgumentException(ERROR, e);
            }})
        .flatMap(Collection::stream)
        .collect(Collectors.toList());

Если вы хотите обработать исключение в потоке и продолжить обработку дополнительных, в DZone есть отличная статья Брайана Вермеера, использующая концепцию Either. Это показывает отличный способ справиться с этой ситуацией. Не хватает только примера кода. Это образец моего исследования с использованием концепций из этой статьи.

@Test
public void whenValuePrinted_thenPrintValue() {

    List<Integer> intStream = Arrays.asList(0, 1, 2, 3, 4, 5, 6);
    intStream.stream().map(Either.liftWithValue(item -> doSomething(item)))
             .map(item -> item.isLeft() ? item.getLeft() : item.getRight())
             .flatMap(o -> {
                 System.out.println(o);
                 return o.isPresent() ? Stream.of(o.get()) : Stream.empty();
             })
             .forEach(System.out::println);
}

private Object doSomething(Integer item) throws Exception {

    if (item == 0) {
        throw new Exception("Zero ain't a number!");
    } else if (item == 4) {
        return Optional.empty();
    }

    return item;
}