Reactor-netty: Unwrap Mono String to a String value

[letkemanpete78]

Expected behavior

With the following code I was expecting to get a string value
Kotlin Code:

var lastName:String = serverWebExchange
            .formData
            .map<String> { form -> return@map form["lastName"]!![0] }
            .block(Duration.ofSeconds(20))

Actual behavior

However I get the following error message

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-nio-2

Steps to reproduce

Unwrap Mono < String > to String value so that it can be used in a function

Reactor Netty version

SpringBoot versions >= 2.0.1

JVM version (e.g. java -version)

Java 9

OS version (e.g. uname -a)

Windows 10 Pro

This code worked with SpringBoot version 2.0.0 and all of it's releases.

[violetagg]

In version 0.7.6 we added a restriction for blocking Reactor Netty threads.
So if the code above is inside Reactor Netty thread then the expected behaviour is java.lang.IllegalStateException

312

[letkemanpete78]

Do you have a work around with an example that I can get a String value from a Mono String? Or should I go back to SpringBoot 2.0.0?

[violetagg]

You have to tell us why do you need to block: what is the scenario that you are trying to achieve, how will you use the extracted string, can't you use some transformation, is it part of response preparation etc.

[violetagg]

you may want to check this #277

[letkemanpete78]

I'm working on a project where the end user will have the ability to add multiple form elements. For example the form will have the following form elements:

image/file upload (input type=file)
description (input type=text)
sort (input type=hidden)

People will be able to add as many sets of those form elements are needed/wanted up to a point.
I figured out that I need to use ServerWebExchange to get the form elements when posting it. I'd like to combine those form elements into a separate set of objects.

So I would like to get the value of 1 to N form elements and serialize those values into a list of POJO or POKO (K = Kotlin)

[rstoyanchev]

Do you have a work around with an example that I can get a String value from a Mono String?

It's not a workaround really. You're not supposed to block.

First question is, why not use data binding, and map the form input to an @ModelAttribute argument?

Second, if that doesn't work for you for some reason, then you need to show a complete, sample controller method implementation, for us to suggest anything.

[letkemanpete78]

Thank you for the suggestion and I will try it out right now. Hopefully I can provide a sample project with this working or not working in the next day or so.

[letkemanpete78]

I tried using the @ModelAttribute and I think I found a different issue as noted here #352

[bhansu]

I have the same issue . Is it specific to only running app inside Netty server or across all servers ?
whats the work around for this ?

I am calling a REST through WebClient and want to return the result from it.

// client
Mono ingestRes = client.get()
.uri("/ingest")
.accept(MediaType.APPLICATION_JSON)
.exchange();
return ingestRes.flatMap(res -> res.bodyToMono(Response.class)).block();

// handler :
WebClient webClient = WebClient.builder()
.baseUrl("")
.defaultHeader("Authorization", "AAAAAAAAAAA")
.build();

    Mono<ClientResponse> result = webClient.get()
        .uri("a/b/c")
        .exchange();

    result.block();
    return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
            .body(BodyInserters.fromObject(result));

I want the handler to wait till it receives response from rest call and then return it back . how can i do it ?

[letkemanpete78]

I'm not too sure what the solution/work around is as I got sidetracked with
other programming on a different part of the project.

On Wed, May 23, 2018, 6:28 AM bhansu, [e-mail] wrote:

I have the same issue . Is it specific to only running app inside Netty
server or across all servers ?
whats the work around for this ?

—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub
https://github.com/reactor/reactor-netty/issues/351#issuecomment-391299418,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AVXUwLlApel5_jztEkti_t9biN9sfyTzks5t1Tm8gaJpZM4T3Aap
.

[rajanashutosh]

Similar kind off issue even i am facing, if am trying to block in-order to receive a response exception
java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-nio-2
at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:77)

else if I am proceeding with non blocking- I am getting and Error Mono

The underlying HTTP client completed without emitting a response and call to another reactive service is not made or completed.

