基于被动Mongo和Web客户端的非阻塞函数方法

Question

我有一个微服务，它使用reactivemongorepository接口从数据库中读取对象。

目标是获取这些对象中的每一个，并将其推送给AWS Lambda函数（在将其转换为DTO之后）。如果lambda函数的结果在200范围内，则将对象标记为成功，否则忽略。

在以前使用简单的Mongo存储库和RestTemplate的时候，这将是一项琐碎的任务。然而，我试图理解这种被动的交易，并避免阻塞。

@Override
public Flux<Video> index() {
    return videoRepository.findAllByIndexedIsFalse().flatMap(video -> {
        final SearchDTO searchDTO = SearchDTO.builder()
                .name(video.getName())
                .canonicalPath(video.getCanonicalPath())
                .objectID(video.getObjectID())
                .userId(video.getUserId())
                .build();

        // Blocking call
        final HttpStatus httpStatus = webClient.post()
                .uri(URI.create(LAMBDA_ENDPOINT))
                .body(BodyInserters.fromObject(searchDTO)).exchange()
                .block()
                .statusCode();

        if (httpStatus.is2xxSuccessful()) {
            video.setIndexed(true);
        }

        return videoRepository.save(video);
    });
}

@Scheduled(fixedDelay = 60000)
public void indexTask() {
    indexService
            .index()
            .log()
            .subscribe();
}

我读过很多关于这个主题的博客文章，但它们都只是简单的CRUD操作，中间没有发生任何事情，所以我并没有真正给出一个如何实现这些事情的完整画面。

有人帮忙吗？

Answer 1

你的解决方案其实相当接近。在这些情况下，您应该尝试分步骤分解反应链，并毫不犹豫地将位转换为独立的方法以求清晰。

@Override
public Flux<Video> index() {

    Flux<Video> unindexedVideos = videoRepository.findAllByIndexedIsFalse();
    return unindexedVideos.flatMap(video -> {
        final SearchDTO searchDTO = SearchDTO.builder()
                .name(video.getName())
                .canonicalPath(video.getCanonicalPath())
                .objectID(video.getObjectID())
                .userId(video.getUserId())
                .build();

        Mono<ClientResponse> indexedResponse = webClient.post()
            .uri(URI.create(LAMBDA_ENDPOINT))
            .body(BodyInserters.fromObject(searchDTO)).exchange()
            .filter(res -> res.statusCode().is2xxSuccessful());

        return indexedResponse.flatMap(response -> {
            video.setIndexed(true);
            return videoRepository.save(video);
        });
    });