distributed/src/main/java/com/orientechnologies/orient/server/distributed/impl/ODistributedMessageServiceImpl.java
public void dispatchResponseToThread(final ODistributedResponse response) {
    try {
      final long msgId = response.getRequestId().getMessageId();

      // GET ASYNCHRONOUS MSG MANAGER IF ANY
      final ODistributedResponseManager asynchMgr = responsesByRequestIds.get(msgId);
      if (asynchMgr == null) {
        if (ODistributedServerLog.isDebugEnabled())
          ODistributedServerLog.debug(this, manager.getLocalNodeName(), response.getExecutorNodeName(), DIRECTION.IN,
              "received response for message %d after the timeout (%dms)", msgId,
              OGlobalConfiguration.DISTRIBUTED_ASYNCH_RESPONSES_TIMEOUT.getValueAsLong());
      } else if (asynchMgr.collectResponse(response)) {
        // ALL RESPONSE RECEIVED, REMOVE THE RESPONSE MANAGER WITHOUT WAITING THE PURGE THREAD REMOVE THEM FOR TIMEOUT
        responsesByRequestIds.remove(msgId);
      }
    } finally {
      Orient.instance().getProfiler()
          .updateCounter("distributed.node.msgReceived", "Number of replication messages received in current node", +1,
              "distributed.node.msgReceived");

      Orient.instance().getProfiler().updateCounter("distributed.node." + response.getExecutorNodeName() + ".msgReceived",
          "Number of replication messages received in current node from a node", +1, "distributed.node.*.msgReceived");
    }
  }
public void dispatchResponseToThread(final ODistributedResponse response) {
    try {
      final long msgId = response.getRequestId().getMessageId();

      // GET ASYNCHRONOUS MSG MANAGER IF ANY
      final ODistributedResponseManager asynchMgr = responsesByRequestIds.get(msgId);
      if (asynchMgr == null) {
        if (ODistributedServerLog.isDebugEnabled())
          ODistributedServerLog.debug(this, manager.getLocalNodeName(), response.getExecutorNodeName(), DIRECTION.IN,
              "received response for message %d after the timeout (%dms)", msgId,
              OGlobalConfiguration.DISTRIBUTED_ASYNCH_RESPONSES_TIMEOUT.getValueAsLong());
      } else if (asynchMgr.collectResponse(response)) {
        // ALL RESPONSE RECEIVED, REMOVE THE RESPONSE MANAGER WITHOUT WAITING THE PURGE THREAD REMOVE THEM FOR TIMEOUT
        responsesByRequestIds.remove(msgId);
      }
    } finally {
      Orient.instance().getProfiler()
          .updateCounter("distributed.node.msgReceived", "Number of replication messages received in current node", +1,
              "distributed.node.msgReceived");

      Orient.instance().getProfiler().updateCounter("distributed.node." + response.getExecutorNodeName() + ".msgReceived",
          "Number of replication messages received in current node from a node", +1, "distributed.node.*.msgReceived");
    }
  }
public void dispatchResponseToThread ( final ODistributedResponse response ) { try { final long msgId = response . getRequestId ( ) . getMessageId ( ) ; // GET ASYNCHRONOUS MSG MANAGER IF ANY final ODistributedResponseManager asynchMgr = responsesByRequestIds . get ( msgId ) ; if ( asynchMgr == null ) { if ( ODistributedServerLog . isDebugEnabled ( ) ) ODistributedServerLog . debug ( this , manager . getLocalNodeName ( ) , response . getExecutorNodeName ( ) , DIRECTION . IN , "received response for message %d after the timeout (%dms)" , msgId , OGlobalConfiguration . DISTRIBUTED_ASYNCH_RESPONSES_TIMEOUT . getValueAsLong ( ) ) ; } else if ( asynchMgr . collectResponse ( response ) ) { // ALL RESPONSE RECEIVED, REMOVE THE RESPONSE MANAGER WITHOUT WAITING THE PURGE THREAD REMOVE THEM FOR TIMEOUT responsesByRequestIds . remove ( msgId ) ; } } finally { Orient . instance ( ) . getProfiler ( ) . updateCounter ( "distributed.node.msgReceived" , "Number of replication messages received in current node" , + 1 , "distributed.node.msgReceived" ) ; Orient . instance ( ) . getProfiler ( ) . updateCounter ( "distributed.node." + response . getExecutorNodeName ( ) + ".msgReceived" , "Number of replication messages received in current node from a node" , + 1 , "distributed.node.*.msgReceived" ) ; } }
Not synchronized, it's called when a message arrives
Not synchronized it s called when a message arrives