@Test @SneakyThrows public void testConsumeThrowsException() { when(bufferedReaderMock.readLine()) .thenReturn("aaa") .thenThrow(new IOException()); final Flux<String> result = bufferedReaderAdapter.consume().apply(bufferedReaderMock); StepVerifier.create(result) .consumeNextWith(a -> assertThat(a, is("aaa"))) .verifyError(); }
public Function<BufferedReader, Flux<String>> consume() { return (reader) -> Flux.create(sink -> { try { String s; while (Objects.nonNull(s = reader.readLine())) { sink.next(s); } sink.complete(); } catch (final IOException e) { sink.error(e); } }); }
BufferedReaderAdapter { public Function<BufferedReader, Flux<String>> consume() { return (reader) -> Flux.create(sink -> { try { String s; while (Objects.nonNull(s = reader.readLine())) { sink.next(s); } sink.complete(); } catch (final IOException e) { sink.error(e); } }); } }
BufferedReaderAdapter { public Function<BufferedReader, Flux<String>> consume() { return (reader) -> Flux.create(sink -> { try { String s; while (Objects.nonNull(s = reader.readLine())) { sink.next(s); } sink.complete(); } catch (final IOException e) { sink.error(e); } }); }  }
BufferedReaderAdapter { public Function<BufferedReader, Flux<String>> consume() { return (reader) -> Flux.create(sink -> { try { String s; while (Objects.nonNull(s = reader.readLine())) { sink.next(s); } sink.complete(); } catch (final IOException e) { sink.error(e); } }); }  Function<BufferedReader, Flux<String>> consume(); Consumer<BufferedReader> close(); }
BufferedReaderAdapter { public Function<BufferedReader, Flux<String>> consume() { return (reader) -> Flux.create(sink -> { try { String s; while (Objects.nonNull(s = reader.readLine())) { sink.next(s); } sink.complete(); } catch (final IOException e) { sink.error(e); } }); }  Function<BufferedReader, Flux<String>> consume(); Consumer<BufferedReader> close();  }