vertx/src/main/java/eu/internetofus/common/vertx/Repository.java
/**
   * Migrate the documents of a collection to the current schema.
   *
   * @param collectionName name of the collection to migrate.
   * @param type           of the documents on the collection.
   * @param version        of the documents to migrate.
   *
   * @param <T>            type of the documents.
   *
   * @return the future that will inform if has migrated or not the documents.
   */
Migrate the documents of a collection to the current schema.
@param collectionName name of the collection to migrate.
@param type           of the documents on the collection.
@param version        of the documents to migrate.

@param             type of the documents.

@return the future that will inform if has migrated or not the documents.
Migrate the documents of a collection to the current schema . @param collectionName name of the collection to migrate . @param type of the documents on the collection . @param version of the documents to migrate . @param type of the documents . @return the future that will inform if has migrated or not the documents .
protected <T extends Model> Future<Void> migrateCollection(final String collectionName, final Class<T> type,
      final String version) {
    final var query = this.createQueryToReturnDocumentsWithAVersionLessThan(version);
    return this.pool.count(collectionName, query).compose(maxDocuments -> {
      if (maxDocuments > 0) {
        Logger.trace("Start to migrate {} documents", maxDocuments);
        final Promise<Void> promise = Promise.promise();
        final var consumerId = UUID.randomUUID().toString();
        final var consumer = this.vertx.eventBus().localConsumer(consumerId);
        consumer.handler(msg -> {
          final var body = msg.body();
          if (body instanceof Number) {
            this.migrateOneDocument(consumerId, collectionName, type, query, ((Number) body).longValue());
          } else {
            if (body instanceof String) {
              promise.fail((String) body);
            } else {
              promise.complete();
            }
            consumer.unregister().onComplete(unregister -> {
              Logger.trace(unregister.cause(), "Finished documents migrations");
            });
          }
        });
        consumer.completionHandler(
            register -> this.migrateOneDocument(consumerId, collectionName, type, query, maxDocuments));
        return promise.future();
      } else {
        return Future.succeededFuture();
      }
    });
  }
protected < T extends Model > Future < Void > migrateCollection ( final String collectionName , final Class < T > type , final String version ) { final var query = this . createQueryToReturnDocumentsWithAVersionLessThan ( version ) ; return this . pool . count ( collectionName , query ) . compose ( maxDocuments -> { if ( maxDocuments > 0 ) { Logger . trace ( "Start to migrate {} documents" , maxDocuments ) ; final Promise < Void > promise = Promise . promise ( ) ; final var consumerId = UUID . randomUUID ( ) . toString ( ) ; final var consumer = this . vertx . eventBus ( ) . localConsumer ( consumerId ) ; consumer . handler ( msg -> { final var body = msg . body ( ) ; if ( body instanceof Number ) { this . migrateOneDocument ( consumerId , collectionName , type , query , ( ( Number ) body ) . longValue ( ) ) ; } else { if ( body instanceof String ) { promise . fail ( ( String ) body ) ; } else { promise . complete ( ) ; } consumer . unregister ( ) . onComplete ( unregister -> { Logger . trace ( unregister . cause ( ) , "Finished documents migrations" ) ; } ) ; } } ) ; consumer . completionHandler ( register -> this . migrateOneDocument ( consumerId , collectionName , type , query , maxDocuments ) ) ; return promise . future ( ) ; } else { return Future . succeededFuture ( ) ; } } ) ; }
Migrate the documents of a collection to the current schema.
Migrate the documents of a collection to the current schema .
// nothing to migrate