Spring WebFlux: разрешен только один абонент

Я пишу простое приложение с Spring 5 Webflux и Kotlin. Я пытаюсь реализовать конечную точку PUT следующим образом:

PUT("/confs/{id}", { val id = it.pathVariable("id") ServerResponse.ok().body(service.save(it.bodyToMono(Item::class.java)), Item::class.java) }) 

Трюк в спасении заключается в том, что я пытаюсь прочитать название города из объекта, разрешить геокоординаты, перезаписать их в исходном элементе, а затем сохранить в Mongo, используя Spring Data Mongo Reactive repo.

 fun save(item: Mono<Item>): Mono<Item> { val geo = item.flatMap { val city = it.location?.city ?: "Somewhere" geoService.resolveGeoFromCity(city) } val zipped = item.zipWith(geo) .map { it.t1.location?.geo = it.t2 it.t1 } return repo.saveAll(zipped) .toMono() } 

Код для решения геокоманд здесь:

 @Service class GeoService() { val client = WebClient.create("https://maps.googleapis.com/maps/api/geocode/") fun resolveGeoFromCity(city: String): Mono<Geo> { return client.get() .uri("json?address=$city&key=$API_KEY&language=en") .exchange() .flatMap { it.bodyToMono(String::class.java) } .map { parse(it) } } private fun parse(response: String): Geo { val locationMap = JsonPath.read<Map<String, Double>>(response, "$.results[0].geometry.location") return Geo(locationMap["lat"] ?: 0.0, locationMap["lng"] ?: 0.0) } } 

Проблема в том, что если вы хотите выполнить запрос PUT, я получил следующую команду stacktrace. Я попытался val geo = Mono.just(Geo(0.0, 0.0)) Mono с val geo = Mono.just(Geo(0.0, 0.0)) (без использования WebClient), а затем он отлично работает.

