public boolean collectResponse(final ODistributedResponse response) {
    final String executorNode = response.getExecutorNodeName();
    final String senderNode = response.getSenderNodeName();
    response.setDistributedResponseManager(this);

    synchronousResponsesLock.lock();
    try {
      if (!executorNode.equals(dManager.getLocalNodeName()) && !responses.containsKey(executorNode)) {
        ODistributedServerLog.warn(this, senderNode, executorNode, DIRECTION.IN,
            "Received response for request (%s) from unexpected node. Expected are: %s", request, getExpectedNodes());

        Orient.instance().getProfiler()
            .updateCounter("distributed.node.unexpectedNodeResponse", "Number of responses from unexpected nodes", +1);

        return false;
      }

      dManager.getMessageService().updateLatency(executorNode, sentOn);

      responses.put(executorNode, response);
      receivedResponses++;

      if (waitForLocalNode && executorNode.equals(senderNode))
        receivedCurrentNode = true;

      if (ODistributedServerLog.isDebugEnabled())
        ODistributedServerLog.debug(this, senderNode, executorNode, DIRECTION.IN,
            "Received response '%s' for request (%s) (receivedCurrentNode=%s receivedResponses=%d totalExpectedResponses=%d quorum=%d)",
            response, request, receivedCurrentNode, receivedResponses, totalExpectedResponses, quorum);

      if (groupResponsesByResult) {
        // PUT THE RESPONSE IN THE RIGHT RESPONSE GROUP
        // TODO: AVOID TO KEEP ALL THE RESULT FOR THE SAME RESP GROUP, BUT RATHER THE FIRST ONE + COUNTER
        final Object responsePayload = response.getPayload();

        boolean foundBucket = false;
        for (int i = 0; i < responseGroups.size(); ++i) {
          final List<ODistributedResponse> responseGroup = responseGroups.get(i);

          if (responseGroup.isEmpty())
            // ABSENT
            foundBucket = true;
          else {
            final Object rgPayload = responseGroup.get(0).getPayload();

            if (rgPayload == null && responsePayload == null)
              // BOTH NULL
              foundBucket = true;
            else if (rgPayload != null) {
              if (rgPayload instanceof ODocument && responsePayload instanceof ODocument && !((ODocument) rgPayload).getIdentity()
                  .isValid() && ((ODocument) rgPayload).hasSameContentOf((ODocument) responsePayload))
                // SAME RESULT
                foundBucket = true;
              else if (rgPayload.equals(responsePayload))
                // SAME RESULT
                foundBucket = true;
              else if (rgPayload instanceof Collection && responsePayload instanceof Collection) {
                if (OMultiValue.equals((Collection) rgPayload, (Collection) responsePayload))
                  // COLLECTIONS WITH THE SAME VALUES
                  foundBucket = true;
              }
            }
          }

          if (foundBucket) {
            responseGroup.add(response);
            break;
          }
        }

        if (!foundBucket) {
          // CREATE A NEW BUCKET
          final ArrayList<ODistributedResponse> newBucket = new ArrayList<ODistributedResponse>();
          responseGroups.add(newBucket);
          newBucket.add(response);
        }
      }

      // FOR EVERY RESPONSE COLLECTED, COMPUTE THE FINAL QUORUM RESPONSE IF POSSIBLE
      computeQuorumResponse(false);

      return checkForCompletion();

    } finally {
      synchronousResponsesLock.unlock();
    }
  }
public boolean collectResponse ( final ODistributedResponse response ) { final String executorNode = response . getExecutorNodeName ( ) ; final String senderNode = response . getSenderNodeName ( ) ; response . setDistributedResponseManager ( this ) ; synchronousResponsesLock . lock ( ) ; try { if ( ! executorNode . equals ( dManager . getLocalNodeName ( ) ) && ! responses . containsKey ( executorNode ) ) { ODistributedServerLog . warn ( this , senderNode , executorNode , DIRECTION . IN , "Received response for request (%s) from unexpected node. Expected are: %s" , request , getExpectedNodes ( ) ) ; Orient . instance ( ) . getProfiler ( ) . updateCounter ( "distributed.node.unexpectedNodeResponse" , "Number of responses from unexpected nodes" , + 1 ) ; return false ; } dManager . getMessageService ( ) . updateLatency ( executorNode , sentOn ) ; responses . put ( executorNode , response ) ; receivedResponses ++ ; if ( waitForLocalNode && executorNode . equals ( senderNode ) ) receivedCurrentNode = true ; if ( ODistributedServerLog . isDebugEnabled ( ) ) ODistributedServerLog . debug ( this , senderNode , executorNode , DIRECTION . IN , "Received response '%s' for request (%s) (receivedCurrentNode=%s receivedResponses=%d totalExpectedResponses=%d quorum=%d)" , response , request , receivedCurrentNode , receivedResponses , totalExpectedResponses , quorum ) ; if ( groupResponsesByResult ) { final Object responsePayload = response . getPayload ( ) ; boolean foundBucket = false ; for ( int i = 0 ; i < responseGroups . size ( ) ; ++ i ) { final List < ODistributedResponse > responseGroup = responseGroups . get ( i ) ; if ( responseGroup . isEmpty ( ) ) foundBucket = true ; else { final Object rgPayload = responseGroup . get ( 0 ) . getPayload ( ) ; if ( rgPayload == null && responsePayload == null ) foundBucket = true ; else if ( rgPayload != null ) { if ( rgPayload instanceof ODocument && responsePayload instanceof ODocument && ! ( ( ODocument ) rgPayload ) . getIdentity ( ) . isValid ( ) && ( ( ODocument ) rgPayload ) . hasSameContentOf ( ( ODocument ) responsePayload ) ) foundBucket = true ; else if ( rgPayload . equals ( responsePayload ) ) foundBucket = true ; else if ( rgPayload instanceof Collection && responsePayload instanceof Collection ) { if ( OMultiValue . equals ( ( Collection ) rgPayload , ( Collection ) responsePayload ) ) foundBucket = true ; } } } if ( foundBucket ) { responseGroup . add ( response ) ; break ; } } if ( ! foundBucket ) { final ArrayList < ODistributedResponse > newBucket = new ArrayList < ODistributedResponse > ( ) ; responseGroups . add ( newBucket ) ; newBucket . add ( response ) ; } } computeQuorumResponse ( false ) ; return checkForCompletion ( ) ; } finally { synchronousResponsesLock . unlock ( ) ; } }
Not synchronized, it's called when a message arrives

@param response Received response to collect

@return True if all the nodes responded, otherwise false