distributed-v2/src/main/java/com/orientechnologies/orient/distributed/impl/ODistributedWorker.java
protected void onMessage(final ODistributedRequest iRequest) {
    String senderNodeName = null;
    for (int retry = 0; retry < 10; retry++) {
      senderNodeName = manager.getNodeNameById(iRequest.getId().getNodeId());
      if (senderNodeName != null)
        break;

      try {
        Thread.sleep(200);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();

        throw OException.wrapException(new ODistributedException("Execution has been interrupted"), e);
      }
    }

    if (senderNodeName == null) {
      ODistributedServerLog.warn(this, localNodeName, senderNodeName, DIRECTION.IN,
          "Sender server id %d is not registered in the cluster configuration, discard the request: (%s) (worker=%d)",
          iRequest.getId().getNodeId(), iRequest, id);
      sendResponseBack(iRequest, new ODistributedException("Sender server id " + iRequest.getId().getNodeId()
          + " is not registered in the cluster configuration, discard the request"));
      return;
    }

    final ORemoteTask task = iRequest.getTask();

    if (ODistributedServerLog.isDebugEnabled())
      ODistributedServerLog
          .debug(this, localNodeName, senderNodeName, DIRECTION.IN, "Received request: (%s) (worker=%d)", iRequest, id);

    // EXECUTE IT LOCALLY
    Object responsePayload = null;
    OSecurityUser origin = null;
    try {
      waitNodeIsOnline();

      distributed.waitIsReady(task);

      // EXECUTE THE TASK
      for (int retry = 1; running; ++retry) {
        if (task.isUsingDatabase()) {
          initDatabaseInstance();
          if (database == null)
            throw new ODistributedOperationException(
                "Error on executing remote request because the database '" + databaseName + "' is not available");
          // keep original user in database, check the username passed in request and set new user in DB, after document saved,
          // reset to original user
          if (database != null) {
            database.activateOnCurrentThread();
            origin = database.getUser();
            try {
              if (iRequest.getUserRID() != null && iRequest.getUserRID().isValid() && (lastUser == null || !(lastUser.getIdentity())
                  .equals(iRequest.getUserRID()))) {
                lastUser = database.getMetadata().getSecurity().getUser(iRequest.getUserRID());
                database.setUser(lastUser);// set to new user
              } else
                origin = null;

            } catch (Exception ex) {
              OLogManager.instance().error(this, "Failed on user switching database. ", ex);
            }
          }

        }

        responsePayload = manager.executeOnLocalNode(iRequest.getId(), iRequest.getTask(), database);

        if (responsePayload instanceof OModificationOperationProhibitedException) {
          // RETRY
          try {
            ODistributedServerLog.info(this, localNodeName, senderNodeName, DIRECTION.IN,
                "Database is frozen, waiting and retrying. Request %s (retry=%d, worker=%d)", iRequest, retry, id);

            Thread.sleep(1000);
          } catch (InterruptedException e) {
          }
        } else {
          // OPERATION EXECUTED (OK OR ERROR), NO RETRY NEEDED
          if (retry > 1)
            ODistributedServerLog
                .info(this, localNodeName, senderNodeName, DIRECTION.IN, "Request %s succeed after retry=%d", iRequest, retry);

          break;
        }

      }

    } catch (RuntimeException e) {
      if (task.hasResponse())
        sendResponseBack(iRequest, e);
      throw e;

    } finally {
      if (database != null && !database.isClosed()) {
        database.activateOnCurrentThread();
        if (!database.isClosed()) {
          database.rollback();
          database.getLocalCache().clear();
          if (origin != null)
            database.setUser(origin);
        }
      }
    }

    if (task.hasResponse()) {
      if (!sendResponseBack(iRequest, responsePayload)) {
        handleError(iRequest, responsePayload);
      }
    }
  }
