UserCollectionChangeStreamService

UserCollectionChangeStreamService.java

reactiveMongoTemplate
        .changeStream(User.class)
        .watchCollection("user")
        .listen()
        .subscribe(event -> {
            log.info("change stream event {}", event);
            BsonDocument newResumeToken = Objects.requireNonNull(event.getRaw()).getResumeToken().asDocument();
            BsonString token = newResumeToken.getString("_data");
            var timeStampInMs = Objects.requireNonNull(event.getBsonTimestamp()).getTime() * 1000L;

            syncToPostgresService.syncInASingleTx(event.getBody(), token.getValue(), timeStampInMs);
        });