/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.cache.tier.sockets;

import static java.lang.Math.min;
import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;

import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Semaphore;
import java.util.regex.Pattern;

import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import org.apache.geode.CopyException;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.SerializationException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.InterestResultPolicy;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.TransactionException;
import org.apache.geode.cache.persistence.PartitionOfflineException;
import org.apache.geode.cache.query.types.CollectionType;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.CachedDeserializable;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.EntrySnapshot;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.FindVersionTagOperation;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalRegion;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.NonTXEntry;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.PartitionedRegionHelper;
import org.apache.geode.internal.cache.TXManagerImpl;
import org.apache.geode.internal.cache.TXStateProxy;
import org.apache.geode.internal.cache.Token;
import org.apache.geode.internal.cache.VersionTagHolder;
import org.apache.geode.internal.cache.execute.ServerToClientFunctionResultSender;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.InterestType;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.versions.VersionStamp;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.offheap.OffHeapHelper;
import org.apache.geode.internal.security.SecurityService;
import org.apache.geode.internal.sequencelog.EntryLogger;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.security.GemFireSecurityException;
import org.apache.geode.util.internal.GeodeGlossary;

public abstract class BaseCommand implements Command {
  protected static final Logger logger = LogService.getLogger();

  @Immutable
  private static final byte[] OK_BYTES = new byte[] {0};

  public static final int MAXIMUM_CHUNK_SIZE =
      Integer.getInteger("BridgeServer.MAXIMUM_CHUNK_SIZE", 100);