protected void onMessage(final ODistributedRequest iRequest) {
    String senderNodeName = null;
    for (int retry = 0; retry < 10; retry++) {
      senderNodeName = manager.getNodeNameById(iRequest.getId().getNodeId());
      if (senderNodeName != null)
        break;

      try {
        Thread.sleep(200);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();

        throw OException.wrapException(new ODistributedException("Execution has been interrupted"), e);
      }
    }

    if (senderNodeName == null) {
      ODistributedServerLog.warn(this, localNodeName, senderNodeName, DIRECTION.IN,
          "Sender server id %d is not registered in the cluster configuration, discard the request: (%s) (worker=%d)",
          iRequest.getId().getNodeId(), iRequest, id);
      sendResponseBack(iRequest, new ODistributedException("Sender server id " + iRequest.getId().getNodeId()
          + " is not registered in the cluster configuration, discard the request"));
      return;
    }

    final ORemoteTask task = iRequest.getTask();

    if (ODistributedServerLog.isDebugEnabled())
      ODistributedServerLog
          .debug(this, localNodeName, senderNodeName, DIRECTION.IN, "Received request: (%s) (worker=%d)", iRequest, id);

    // EXECUTE IT LOCALLY
    Object responsePayload = null;
    OSecurityUser origin = null;
    try {
      waitNodeIsOnline();

      distributed.waitIsReady(task);

      // EXECUTE THE TASK
      for (int retry = 1; running; ++retry) {
        if (task.isUsingDatabase()) {
          initDatabaseInstance();
          if (database == null)
            throw new ODistributedOperationException(
                "Error on executing remote request because the database '" + databaseName + "' is not available");
          // keep original user in database, check the username passed in request and set new user in DB, after document saved,
          // reset to original user
          if (database != null) {
            database.activateOnCurrentThread();
            origin = database.getUser();
            try {
              if (iRequest.getUserRID() != null && iRequest.getUserRID().isValid() && (lastUser == null || !(lastUser.getIdentity())
                  .equals(iRequest.getUserRID()))) {
                lastUser = database.getMetadata().getSecurity().getUser(iRequest.getUserRID());
                database.setUser(lastUser);// set to new user
              } else
                origin = null;

            } catch (Exception ex) {
              OLogManager.instance().error(this, "Failed on user switching database. ", ex);
            }
          }

        }

        responsePayload = manager.executeOnLocalNode(iRequest.getId(), iRequest.getTask(), database);

        if (responsePayload instanceof OModificationOperationProhibitedException) {
          // RETRY
          try {
            ODistributedServerLog.info(this, localNodeName, senderNodeName, DIRECTION.IN,
                "Database is frozen, waiting and retrying. Request %s (retry=%d, worker=%d)", iRequest, retry, id);

            Thread.sleep(1000);
          } catch (InterruptedException e) {
          }
        } else {
          // OPERATION EXECUTED (OK OR ERROR), NO RETRY NEEDED
          if (retry > 1)
            ODistributedServerLog
                .info(this, localNodeName, senderNodeName, DIRECTION.IN, "Request %s succeed after retry=%d", iRequest, retry);

          break;
        }

      }

    } catch (RuntimeException e) {
      if (task.hasResponse())
        sendResponseBack(iRequest, e);
      throw e;

    } finally {
      if (database != null && !database.isClosed()) {
        database.activateOnCurrentThread();
        if (!database.isClosed()) {
          database.rollback();
          database.getLocalCache().clear();
          if (origin != null)
            database.setUser(origin);
        }
      }
    }

    if (task.hasResponse()) {
      if (!sendResponseBack(iRequest, responsePayload)) {
        handleError(iRequest, responsePayload);
      }
    }
  }
