Streams { public static <T> Stream<T> toParallelStream(final Iterable<T> iterable) { if (iterable instanceof StreamIterable) { return ((StreamIterable<T>) iterable).getStream().parallel(); } return StreamSupport.stream(iterable.spliterator(), true) .onClose(() -> CloseableUtil.close(iterable)); } private  Streams(); static Stream<T> toStream(final Iterable<T> iterable); static Stream<T> toStream(final T[] array); static Stream<T> toStream(final Iterator<T> iterator); static Stream<T> toParallelStream(final Iterable<T> iterable); static Stream<T> toParallelStream(final Iterator<T> iterator); static Stream<T> toParallelStream(final T[] array);  }
@Test public void shouldCloseIteratorWhenParallelStreamIsClosed() throws Throwable { final CloseableIterator<String> iterator = mock(CloseableIterator.class); given(iterator.hasNext()).willReturn(true, false); final String first = "first item"; given(iterator.next()).willReturn(first, null, null); final Object firstResult; try (final Stream stream = Streams.toParallelStream(iterator)) { firstResult = stream.findFirst().orElseThrow(RuntimeException::new); } assertEquals("first item", firstResult); verify(iterator).close(); }
@Test public void shouldCloseIterableWhenParallelStreamIsClosed() throws Throwable { final CloseableIterable<String> iterable = mock(CloseableIterable.class); final CloseableIterator<String> iterator = mock(CloseableIterator.class); given(iterable.spliterator()).willReturn(Spliterators.spliteratorUnknownSize(iterator, 0)); given(iterator.hasNext()).willReturn(true, false); final String first = "first item"; given(iterator.next()).willReturn(first, null, null); final Object firstResult; try (final Stream stream = Streams.toParallelStream(iterable)) { firstResult = stream.findFirst().orElseThrow(RuntimeException::new); } assertEquals("first item", firstResult); verify(iterable).close(); }