Using combineLatest with a single non-fuseable source results in ClassCastException

[mlex]

Expected behavior

The below test passes.

Actual behavior

The test fails with the following exception

java.lang.ClassCastException: reactor.core.publisher.FluxMap$MapSubscriber cannot be cast to reactor.core.Fuseable$QueueSubscription

	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:89)
	at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:86)
	at reactor.core.publisher.FluxHide$HideSubscriber.onSubscribe(FluxHide.java:69)
	at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:53)
	at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:59)
	at reactor.core.publisher.FluxHide.subscribe(FluxHide.java:39)
	at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62)
	at reactor.core.publisher.FluxCombineLatest.subscribe(FluxCombineLatest.java:164)
	at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63)
	at reactor.core.publisher.MonoCollectList.subscribe(MonoCollectList.java:59)
	at reactor.core.publisher.Mono.block(Mono.java:1174)
	at FluxCombineLatestFuseableBugTest.failsWithClassCastException(FluxCombineLatestFuseableBugTest.java:20)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
	at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
	at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
	at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
	at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
	at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)
	at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
	at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)

Steps to reproduce

  @Test
  public void failsWithClassCastException() {
    List<String> result = Flux.combineLatest(
        Arrays.asList(Flux.just(1, 2, 3).hide()),
        (arr) -> arr[0].toString())
        //.hide() // adding this makes the test pass
        .map(x -> x + "!")
        .collectList()
        .block();
    assertEquals(result, Arrays.asList("1!", "2!", "3!"));
  }

Reactor Core version

Latest master (commit e337f0e) and 3.1.6.RELEASE

JVM version (e.g. java -version)

1.8.0_181-b13

[simonbasle]

This is a bug, but quite a niche one I think, as you have to combine 3 low-probability conditions to trigger it:

• combineLatest with a single source (defeating the purpose a bit, especially if you have a way of knowing if the iterable has more than 1 publishers)
• single source is not Fuseable (eg. using .hide() like here)
• operator after combineLatest is Fuseable

Judging from your combinator function, you seem to expect a singleton list. In that case, combineLatest makes no sense, just use map.

If your production code is more dynamic but you can detect the "iterable has one publisher" case, make it use map there also.

@smaldini what do you think? Since the combineLatest is always detected as Fuseable by downstream, shouldn't the shortcircuit connection in the "1 source" case here only happen if said source is also Fuseable?

[helmbold]

I've encountered a similar issue, also using combineLatest with a single source. As you've said this is a pointless usage of combineLatest but it is configurable by the use in this case. It is a really surprising behavior since you need to know internals to understand what's going on.

I fixed the error by replacing combineLatest with map if the user configured only a single source -- thanks to your answer @simonbasle !

If the reported behavior can't be fixed, I would prefer to enforce multiple sources in combineLatest and throw an IllegalArgument exception otherwise.