private <X> Publisher<WatchEvent<X>> watch(StandardOperations.Watch operation) {
    final MongoCollection<X> collection = (MongoCollection<X>) this.collection;

    if (operation.query().hasProjections()) {
      return Flowable.error(new UnsupportedOperationException("Projections are not yet supported with watch operation"));
    }

    ChangeStreamPublisher<X> watch;
    if (!operation.query().filter().isPresent()) {
      // watch without filter
      watch = collection.watch(collection.getDocumentClass());
    } else {
      // prefix all attributes with 'fullDocument.'
      PathNaming naming = path -> "fullDocument." + this.pathNaming.name(path);
      AggregationQuery agg = new AggregationQuery(operation.query(), naming); // reuse aggregation pipeline
      watch = collection.watch(agg.toPipeline(), collection.getDocumentClass());
    }

    return Flowable.fromPublisher(watch.fullDocument(FullDocument.UPDATE_LOOKUP))
            .map(MongoWatchEvent::fromChangeStream);
  }
Uses <a href="https://docs.mongodb.com/manual/reference/method/db.collection.replaceOne/">replaceOne</a> operation
with <a href="https://docs.mongodb.com/manual/reference/method/db.collection.bulkWrite/">bulkWrite</a>. Right now has to convert
object to BsonDocument to extract {@code _id} attribute.
criteria/mongo/src/org/immutables/criteria/mongo/MongoSession.java