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.

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, я получу следующую трассировку стека. Я попытался заглушить Mono с помощью val geo = Mono.just(Geo(0.0, 0.0)) (без использования WebClient), и тогда он отлично работает.

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

    2018-01-01 01:41:00.595 ERROR 15120 --- [ctor-http-nio-4] .a.w.r.e.DefaultErrorWebExceptionHandler : 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)

spring-boot kotlin spring-webflux project-reactor reactor-netty
person Nikolay Kuznetsov    schedule 31.12.2017    source источник

Ответы (3)


arrow_upward
14
arrow_downward

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 - это холодный оператор, он позволяет как можно большему количеству абонентов получать одни и те же данные в любой момент времени. Вот почему, когда вы дважды пытались использовать один и тот же блок данных из соединения, вы не получали никаких исключений.

person Oleh Dokuka    schedule 02.01.2018
comment
Привет, Олег. Я думаю, что проблема скорее в зависимости geo Mono от элемента Mono. При подписке geo генерирует вторую подписку на элемент. И это запрещено rx-netty. - person Nikolay Kuznetsov; 02.01.2018
comment
В моей версии Reactor есть оператор share. cache не помогает - person Nikolay Kuznetsov; 02.01.2018
comment
Итак, помогли ли некоторые из предложенных приемов решить вашу проблему? - person Oleh Dokuka; 03.01.2018
comment
Кстати, из вашего стека я не вижу кода, связанного с ConferenceService, но стек ошибок напрямую связан с этой частью кода - person Oleh Dokuka; 03.01.2018
comment
Я применил оператор cache к самому верхнему элементу, тогда он работает нормально. - person Nikolay Kuznetsov; 05.01.2018
comment
@OlehDokuka Этот оператор share, похоже, недоступен в новых выпусках Reactor, и оператор cache по-прежнему выдает ту же ошибку. Была ли эта функция удалена в более поздних выпусках? Если да, то есть ли что-то новое, чтобы это разрешить? Связанный с этим вопрос, который я разместил, немного отличается, но приводит к той же ошибке stackoverflow.com/questions/56238895/ - person Bobbake4; 21.05.2019
comment
@ Bobbake4, пожалуйста, поделитесь образцом кода в своем вопросе. Я сделаю все возможное, чтобы помочь тебе. - person Oleh Dokuka; 21.05.2019
comment
Привет, @OlehDokuka, у нас такая же проблема, но не с Webclient. Не могли бы вы обратиться к нужной документации для понимания, регистрации и устранения неисправностей этого типа. - person Ivan; 30.04.2021

arrow_upward
1
arrow_downward

У меня была аналогичная проблема. Исправление заключалось в том, чтобы указать эту зависимость:

org.springframework: весна-webflux: 5.1.4.RELEASE

Поскольку я использовал spring -boot, он развертывает его предыдущую версию. К сожалению, ссылки на этот вопрос сейчас нет.

Итак, теперь мой градиент выглядит так:

compile('org.springframework.boot:spring-boot-starter-data-mongodb-reactive')
compile('org.springframework.boot:spring-boot-starter-webflux')
// Next 2 dependencies are temporally here until the one above does not resolves next to at least 5.1.4 - where 
// webflux issue is resolved:
// https://github.com/rstoyanchev/spr-issue-migration-test-2/issues/17323
compile('org.springframework:spring-webflux:5.1.4.RELEASE')
compile('org.springframework:spring-web:5.1.4.RELEASE')
person Vitaliy Tsimbaluk    schedule 26.03.2019

arrow_upward
0
arrow_downward

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

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

Я использовал и из 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)

Подробнее об этом пример

person Juan Medina    schedule 01.01.2018