反应式 Spring Data Mongodb Query 不应该返回旧数据答案

【问题标题】：Reactive Spring Data Mongodb Query is returning old data when it should not反应式 Spring Data Mongodb Query 不应该返回旧数据
【发布时间】：2019-12-13 07:05:21
【问题描述】：

我正在尝试仅使用 Spring Data 从反应式 Mongodb 存储库中获取新创建的消息。

客户端正在通过 SSE 获取消息。我正在使用“after”查询，它应该只返回在“LocalDateTime.now()”之后发送的消息。

不幸的是，SSE 也推出了早于“现在”的旧消息。我不知道为什么它会返回那些旧消息。

我的控制器方法：

    @GetMapping(value = "/receiving-sse", produces = "text/event-stream")
    public Flux<Message> streamEvents() {
        Mono<String> username = getUsernameFromAuth();

        Flux<Message> message = findOrCreateUser(username)
                .flatMapMany(user -> messageRepository
                        .findAllBySenderIdOrReceiverIdAndSentAtAfter(user.getId(), user.getId(), LocalDateTime.now()));

        Flux<Message> heartBeat = Flux.interval(Duration.ofSeconds(30)).map(sequence -> {
            Message heartBeatMessage = new Message();
            heartBeatMessage.setHeartbeat(true);
            return heartBeatMessage;
        });

        return Flux.merge(message, heartBeat);
    }

我的仓库：

public interface MessageRepository extends ReactiveMongoRepository<Message, String> {

    Flux<Message> findAllByReceiverId(String receiverId);

    @Tailable
    Flux<Message> findAllBySenderIdOrReceiverIdAndSentAtAfter(String senderId, String receiverId, LocalDateTime sentAt);

    Flux<Message> findAllBySenderId(String senderId);

    Flux<Message> findAllByIdIn(Collection<String> ids);

}

还有我的文件：

@Data
@Document
public class Message {

    private String id;

    private LocalDateTime sentAt;

    private String message;

    private boolean heartbeat;

    @DBRef
    private User sender;

    @DBRef
    private User receiver;
}

非常感谢有关 repo 获取“sentAt”早于“LocalDateTime.now()”的消息的任何提示。

【问题讨论】：

标签： java mongodb spring-data spring-webflux

【解决方案1】：

问题出在LocalDateTime.now()。 LocalDateTime 在Flux<Message> message 变量的初始化期间创建一个时间。

您需要将其替换为 Mono.defer() 之类的构造。它将在每个订阅中创建 LocalDateTime。

更多你可以阅读here

【讨论】：