    @Test
    void subscribeSingleFuture_cancel_after_next() throws InterruptedException {

        final CountDownLatch cancelLatch = new CountDownLatch(1);
        final CountDownLatch completeLatch = new CountDownLatch(1);
        final Flux<? extends CharSequence> flux = Flux.<CharSequence>create(emitter -> {
            emitter.next(new StringBuilder("single"));
            emitter.next("next0");
            try {
                assertTrue(cancelLatch.await(100, TimeUnit.MILLISECONDS));
            } catch (final InterruptedException e) {
                // ignore
            }
            assertTrue(emitter.isCancelled());
            completeLatch.countDown();
        }, FluxSink.OverflowStrategy.IGNORE).subscribeOn(Schedulers.single());

        final FluxWithSingle<String, StringBuilder> fluxWithSingle =
                new FluxWithSingleSplit<>(flux, String.class, StringBuilder.class);

        final CountDownLatch onNextLatch = new CountDownLatch(1);
        final CompletableFuture<StringBuilder> future =
                fluxWithSingle.subscribeSingleFuture(s -> onNextLatch.countDown());
        assertTimeout(Duration.ofMillis(100), () -> assertEquals("single", future.get().toString()));
        assertTrue(onNextLatch.await(100, TimeUnit.MILLISECONDS));
        future.cancel(false);
        cancelLatch.countDown();
        assertTrue(completeLatch.await(100, TimeUnit.MILLISECONDS));
        assertTrue(future.isDone());
        assertFalse(future.isCancelled());
        assertFalse(future.isCompletedExceptionally());
    }