title: Spring WebFlux - how to get data from DB to use in the next step
body: <p>I use Spring WebFlux (Project Reactor) and I'm facing the following problem:
  I have to get some data from db to use them to call another service - everything in one stream. How to do that?</p>
  
  <pre><code>public Mono&lt;MyObj&gt; saveObj(Mono&lt;MyObj&gt; obj) {
      return obj
          .flatMap(
              ob-&gt;
                  Mono.zip(
                          repo1.save(
                            ...),
                          repo2
                              .saveAll(...)
                              .collectList(),
                          repo3
                              .saveAll(...)
                              .collectList())
                      .map(this::createSpecificObject))
          .doOnNext(item-&gt; createObjAndCallAnotherService(item));
    }
  
  
  
  private void createObjAndCallAnotherService(Prot prot){
  myRepository
          .findById(
              prot.getDomCred().stream()
                  .filter(Objects::nonNull)
                  .findFirst()
                  .map(ConfDomCred::getCredId)
                  .orElse(UUID.fromString("00000000-0000-0000-0000-000000000000")))
          .doOnNext( //one value is returned from myRepository -&gt; Flux&lt;MyObjectWithNeededData&gt;
              confCred-&gt; {//from this point the code is unreachable!!! - why????
                Optional&lt;ConfDomCred&gt; confDomCred=
                    prot.getDomCreds().stream().filter(Objects::nonNull).findFirst();
  
                confDomCred.ifPresent(
                    domCred -&gt; {
                      ProtComDto com=
                          ProtComDto.builder()
                              .userName(confCred.getUsername())
                              .password(confCred.getPassword())                          
                              .build();
                      clientApiToAnotherService.callEndpintInAnotherService(com); //this is a client like Feign that invokes method in another service
                    });
              });
  }
  </code></pre>
  
  <p><strong>UPDATE</strong></p>
  
  <p>When I invoke</p>
  
  <pre><code>    Flux&lt;MyObj&gt; myFlux =  myRepository
              .findById(
                  prot.getDomCred().stream()
                      .filter(Objects::nonNull)
                      .findFirst()
                      .map(ConfDomCred::getCredId)
                      .orElse(UUID.fromString("00000000-0000-0000-0000-000000000000")));
  
  myFlux.subscribe(e -&gt; e.getPassword()) 
  </code></pre>
  
  <p>then the value is printed</p>
  
  <p><strong>UPDATE2</strong></p>
  
  <p>So as a recap - I think the code below is asynchronous/non-blocking - am I right?
  In my </p>
  
  <blockquote>
    <p>ProtectionCommandService</p>
  </blockquote>
  
  <p>I had to use subscribe() twice - only then I can call my other service and store them my object: <code>commandControllerApi.createNewCommand</code></p>
  
  <pre><code>public Mono&lt;Protection&gt; saveProtection(Mono&lt;Protection&gt; newProtection) {
      return newProtection.flatMap(
          protection -&gt;
              Mono.zip(
                      protectorRepository.save(//some code),
                      domainCredentialRepository
                          .saveAll(//some code)
                          .collectList(),
                      protectionSetRepository
                          .saveAll(//some code)
                          .collectList())
                  .map(this::createNewObjectWrapper)
                  .doOnNext(protectionCommandService::createProtectionCommand));
    }
  </code></pre>
  
  <p>ProtectionCommandService class:</p>
  
  <pre><code>public class ProtectionCommandService {
  
    private final ProtectionCommandStrategyFactory protectionCommandFactory;
    private final CommandControllerApi commandControllerApi;
  
    public Mono&lt;ProtectionObjectsWrapper&gt; createProtectionCommand(
        ProtectionObjectsWrapper protection) {
      ProductType productType = protection.getProtector().getProductType();
  
      Optional&lt;ProtectionCommandFactory&gt; commandFactory = protectionCommandFactory.get(productType);
  
      commandFactory
          .get()
          .createCommandFromProtection(protection)
          .subscribe(command -&gt; commandControllerApi.createNewCommand(command).subscribe());
      return Mono.just(protection);
    }
  }
  </code></pre>
  
  <p>And one of 2 factories:</p>
  
  <pre><code>@Component
  @AllArgsConstructor
  @Slf4j
  public class VmWareProtectionCommandFactory implements ProtectionCommandFactory {
  
    private static final Map&lt;ProductType, CommandTypeEnum&gt; productTypeToCommandType =
        ImmutableMap.of(...//some values);
  
    private final ConfigurationCredentialRepository configurationCredentialRepository;
  
    @Override
    public Mono&lt;CommandDetails&gt; createCommandFromProtection(ProtectionObjectsWrapper protection) {
      Optional&lt;DomainCredential&gt; domainCredential =
          protection.getDomainCredentials().stream().findFirst();
  
      return configurationCredentialRepository
          .findByOwnerAndId(protection.getOwner(), domainCredential.get().getCredentialId())
          .map(credential -&gt; createCommand(protection, credential, domainCredential.get()));
    }
  </code></pre>
  
  <p>and createCommand method returns Mono object as a result of this factory.</p>
  
  <pre><code>private Mono&lt;CommandDetails&gt; createCommand(Protection protection
       //other parameters) {
  
      CommandDto commandDto =
          buildCommandDto(protection, confCredential, domainCredentials);
  
      String commands = JsonUtils.toJson(commandDto);
      CommandDetails details = new CommandDetails();
      details.setAgentId(protection.getProtector().getAgentId().toString());
      details.setCommandType(///some value);
      details.setArguments(//some value);
      return Mono.just(details);
  </code></pre>
  
  <p><strong>UPDATE3</strong></p>
  
  <p>My main method that calls everything has been changed a little bit:</p>
  
  <pre><code>public Mono&lt;MyObj&gt; saveObj(Mono&lt;MyObj&gt; obj) {
      return obj
          .flatMap(
              ob-&gt;
                  Mono.zip(
                          repo1.save(
                            ...),
                          repo2
                              .saveAll(...)
                              .collectList(),
                          repo3
                              .saveAll(...)
                              .collectList())
  .map(this::wrapIntoAnotherObject)
  .flatMap(protectionCommandService::createProtectionCommand)
  .map(this::createMyObj));
  </code></pre>