@Test public void streamReuseCloseCrash() { Stream<Integer> stream = Stream.of(1).onClose(() -> { throw new IllegalArgumentException(); }); stream.iterator(); TestSubscriber<Integer> ts = new TestSubscriber<>(); Multi.create(stream) .subscribe(ts); ts.assertFailure(IllegalStateException.class); assertThat(ts.getLastError().getSuppressed()[0], instanceOf(IllegalArgumentException.class)); }
@Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); Iterator<T> iterator; boolean hasFirst; try { iterator = stream.iterator(); hasFirst = iterator.hasNext(); } catch (Throwable ex) { subscriber.onSubscribe(EmptySubscription.INSTANCE); try { stream.close(); } catch (Throwable exc) { if (exc != ex) { ex.addSuppressed(exc); } } subscriber.onError(ex); return; } if (!hasFirst) { subscriber.onSubscribe(EmptySubscription.INSTANCE); try { stream.close(); } catch (Throwable ex) { subscriber.onError(ex); return; } subscriber.onComplete(); return; } subscriber.onSubscribe(new IteratorSubscription<>(subscriber, iterator, stream)); }
MultiFromStream implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); Iterator<T> iterator; boolean hasFirst; try { iterator = stream.iterator(); hasFirst = iterator.hasNext(); } catch (Throwable ex) { subscriber.onSubscribe(EmptySubscription.INSTANCE); try { stream.close(); } catch (Throwable exc) { if (exc != ex) { ex.addSuppressed(exc); } } subscriber.onError(ex); return; } if (!hasFirst) { subscriber.onSubscribe(EmptySubscription.INSTANCE); try { stream.close(); } catch (Throwable ex) { subscriber.onError(ex); return; } subscriber.onComplete(); return; } subscriber.onSubscribe(new IteratorSubscription<>(subscriber, iterator, stream)); } }
MultiFromStream implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); Iterator<T> iterator; boolean hasFirst; try { iterator = stream.iterator(); hasFirst = iterator.hasNext(); } catch (Throwable ex) { subscriber.onSubscribe(EmptySubscription.INSTANCE); try { stream.close(); } catch (Throwable exc) { if (exc != ex) { ex.addSuppressed(exc); } } subscriber.onError(ex); return; } if (!hasFirst) { subscriber.onSubscribe(EmptySubscription.INSTANCE); try { stream.close(); } catch (Throwable ex) { subscriber.onError(ex); return; } subscriber.onComplete(); return; } subscriber.onSubscribe(new IteratorSubscription<>(subscriber, iterator, stream)); } MultiFromStream(Stream<T> stream); }
MultiFromStream implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); Iterator<T> iterator; boolean hasFirst; try { iterator = stream.iterator(); hasFirst = iterator.hasNext(); } catch (Throwable ex) { subscriber.onSubscribe(EmptySubscription.INSTANCE); try { stream.close(); } catch (Throwable exc) { if (exc != ex) { ex.addSuppressed(exc); } } subscriber.onError(ex); return; } if (!hasFirst) { subscriber.onSubscribe(EmptySubscription.INSTANCE); try { stream.close(); } catch (Throwable ex) { subscriber.onError(ex); return; } subscriber.onComplete(); return; } subscriber.onSubscribe(new IteratorSubscription<>(subscriber, iterator, stream)); } MultiFromStream(Stream<T> stream); @Override void subscribe(Flow.Subscriber<? super T> subscriber); }
MultiFromStream implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); Iterator<T> iterator; boolean hasFirst; try { iterator = stream.iterator(); hasFirst = iterator.hasNext(); } catch (Throwable ex) { subscriber.onSubscribe(EmptySubscription.INSTANCE); try { stream.close(); } catch (Throwable exc) { if (exc != ex) { ex.addSuppressed(exc); } } subscriber.onError(ex); return; } if (!hasFirst) { subscriber.onSubscribe(EmptySubscription.INSTANCE); try { stream.close(); } catch (Throwable ex) { subscriber.onError(ex); return; } subscriber.onComplete(); return; } subscriber.onSubscribe(new IteratorSubscription<>(subscriber, iterator, stream)); } MultiFromStream(Stream<T> stream); @Override void subscribe(Flow.Subscriber<? super T> subscriber);  }