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();  }
@Test @SneakyThrows public void testConsumeReturnsLinesCorrectly() { when(bufferedReaderMock.readLine()).thenReturn("aaa", "bbb", "ccc", null); final Flux<String> result = bufferedReaderAdapter.consume().apply(bufferedReaderMock); StepVerifier.create(result) .consumeNextWith(a -> assertThat(a, is("aaa"))) .consumeNextWith(a -> assertThat(a, is("bbb"))) .consumeNextWith(a -> assertThat(a, is("ccc"))) .verifyComplete(); }
@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(); }