protected void onMessage ( final ODistributedRequest iRequest ) { String senderNodeName = null ; for ( int retry = 0 ; retry < 10 ; retry ++ ) { senderNodeName = manager . getNodeNameById ( iRequest . getId ( ) . getNodeId ( ) ) ; if ( senderNodeName != null ) break ; try { Thread . sleep ( 200 ) ; } catch ( InterruptedException e ) { Thread . currentThread ( ) . interrupt ( ) ; throw OException . wrapException ( new ODistributedException ( "Execution has been interrupted" ) , e ) ; } } if ( senderNodeName == null ) { ODistributedServerLog . warn ( this , localNodeName , senderNodeName , DIRECTION . IN , "Sender server id %d is not registered in the cluster configuration, discard the request: (%s) (worker=%d)" , iRequest . getId ( ) . getNodeId ( ) , iRequest , id ) ; sendResponseBack ( iRequest , new ODistributedException ( "Sender server id " + iRequest . getId ( ) . getNodeId ( ) + " is not registered in the cluster configuration, discard the request" ) ) ; return ; } final ORemoteTask task = iRequest . getTask ( ) ; if ( ODistributedServerLog . isDebugEnabled ( ) ) ODistributedServerLog . debug ( this , localNodeName , senderNodeName , DIRECTION . IN , "Received request: (%s) (worker=%d)" , iRequest , id ) ; // EXECUTE IT LOCALLY Object responsePayload = null ; OSecurityUser origin = null ; try { waitNodeIsOnline ( ) ; distributed . waitIsReady ( task ) ; // EXECUTE THE TASK for ( int retry = 1 ; running ; ++ retry ) { if ( task . isUsingDatabase ( ) ) { initDatabaseInstance ( ) ; if ( database == null ) throw new ODistributedOperationException ( "Error on executing remote request because the database '" + databaseName + "' is not available" ) ; // keep original user in database, check the username passed in request and set new user in DB, after document saved, // reset to original user if ( database != null ) { database . activateOnCurrentThread ( ) ; origin = database . getUser ( ) ; try { if ( iRequest . getUserRID ( ) != null && iRequest . getUserRID ( ) . isValid ( ) && ( lastUser == null || ! ( lastUser . getIdentity ( ) ) . equals ( iRequest . getUserRID ( ) ) ) ) { lastUser = database . getMetadata ( ) . getSecurity ( ) . getUser ( iRequest . getUserRID ( ) ) ; database . setUser ( lastUser ) ; // set to new user } else origin = null ; } catch ( Exception ex ) { OLogManager . instance ( ) . error ( this , "Failed on user switching database. " , ex ) ; } } } responsePayload = manager . executeOnLocalNode ( iRequest . getId ( ) , iRequest . getTask ( ) , database ) ; if ( responsePayload instanceof OModificationOperationProhibitedException ) { // RETRY try { ODistributedServerLog . info ( this , localNodeName , senderNodeName , DIRECTION . IN , "Database is frozen, waiting and retrying. Request %s (retry=%d, worker=%d)" , iRequest , retry , id ) ; Thread . sleep ( 1000 ) ; } catch ( InterruptedException e ) { } } else { // OPERATION EXECUTED (OK OR ERROR), NO RETRY NEEDED if ( retry > 1 ) ODistributedServerLog . info ( this , localNodeName , senderNodeName , DIRECTION . IN , "Request %s succeed after retry=%d" , iRequest , retry ) ; break ; } } } catch ( RuntimeException e ) { if ( task . hasResponse ( ) ) sendResponseBack ( iRequest , e ) ; throw e ; } finally { if ( database != null && ! database . isClosed ( ) ) { database . activateOnCurrentThread ( ) ; if ( ! database . isClosed ( ) ) { database . rollback ( ) ; database . getLocalCache ( ) . clear ( ) ; if ( origin != null ) database . setUser ( origin ) ; } } } if ( task . hasResponse ( ) ) { if ( ! sendResponseBack ( iRequest , responsePayload ) ) { handleError ( iRequest , responsePayload ) ; } } }
Executes the remote call on the local node and send back the result
Executes the remote call on the local node and send back the result