  /** Whether to suppress logging of IOExceptions */
  private static final boolean SUPPRESS_IO_EXCEPTION_LOGGING =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "bridge.suppressIOExceptionLogging");

  /**
   * Maximum number of concurrent incoming client message bytes that a cache server will allow.
   * Once a server is working on this number additional incoming client messages will wait until one
   * of them completes or fails. The bytes are computed based in the size sent in the incoming msg
   * header.
   */
  private static final int MAX_INCOMING_DATA =
      Integer.getInteger("BridgeServer.MAX_INCOMING_DATA", -1);

  /**
   * Maximum number of concurrent incoming client messages that a cache server will allow. Once a
   * server is working on this number additional incoming client messages will wait until one of
   * them completes or fails.
   */
  private static final int MAX_INCOMING_MESSAGES =
      Integer.getInteger("BridgeServer.MAX_INCOMING_MSGS", -1);

  @MakeNotStatic
  private static final Semaphore INCOMING_DATA_LIMITER;

  @MakeNotStatic
  private static final Semaphore INCOMING_MSG_LIMITER;

  static {
    Semaphore semaphore;
    if (MAX_INCOMING_DATA > 0) {
      // backport requires that this is fair since we inc by values > 1
      semaphore = new Semaphore(MAX_INCOMING_DATA, true);
    } else {
      semaphore = null;
    }
    INCOMING_DATA_LIMITER = semaphore;
    if (MAX_INCOMING_MESSAGES > 0) {
      // unfair for best performance
      semaphore = new Semaphore(MAX_INCOMING_MESSAGES, false);
    } else {
      semaphore = null;
    }
    INCOMING_MSG_LIMITER = semaphore;
  }

  protected static byte[] okBytes() {
    return OK_BYTES;
  }

  protected boolean setLastResultReceived(
      final @Nullable ServerToClientFunctionResultSender resultSender) {

    if (resultSender != null) {
      synchronized (resultSender) {
        if (resultSender.isLastResultReceived()) {
          return false;
        } else {
          resultSender.setLastResultReceived(true);
        }
      }
    }
    return true;
  }

  @Override
  public void execute(final @NotNull Message clientMessage,
      final @NotNull ServerConnection serverConnection,
      final @NotNull SecurityService securityService) {
    // Read the request and update the statistics
    long start = DistributionStats.getStatTime();
    if (EntryLogger.isEnabled()) {
      EntryLogger.setSource(serverConnection.getMembershipID(), "c2s");
    }
    boolean shouldMasquerade = shouldMasqueradeForTx(clientMessage);
    try {
      if (shouldMasquerade) {
        InternalCache cache = serverConnection.getCache();
        InternalDistributedMember member =
            (InternalDistributedMember) serverConnection.getProxyID().getDistributedMember();
        TXManagerImpl txMgr = cache.getTxManager();
        TXStateProxy tx = null;
        try {
          tx = txMgr.masqueradeAs(clientMessage, member, false);
          cmdExecute(clientMessage, serverConnection, securityService, start);
          tx.updateProxyServer(txMgr.getMemberId());
        } finally {
          txMgr.unmasquerade(tx);
        }
      } else {
        cmdExecute(clientMessage, serverConnection, securityService, start);
      }

    } catch (TransactionException | CopyException | SerializationException | CacheWriterException
        | CacheLoaderException | GemFireSecurityException | PartitionOfflineException
        | MessageTooLargeException e) {
      handleExceptionNoDisconnect(clientMessage, serverConnection, e);
    } catch (EOFException eof) {
      BaseCommand.handleEOFException(clientMessage, serverConnection, eof);
    } catch (InterruptedIOException e) { // Solaris only
      BaseCommand.handleInterruptedIOException(serverConnection, e);
    } catch (IOException e) {
      BaseCommand.handleIOException(clientMessage, serverConnection, e);
    } catch (DistributedSystemDisconnectedException e) {
      BaseCommand.handleShutdownException(clientMessage, serverConnection, e);
    } catch (VirtualMachineError err) {
      SystemFailure.initiateFailure(err);
      // If this ever returns, rethrow the error. We're poisoned
      // now, so don't let this thread continue.
      throw err;
    } catch (Throwable e) {
      handleThrowable(clientMessage, serverConnection, e);
    } finally {
      EntryLogger.clearSource();
    }
  }

  /**
   * checks to see if this thread needs to masquerade as a transactional thread.
   *
   * @return true if thread should masquerade as a transactional thread.
   */
  protected boolean shouldMasqueradeForTx(Message clientMessage) {
    return clientMessage.getTransactionId() > TXManagerImpl.NOTX;
  }

  /**
   * If an operation is retried then some server may have seen it already. We cannot apply this
   * operation to the cache without knowing whether a version tag has already been created for it.
   * Otherwise caches that have seen the event already will reject it but others will not, but will
   * have no version tag with which to perform concurrency checks.
   * <p>
   * The client event should have the event identifier from the client and the region affected by
   * the operation.
   */
  public boolean recoverVersionTagForRetriedOperation(EntryEventImpl clientEvent) {
    InternalRegion r = clientEvent.getRegion();
    VersionTag<?> tag = r.findVersionTagForEvent(clientEvent.getEventId());
    if (tag == null) {
      if (r instanceof DistributedRegion || r instanceof PartitionedRegion) {
        // TODO this could be optimized for partitioned regions by sending the key
        // so that the PR could look at an individual bucket for the event
        tag = FindVersionTagOperation.findVersionTag(r, clientEvent.getEventId(), false);
      }
    }
    if (tag != null) {
      if (logger.isDebugEnabled()) {
        logger.debug("recovered version tag {} for replayed operation {}", tag,
            clientEvent.getEventId());
      }
      clientEvent.setVersionTag(tag);
    }
    return tag != null;
  }

  /**
   * If an operation is retried then some server may have seen it already. We cannot apply this
   * operation to the cache without knowing whether a version tag has already been created for it.
   * Otherwise caches that have seen the event already will reject it but others will not, but will
   * have no version tag with which to perform concurrency checks.
   * <p>
   * The client event should have the event identifier from the client and the region affected by
   * the operation.
   */
  protected VersionTag<?> findVersionTagsForRetriedBulkOp(LocalRegion region, EventID eventID) {
    VersionTag<?> tag = region.findVersionTagForClientBulkOp(eventID);
    if (tag != null) {
      if (logger.isDebugEnabled()) {
        logger.debug("recovered version tag {} for replayed bulk operation {}", tag, eventID);
      }
      return tag;
    }
    if (region instanceof DistributedRegion || region instanceof PartitionedRegion) {
      // TODO this could be optimized for partitioned regions by sending the key
      // so that the PR could look at an individual bucket for the event
      tag = FindVersionTagOperation.findVersionTag(region, eventID, true);
    }
    if (tag != null) {
      if (logger.isDebugEnabled()) {
        logger.debug("recovered version tag {} for replayed bulk operation {}", tag, eventID);
      }
    }
    return tag;
  }

  public abstract void cmdExecute(final @NotNull Message clientMessage,
      final @NotNull ServerConnection serverConnection,
      final @NotNull SecurityService securityService,
      final long start) throws IOException, ClassNotFoundException, InterruptedException;

  protected void writeReply(final @NotNull Message origMsg,
      final @NotNull ServerConnection serverConnection) throws IOException {
    Message replyMsg = serverConnection.getReplyMessage();
    serverConnection.getCache().getCancelCriterion().checkCancelInProgress(null);
    replyMsg.setMessageType(MessageType.REPLY);
    replyMsg.setNumberOfParts(1);
    replyMsg.setTransactionId(origMsg.getTransactionId());
    replyMsg.addBytesPart(okBytes());
    replyMsg.send(serverConnection);
    if (logger.isTraceEnabled()) {
      logger.trace("{}: rpl tx: {}", serverConnection.getName(), origMsg.getTransactionId());
    }
  }

  private static void handleEOFException(final @Nullable Message msg,
      final @NotNull ServerConnection serverConnection,
      Exception eof) {
    CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
    CacheServerStats stats = serverConnection.getCacheServerStats();
    boolean potentialModification = serverConnection.getPotentialModification();
    if (!crHelper.isShutdown()) {
      if (potentialModification) {
        stats.incAbandonedWriteRequests();
      } else {
        stats.incAbandonedReadRequests();
      }
      if (!SUPPRESS_IO_EXCEPTION_LOGGING) {
        if (potentialModification) {
          int transId = msg != null ? msg.getTransactionId() : Integer.MIN_VALUE;
          logger.warn(
              "{}: EOFException during a write operation on region : {} key: {} messageId: {}",
              new Object[] {serverConnection.getName(), serverConnection.getModRegion(),
                  serverConnection.getModKey(), transId});
        } else {
          logger.debug("EOF exception", eof);
          logger.warn("{}: connection disconnect detected by EOF.",
              serverConnection.getName());
        }
      }
    }
    serverConnection.setFlagProcessMessagesAsFalse();
    serverConnection.setClientDisconnectedException(eof);
  }

  private static void handleInterruptedIOException(final @NotNull ServerConnection serverConnection,
      final @NotNull Exception e) {
    CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
    if (!crHelper.isShutdown() && serverConnection.isOpen()) {
      if (!SUPPRESS_IO_EXCEPTION_LOGGING) {
        if (logger.isDebugEnabled()) {
          logger.debug("Aborted message due to interrupt: {}", e.getMessage(), e);
        }
      }
    }
    serverConnection.setFlagProcessMessagesAsFalse();
    serverConnection.setClientDisconnectedException(e);
  }

  private static void handleIOException(final @Nullable Message msg,
      final @NotNull ServerConnection serverConnection,
      Exception e) {
    CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
    boolean potentialModification = serverConnection.getPotentialModification();

    if (!crHelper.isShutdown() && serverConnection.isOpen()) {
      if (!SUPPRESS_IO_EXCEPTION_LOGGING) {
        if (potentialModification) {
          int transId = msg != null ? msg.getTransactionId() : Integer.MIN_VALUE;
          logger.warn(String.format(
              "%s: Unexpected IOException during operation for region: %s key: %s messId: %s",
              serverConnection.getName(), serverConnection.getModRegion(),
              serverConnection.getModKey(), transId),
              e);
        } else {
          logger.warn(String.format("%s: Unexpected IOException: ",
              serverConnection.getName()), e);
        }
      }
    }
    serverConnection.setFlagProcessMessagesAsFalse();
    serverConnection.setClientDisconnectedException(e);
  }

  private static void handleShutdownException(final @Nullable Message msg,
      final @NotNull ServerConnection serverConnection,
      final @NotNull Exception e) {
    CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
    boolean potentialModification = serverConnection.getPotentialModification();

    if (!crHelper.isShutdown()) {
      if (potentialModification) {
        int transId = msg != null ? msg.getTransactionId() : Integer.MIN_VALUE;
        logger.warn(String.format(
            "%s: Unexpected ShutdownException during operation on region: %s key: %s messageId: %s",
            serverConnection.getName(), serverConnection.getModRegion(),
            serverConnection.getModKey(), transId),
            e);
      } else {
        logger.warn(String.format("%s: Unexpected ShutdownException: ",
            serverConnection.getName()),
            e);
      }
    }
    serverConnection.setFlagProcessMessagesAsFalse();
    serverConnection.setClientDisconnectedException(e);
  }

  private static void handleExceptionNoDisconnect(final @NotNull Message msg,
      final @NotNull ServerConnection serverConnection,
      final @NotNull Exception e) {
    boolean requiresResponse = serverConnection.getTransientFlag(REQUIRES_RESPONSE);
    boolean responded = serverConnection.getTransientFlag(RESPONDED);
    boolean requiresChunkedResponse = serverConnection.getTransientFlag(REQUIRES_CHUNKED_RESPONSE);
    boolean potentialModification = serverConnection.getPotentialModification();

    try {
      boolean wroteExceptionResponse = false;
      try {
        if (requiresResponse && !responded) {
          if (requiresChunkedResponse) {
            writeChunkedException(msg, e, serverConnection);
          } else {
            writeException(msg, e, false, serverConnection);
          }
          wroteExceptionResponse = true;
          serverConnection.setAsTrue(RESPONDED);
        }
      } finally { // inner try-finally to ensure proper ordering of logging
        if (potentialModification) {
          int transId = msg.getTransactionId();
          if (!wroteExceptionResponse) {
            logger.warn(String.format(
                "%s: Unexpected Exception during operation on region: %s key: %s messageId: %s",
                serverConnection.getName(), serverConnection.getModRegion(),
                serverConnection.getModKey(), transId),
                e);
          } else {
            if (logger.isDebugEnabled()) {
              logger.debug("{}: Exception during operation on region: {} key: {} messageId: {}",
                  serverConnection.getName(), serverConnection.getModRegion(),
                  serverConnection.getModKey(), transId, e);
            }
          }
        } else {
          if (!wroteExceptionResponse) {
            logger.warn(String.format("%s: Unexpected Exception",
                serverConnection.getName()), e);
          } else {
            if (logger.isDebugEnabled()) {
              logger.debug("{}: Exception: {}", serverConnection.getName(), e.getMessage(), e);
            }
          }
        }
      }
    } catch (IOException ioe) {
      if (logger.isDebugEnabled()) {
        logger.debug("{}: Unexpected IOException writing exception: {}", serverConnection.getName(),
            ioe.getMessage(), ioe);
      }
    }
  }

  private static void handleThrowable(final @Nullable Message msg,
      final @NotNull ServerConnection serverConnection,
      final @NotNull Throwable th) {
    boolean requiresResponse = serverConnection.getTransientFlag(REQUIRES_RESPONSE);
    boolean responded = serverConnection.getTransientFlag(RESPONDED);
    boolean requiresChunkedResponse = serverConnection.getTransientFlag(REQUIRES_CHUNKED_RESPONSE);
    boolean potentialModification = serverConnection.getPotentialModification();

    try {
      try {
        if (th instanceof Error) {
          logger.fatal(String.format("%s : Unexpected Error on server",
              serverConnection.getName()),
              th);
        }
        if (requiresResponse && !responded && null != msg) {
          if (requiresChunkedResponse) {
            writeChunkedException(msg, th, serverConnection);
          } else {
            writeException(msg, th, false, serverConnection);
          }
          serverConnection.setAsTrue(RESPONDED);
        }
      } finally { // inner try-finally to ensure proper ordering of logging
        if (!(th instanceof Error || th instanceof CacheLoaderException)) {
          if (potentialModification) {
            int transId = msg != null ? msg.getTransactionId() : Integer.MIN_VALUE;
            logger.warn(String.format(
                "%s: Unexpected Exception during operation on region: %s key: %s messageId: %s",
                serverConnection.getName(), serverConnection.getModRegion(),
                serverConnection.getModKey(), transId),
                th);
          } else {
            logger.warn(String.format("%s: Unexpected Exception",
                serverConnection.getName()), th);
          }
        }
      }
    } catch (IOException ioe) {
      if (logger.isDebugEnabled()) {
        logger.debug("{}: Unexpected IOException writing exception: {}", serverConnection.getName(),
            ioe.getMessage(), ioe);
      }
    } finally {
      serverConnection.setFlagProcessMessagesAsFalse();
      serverConnection.setClientDisconnectedException(th);
    }
  }

  protected static void writeChunkedException(final @NotNull Message origMsg,
      final @NotNull Throwable e,
      final @NotNull ServerConnection serverConnection) throws IOException {
    writeChunkedException(origMsg, e, serverConnection,
        serverConnection.getChunkedResponseMessage());
  }

  protected static void writeChunkedException(final @NotNull Message origMsg,
      final @NotNull Throwable e,
      final @NotNull ServerConnection serverConnection,
      final @NotNull ChunkedMessage originalResponse) throws IOException {
    writeChunkedException(origMsg, e, serverConnection, originalResponse, 2);
  }

  private static void writeChunkedException(final @NotNull Message origMsg,
      final @NotNull Throwable exception,
      final @NotNull ServerConnection serverConnection,
      final @NotNull ChunkedMessage originalResponse, final int numOfParts)
      throws IOException {
    Throwable e = getClientException(serverConnection, exception);
    ChunkedMessage chunkedResponseMsg = serverConnection.getChunkedResponseMessage();
    chunkedResponseMsg.setServerConnection(serverConnection);
    if (originalResponse.headerHasBeenSent()) {
      chunkedResponseMsg.setNumberOfParts(numOfParts);
      chunkedResponseMsg.setLastChunkAndNumParts(true, numOfParts);
      chunkedResponseMsg.addObjPart(e);
      if (numOfParts == 2) {
        chunkedResponseMsg.addStringPart(getExceptionTrace(e));
      }
      if (logger.isDebugEnabled()) {
        logger.debug("{}: Sending exception chunk while reply in progress: {}",
            serverConnection.getName(), e.getMessage(), e);
      }
    } else {
      chunkedResponseMsg.setMessageType(MessageType.EXCEPTION);
      chunkedResponseMsg.setNumberOfParts(numOfParts);
      chunkedResponseMsg.setLastChunkAndNumParts(true, numOfParts);
      chunkedResponseMsg.setTransactionId(origMsg.getTransactionId());
      chunkedResponseMsg.sendHeader();
      chunkedResponseMsg.addObjPart(e);
      if (numOfParts == 2) {
        chunkedResponseMsg.addStringPart(getExceptionTrace(e));
      }
      if (logger.isDebugEnabled()) {
        logger.debug("{}: Sending exception chunk: {}", serverConnection.getName(), e.getMessage(),
            e);
      }
    }
    chunkedResponseMsg.sendChunk(serverConnection);
  }

  // Get the exception stacktrace for native clients
  public static String getExceptionTrace(final @NotNull Throwable ex) {
    StringWriter sw = new StringWriter();
    PrintWriter pw = new PrintWriter(sw);
    ex.printStackTrace(pw);
    pw.close();
    return sw.toString();
  }

  protected static void writeException(final @NotNull Message origMsg, final @NotNull Throwable e,
      final boolean isSevere,
      ServerConnection serverConnection) throws IOException {
    writeException(origMsg, MessageType.EXCEPTION, e, isSevere, serverConnection);
  }

  private static Throwable getClientException(final @NotNull ServerConnection serverConnection,
      final @NotNull Throwable e) {
    InternalCache cache = serverConnection.getCache();
    if (cache != null) {
      OldClientSupportService svc = cache.getService(OldClientSupportService.class);
      if (svc != null) {
        return svc.getThrowable(e, serverConnection.getClientVersion());
      }
    }
    return e;
  }

  protected static void writeException(final @NotNull Message origMsg, final MessageType msgType,
      final @NotNull Throwable e, final boolean isSevere,
      final @NotNull ServerConnection serverConnection) throws IOException {
    Throwable theException = getClientException(serverConnection, e);
    Message errorMsg = serverConnection.getErrorResponseMessage();
    errorMsg.setMessageType(msgType);
    errorMsg.setNumberOfParts(2);
    errorMsg.setTransactionId(origMsg.getTransactionId());
    if (isSevere) {
      String msg = theException.getMessage();
      if (msg == null) {
        msg = theException.toString();
      }
      logger.fatal("Severe cache exception : {}", msg);
    }
    errorMsg.addObjPart(theException);
    errorMsg.addStringPart(getExceptionTrace(theException));
    errorMsg.send(serverConnection);
    if (logger.isDebugEnabled()) {
      logger.debug("{}: Wrote exception: {}", serverConnection.getName(), e.getMessage(), e);
    }
    if (e instanceof MessageTooLargeException) {
      throw (IOException) e;
    }
  }

  protected static void writeErrorResponse(final Message origMsg, final MessageType messageType,
      final @NotNull ServerConnection serverConnection) throws IOException {
    Message errorMsg = serverConnection.getErrorResponseMessage();
    errorMsg.setMessageType(messageType);
    errorMsg.setNumberOfParts(1);
    errorMsg.setTransactionId(origMsg.getTransactionId());
    errorMsg.addStringPart(
        "Invalid data received. Please see the cache server log file for additional details.");
    errorMsg.send(serverConnection);
  }

  protected static void writeErrorResponse(final @NotNull Message origMsg,
      final MessageType messageType,
      final @NotNull String msg,
      final @NotNull ServerConnection serverConnection) throws IOException {
    Message errorMsg = serverConnection.getErrorResponseMessage();
    errorMsg.setMessageType(messageType);
    errorMsg.setNumberOfParts(1);
    errorMsg.setTransactionId(origMsg.getTransactionId());
    errorMsg.addStringPart(msg);
    errorMsg.send(serverConnection);
  }

  protected static void writeRegionDestroyedEx(final @NotNull Message msg,
      final @NotNull String regionName, final @NotNull String reason,
      final @NotNull ServerConnection serverConnection) throws IOException {
    String exceptionMessage = serverConnection.getName() + ": Region named " + regionName + reason;
    RegionDestroyedException ex = new RegionDestroyedException(exceptionMessage, regionName);
    if (serverConnection.getTransientFlag(REQUIRES_CHUNKED_RESPONSE)) {
      writeChunkedException(msg, ex, serverConnection);
    } else {
      writeException(msg, ex, false, serverConnection);
    }
  }

  public void writeResponse(final @Nullable Object data, final @Nullable Object callbackArg,
      final @NotNull Message origMsg,
      final boolean isObject, final @NotNull ServerConnection serverConnection) throws IOException {
    Message responseMsg = serverConnection.getResponseMessage();
    responseMsg.setMessageType(MessageType.RESPONSE);
    responseMsg.setTransactionId(origMsg.getTransactionId());


    if (callbackArg == null) {
      responseMsg.setNumberOfParts(1);
    } else {
      responseMsg.setNumberOfParts(2);
    }
    if (data instanceof byte[]) {
      responseMsg.addRawPart((byte[]) data, isObject);
    } else {
      Assert.assertTrue(isObject, "isObject should be true when value is not a byte[]");
      responseMsg.addObjPart(data, false);
    }
    if (callbackArg != null) {
      responseMsg.addObjPart(callbackArg);
    }
    serverConnection.getCache().getCancelCriterion().checkCancelInProgress(null);
    responseMsg.send(serverConnection);
    origMsg.clearParts();
  }

  protected static void writeResponseWithFunctionAttribute(final byte @NotNull [] data,
      final @NotNull Message origMsg,
      final @NotNull ServerConnection serverConnection) throws IOException {
    Message responseMsg = serverConnection.getResponseMessage();
    responseMsg.setMessageType(MessageType.RESPONSE);
    responseMsg.setTransactionId(origMsg.getTransactionId());
    responseMsg.setNumberOfParts(1);
    responseMsg.addBytesPart(data);
    serverConnection.getCache().getCancelCriterion().checkCancelInProgress(null);
    responseMsg.send(serverConnection);
    origMsg.clearParts();
  }

  protected static void checkForInterrupt(final @NotNull ServerConnection serverConnection,
      final @NotNull Exception e)
      throws InterruptedException, InterruptedIOException {
    serverConnection.getCachedRegionHelper().checkCancelInProgress(e);
    if (e instanceof InterruptedException) {
      throw (InterruptedException) e;
    }
    if (e instanceof InterruptedIOException) {
      throw (InterruptedIOException) e;
    }
  }

  static void writeQueryResponseChunk(final @NotNull Object queryResponseChunk,
      final @Nullable CollectionType collectionType,
      final boolean lastChunk, final @NotNull ServerConnection serverConnection)
      throws IOException {
    ChunkedMessage queryResponseMsg = serverConnection.getQueryResponseMessage();
    queryResponseMsg.setNumberOfParts(2);
    queryResponseMsg.setLastChunk(lastChunk);
    queryResponseMsg.addObjPart(collectionType, false);
    queryResponseMsg.addObjPart(queryResponseChunk, false);
    queryResponseMsg.sendChunk(serverConnection);
  }

  protected static void writeQueryResponseException(final @NotNull Message origMsg,
      final @NotNull Throwable exception,
      final @NotNull ServerConnection serverConnection) throws IOException {
    Throwable e = getClientException(serverConnection, exception);
    ChunkedMessage queryResponseMsg = serverConnection.getQueryResponseMessage();
    ChunkedMessage chunkedResponseMsg = serverConnection.getChunkedResponseMessage();
    if (queryResponseMsg.headerHasBeenSent()) {
      // fix for bug 35442
      // This client is expecting 2 parts in this message so send 2 parts
      queryResponseMsg.setServerConnection(serverConnection);
      queryResponseMsg.setNumberOfParts(2);
      queryResponseMsg.setLastChunkAndNumParts(true, 2);
      queryResponseMsg.addObjPart(e);
      queryResponseMsg.addStringPart(getExceptionTrace(e));
      if (logger.isDebugEnabled()) {
        logger.debug("{}: Sending exception chunk while reply in progress: {}",
            serverConnection.getName(), e.getMessage(), e);
      }
      queryResponseMsg.sendChunk(serverConnection);
    } else {
      chunkedResponseMsg.setServerConnection(serverConnection);
      chunkedResponseMsg.setMessageType(MessageType.EXCEPTION);
      chunkedResponseMsg.setNumberOfParts(2);
      chunkedResponseMsg.setLastChunkAndNumParts(true, 2);
      chunkedResponseMsg.setTransactionId(origMsg.getTransactionId());
      chunkedResponseMsg.sendHeader();
      chunkedResponseMsg.addObjPart(e);
      chunkedResponseMsg.addStringPart(getExceptionTrace(e));
      if (logger.isDebugEnabled()) {
        logger.debug("{}: Sending exception chunk: {}", serverConnection.getName(), e.getMessage(),
            e);
      }
      chunkedResponseMsg.sendChunk(serverConnection);
    }
  }

  protected static void writeChunkedErrorResponse(final @NotNull Message origMsg,
      final @NotNull MessageType messageType, final @NotNull String message,
      final @NotNull ServerConnection serverConnection) throws IOException {
    // Send chunked response header identifying error message
    ChunkedMessage chunkedResponseMsg = serverConnection.getChunkedResponseMessage();
    if (logger.isDebugEnabled()) {
      logger.debug("{}: Sending error message header type: {} transaction: {}",
          serverConnection.getName(), messageType, origMsg.getTransactionId());
    }
    chunkedResponseMsg.setMessageType(messageType);
    chunkedResponseMsg.setTransactionId(origMsg.getTransactionId());
    chunkedResponseMsg.sendHeader();

    // Send actual error
    if (logger.isDebugEnabled()) {
      logger.debug("{}: Sending error message chunk: {}", serverConnection.getName(), message);
    }
    chunkedResponseMsg.setNumberOfParts(1);
    chunkedResponseMsg.setLastChunk(true);
    chunkedResponseMsg.addStringPart(message);
    chunkedResponseMsg.sendChunk(serverConnection);
  }

