private void readHeaderAndBody(boolean setHeaderReadTimeout, int headerReadTimeoutMillis)
      throws IOException {
    clearParts();
    // TODO: for server changes make sure sc is not null as this class also used by client

    int oldTimeout = -1;
    if (setHeaderReadTimeout) {
      oldTimeout = socket.getSoTimeout();
      socket.setSoTimeout(headerReadTimeoutMillis);
    }
    try {
      fetchHeader();
    } finally {
      if (setHeaderReadTimeout) {
        socket.setSoTimeout(oldTimeout);
      }
    }

    final ByteBuffer cb = getCommBuffer();
    final int type = cb.getInt();
    final int len = cb.getInt();
    final int numParts = cb.getInt();
    final int txid = cb.getInt();
    byte bits = cb.get();
    cb.clear();

    if (!MessageType.validate(type)) {
      throw new IOException(String.format("Invalid message type %s while reading header",
          type));
    }

    int timeToWait = 0;
    if (serverConnection != null) {
      // Keep track of the fact that a message is being processed.
      serverConnection.setProcessingMessage();
      timeToWait = serverConnection.getClientReadTimeout();
    }
    readHeader = true;

    if (messageLimiter != null) {
      for (;;) {
        serverConnection.getCachedRegionHelper().checkCancelInProgress(null);
        boolean interrupted = Thread.interrupted();
        try {
          if (timeToWait == 0) {
            messageLimiter.acquire(1);
          } else {
            if (!messageLimiter.tryAcquire(1, timeToWait, TimeUnit.MILLISECONDS)) {
              if (messageStats instanceof CacheServerStats) {
                ((CacheServerStats) messageStats).incConnectionsTimedOut();
              }
              throw new IOException(
                  String.format(
                      "Operation timed out on server waiting on concurrent message limiter after waiting %s milliseconds",
                      timeToWait));
            }
          }
          break;
        } catch (InterruptedException ignore) {
          interrupted = true;
        } finally {
          if (interrupted) {
            Thread.currentThread().interrupt();
          }
        }
      } // for
    }

    if (len > 0) {
      if (maxIncomingMessageLength > 0 && len > maxIncomingMessageLength) {
        throw new IOException(String.format("Message size %s exceeded max limit of %s",
            len, maxIncomingMessageLength));
      }

      if (dataLimiter != null) {
        for (;;) {
          if (serverConnection != null) {
            serverConnection.getCachedRegionHelper().checkCancelInProgress(null);
          }
          boolean interrupted = Thread.interrupted();
          try {
            if (timeToWait == 0) {
              dataLimiter.acquire(len);
            } else {
              int newTimeToWait = timeToWait;
              if (messageLimiter != null) {
                // may have waited for msg limit so recalc time to wait
                newTimeToWait -= (int) serverConnection.getCurrentMessageProcessingTime();
              }
              if (newTimeToWait <= 0
                  || !messageLimiter.tryAcquire(1, newTimeToWait, TimeUnit.MILLISECONDS)) {
                throw new IOException(
                    String.format(
                        "Operation timed out on server waiting on concurrent data limiter after waiting %s milliseconds",
                        timeToWait));
              }
            }
            // makes sure payloadLength gets set now so we will release the semaphore
            payloadLength = len;
            break; // success
          } catch (InterruptedException ignore) {
            interrupted = true;
          } finally {
            if (interrupted) {
              Thread.currentThread().interrupt();
            }
          }
        }
      }
    }
    if (messageStats != null) {
      messageStats.incMessagesBeingReceived(len);
      payloadLength = len; // makes sure payloadLength gets set now so we will dec on clear
    }

    isRetry = (bits & MESSAGE_IS_RETRY) != 0;
    bits &= MESSAGE_IS_RETRY_MASK;
    flags = bits;
    messageType = MessageType.valueOf(type);

    readPayloadFields(numParts, len);

    // Set the header and payload fields only after receiving all the
    // socket data, providing better message consistency in the face
    // of exceptional conditions (e.g. IO problems, timeouts etc.)
    payloadLength = len;
    // this.numberOfParts = numParts; Already set in setPayloadFields via setNumberOfParts
    transactionId = txid;
    flags = bits;
    if (serverConnection != null) {
      // Keep track of the fact that a message is being processed.
      serverConnection.updateProcessingMessage();
    }
  }
Sends this message out on its socket.
geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/Message.java