Как исправить это, не жертвуя функциональностью?)

  2018-01-01 01:41:00.595 ERROR 15120 --- [ctor-http-nio-4] .awreDefaultErrorWebExceptionHandler : Failed to handle request [PUT http://localhost:8097/confs/5a49675c910d123b1057207a] java.lang.IllegalStateException: Only one connection receive subscriber allowed. at reactor.ipc.netty.channel.FluxReceive.startReceiver(FluxReceive.java:276) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE] at reactor.ipc.netty.channel.FluxReceive.subscribe(FluxReceive.java:124) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE] at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.ipc.netty.ByteBufFlux.subscribe(ByteBufFlux.java:242) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE] at reactor.core.publisher.FluxPeek.subscribe(FluxPeek.java:83) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.ipc.netty.ByteBufFlux.subscribe(ByteBufFlux.java:242) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE] at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.FluxFlatMap.subscribe(FluxFlatMap.java:97) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.FluxDoFinallyFuseable.subscribe(FluxDoFinallyFuseable.java:48) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~ [reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.Mono.subscribe(Mono.java:3008) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:167) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:71) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE] at reactor.ipc.netty.channel.ChannelOperations.applyHandler(ChannelOperations.java:383) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE] at reactor.ipc.netty.http.server.HttpServerOperations.onHandlerStart(HttpServerOperations.java:359) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE] at io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:163) ~[netty-common-4.1.17.Final.jar:4.1.17.Final] at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java) ~[netty-common-4.1.17.Final.jar:4.1.17.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:403) ~[netty-common-4.1.17.Final.jar:4.1.17.Final] at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:463) ~[netty-transport-4.1.17.Final.jar:4.1.17.Final] at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858) ~[netty-common-4.1.17.Final.jar:4.1.17.Final] at java.lang.Thread.run(Thread.java:745) ~[na:1.8.0_112] Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: Assembly trace from producer [reactor.core.publisher.FluxMap] : reactor.core.publisher.Flux.map(Flux.java:4855) reactor.ipc.netty.ByteBufFlux.fromInbound(ByteBufFlux.java:68) reactor.ipc.netty.NettyInbound.receive(NettyInbound.java:90) org.springframework.http.server.reactive.ReactorServerHttpRequest.getBody(ReactorServerHttpRequest.java:148) org.springframework.http.codec.DecoderHttpMessageReader.readMono(DecoderHttpMessageReader.java:93) org.springframework.http.codec.DecoderHttpMessageReader.readMono(DecoderHttpMessageReader.java:123) org.springframework.web.reactive.function.BodyExtractors.lambda$null$0(BodyExtractors.java:101) java.util.Optional.map(Optional.java:215) org.springframework.web.reactive.function.BodyExtractors.readWithMessageReaders(BodyExtractors.java:256) org.springframework.web.reactive.function.BodyExtractors.lambda$toMono$2(BodyExtractors.java:96) org.springframework.web.reactive.function.server.DefaultServerRequest.body(DefaultServerRequest.java:126) org.springframework.web.reactive.function.server.DefaultServerRequest.body(DefaultServerRequest.java:120) org.springframework.web.reactive.function.server.DefaultServerRequest.bodyToMono(DefaultServerRequest.java:145) com.example.confs.web.ConferenceRouter$routes$1$2.invoke(ConferenceRouter.kt:31) com.example.confs.web.ConferenceRouter$routes$1$2.invoke(ConferenceRouter.kt:16) org.springframework.web.reactive.function.server.RouterFunctionDsl$PUT$1.handle(RouterFunctionDsl.kt:200) org.springframework.web.reactive.function.server.support.HandlerFunctionAdapter.handle(HandlerFunctionAdapter.java:61) org.springframework.web.reactive.DispatcherHandler.invokeHandler(DispatcherHandler.java:168) org.springframework.web.reactive.DispatcherHandler.lambda$handle$1(DispatcherHandler.java:160) reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:118) reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:76) reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:271) reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:803) reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:1649) reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1463) reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:1337) reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54) reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) reactor.core.publisher.Mono.subscribe(Mono.java:3008) reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onComplete(FluxSwitchIfEmpty.java:75) reactor.core.publisher.Operators.complete(Operators.java:125) reactor.core.publisher.MonoEmpty.subscribe(MonoEmpty.java:45) reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44) reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44) reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44) reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44) reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59) reactor.core.publisher.Mono.subscribe(Mono.java:3008) reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:418) reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onSubscribe(FluxConcatMap.java:210) reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:128) reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:61) reactor.core.publisher.FluxConcatMap.subscribe(FluxConcatMap.java:121) reactor.core.publisher.MonoNext.subscribe(MonoNext.java:40) reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44) reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60) reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60) reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74) reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) reactor.core.publisher.Mono.subscribe(Mono.java:3008) reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:167) reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) reactor.ipc.netty.channel.ChannelOperations.applyHandler(ChannelOperations.java:383) reactor.ipc.netty.http.server.HttpServerOperations.onHandlerStart(HttpServerOperations.java:359) io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:163) io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:403) io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:463) io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858) Error has been observed by the following operator(s): |_ Flux.map(ByteBufFlux.java:68) |_ Flux.doOnNext(ByteBufFlux.java:230) |_ Flux.map(ReactorServerHttpRequest.java:148) |_ Flux.flatMap(AbstractJackson2Decoder.java:95) |_ Flux.doFinally(AbstractJackson2Decoder.java:95) |_ Flux.map(AbstractJackson2Decoder.java:117) |_ Flux.singleOrEmpty(AbstractJackson2Decoder.java:87) |_ Operators.error(FluxReceive.java:276) |_ Mono.onErrorMap(DefaultServerRequest.java:146) |_ Mono.map(ConferenceService.kt:27) |_ Mono.map(ConferenceService.kt:32) |_ Mono.zipWith(ConferenceService.kt:47) |_ Mono.map(ConferenceService.kt:48) |_ Flux.flatMap(SimpleReactiveMongoRepository.java:318) |_ MonoExtensionsKt.toMono(ConferenceService.kt:55) |_ Mono.map(ConferenceService.kt:56) |_ Flux.map(AbstractJackson2Encoder.java:99) |_ Mono.flatMap(DispatcherHandler.java:177) |_ Mono.onErrorResume(DispatcherHandler.java:177) |_ Mono.flatMap(DispatcherHandler.java:161) |_ Mono.defer(DefaultWebFilterChain.java:71) |_ Mono.doOnSuccess(MetricsWebFilter.java:59) |_ Mono.doOnError(MetricsWebFilter.java:60) |_ Mono.compose(MetricsWebFilter.java:54) |_ Mono.defer(DefaultWebFilterChain.java:71) |_ Mono.defer(DefaultWebFilterChain.java:71) 

Здесь я сделал очень похожий пример:

Этот маршрутизатор получит геолокацию от имени, а с другой службой – время восхода и захода солнца:

Я использую метод и метод из Mono.

 internal fun buildResponse(address: Mono<String>) = address.transform(geoLocationService::fromAddress).and(this::sunriseSunset, ::LocationResponse) internal fun sunriseSunset(geographicCoordinates: GeographicCoordinates) = geographicCoordinates.toMono().transform(sunriseSunsetService::fromGeographicCoordinates) 

“ `

Более подробная информация о примере

Поток результатов WebClient.exchange() является одноадресной

Проблема в том, что WebClient позволяет только одному абоненту на соединение. Если вы попытаетесь подписаться на одно и то же обменное соединение дважды – вы получите java.lang.IllegalStateException: Only one connection receive subscriber allowed.

Несмотря на то, что я не вижу, где вы пытались повторно использовать одно и то же соединение дважды, я считаю, что вы можете решить эту проблему, используя следующую комбинацию операторов:

 class GeoService() { val client = WebClient.create("https://maps.googleapis.com/maps/api/geocode/") fun resolveGeoFromCity(city: String): Mono<Geo> { return client.get() .uri("json?address=$city&key=$API_KEY&language=en") .exchange() .flatMap { it.bodyToMono(String::class.java) } .map { parse(it) } .share(); } ... } 

в этом примере поток сконфигурирован для многоадресной рассылки (долей) исходного источника, если будет подписан хотя бы один Subscriber . Если вам нужно, чтобы все подписчики получали ту же дату, вы можете заменить .share на .cache .

Кроме того, существует альтернатива вышеописанному методу. Вы можете заменить упомянутого оператора процессором и получить такую ​​же возможность совместного использования:

 class GeoService() { val client = WebClient.create("https://maps.googleapis.com/maps/api/geocode/") fun resolveGeoFromCity(city: String): Mono<Geo> { return client.get() .uri("json?address=$city&key=$API_KEY&language=en") .exchange() .flatMap { it.bodyToMono(String::class.java) } .map { parse(it) } .subscribeWith(DirectProcessor.create()); } ... } 

В этом случае вы подписываете и используете потребление данных источника прямо сразу после вызова subscribeWith , поэтому потенциально в этом случае вы можете потерять часть данных и т. Д.

Почему с Mono.just(..) все работает нормально?

Прежде всего, это просто холодный оператор, он позволяет как можно .just абонентов получать одни и те же данные в любой момент времени. Вот почему, когда вы пытались использовать один и тот же фрагмент данных из соединения дважды, вы не получили никаких исключений.