private int processCqs(Message clientMessage, int startMessagePart, int numCqParts,
      MessageType messageType, Object key, Object value, byte[] delta, EventID eventId) {
    HashMap<String, MessageType> cqs = new HashMap<>();
    final boolean isDebugEnabled = logger.isDebugEnabled();

    for (int cqCnt = 0; cqCnt < numCqParts;) {
      StringBuilder sb = null;
      if (isDebugEnabled) {
        sb = new StringBuilder(100);
        sb.append("found these queries: ");
      }
      try {
        // Get CQ Name.
        Part cqNamePart = clientMessage.getPart(startMessagePart + cqCnt++);
        // Get CQ Op.
        Part cqOpPart = clientMessage.getPart(startMessagePart + cqCnt++);
        final String cqName = cqNamePart.getString();
        final MessageType cqOp = MessageType.valueOf(cqOpPart.getInt());
        cqs.put(cqName, cqOp);

        if (sb != null) {
          sb.append(cqName).append(" op=").append(cqOp).append("  ");
        }
      } catch (Exception ignore) {
        logger.warn(
            "Error while processing the CQ Message. Problem with reading message for CQ# : {}",
            cqCnt);
      }
      if (isDebugEnabled) {
        logger.debug(sb);
      }
    }

    CqService cqService = cache.getCqService();
    try {
      cqService.dispatchCqListeners(cqs, messageType, key, value, delta, qManager, eventId);
    } catch (Exception ex) {
      logger.warn("Failed to invoke CQ Dispatcher. Error :  {}",
          ex.getMessage());
      if (isDebugEnabled) {
        logger.debug("Failed to invoke CQ Dispatcher.", ex);
      }
    }

    return startMessagePart + numCqParts;
  }
Processes message to invoke CQ listeners.
geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java