Почему не вызывается метод doOnComplete издателя Flux?

Question

Я пытаюсь выполнить метод после выполнения другого с помощью издателя Flux, но метод doOnComplete никогда не вызывается. Код выглядит так:

public class Client implements Serializable {
    private Long id;
    private String category;
    // other properties, getters and setters
}

interface ClientRepository extends JpaRepository<Client,Long> {
    List<Client> findAllByCategory(String category);
    @Transactional
    void deleteByCategory(String category);
}

class ClientResponse {
    private Long status;
    private String message;
}

@Component
class ClientService {
    @Autowired
    ClientRepository clientRepository;
    @Autowired
    WebClient webClient;

  public Mono<ClientResponse> deleteRemoteClient(Long idClient) {
    return webClient.post()
      .uri("/api/remoteClient/{idClient}",idClient)
      .retrieve()
      .bodyToMono(ClientResponse.class)
      .doOnSuccess(ok -> System.out.println(
                          "Delete success for client= " + idClient))
      .doOnError(err -> System.out.println(
                          "Delete failed for client= " + idClient + ", err =" + err));
  }

  /**
   * Get All clients by category, then delete them remotely one by one.
   * When everything goes well, delete all clients locally in one shot by category
   **/
  public Flux<ClientResponse> deleteLocalClientByCategory(String category) {
    return Flux.fromStream(clientRepository.findAllByCategory(category).stream())
      .flatMap(client -> deleteRemoteClient(client.getId()))
      .doOnComplete(() -> clientRepository.deleteByCategory(category));
  }
}

@Component
class ClientHandler {
    @Autowired 
    ClientService service;

    public Mono<ClientResponse> deleteByCatgeory(ServerRequest request) {
        return service.deleteLocalClientByCategory(
                   Long.parseLong(request.queryParam("category").get()))
               .publishNext()
               .flatMap(response -> ServerResponse.ok().build());
    }
}

Как я упоминал ранее, метод deleteRemoteClient(client.getId()), но не clientRepository.deleteByCategory(category).

Answer 1

Итак, использовать .doOnComplite для этой цели - не лучшая идея, попробуйте оператор .then, например:

  ...
  .flatMap(client -> deleteRemoteClient(client.getId()))
  .then(clientRepository.deleteByCategory(category));

Answer 2

Когда вы вызываете publishNext, вы отменяете свой поток после первого элемента. См. Иллюстрацию здесь .

Итак, ваш поток никогда не завершается. Вы должны использовать doOnCancel или doFinally.