title: How to map a flux with mono?
body: <p>I have these two requests:</p>
  
  <pre><code>Flux&lt;ProductID&gt; getProductIds() {
      return this.webClient.get()
              .uri(PRODUCT_ID_URI)
              .accept(MediaType.APPLICATION_STREAM_JSON)
              .retrieve()
              .bodyToFlux(ProductID.class);
  }
  
  Mono&lt;Product&gt; getProduct(String id) {
      return this.itemServiceWebClient.get()
              .uri(uriBuilder -&gt; uriBuilder.path(PRODUCT_URI + "/{id}")
                      .build(id))
              .accept(MediaType.APPLICATION_STREAM_JSON)
              .exchange()
              .flatMap(clientResponse -&gt; clientResponse.bodyToMono(Product.class));
  }
  </code></pre>
  
  <p>And with these I want to do the following:</p>
  
  <pre><code>Flux&lt;Product&gt; getProducts() {
      return Flux.create(sink -&gt; this.gateway.getProductIds()
              .doOnComplete(() -&gt; {
                  log.info("PRODUCTS COMPLETE");
  
                  sink.complete();
              })
              .flatMap(productId -&gt; this.getProduct(productId.getID()))
              .subscribe(product -&gt; {
                  log.info("NEW PRODUCT: " + product);
  
                  sink.next(product);
              }));
  }
  </code></pre>
  
  <p>When I call this I get the following output:</p>
  
  <pre><code>PRODUCTS COMPLETE
  NEW PRODUCT: ...
  NEW PRODUCT: ...
  ....
  </code></pre>
  
  <p>Ofcourse the stream is closing before the results are actually there because the the async mono mapping. How can I keep this non-blocking but also making sure that the results arrive before the on complete is called?</p>