@Override
    public <T> Flux<ChangeStreamDocument<T>> watch(Class<T> clazz, FullDocument fullDocument) {
        MongoCollection<T> collection = context.getCollection(clazz);
        Publisher<ChangeStreamDocument<T>> source = collection.watch(clazz)
                .fullDocument(fullDocument);
        // Use single thread for onNext calls to ensure they happen in sequence
        // without the need of "synchronized" one by one

        // The code is a little hacky because only using one thread bases on the fact
        // that we know all our usages are so fast that they can run in a thread
        return Flux.from(source)
                .publishOn(WATCH_SCHEDULER);
    }
apply the shard key to the filter so that mongos know to which MongoDB the command needs to
be sent for better performance
turms-server-common/src/main/java/im/turms/server/common/storage/mongo/operation/TurmsMongoOperations.java