[rajanashutosh]

This has been sorted out, key is to map with the appropriate subsciber without going for a blocking operation.

[zpf7879]

I cannot see how this is sorted out and why the issue is closed. How we can block and get a result from Mono?

[17610070778]

我写代码，也遇到了这样的问题，查了许多资料，依旧不知道怎么解决，请问这个原因是为什么？

[17610070778]

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-nio-3
at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:77) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Mono.block(Mono.java:1175) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at com.ymhq.nomad.controller.InvoiceController.lambda$invoice$9(InvoiceController.java:145) ~[classes/:na]
at reactor.core.publisher.FluxZip$PairwiseZipper.apply(FluxZip.java:960) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxZip$PairwiseZipper.apply(FluxZip.java:949) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoZip$ZipCoordinator.signal(MonoZip.java:237) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoZip$ZipInner.onNext(MonoZip.java:329) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:1640) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoZip$ZipInner.onSubscribe(MonoZip.java:318) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Mono.subscribe(Mono.java:3080) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:128) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoMap.subscribe(MonoMap.java:55) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoMap.subscribe(MonoMap.java:55) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoMap.subscribe(MonoMap.java:55) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoMap.subscribe(MonoMap.java:55) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoMap.subscribe(MonoMap.java:55) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxSourceMono.subscribe(FluxSourceMono.java:46) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxPeek.subscribe(FluxPeek.java:83) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at org.springframework.http.server.reactive.ChannelSendOperator.subscribe(ChannelSendOperator.java:74) ~[spring-web-5.0.7.RELEASE.jar:5.0.7.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1083) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:198) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:198) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1083) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoIgnoreThen$ThenAcceptInner.onNext(MonoIgnoreThen.java:295) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1083) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:144) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1083) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoZip$ZipCoordinator.signal(MonoZip.java:247) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoZip$ZipInner.onNext(MonoZip.java:329) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:185) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxDefaultIfEmpty$DefaultIfEmptySubscriber.onNext(FluxDefaultIfEmpty.java:92) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:1640) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1454) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:1328) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoDefaultIfEmpty.subscribe(MonoDefaultIfEmpty.java:37) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoPeek.subscribe(MonoPeek.java:71) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Mono.subscribe(Mono.java:3080) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:128) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:53) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:153) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:76) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:271) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:803) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:1640) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1454) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:1328) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Mono.subscribe(Mono.java:3080) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:418) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onSubscribe(FluxConcatMap.java:210) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:140) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:64) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.FluxConcatMap.subscribe(FluxConcatMap.java:121) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoNext.subscribe(MonoNext.java:40) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:53) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1083) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoZip$ZipCoordinator.signal(MonoZip.java:247) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoZip$ZipInner.onNext(MonoZip.java:329) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:1640) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoZip$ZipInner.onSubscribe(MonoZip.java:318) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Mono.subscribe(Mono.java:3080) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:128) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:53) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:53) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:53) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:53) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:53) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.Mono.subscribe(Mono.java:3080) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:172) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:70) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]
at reactor.ipc.netty.channel.ChannelOperations.applyHandler(ChannelOperations.java:380) ~[reactor-netty-0.7.8.RELEASE.jar:0.7.8.RELEASE]
at reactor.ipc.netty.http.server.HttpServerOperations.onHandlerStart(HttpServerOperations.java:398) ~[reactor-netty-0.7.8.RELEASE.jar:0.7.8.RELEASE]
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:163) ~[netty-common-4.1.25.Final.jar:4.1.25.Final]
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java) ~[netty-common-4.1.25.Final.jar:4.1.25.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:404) ~[netty-common-4.1.25.Final.jar:4.1.25.Final]
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:465) ~[netty-transport-4.1.25.Final.jar:4.1.25.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:884) ~[netty-common-4.1.25.Final.jar:4.1.25.Final]
at java.base/java.lang.Thread.run(Thread.java:844) ~[na:na]

