Open mp911de opened 4 years ago
The BlobToByteBufferCodec
also should be noted, looks like it use same way to blocking Flux
.
Fixing the reading side is probably quite straight-forward as H2 isn't backed by anything reactive. The complexity comes into play when trying to write values to H2. We need to resolve each parameter that provides a Publisher
through an additional concatMap
during Statement.execute(…)
.
Similar error occurs at
io.r2dbc.h2.codecs.BlobToByteBufferCodec.doEncode(BlobToByteBufferCodec.java:65)
when using Spring Security Oauth2 client configured with H2 and R2dbcReactiveOAuth2AuthorizedClientService
java.lang.IllegalStateException: Iterating over a toIterable() / toStream() is blocking, which is not supported in thread reactor-http-nio-2
at reactor.core.publisher.BlockingIterable$SubscriberIterator.hasNext(BlockingIterable.java:161) ~[reactor-core-3.4.12.jar:3.4.12]
Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Error has been observed at the following site(s):
*__checkpoint ⇢ org.springframework.security.oauth2.client.web.server.authentication.OAuth2LoginAuthenticationWebFilter [DefaultWebFilterChain]
*__checkpoint ⇢ org.springframework.security.oauth2.client.web.server.OAuth2AuthorizationRequestRedirectWebFilter [DefaultWebFilterChain]
*__checkpoint ⇢ org.springframework.security.web.server.context.ReactorContextWebFilter [DefaultWebFilterChain]
*__checkpoint ⇢ org.springframework.security.web.server.header.HttpHeaderWriterWebFilter [DefaultWebFilterChain]
*__checkpoint ⇢ org.springframework.security.config.web.server.ServerHttpSecurity$ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
*__checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
*__checkpoint ⇢ HTTP GET "/login/oauth2/code/registrationid?code=redacted&state=alsoredacted" [ExceptionHandlingWebHandler]
Original Stack Trace:
at reactor.core.publisher.BlockingIterable$SubscriberIterator.hasNext(BlockingIterable.java:161) ~[reactor-core-3.4.12.jar:3.4.12]
at io.r2dbc.h2.codecs.BlobToByteBufferCodec$BlobInputStreamEnumeration.hasMoreElements(BlobToByteBufferCodec.java:92) ~[r2dbc-h2-0.8.4.RELEASE.jar:0.8.4.RELEASE]
at java.base/java.io.SequenceInputStream.peekNextStream(SequenceInputStream.java:100) ~[na:na]
at java.base/java.io.SequenceInputStream.<init>(SequenceInputStream.java:67) ~[na:na]
at io.r2dbc.h2.codecs.BlobToByteBufferCodec.doEncode(BlobToByteBufferCodec.java:65) ~[r2dbc-h2-0.8.4.RELEASE.jar:0.8.4.RELEASE]
Same case as with #127. To solve that issue, we need to defer the stream consumption until statement execution.
Reproducer:
Code must be executed on a
NonBlocking
Reactor Thread such asSchedulers.parallel()
to reproduce the issue.