FluxSink emitting does not happen when next called during Flux create [SPR-16969]

[spring-projects-issues]

Chris Roberts opened SPR-16969 and commented

Not sure if an issue, or I'm misunderstanding webflux operation. I'm trying to create a Flux<String> which is used by the client when data is available, ie: streamed string by string

Spring boot 2.0.1, using chrome

http://localhost:8080/demo -> Flux stream , streams the results

http://localhost:8080/demo2 -> Flux create, streams the results

 

Spring boot 2.0.3:

http://localhost:8080/demo -> Flux stream , streams the results

http://localhost:8080/demo2 -> Flux create, does not stream but waits for the all results

 

The reactor docs say: 

"The more advanced form of programmatic creation of a Flux, create can work asynchronously or synchronously and is suitable for multiple emissions per round."

 

 

 

 

 

 

---

Affects: 5.0.7

Reference URL: http://projectreactor.io/docs/core/release/reference/

Attachments:

• DemoController.java (1.15 kB)
• pom.xml (1.35 kB)

[spring-projects-issues]

Rossen Stoyanchev commented

I will give it a try but keep in mind you're doing this on a server thread and blocking it, which is not okay. See the reference.

[spring-projects-issues]

Chris Roberts commented

So if I use subscribeOn (slow producer with blocking IO and fast consumer), the first emission is as expected, but remaining emitted on complete

@GetMapping(path = "/demo3")
public Flux<String> demo3() {
   return Flux.from(demo2()).subscribeOn(Schedulers.elastic());
}

[spring-projects-issues]

Rossen Stoyanchev commented

There are two issues. First, streaming is driven by MediaType, and explicit flushing is done for known streaming media types only, see HTTP streaming. That means neither /demo, nor /demo2 should work reliably as-is.

Second, it says on the Javadoc of subscribeOn:

Note that if you are using an eager or blocking create(Consumer, FluxSink.OverflowStrategy) as the source, it can lead to deadlocks due to requests piling up behind the emitter. Thus this operator has a requestOnSeparateThread parameter, which should be set to false in this case.

Basically without subscribeOn you block the event loop thread from Undertow, which is not expected, and you won't be called again until you release it. With subscribeOn, you get off the Undertow thread, and that's why you see the first item, but you're blocking "request" signals and that causes items to be held. Setting requestOneSeparateThread=true allows request signals to come through on the Undertow thread (instead of on the elastic scheduler one). So this works:

@GetMapping(path = "/demo2", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Flux<?> demo2() {
	return Flux.create(e -> {
		try {
			logger.debug("doing 1");
			e.next(Collections.singletonMap("item", "1"));
			Thread.sleep(2000);
			logger.debug("doing 2");
			e.next(Collections.singletonMap("item", "2"));
			Thread.sleep(2000);
			logger.debug("doing 3");
			e.next(Collections.singletonMap("item", "3"));
			Thread.sleep(2000);
			logger.debug("doing 4");
			e.next(Collections.singletonMap("item", "4"));
			e.complete();
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
	}).subscribeOn(Schedulers.elastic(), true);
}

 

[spring-projects-issues]

Rossen Stoyanchev commented

Note that I've created ticket #1266 for an improvement in the Reactor Netty docs.

[spring-projects-issues]

Chris Roberts commented

Thanks for taking the time to work through this. My understanding has improved somewhat and implementing what you suggested works as expected. (NOTE: replacing the 'true' in subsciberOn with 'false')

My use case, to get familiar with reactive, is a command line deployment utility calling curl which gives progress updates as the operational progresses. Quite nifty.