### Question:
FlowAdapter { public static <T> Flow.Publisher<T> toPublisher(final AsyncIterator<? extends T> asyncIterator) { return new IteratorBackedPublisher<>(asyncIterator); } private  FlowAdapter(); static AsyncIterator<T> toAsyncIterator(final Flow.Publisher<? extends T> publisher); static Flow.Publisher<T> toPublisher(final AsyncIterator<? extends T> asyncIterator); static Flow.Publisher<T> toPublisher(
      final Supplier<AsyncIterator<? extends T>> asyncIteratorSupplier);  }### Answer:
@Test(expected = IOException.class) public void testCancelCloseException() throws Throwable { final ConsumingSubscriber<Long> subscriber = new ConsumingSubscriber<Long>() { @Override public void onNext(final Long arg0) {} }; final Publisher<Long> publisher = FlowAdapter.toPublisher(new ExceptionalIterator(10, null, new IOException())); publisher.subscribe(subscriber); subscriber.request(); subscriber.request(); Assert.assertFalse(subscriber.isDone()); subscriber.cancel(); Assert.assertTrue(subscriber.isDone()); FlowAdapterTest.unwrap(subscriber); }
@Test(expected = IllegalStateException.class) public void testDoubleSubscription() throws Throwable { final Flow.Publisher<Long> publisher = FlowAdapter.toPublisher(AsyncIterator.range(0, 5)); final ConsumingSubscriber<Long> s = new ConsumingSubscriber<>(); publisher.subscribe(s); s.join(); final ConsumingSubscriber<Long> s2 = new ConsumingSubscriber<>(); try { publisher.subscribe(s2); } catch (final Throwable e) { Assert.fail("failure should be notified via onError, got: " + e); } FlowAdapterTest.unwrap(s2); }