[17610070778]

I cannot see how this is sorted out and why the issue is closed. How we can block and get a result from Mono?

你的解决了吗？

[violetagg]

The solution depends on the use case. You may want to ask on the gitter channel: https://gitter.im/reactor/reactor

One very basic scenario is the following:

When you "echo" the received request body.

If you write the handler in the following way you will receive the exception above:

HttpServer.create()
    .port(8080)
    .handle((req, res) -> {
        String requestBody = req.receive().aggregate().asString().block();
        return res.sendString(Mono.just(requestBody));
    })

If you rewrite the code to the one below you'll be able to achieve echo of the request body without blocking. (you specify send the response when you receive the request)

HttpServer.create()
    .port(8080)
    .handle((req, res) -> res.sendString(req.receive().aggregate().asString()))

[zpf7879]

Sort of the same idea with @violetagg, I write my code as below:

return WebClient.builder() .baseUrl() .build() .post() .retrieve() .bodyToMono(Map.class) .flatMap(e -> getAuthentication(e, authentication));

And:
private Mono<? extends Authentication> getAuthentication(Map sspResponse, Authentication authentication) { if (sspResponse != null) { Authentication auth = new UsernamePasswordAuthenticationToken(authentication.getPrincipal(), authentication.getCredentials(), authorities); return Mono.just(auth); } return Mono.error(new FailedLoginException("Authentication failed!")); }

The expected return of my function is Mono of Authentication. As you can see, the idea is that you should return a converted Mono (by mapping) instead of get the content of the Mono (by block) and construct Mono again. I think the latter approach is not "reactive".

This is a mindset change.

However, the specification (and even many tutorials) says Mono.block() should be supported. Apparently, in this case, the implementation deviates the specification even if it is justified.

[violetagg]

However, the specification (and even many tutorials) says Mono.block() should be supported. Apparently, in this case, the implementation deviates the specification even if it is justified.

If you really want to block you can achieve that with additional external threads. The idea here is that Reactor Netty works with very small number of threads and you should not block these threads.

[17610070778]

（这段代码是跑在Netty中的）问题已经得到解决。是因为Mono.just()返回的是MonoJust，而执行map方法后，返回的是MonoMap，执行zipWith后，返回的是MonoZip。而MonoJust对象在block()时，直接返回的值，MonoMap以及MonoZip在执行block()时，首先检查当前线程是否是NoBlocking的对象，如果是则不能阻塞，报异常。调试的时候发现Netty的主线程是NoBlocking的代理对象。最后说一下Netty的线程是不可以阻塞的对吗？

[zpf7879]

@17610070778 According to @violetagg The idea here is that Reactor Netty works with very small number of threads and you should not block these threads. So, it seems like "Netty的线程是不可以阻塞的".

[remigiusv]

I am having same issue of block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-3

I am inside the Kafka reactive subscription which is NonBlocking thread. So WebClient block inside such subscription is failing with the java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-3 at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:111) ~[reactor-core-3.2.10.RELEASE.jar:3.2.10.RELEASE]

Any workaround?

[violetagg]

@remigiusv For question use Gitter/Stackoverflow

[17610070778]

我是用的是netty，netty的boss线程标记了状态，不可阻塞，所以阻塞会报错。

------------------ Original ------------------
From: remigiusv [e-mail]
Date: Tue,Jul 2,2019 7:14 PM
To: reactor/reactor-netty [e-mail]
Cc: 17610070778 [e-mail], Mention [e-mail]
Subject: Re: [reactor/reactor-netty] Unwrap Mono String to a String value (#351)

I am having same issue of block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-3

I am inside the Kafka reactive subscription which is NonBlocking thread. So WebClient block inside such subscription is failing with the java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-3 at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:111) ~[reactor-core-3.2.10.RELEASE.jar:3.2.10.RELEASE]

Any workaround?

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub, or mute the thread.