### Question:
SourceAdapters { public static <T> Publisher<T> fromSource(PublisherSource<T> source) { if (source instanceof Publisher) { return uncheckedCast(source); } return new PublisherSourceToPublisher<>(source); } private  SourceAdapters(); static PublisherSource<T> toSource(Publisher<T> publisher); static SingleSource<T> toSource(Single<T> single); static CompletableSource toSource(Completable completable); static Publisher<T> fromSource(PublisherSource<T> source); static Single<T> fromSource(SingleSource<T> source); static Completable fromSource(CompletableSource source);  }### Answer:
@Test public void publisherFromSourceSuccess() throws Exception { PublisherSource<Integer> src = s -> s.onSubscribe(new ScalarValueSubscription<>(1, s)); Integer result = fromSource(src).firstOrElse(() -> null).toFuture().get(); assertThat("Unexpected result.", result, is(1)); }
@Test public void publisherFromSourceError() throws Exception { PublisherSource<Integer> src = s -> { s.onSubscribe(EMPTY_SUBSCRIPTION); s.onError(DELIBERATE_EXCEPTION); }; Future<Integer> future = fromSource(src).firstOrElse(() -> null).toFuture(); expectedException.expect(ExecutionException.class); expectedException.expectCause(sameInstance(DELIBERATE_EXCEPTION)); future.get(); }
@Test public void publisherFromSourceCancel() { PublisherSource.Subscription srcSubscription = mock(PublisherSource.Subscription.class); PublisherSource<Integer> source = s -> s.onSubscribe(srcSubscription); fromSource(source).firstOrElse(() -> null).toFuture().cancel(true); verify(srcSubscription).cancel(); }
@Test public void singleFromSourceSuccess() throws Exception { SingleSource<Integer> src = s -> { s.onSubscribe(IGNORE_CANCEL); s.onSuccess(1); }; Integer result = fromSource(src).toFuture().get(); assertThat("Unexpected result.", result, is(1)); }
@Test public void singleFromSourceError() throws Exception { SingleSource<Integer> src = s -> { s.onSubscribe(IGNORE_CANCEL); s.onError(DELIBERATE_EXCEPTION); }; Future<Integer> future = fromSource(src).toFuture(); expectedException.expect(ExecutionException.class); expectedException.expectCause(sameInstance(DELIBERATE_EXCEPTION)); future.get(); }
@Test public void singleFromSourceCancel() { Cancellable srcCancellable = mock(Cancellable.class); SingleSource<Integer> source = s -> s.onSubscribe(srcCancellable); fromSource(source).toFuture().cancel(true); verify(srcCancellable).cancel(); }
@Test public void completableFromSourceSuccess() throws Exception { CompletableSource src = s -> { s.onSubscribe(IGNORE_CANCEL); s.onComplete(); }; fromSource(src).toFuture().get(); }
@Test public void completableFromSourceError() throws Exception { CompletableSource src = s -> { s.onSubscribe(IGNORE_CANCEL); s.onError(DELIBERATE_EXCEPTION); }; Future<Void> future = fromSource(src).toFuture(); expectedException.expect(ExecutionException.class); expectedException.expectCause(sameInstance(DELIBERATE_EXCEPTION)); future.get(); }
@Test public void completableFromSourceCancel() { Cancellable srcCancellable = mock(Cancellable.class); CompletableSource source = s -> s.onSubscribe(srcCancellable); fromSource(source).toFuture().cancel(true); verify(srcCancellable).cancel(); }