private void processMessages() {
    final boolean isDebugEnabled = logger.isDebugEnabled();

    final int headerReadTimeout = (int) Math.round(serverQueueStatus.getPingInterval()
        * qManager.getPool().getSubscriptionTimeoutMultiplier() * 1.25);

    try {
      Message clientMessage = initializeMessage();

      if (quitting()) {
        if (isDebugEnabled) {
          logger.debug("processMessages quitting early because we have stopped");
        }
        // our caller calls close which will notify all waiters for our init
        return;
      }

      logger.info("{} : ready to process messages.", this);

      while (continueProcessing.get()) {
        if (quitting()) {
          if (isDebugEnabled) {
            logger.debug("termination detected");
          }
          // our caller calls close which will notify all waiters for our init
          return;
        }

        // the endpoint died while this thread was sleeping.
        if (endpoint.isClosed()) {
          if (isDebugEnabled) {
            logger.debug("endpoint died");
          }
          continueProcessing.set(false);// = false;
          break;
        }

        try {
          // Read the message
          clientMessage.receiveWithHeaderReadTimeout(headerReadTimeout);

          // Wait for the previously failed cache client updater
          // to finish. This will avoid out of order messages.
          waitForFailedUpdater();
          cache.waitForRegisterInterestsInProgress();
          if (quitting()) {
            if (isDebugEnabled) {
              logger.debug("processMessages quitting before processing message");
            }
            break;
          }

          // If the message is a ping, ignore it
          if (clientMessage.getMessageType() == SERVER_TO_CLIENT_PING) {
            if (isDebugEnabled) {
              logger.debug("{}: Received ping", this);
            }
            continue;
          }

          boolean isDeltaSent = false;
          boolean isCreateOrUpdate = clientMessage.getMessageType() == LOCAL_CREATE
              || clientMessage.getMessageType() == LOCAL_UPDATE;
          if (isCreateOrUpdate) {
            isDeltaSent = (Boolean) clientMessage.getPart(2).getObject();
          }

          // extract the eventId and verify if it is a duplicate event
          // if it is a duplicate event, ignore
          // @since GemFire 5.1
          int numberOfParts = clientMessage.getNumberOfParts();
          Part eid = clientMessage.getPart(numberOfParts - 1);

          // TODO the message handling methods also deserialized the eventID - inefficient
          EventID eventId = (EventID) eid.getObject();

          // no need to verify if the instantiator msg is duplicate or not
          if (clientMessage.getMessageType() != REGISTER_INSTANTIATORS
              && clientMessage.getMessageType() != REGISTER_DATASERIALIZERS) {
            if (qManager.getState().verifyIfDuplicate(eventId,
                !(isDurableClient || isDeltaSent))) {
              continue;
            }
          }

          if (logger.isTraceEnabled(LogMarker.BRIDGE_SERVER_VERBOSE)) {
            logger.trace(LogMarker.BRIDGE_SERVER_VERBOSE, "Processing event with id {}",
                eventId.expensiveToString());
          }

          isOpCompleted = true;

          // Process the message
          switch (clientMessage.getMessageType()) {
            case LOCAL_CREATE:
            case LOCAL_UPDATE:
              handleUpdate(clientMessage);
              break;
            case LOCAL_INVALIDATE:
              handleInvalidate(clientMessage);
              break;
            case LOCAL_DESTROY:
              handleDestroy(clientMessage);
              break;
            case CLIENT_RE_AUTHENTICATE:
              handleAuthenticate();
              break;
            case LOCAL_DESTROY_REGION:
              handleDestroyRegion(clientMessage);
              break;
            case CLEAR_REGION:
              handleClearRegion(clientMessage);
              break;
            case REGISTER_INSTANTIATORS:
              handleRegisterInstantiator(clientMessage, eventId);
              break;
            case REGISTER_DATASERIALIZERS:
              handleRegisterDataSerializer(clientMessage, eventId);
              break;
            case CLIENT_MARKER:
              handleMarker(clientMessage);
              break;
            case INVALIDATE_REGION:
              handleInvalidateRegion(clientMessage);
              break;
            case CLIENT_REGISTER_INTEREST:
              handleRegisterInterest(clientMessage);
              break;
            case CLIENT_UNREGISTER_INTEREST:
              handleUnregisterInterest(clientMessage);
              break;
            case TOMBSTONE_OPERATION:
              handleTombstoneOperation(clientMessage);
              break;
            default:
              logger.warn("{}: Received an unsupported message (type={})",
                  new Object[] {this, clientMessage.getMessageType()});
              break;
          }

          if (isOpCompleted && (isDurableClient || isDeltaSent)) {
            qManager.getState().verifyIfDuplicate(eventId, true);
          }

          // TODO we should maintain the client's "live" view of the server
          // but we don't because the server health monitor needs traffic
          // originating from the client
          // and by updating the last update stat, the ServerMonitor is less
          // likely to send pings...
          // and the ClientHealthMonitor will cause a disconnect

        } catch (InterruptedIOException ignore) {
          // Per Sun's support web site, this exception seems to be peculiar
          // to Solaris, and may eventually not even be generated there.
          //
          // When this exception is thrown, the thread has been interrupted, but
          // isInterrupted() is false. (How very odd!)
          //
          // We regard it the same as an InterruptedException

          continueProcessing.set(false);
          if (isDebugEnabled) {
            logger.debug("InterruptedIOException");
          }

        } catch (IOException e) {
          // Either the server went away, or we caught a closing condition.
          if (!quitting()) {
            // Server departed; print a message.
            ClientServerObserver clientServerObserver = ClientServerObserverHolder.getInstance();
            clientServerObserver.beforeFailoverByCacheClientUpdater(location);
            eManager.serverCrashed(endpoint);
            if (isDebugEnabled) {
              logger.debug("Caught the following exception and will exit", e);
            }
          } // !quitting

          // In any event, terminate this thread.
          continueProcessing.set(false);
          if (isDebugEnabled) {
            logger.debug("terminated due to IOException");
          }

        } catch (Exception e) {
          if (!quitting()) {
            ClientServerObserver clientServerObserver = ClientServerObserverHolder.getInstance();
            clientServerObserver.beforeFailoverByCacheClientUpdater(location);
            eManager.serverCrashed(endpoint);
            String message = ": Caught the following exception and will exit: ";
            handleException(message, e);
          }

          // In any event, terminate this thread.
          continueProcessing.set(false);// = false; // force termination
          if (isDebugEnabled) {
            logger.debug("CCU terminated due to Exception");
          }

        } finally {
          clientMessage.clear();
        }
      } // while

    } finally {
      if (isDebugEnabled) {
        logger.debug("has stopped and cleaning the helper ..");
      }
      close(); // added to fix some race conditions associated with 38382
      // this will make sure that if this thread dies without starting QueueMgr then it will start..
      // 1. above we ignore InterruptedIOException and this thread dies without informing QueueMgr
      // 2. if there is some other race condition with continueProcessing flag
      qManager.checkEndpoint(this, endpoint);
    }
  }
Processes messages received from the server.

Only certain types of messages are handled.

TODO: Method 'processMessages' is too complex to analyze by data flow algorithm

@see MessageType#CLIENT_MARKER
@see MessageType#LOCAL_CREATE
@see MessageType#LOCAL_UPDATE
@see MessageType#LOCAL_INVALIDATE
@see MessageType#LOCAL_DESTROY
@see MessageType#LOCAL_DESTROY_REGION
@see MessageType#CLEAR_REGION
@see ClientUpdateMessage
geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java