title: Spring WebFlux: Only one connection receive subscriber allowed
body: <p>I am writing a simple app with Spring 5 Webflux and Kotlin. I am trying to implement PUT endpoint in a following way:</p>
  
  <pre><code>PUT("/confs/{id}", {
      val id = it.pathVariable("id")
      ServerResponse.ok().body(service.save(it.bodyToMono(Item::class.java)), Item::class.java)
  })
  </code></pre>
  
  <p>The trick on save is that I try to read a city name from item, resolve geo coordinates, overwrite them in original item and then save to Mongo using Spring Data Mongo Reactive repo.</p>
  
  <pre><code>fun save(item: Mono&lt;Item&gt;): Mono&lt;Item&gt; {
      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()
  }
  </code></pre>
  
  <p>Code to resolve geo coordinates is here:</p>
  
  <pre><code>@Service
  class GeoService() {
  
      val client = WebClient.create("https://maps.googleapis.com/maps/api/geocode/")
  
      fun resolveGeoFromCity(city: String): Mono&lt;Geo&gt; {
          return client.get()
                  .uri("json?address=$city&amp;key=$API_KEY&amp;language=en")
                  .exchange()
                  .flatMap { it.bodyToMono(String::class.java) }
                  .map { parse(it) }
      }
  
      private fun parse(response: String): Geo {
          val locationMap = JsonPath.read&lt;Map&lt;String, Double&gt;&gt;(response, "$.results[0].geometry.location")
          return Geo(locationMap["lat"] ?: 0.0, locationMap["lng"] ?: 0.0)
      }
  
  }
  </code></pre>
  
  <p>The problem is that if make a PUT request I got following stacktrace. I have tried to stub Mono with <code>val geo = Mono.just(Geo(0.0, 0.0))</code> (without using WebClient) and then it works fine.</p>
  
  <p>How to fix it without sacrifying functionality?)</p>
  
  <pre><code>    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)
  </code></pre>