Webflux在何时重复或何时重试

Question

我使用SpringBoot和Webflux的反应式编程。我想重复对endpoint的一些调用，直到数据可用为止（将返回一些内容）。

我想调用commandControllerApi。findById直到displayCommand返回状态为==SUCCESS。如何告诉Webflux我的链的这一部分应该被调用5次，例如，因为我的数据库中的数据应该在5-10秒后出现。。。

我认为当前代码会导致再次调用整个链，而不仅仅是调用我链的适当部分（.flatMap（commandResponse））-

我的代码：

public Mono<Boolean> validateCredentials(FlowConfCredentials flowCredentials, UUID agentId) {
    return securityService
        .getUser()
        .flatMap(
            user -> {
              Command command = new Command ();
              command.setAgentId(agentId.toString());
              command.setCommandType(COMMAND_TYPE);
              command.setArguments(createArguments());
              command.setCreatedBy(
                  user.getEmail());
              return commandControllerApi.saveCommand(command);
            })
        //       .retryWhen(Retry.fixedDelay(5, Duration.ofSeconds(5)))
      .flatMap(commandResponse -> commandControllerApi.findById(commandResponse.getCommandId()))
        .filter(displayCommand -> displayCommand.getStatus().equals(OaCommandStatus.SUCCESS))
        .retryWhen(Retry.fixedDelay(5, Duration.ofSeconds(5)))
//        .repeatWhenEmpty(
//            Repeat.onlyIf(repeatContext -> true)
//                .exponentialBackoff(Duration.ofSeconds(5), Duration.ofSeconds(5))
//                .timeout(Duration.ofSeconds(30)))
        .filter(
            commandResponse ->
                commandResponse.getStatus() != null
                    && commandResponse.getStatus().equals(CommandStatus.SUCCESS))
        .map(commandResponse -> true)
        .switchIfEmpty(Mono.just(false));
  }

下面是调用上面的metohd的方法：

 public Flux<CredConfiguration> saveCredentials(
      Mono<FlowConfCredentials> flowCredentials, UUID agentId) {
    return flowCredentials
        .filterWhen(cred -> validationService.validateCredentials(cred, agentId))
        .flatMapMany(
            flowConfCredentials -> {
              if (condition1()) {
                return caveCredentials(flowConfCredentials);
              }
              if (condition2()) {
                return saveCredentialsForUser(flowConfCredentials.getExistingCredentials());
              }
              return Flux.error(new EmptyConfigurationException(CREDENTIALS_MESSAGE));
            });
  }




Answer 1

要仅重复订阅findById返回的mono而不重新订阅上游saveCommand/getUser，请将筛选器/repeatWhenEmpty移动到调用findById的flatMap中。

public Mono<Boolean> validateCredentials(FlowConfCredentials flowCredentials, UUID agentId) {
    return securityService
        .getUser()
        .flatMap(
            user -> {
              Command command = new Command();
              command.setAgentId(agentId.toString());
              command.setCommandType(COMMAND_TYPE);
              command.setArguments(createArguments());
              command.setCreatedBy(
                  user.getEmail());
              return commandControllerApi.saveCommand(command);
            })
        .flatMap(saveResponse -> commandControllerApi.findById(saveResponse.getCommandId())
           .filter(findResponse -> findResponse.getStatus().equals(OaCommandStatus.SUCCESS))
           .repeatWhenEmpty(
                Repeat.onlyIf(repeatContext -> true)
                    .exponentialBackoff(Duration.ofSeconds(5), Duration.ofSeconds(5))
                    .timeout(Duration.ofSeconds(30))))
        .hasElement();
}