StreamComposition { @SuppressWarnings("varargs") @SafeVarargs public static <T> Stream<T> concatAndAutoClose(Stream<T>... streams) { requireNonNullElements(streams); return configureAutoCloseStream(AutoClosingStream.of(Stream.of(streams).flatMap(Function.identity())), streams); } private  StreamComposition(); @SuppressWarnings("varargs") @SafeVarargs // Creating a Stream of an array is safe. static Stream<T> concatAndAutoClose(Stream<T>... streams); @SuppressWarnings("varargs") @SafeVarargs // Creating a Stream of an array is safe. static Stream<T> concat(Stream<T>... streams);  }
@Test void testPureStream() { List<String> result = StreamComposition.concatAndAutoClose(a, b, c).collect(toList()); assertEquals(Arrays.asList("A", "B", "C"), result); assertAllClosed(); }
@Test void testParallelStream() { List<String> result = StreamComposition.concatAndAutoClose(a.parallel(), b, c).collect(toList()); assertEquals(Arrays.asList("A", "B", "C"), result); assertAllClosed(); }
@Test void testPartialIteration() { Optional<String> result = StreamComposition.concatAndAutoClose(a, b, c).findFirst(); assertEquals(Optional.of("A"), result); assertEquals(Collections.singletonList("A"), producedItems); assertAllClosed(); }
@Test void testException() { final AtomicBoolean fClosed = new AtomicBoolean(); boolean gotException = false; final Stream<String> f = Stream.of("F").peek(this::produceException).onClose(() -> fClosed.set(true)); List<String> result; try { result = StreamComposition.concatAndAutoClose(a, b, f, c).collect(toList()); } catch (Exception e) { gotException = true; } assertTrue(gotException); assertAllClosed(); assertTrue(fClosed.get()); }
@Test void testExceptionInClose() { final AtomicBoolean fClosed = new AtomicBoolean(); final AtomicBoolean gClosed = new AtomicBoolean(); boolean gotException = false; final Stream<String> f = Stream.of("F") .onClose(() -> fClosed.set(true)) .onClose(() -> produceException("F")); final Stream<String> g = Stream.of("G") .onClose(() -> gClosed.set(true)) .onClose(() -> produceException("G")); List<String> result; try { result = StreamComposition.concatAndAutoClose(a, b, f, g, c).collect(toList()); } catch (Exception e) { gotException = true; } assertTrue(gotException); assertAllClosed(); assertTrue(fClosed.get()); assertTrue(gClosed.get()); }
@Test void testChainedStreams() { final AtomicBoolean fClosed = new AtomicBoolean(); final Stream<String> f = StreamComposition.concatAndAutoClose( Stream.of("F").onClose(() -> fClosed.set(true)) ); List<Integer> result = f .mapToInt(String::length) .boxed() .mapToLong(l -> l) .mapToInt((long l) -> (int) l) .boxed() .collect(toList()); assertEquals(Collections.singletonList(1), result); assertTrue(fClosed.get()); }