@VisibleForTesting
  protected void runDispatcher() {
    boolean exceptionOccurred = false;
    _isStopped = false;

    if (logger.isDebugEnabled()) {
      logger.debug("{}: Beginning to process events", this);
    }

    ClientMessage clientMessage = null;

    while (!isStopped()) {
      // SystemFailure.checkFailure(); DM's stopper does this
      if (getCache().getCancelCriterion().isCancelInProgress()) {
        break;
      }
      try {
        // If paused, wait to be told to resume (or interrupted if stopped)
        if (getProxy().isPaused()) {
          // ARB: Before waiting for resumption, process acks from client.
          // This will reduce the number of duplicates that a client receives after
          // reconnecting.
          synchronized (_pausedLock) {
            try {
              logger.info("available ids = " + _messageQueue.size() + " , isEmptyAckList ="
                  + _messageQueue.isEmptyAckList() + ", peekInitialized = "
                  + _messageQueue.isPeekInitialized());
              while (!_messageQueue.isEmptyAckList()
                  && _messageQueue.isPeekInitialized()) {
                _messageQueue.remove();
              }
            } catch (InterruptedException ex) {
              logger.warn("{}: sleep interrupted.", this);
            }
          }
          waitForResumption();
        }

        // if message is not delivered due to authentication expiation, continue to try to
        // deliver the same message. Always retrieve a new message from the queue if we are not
        // waiting for the re-auth to happen.
        if (waitForReAuthenticationStartTime == -1) {
          try {
            clientMessage = (ClientMessage) _messageQueue.peek();
          } catch (RegionDestroyedException skipped) {
            break;
          }
        }

        getStatistics().setQueueSize(_messageQueue.size());
        if (isStopped()) {
          break;
        }

        if (clientMessage == null) {
          _messageQueue.remove();
          continue;
        }

        // Process the message
        long start = getStatistics().startTime();
        try {
          if (dispatchMessage(clientMessage)) {
            getStatistics().endMessage(start);
            _messageQueue.remove();
            if (clientMessage instanceof ClientMarkerMessageImpl) {
              getProxy().setMarkerEnqueued(false);
            }
          }
          clientMessage = null;
          waitForReAuthenticationStartTime = -1;
        } catch (NotAuthorizedException notAuthorized) {
          // behave as if the message is dispatched, remove from the queue
          logger.warn("skip delivering message: " + clientMessage, notAuthorized);
          _messageQueue.remove();
          clientMessage = null;
        } catch (AuthenticationExpiredException expired) {
          exceptionOccurred = handleAuthenticationExpiredException(expired);
        }
      } catch (MessageTooLargeException e) {
        logger.warn("Message too large to send to client: {}, {}", clientMessage, e.getMessage());
      } catch (IOException e) {
        // Added the synchronization below to ensure that exception handling
        // does not occur while stopping the dispatcher and vice versa.
        synchronized (_stopDispatchingLock) {
          // An IOException occurred while sending a message to the
          // client. If the processor is not already stopped, assume
          // the client is dead and stop processing.
          if (!isStopped() && !getProxy().isPaused()) {
            if ("Broken pipe".equals(e.getMessage())) {
              logger.warn("{}: Proxy closing due to unexpected broken pipe on socket connection.",
                  this);
            } else if ("Connection reset".equals(e.getMessage())) {
              logger.warn("{}: Proxy closing due to unexpected reset on socket connection.",
                  this);
            } else if ("Connection reset by peer".equals(e.getMessage())) {
              logger.warn(
                  "{}: Proxy closing due to unexpected reset by peer on socket connection.",
                  this);
            } else if ("Socket is closed".equals(e.getMessage())
                || "Socket Closed".equals(e.getMessage())) {
              logger.info("{}: Proxy closing due to socket being closed locally.",
                  this);
            } else {
              logger.warn(String.format(
                  "%s: An unexpected IOException occurred so the proxy will be closed.",
                  this),
                  e);
            }
            // Let the CacheClientNotifier discover the proxy is not alive.
            // See isAlive().
            // getProxy().close(false);
            pauseOrUnregisterProxy(e);
          } // _isStopped
        } // synchronized
        exceptionOccurred = true;
      } // IOException
      catch (InterruptedException e) {
        // If the thread is paused, ignore the InterruptedException and
        // continue. The proxy is null if stopDispatching has been called.
        if (getProxy().isPaused()) {
          if (logger.isDebugEnabled()) {
            logger.debug(
                "{}: interrupted because it is being paused. It will continue and wait for resumption.",
                this);
          }
          Thread.interrupted();
          continue;
        }

        // no need to reset the bit; we're exiting
        if (logger.isDebugEnabled()) {
          logger.debug("{}: interrupted", this);
        }
        break;
      } catch (CancelException e) {
        if (logger.isDebugEnabled()) {
          logger.debug("{}: shutting down due to cancellation", this);
        }
        exceptionOccurred = true; // message queue is defunct, don't try to read it.
        break;
      } catch (RegionDestroyedException e) {
        if (logger.isDebugEnabled()) {
          logger.debug("{}: shutting down due to loss of message queue", this);
        }
        exceptionOccurred = true; // message queue is defunct, don't try to read it.
        break;
      } catch (Exception e) {
        // An exception occurred while processing a message. Since it
        // is not an IOException, the client may still be alive, so
        // continue processing.
        if (!isStopped()) {
          logger.fatal(String.format("%s : An unexpected Exception occurred", this),
              e);
        }
      }
    }

    // Processing gets here if isStopped=true. What is this code below doing?
    if (!exceptionOccurred) {
      dispatchResidualMessages();
    }
    if (logger.isTraceEnabled()) {
      logger.trace("{}: Dispatcher thread is ending", this);
    }
  }
Runs the dispatcher by taking a message from the queue and sending it to the client attached
to this proxy.
geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/MessageDispatcher.java