@Test
public void castCrashUnsubscribes() {

    PublishProcessor<Integer> pp = PublishProcessor.create();
    TestSubscriber<Integer> ts = TestSubscriber.create();
    pp.flatMap(new Function<Integer, Publisher<Integer>>() {
        @Override
        public Publisher<Integer> apply(Integer t) {
            throw new TestException();
        }
    }, new BiFunction<Integer, Integer, Integer>() {
        @Override
        public Integer apply(Integer t1, Integer t2) {
            return t1;
        }
    }).subscribe(ts);
    Assert.assertTrue("Not subscribed?", pp.hasSubscribers());
    pp.onNext(1);
    Assert.assertFalse("Subscribed?", pp.hasSubscribers());
    ts.assertError(TestException.class);
}