java - 响应式(Reactive) Spring Data Mongodb 查询在不应该返回旧数据的情况下返回旧数据

Question

我尝试仅使用 Spring Data 从响应式(Reactive) Mongodb 存储库中获取新创建的消息。

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

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

我的 Controller 方法:

    @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;
}

任何关于为什么存储库正在获取“sentAt”早于“LocalDateTime.now()”的消息的任何提示都非常感谢。

Answer 1

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

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

您可以阅读更多内容 here