private void processInputBuffer(AbstractExecutor threadMonitorExecutor)
      throws ConnectionException, IOException {
    try (final ByteBufferSharing inputSharing = inputBufferVendor.open()) {
      // can't be final because in some cases we expand the buffer (resulting in a new object)
      ByteBuffer inputBuffer = inputSharing.getBuffer();
      inputBuffer.flip();

      try (final ByteBufferSharing sharedBuffer = ioFilter.unwrap(inputBuffer)) {
        final ByteBuffer peerDataBuffer = sharedBuffer.getBuffer();

        peerDataBuffer.flip();

        boolean done = false;

        while (!done && connected) {
          owner.getConduit().getCancelCriterion().checkCancelInProgress(null);
          int remaining = peerDataBuffer.remaining();
          if (lengthSet || remaining >= MSG_HEADER_BYTES) {
            if (!lengthSet) {
              if (readMessageHeader(peerDataBuffer)) {
                break;
              }
            }
            if (remaining >= messageLength + MSG_HEADER_BYTES) {
              lengthSet = false;
              peerDataBuffer.position(peerDataBuffer.position() + MSG_HEADER_BYTES);
              // don't trust the message deserialization to leave the position in
              // the correct spot. Some of the serialization uses buffered
              // streams that can leave the position at the wrong spot
              int startPos = peerDataBuffer.position();
              int oldLimit = peerDataBuffer.limit();
              peerDataBuffer.limit(startPos + messageLength);

              if (handshakeRead) {
                try {
                  readMessage(peerDataBuffer, threadMonitorExecutor);
                } catch (SerializationException e) {
                  logger.info("input buffer startPos {} oldLimit {}", startPos, oldLimit);
                  throw e;
                }
              } else {
                try (ByteBufferInputStream bbis = new ByteBufferInputStream(peerDataBuffer);
                    DataInputStream dis = new DataInputStream(bbis)) {
                  if (!isReceiver) {
                    // we read the handshake and then stop processing since we don't want
                    // to process the input buffer anymore in a handshake thread
                    readHandshakeForSender(dis, peerDataBuffer);
                    return;
                  }
                  if (readHandshakeForReceiver(dis)) {
                    ioFilter.doneReading(peerDataBuffer);
                    return;
                  }
                }
              }
              if (!connected) {
                continue;
              }
              accessed();
              peerDataBuffer.limit(oldLimit);
              peerDataBuffer.position(startPos + messageLength);
            } else {
              done = true;
              if (getConduit().useSSL()) {
                ioFilter.doneReading(peerDataBuffer);
              } else {
                // compact or resize the buffer
                final int oldBufferSize = inputBuffer.capacity();
                final int allocSize = messageLength + MSG_HEADER_BYTES;
                if (oldBufferSize < allocSize) {
                  // need a bigger buffer
                  logger.info(
                      "Allocating larger network read buffer, new size is {} old size was {}.",
                      allocSize, oldBufferSize);
                  inputBuffer = inputSharing.expandReadBufferIfNeeded(allocSize);
                  makeReadableBufferWriteable(inputBuffer);
                } else {
                  if (inputBuffer.position() != 0) {
                    inputBuffer.compact();
                  } else {
                    makeReadableBufferWriteable(inputBuffer);
                  }
                }
              }
            }
          } else {
            ioFilter.doneReading(peerDataBuffer);
            done = true;
          }
        }
      }
    }
  }
processes the current NIO buffer. If there are complete messages in the buffer, they are
deserialized and passed to TCPConduit for further processing

pre-condition: inputBuffer (from inputSharing.getBuffer()) is in WRITABLE mode
post-condition: inputBuffer is in WRITABLE mode
geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java