vertx/src/main/java/eu/internetofus/common/vertx/Repository.java
/**
   * Called when want to migrate one document.
   *
   * @param consumerId     address to send the message to continue migrating.
   * @param collectionName name of the collection to migrate.
   * @param type           of the documents on the collection.
   * @param query          to obtain the documents to migrate.
   * @param maxDocuments   number of document that has to migrate.
   *
   * @param <T>            type of the documents.
   */
Called when want to migrate one document.
@param consumerId     address to send the message to continue migrating.
@param collectionName name of the collection to migrate.
@param type           of the documents on the collection.
@param query          to obtain the documents to migrate.
@param maxDocuments   number of document that has to migrate.

@param             type of the documents.
Called when want to migrate one document . @param consumerId address to send the message to continue migrating . @param collectionName name of the collection to migrate . @param type of the documents on the collection . @param query to obtain the documents to migrate . @param maxDocuments number of document that has to migrate . @param type of the documents .
protected <T extends Model> void migrateOneDocument(final String consumerId, final String collectionName,
      final Class<T> type, final JsonObject query, final long maxDocuments) {
    this.findOneDocument(collectionName, query, null, null).onComplete(find -> {
      if (find.failed()) {
        final var cause = find.cause();
        Logger.error(cause, "Cannot find document to migrate");
        this.vertx.eventBus().send(consumerId, cause.getMessage());
      } else {
        final var foundObject = find.result();
        try {
          final var mapper = new ObjectMapper();
          mapper.configure(DeserializationFeature.FAIL_ON_IGNORED_PROPERTIES, false);
          mapper.configure(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES, false);
          mapper.configure(DeserializationFeature.FAIL_ON_INVALID_SUBTYPE, false);
          mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
          final var reader = mapper.readerFor(type);
          @SuppressWarnings("unchecked")
          final var value = (T) reader.readValue(foundObject.encode());
          final var id = foundObject.remove("_id");
          final var updateQuery = new JsonObject().put("_id", id);
          final var model = value.toJsonObject();
          for (final var key : foundObject.fieldNames()) {
            if (!model.containsKey(key)) {
              model.putNull(key);
            }
          }
          this.updateOneDocument(collectionName, updateQuery, model).onComplete(update -> {
            if (update.failed()) {
              final var cause = update.cause();
              Logger.error(cause, "Cannot migrate {}", () -> foundObject.encodePrettily());
              this.vertx.eventBus().send(consumerId, cause.getMessage());
            } else if (maxDocuments == 1) {
              Logger.trace("Migrated all document");
              this.vertx.eventBus().send(consumerId, null);
            } else {
              final var documentsToMigrate = maxDocuments - 1;
              Logger.trace("Migrated a document. There are {} documents to migrate.", documentsToMigrate);
              this.vertx.eventBus().send(consumerId, documentsToMigrate);
            }
          });
        } catch (final Throwable cause) {
          Logger.error(cause, "Cannot migrate {}", () -> foundObject.encodePrettily());
          this.vertx.eventBus().send(consumerId, cause.getMessage());
        }
      }
    });
  }
protected < T extends Model > void migrateOneDocument ( final String consumerId , final String collectionName , final Class < T > type , final JsonObject query , final long maxDocuments ) { this . findOneDocument ( collectionName , query , null , null ) . onComplete ( find -> { if ( find . failed ( ) ) { final var cause = find . cause ( ) ; Logger . error ( cause , "Cannot find document to migrate" ) ; this . vertx . eventBus ( ) . send ( consumerId , cause . getMessage ( ) ) ; } else { final var foundObject = find . result ( ) ; try { final var mapper = new ObjectMapper ( ) ; mapper . configure ( DeserializationFeature . FAIL_ON_IGNORED_PROPERTIES , false ) ; mapper . configure ( DeserializationFeature . FAIL_ON_NULL_FOR_PRIMITIVES , false ) ; mapper . configure ( DeserializationFeature . FAIL_ON_INVALID_SUBTYPE , false ) ; mapper . configure ( DeserializationFeature . FAIL_ON_UNKNOWN_PROPERTIES , false ) ; final var reader = mapper . readerFor ( type ) ; @ SuppressWarnings ( "unchecked" ) final var value = ( T ) reader . readValue ( foundObject . encode ( ) ) ; final var id = foundObject . remove ( "_id" ) ; final var updateQuery = new JsonObject ( ) . put ( "_id" , id ) ; final var model = value . toJsonObject ( ) ; for ( final var key : foundObject . fieldNames ( ) ) { if ( ! model . containsKey ( key ) ) { model . putNull ( key ) ; } } this . updateOneDocument ( collectionName , updateQuery , model ) . onComplete ( update -> { if ( update . failed ( ) ) { final var cause = update . cause ( ) ; Logger . error ( cause , "Cannot migrate {}" , ( ) -> foundObject . encodePrettily ( ) ) ; this . vertx . eventBus ( ) . send ( consumerId , cause . getMessage ( ) ) ; } else if ( maxDocuments == 1 ) { Logger . trace ( "Migrated all document" ) ; this . vertx . eventBus ( ) . send ( consumerId , null ) ; } else { final var documentsToMigrate = maxDocuments - 1 ; Logger . trace ( "Migrated a document. There are {} documents to migrate." , documentsToMigrate ) ; this . vertx . eventBus ( ) . send ( consumerId , documentsToMigrate ) ; } } ) ; } catch ( final Throwable cause ) { Logger . error ( cause , "Cannot migrate {}" , ( ) -> foundObject . encodePrettily ( ) ) ; this . vertx . eventBus ( ) . send ( consumerId , cause . getMessage ( ) ) ; } } } ) ; }
Called when want to migrate one document.
Called when want to migrate one document .
// migrated all the documents