/*
 * 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 org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_ACCESSOR;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_ACCESSOR_PP;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTHENTICATOR;
import static org.apache.geode.internal.monitoring.ThreadsMonitoring.Mode.ServerConnectionExecutor;
import static org.apache.geode.logging.internal.spi.LoggingProvider.SECURITY_LOGGER_NAME;

import java.io.EOFException;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.security.Principal;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;

import org.apache.commons.lang3.mutable.MutableInt;
import org.apache.logging.log4j.Logger;
import org.apache.shiro.subject.Subject;
import org.apache.shiro.util.ThreadState;
import org.jetbrains.annotations.TestOnly;

import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.UnsupportedVersionException;
import org.apache.geode.cache.client.internal.Connection;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.tier.Acceptor;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.CommunicationMode;
import org.apache.geode.internal.cache.tier.InternalClientMembership;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.ServerSideHandshake;
import org.apache.geode.internal.cache.tier.sockets.command.Default;
import org.apache.geode.internal.cache.tier.sockets.command.PutUserCredentials;
import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.monitoring.ThreadsMonitoring;
import org.apache.geode.internal.monitoring.executor.AbstractExecutor;
import org.apache.geode.internal.security.AuthorizeRequest;
import org.apache.geode.internal.security.AuthorizeRequestPP;
import org.apache.geode.internal.security.SecurityService;
import org.apache.geode.internal.serialization.ByteArrayDataInput;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.util.Breadcrumbs;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.security.AuthenticationExpiredException;
import org.apache.geode.security.AuthenticationFailedException;
import org.apache.geode.security.AuthenticationRequiredException;
import org.apache.geode.security.GemFireSecurityException;
import org.apache.geode.security.NotAuthorizedException;

/**
 * Provides an implementation for the server socket end of the hierarchical cache connection. Each
 * server connection runs in its own thread to maximize concurrency and improve response times to
 * edge requests
 *
 * @since GemFire 2.0.2
 */
public class ServerConnection implements Runnable {

  protected static final Logger logger = LogService.getLogger();
  private static final Logger secureLogger = LogService.getLogger(SECURITY_LOGGER_NAME);

  /**
   * This is a buffer that we add to client readTimeout value before we cleanup the connection. This
   * buffer time helps prevent EOF in the client instead of SocketTimeout
   */
  private static final int TIMEOUT_BUFFER_FOR_CONNECTION_CLEANUP_MS = 5000;

  private static final String DISALLOW_INTERNAL_MESSAGES_WITHOUT_CREDENTIALS_NAME =
      "geode.disallow-internal-messages-without-credentials";

  public static final String USER_NOT_FOUND = "User authorization attributes not found.";

  /**
   * When true requires some formerly credential-less messages to carry credentials. See GEODE-3249
   * and ServerConnection.isInternalMessage()
   */
  @MutableForTesting
  public static boolean allowInternalMessagesWithoutCredentials =
      !Boolean.getBoolean(DISALLOW_INTERNAL_MESSAGES_WITHOUT_CREDENTIALS_NAME);

  private Map<MessageType, Command> commands;

  protected final SecurityService securityService;

  protected final CacheServerStats stats;

  private final ServerSideHandshakeFactory handshakeFactory = new ServerSideHandshakeFactory();

  // The key is the size of each ByteBuffer. The value is a queue of byte buffers all of that size.
  @MakeNotStatic
  private static final ConcurrentHashMap<Integer, LinkedBlockingQueue<ByteBuffer>> commBufferMap =
      new ConcurrentHashMap<>(4, 0.75f, 1);

  private ServerConnectionCollection serverConnectionCollection;

  private final ProcessingMessageTimer processingMessageTimer = new ProcessingMessageTimer();

  /**
   * Set to false once handshake has been done
   */
  private boolean doHandshake = true;

  private final ThreadsMonitoring threadMonitoring;
  /**
   * The threadMonitorExecutor for this server connection.
   * This will be null if acceptor is a selector.
   * When a selector is used, the thread pool that is used to process client requests
   * will automatically register with the thread monitor the thread that is processing
   * the request. So no need to create a threadMonitorExecutor.
   */
  private AbstractExecutor threadMonitorExecutor;

  public static ByteBuffer allocateCommBuffer(int size, Socket sock) {
    // I expect that size will almost always be the same value
    if (sock.getChannel() == null) {
      // The socket this commBuffer will be used for is old IO (it has no channel).
      // So the commBuffer should be heap based.
      return ByteBuffer.allocate(size);
    }
    LinkedBlockingQueue<ByteBuffer> q = commBufferMap.get(size);
    ByteBuffer result = null;
    if (q != null) {
      result = q.poll();
    }
    if (result == null) {
      result = ByteBuffer.allocateDirect(size);
    } else {
      result.position(0);
      result.limit(result.capacity());
    }
    return result;
  }

  public static void releaseCommBuffer(ByteBuffer bb) {
    if (bb != null && bb.isDirect()) {
      LinkedBlockingQueue<ByteBuffer> q = commBufferMap.get(bb.capacity());
      if (q == null) {
        q = new LinkedBlockingQueue<>();
        LinkedBlockingQueue<ByteBuffer> oldQ = commBufferMap.putIfAbsent(bb.capacity(), q);
        if (oldQ != null) {
          q = oldQ;
        }
      }
      q.offer(bb);
    }
  }

  public static void emptyCommBufferPool() {
    for (LinkedBlockingQueue<ByteBuffer> q : commBufferMap.values()) {
      q.clear();
    }
  }

  Socket theSocket;
  private ByteBuffer commBuffer;
  protected final CachedRegionHelper crHelper;
  protected String name;

  // IMPORTANT: if new messages are added change setHandshake to initialize them
  // to the correct Version for serializing to the client
  private Message requestMessage = new Message(2, KnownVersion.CURRENT);
  private final Message replyMessage = new Message(1, KnownVersion.CURRENT);
  private final Message responseMessage = new Message(1, KnownVersion.CURRENT);
  private final Message errorMessage = new Message(1, KnownVersion.CURRENT);

  // IMPORTANT: if new messages are added change setHandshake to initialize them
  // to the correct Version for serializing to the client
  private final ChunkedMessage queryResponseMessage = new ChunkedMessage(2, KnownVersion.CURRENT);
  private final ChunkedMessage chunkedResponseMessage = new ChunkedMessage(1, KnownVersion.CURRENT);
  private final ChunkedMessage executeFunctionResponseMessage =
      new ChunkedMessage(1, KnownVersion.CURRENT);
  private final ChunkedMessage registerInterestResponseMessage =
      new ChunkedMessage(1, KnownVersion.CURRENT);
  private final ChunkedMessage keySetResponseMessage = new ChunkedMessage(1, KnownVersion.CURRENT);

  @Deprecated
  private final InternalLogWriter logWriter;
  @Deprecated
  private final InternalLogWriter securityLogWriter;

  final Acceptor acceptor;

  /**
   * Handshake reference uniquely identifying a client
   */
  protected ServerSideHandshake handshake;
  private int handshakeTimeout;
  private final Object handshakeMonitor = new Object();

  /*
   * This timeout is request specific which come with message itself Otherwise, timeout which comes
   * during handshake is used.
   */
  private volatile int requestSpecificTimeout = -1;

  /**
   * Tracks the id of the most recent batch to which a reply has been sent
   */
  private int latestBatchIdReplied = -1;

  /**
   * Client identity from handshake
   */
  private ClientProxyMembershipID proxyId;

  private byte[] memberIdByteArray;

  /**
   * Authorize client requests using this object. This is set when each operation on this connection
   * is authorized in post-operation phase.
   */
  private AuthorizeRequestPP postAuthzRequest;

  /**
   * The communication mode for this {@code ServerConnection}. Valid types include
   * 'client-server', 'gateway-gateway' and 'monitor-server'.
   */
  protected final CommunicationMode communicationMode;

  @MakeNotStatic
  private static final ConcurrentHashMap<ClientProxyMembershipID, ClientUserAuths> proxyIdVsClientUserAuths =
      new ConcurrentHashMap<>();


  private ClientUserAuths clientUserAuths;
  private final Object clientUserAuthsLock = new Object();

  // this is constant(server and client) for first user request, after that it is random
  // this also need to send in handshake
  private long connectionId = Connection.DEFAULT_CONNECTION_ID;

  private final Random randomConnectionIdGen;

  private Part securePart;

  protected Principal principal;

  private MessageIdExtractor messageIdExtractor = new MessageIdExtractor();

  /**
   * A debug flag used for testing Backward compatibility
   */
  @MutableForTesting
  private static boolean TEST_VERSION_AFTER_HANDSHAKE_FLAG;

  /**
   * Creates a new {@code ServerConnection} that processes messages received from an edge
   * client over a given {@code Socket}.
   */
  public ServerConnection(final Socket socket, final InternalCache internalCache,
      final CachedRegionHelper cachedRegionHelper, final CacheServerStats stats,
      final int hsTimeout, final int socketBufferSize, final String communicationModeStr,
      final byte communicationMode, final Acceptor acceptor,
      final SecurityService securityService) {
    StringBuilder buffer = new StringBuilder(100);
    if (acceptor.isGatewayReceiver()) {
      buffer.append("GatewayReceiver connection from [");
    } else {
      buffer.append("Server connection from [");
    }
    buffer.append(communicationModeStr).append(" host address=")
        .append(socket.getInetAddress().getHostAddress()).append("; ").append(communicationModeStr)
        .append(" port=").append(socket.getPort()).append("]");
    name = buffer.toString();

    this.stats = stats;
    this.acceptor = acceptor;
    crHelper = cachedRegionHelper;
    logWriter = (InternalLogWriter) internalCache.getLogger();
    securityLogWriter = (InternalLogWriter) internalCache.getSecurityLoggerI18n();
    this.communicationMode = CommunicationMode.fromModeNumber(communicationMode);
    principal = null;
    postAuthzRequest = null;
    randomConnectionIdGen = new Random(hashCode());

    this.securityService = securityService;

    final boolean isDebugEnabled = logger.isDebugEnabled();
    try {
      theSocket = socket;
      theSocket.setSendBufferSize(socketBufferSize);
      theSocket.setReceiveBufferSize(socketBufferSize);

      if (isDebugEnabled) {
        logger.debug(
            "{}: Accepted client connection from {}[client host={}; client port={}]",
            getName(), communicationModeStr, socket.getInetAddress(),
            socket.getPort());
      }
      handshakeTimeout = hsTimeout;
    } catch (Exception e) {
      if (isDebugEnabled) {
        logger.debug("While creating server connection", e);
      }
    }
    threadMonitoring = getCache().getInternalDistributedSystem().getDM().getThreadMonitoring();
    initStreams(socket, socketBufferSize, stats);
  }

  public Acceptor getAcceptor() {
    return acceptor;
  }

  private static final ThreadLocal<Byte> executeFunctionOnLocalNodeOnly =
      ThreadLocal.withInitial(() -> (byte) 0x00);

  public static void executeFunctionOnLocalNodeOnly(Byte value) {
    byte b = value;
    executeFunctionOnLocalNodeOnly.set(b);
  }

  public static Byte isExecuteFunctionOnLocalNodeOnly() {
    return executeFunctionOnLocalNodeOnly.get();
  }

  @VisibleForTesting
  void setServerConnectionCollection(
      ServerConnectionCollection serverConnectionCollection) {
    this.serverConnectionCollection = serverConnectionCollection;
  }

  private boolean verifyClientConnection() {
    synchronized (handshakeMonitor) {
      if (handshake == null) {
        ServerSideHandshake readHandshake;
        try {
          readHandshake = handshakeFactory.readHandshake(getSocket(), getHandShakeTimeout(),
              getCommunicationMode(), getDistributedSystem(), getSecurityService());

        } catch (SocketTimeoutException timeout) {
          logger.warn("{}: Handshake reply code timeout, not received with in {} ms",
              getName(), handshakeTimeout);
          failConnectionAttempt();
          return false;
        } catch (EOFException | SocketException e) {
          // no need to warn client just gave up on this server before we could
          // handshake
          logger.info("{} {}", getName(), e);
          failConnectionAttempt();
          return false;
        } catch (IOException e) {
          logger.warn(getName() + ": Received no handshake reply code",
              e);
          failConnectionAttempt();
          return false;
        } catch (AuthenticationRequiredException | AuthenticationFailedException ex) {
          handleHandshakeAuthenticationException(ex);
          return false;
        } catch (UnsupportedVersionException uve) {
          // Server logging
          logger.warn("{} {}", getName(), uve.getMessage(), uve);
          handleHandshakeException(uve);
          return false;
        } catch (Exception ex) {
          logger.warn("{} {}", getName(), ex.getLocalizedMessage());
          handleHandshakeException(ex);
          return false;
        }

        setHandshake(readHandshake);
        setProxyId(readHandshake.getMembershipId());
        if (getCommunicationMode().isWAN()) {
          try {
            setAuthAttributes();
          } catch (AuthenticationRequiredException | AuthenticationFailedException ex) {
            handleHandshakeAuthenticationException(ex);
            return false;
          } catch (Exception ex) {
            logger.warn("{} {}", getName(), ex.getLocalizedMessage());
            handleHandshakeException(ex);
            return false;
          }
        }

        // readHandshake will establish a handshake object in this ServerConnection
        if (handshake.isOK()) {
          try {
            return processHandShake();
          } catch (CancelException e) {
            if (!crHelper.isShutdown()) {
              logger.warn(getName() + ": Unexpected cancellation: ", e);
            }
            cleanup();
            return false;
          }
        }
        // is this branch ever taken?
        crHelper.checkCancelInProgress(null);
        logger.warn("Received Unknown handshake reply code.");
        refuseHandshake("Received Unknown handshake reply code.", Handshake.REPLY_INVALID);
        return false;
      }
    }
    return true;
  }

  private void failConnectionAttempt() {
    stats.incFailedConnectionAttempts();
    cleanup();
  }

  private void handleHandshakeException(Exception ex) {
    refuseHandshake(ex.getMessage(), Handshake.REPLY_REFUSED);
    failConnectionAttempt();
  }

  private void handleHandshakeAuthenticationException(Exception ex) {
    if (ex instanceof AuthenticationRequiredException) {
      AuthenticationRequiredException noauth = (AuthenticationRequiredException) ex;
      String exStr = noauth.getLocalizedMessage();
      if (noauth.getCause() != null) {
        exStr += " : " + noauth.getCause().getLocalizedMessage();
      }
      if (securityLogWriter.warningEnabled()) {
        securityLogWriter.warning(String.format("%s",
            getName() + ": Security exception: " + exStr));
      }
      refuseHandshake(noauth.getMessage(), Handshake.REPLY_EXCEPTION_AUTHENTICATION_REQUIRED);
      failConnectionAttempt();
    } else if (ex instanceof AuthenticationFailedException) {
      AuthenticationFailedException failed = (AuthenticationFailedException) ex;
      String exStr = failed.getLocalizedMessage();
      if (failed.getCause() != null) {
        exStr += " : " + failed.getCause().getLocalizedMessage();
      }
      if (securityLogWriter.warningEnabled()) {
        securityLogWriter.warning(String.format("%s",
            getName() + ": Security exception: " + exStr));
      }
      refuseHandshake(failed.getMessage(), Handshake.REPLY_EXCEPTION_AUTHENTICATION_FAILED);
      failConnectionAttempt();
    } else {
      logger.warn(
          "Unexpected exception type in ServerConnection handleHandshakeAuthenticationException");
      throw new RuntimeException(
          "Invalid exception type, must be either AuthenticationRequiredException or AuthenticationFailedException",
          ex);
    }
  }

  protected Map<MessageType, Command> getCommands() {
    return commands;
  }

  protected Socket getSocket() {
    return theSocket;
  }

  private int getHandShakeTimeout() {
    return handshakeTimeout;
  }

  protected DistributedSystem getDistributedSystem() {
    return getCache().getDistributedSystem();
  }

  public InternalCache getCache() {
    return crHelper.getCache();
  }

  public ServerSideHandshake getHandshake() {
    return handshake;
  }

  public void setHandshake(ServerSideHandshake handshake) {
    this.handshake = handshake;
    KnownVersion v = handshake.getVersion();

    replyMessage.setVersion(v);
    requestMessage.setVersion(v);
    responseMessage.setVersion(v);
    errorMessage.setVersion(v);

    queryResponseMessage.setVersion(v);
    chunkedResponseMessage.setVersion(v);
    executeFunctionResponseMessage.setVersion(v);
    registerInterestResponseMessage.setVersion(v);
    keySetResponseMessage.setVersion(v);
  }

  void setRequestMessage(Message requestMessage) {
    this.requestMessage = requestMessage;
  }

  public KnownVersion getClientVersion() {
    return handshake.getVersion();
  }

  protected void setProxyId(ClientProxyMembershipID proxyId) {
    this.proxyId = proxyId;
    memberIdByteArray = EventID.getMembershipId(proxyId);
    name = "Server connection from [" + proxyId + "; port=" + theSocket.getPort() + "]";
  }

  protected void setPrincipal(Principal principal) {
    this.principal = principal;
  }

  private long setUserAuthorizeAndPostAuthorizeRequest(AuthorizeRequest authzRequest,
      AuthorizeRequestPP postAuthzRequest) {
    UserAuthAttributes userAuthAttr = new UserAuthAttributes(authzRequest, postAuthzRequest);
    synchronized (clientUserAuthsLock) {
      if (clientUserAuths == null) {
        initializeClientUserAuths();
      }
      return clientUserAuths.putUserAuth(userAuthAttr);
    }
  }

  @Deprecated
  public InternalLogWriter getSecurityLogWriter() {
    return securityLogWriter;
  }

  private SecurityService getSecurityService() {
    return securityService;
  }

  private boolean incedCleanupTableRef;
  private boolean incedCleanupProxyIdTableRef;

  private final Object chmLock = new Object();
  private boolean chmRegistered;

  private Map<ServerSideHandshake, MutableInt> getCleanupTable() {
    return acceptor.getClientHealthMonitor().getCleanupTable();
  }

  private Map<ClientProxyMembershipID, MutableInt> getCleanupProxyIdTable() {
    return acceptor.getClientHealthMonitor().getCleanupProxyIdTable();
  }

  boolean processHandShake() {
    boolean result = false;
    boolean clientJoined = false;

    final boolean isDebugEnabled = logger.isDebugEnabled();
    try {
      synchronized (getCleanupTable()) {
        MutableInt numRefs = getCleanupTable().get(handshake);
        byte endpointType = (byte) 0;
        int queueSize = 0;

        if (proxyId.isDurable()) {
          if (isDebugEnabled) {
            logger.debug("looking if the Proxy existed for this durable client or not :{}",
                proxyId);
          }
          CacheClientProxy proxy =
              getAcceptor().getCacheClientNotifier().getClientProxy(proxyId);
          if (proxy != null && proxy.waitRemoval()) {
            proxy = getAcceptor().getCacheClientNotifier().getClientProxy(proxyId);
          }
          if (proxy != null) {
            if (isDebugEnabled) {
              logger.debug("Proxy existed for this durable client :{} and proxy : {}", proxyId,
                  proxy);
            }
            if (proxy.isPrimary()) {
              endpointType = (byte) 2;
              queueSize = proxy.getQueueSize();
            } else {
              endpointType = (byte) 1;
              queueSize = proxy.getQueueSize();
            }
          }
          if (numRefs == null) {
            // Check whether this is a durable client first. A durable client with
            // the same id is not allowed. In this case, reject the client.
            if (proxy != null && !proxy.isPaused()) {
              // The handshake refusal message must be smaller than 127 bytes.
              String handshakeRefusalMessage =
                  String.format("Duplicate durable clientId (%s)", proxyId.getDurableId());
              logger.warn("{} : {}", name, handshakeRefusalMessage);
              refuseHandshake(handshakeRefusalMessage,
                  Handshake.REPLY_EXCEPTION_DUPLICATE_DURABLE_CLIENT);
              return result;
            }
          }
        }
        if (numRefs != null) {
          if (acceptHandShake(endpointType, queueSize)) {
            numRefs.increment();
            incedCleanupTableRef = true;
            result = true;
          }
          return result;
        }
        if (acceptHandShake(endpointType, queueSize)) {
          clientJoined = true;
          getCleanupTable().put(handshake, new MutableInt(1));
          incedCleanupTableRef = true;
          stats.incCurrentClients();
          result = true;
        }
        return result;
      }
    } finally {
      if (isTerminated() || !result) {
        return false;
      }
      boolean registerClient = false;
      synchronized (getCleanupProxyIdTable()) {
        MutableInt numRefs = getCleanupProxyIdTable().get(proxyId);
        if (numRefs != null) {
          numRefs.increment();
        } else {
          registerClient = true;
          getCleanupProxyIdTable().put(proxyId, new MutableInt(1));
        }
        incedCleanupProxyIdTableRef = true;
      }

      if (isDebugEnabled) {
        logger.debug("{}registering client {}", registerClient ? "" : "not ", proxyId);
      }
      crHelper.checkCancelInProgress(null);
      if (clientJoined && isFiringMembershipEvents()) {
        InternalClientMembership.notifyClientJoined(proxyId.getDistributedMember());
      }

      ClientHealthMonitor chm = acceptor.getClientHealthMonitor();
      synchronized (chmLock) {
        chmRegistered = true;
      }
      if (registerClient) {
        chm.registerClient(proxyId, acceptor.getMaximumTimeBetweenPings());
      }
      serverConnectionCollection = chm.addConnection(proxyId, this);
      acceptor.getConnectionListener().connectionOpened(registerClient, communicationMode);
    }
  }

  private boolean isFiringMembershipEvents() {
    return acceptor.isRunning()
        && !acceptor.getCachedRegionHelper().getCache().isClosed()
        && !acceptor.getCachedRegionHelper().getCache().getCancelCriterion().isCancelInProgress();
  }

  private void refuseHandshake(String message, byte exception) {
    try {
      acceptor.refuseHandshake(theSocket.getOutputStream(), message, exception);
    } catch (IOException ignore) {
    } finally {
      stats.incFailedConnectionAttempts();
      cleanup();
    }
  }

  private boolean acceptHandShake(byte endpointType, int queueSize) {
    return doHandShake(endpointType, queueSize) && handshakeAccepted();
  }

  protected boolean doHandShake(byte endpointType, int queueSize) {
    try {
      handshake.handshakeWithClient(theSocket.getOutputStream(), theSocket.getInputStream(),
          endpointType, queueSize, communicationMode, principal);
    } catch (IOException ioe) {
      if (!crHelper.isShutdown() && !isTerminated()) {
        logger.warn("{}: Handshake accept failed on socket {}: {}", name, theSocket, ioe);
      }
      cleanup();
      return false;
    }
    return true;
  }

  private boolean handshakeAccepted() {
    if (logger.isDebugEnabled()) {
      logger.debug("{}: Accepted handshake", name);
    }

    if (communicationMode == CommunicationMode.ClientToServerForQueue) {
      stats.incCurrentQueueConnections();
    } else {
      stats.incCurrentClientConnections();
    }
    return true;
  }

  public void setCq(String cqName, boolean isDurable) throws Exception {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    if (requestMessage.isSecureMode()) {
      if (isDebugEnabled) {
        logger.debug("setCq() security header found registering CQname = {}", cqName);
      }
      try {
        byte[] secureBytes = requestMessage.getSecureBytes();

        secureBytes = handshake.getEncryptor().decryptBytes(secureBytes);
        AuthIds aIds = new AuthIds(secureBytes);

        long uniqueId = aIds.getUniqueId();

        CacheClientProxy proxy =
            getAcceptor().getCacheClientNotifier().getClientProxy(proxyId);

        if (proxy != null) {
          proxy.setCQVsUserAuth(cqName, uniqueId, isDurable);
        }
      } catch (Exception ex) {
        if (isDebugEnabled) {
          logger.debug("While setting cq got exception ", ex);
        }
        throw ex;
      }
    } else {
      if (isDebugEnabled) {
        logger.debug("setCq() security header is not found ");
      }
    }
  }

  public void removeCq(String cqName, boolean isDurable) {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    if (requestMessage.isSecureMode()) {
      if (isDebugEnabled) {
        logger.debug("removeCq() security header found registering CQname = {}", cqName);
      }
      try {
        clientUserAuths.removeUserAuthAttributesForCq(cqName, isDurable);
      } catch (Exception ex) {
        if (isDebugEnabled) {
          logger.debug("While setting cq got exception ", ex);
        }
      }
    } else {
      if (isDebugEnabled) {
        logger.debug("removeCq() security header is not found");
      }
    }
  }

  /**
   * @return whether this is a connection to a client, regardless of protocol.
   */
  public boolean isClientServerConnection() {
    return communicationMode.isClientToServerOrSubscriptionFeed();
  }

  private boolean clientDisconnectedCleanly;
  private Throwable clientDisconnectedException;
  private int failureCount;
  private volatile boolean processMessages = true;

  public boolean getProcessMessages() {
    return processMessages;
  }

  @VisibleForTesting
  void setProcessMessages(boolean processMessages) {
    this.processMessages = processMessages;
  }

  protected void doHandshake() {
    if (verifyClientConnection()) {
      initializeCommands();
      if (!getCommunicationMode().isWAN()) {
        synchronized (clientUserAuthsLock) {
          initializeClientUserAuths();
        }
      }
    }
    if (TEST_VERSION_AFTER_HANDSHAKE_FLAG) {
      short testVersionAfterHandshake = 4;
      Assert.assertTrue(handshake.getVersion().ordinal() == testVersionAfterHandshake,
          "Found different version after handshake");
      TEST_VERSION_AFTER_HANDSHAKE_FLAG = false;
    }
  }

  void doNormalMessage() {
    if (serverConnectionCollection == null) {
      logger.warn("Continued processing ServerConnection after handshake failed");
      processMessages = false;
      return;
    }

    Message message = getMessage();
    if (!serverConnectionCollection.incrementConnectionsProcessing()) {
      // Client is being disconnected, don't try to process message.
      processMessages = false;
      return;
    }

    if (isTerminated()) {
      // Client is being terminated, don't try to process message.
      processMessages = false;
      return;
    }

    ThreadState threadState = null;
    resumeThreadMonitoring();
    try {
      if (message != null) {
        // Since this thread is not interrupted when the cache server is shutdown, test again after
        // a message has been read. This is a bit of a hack. I think this thread should be
        // interrupted, but currently AcceptorImpl doesn't keep track of the threads that it
        // launches.
        if (!processMessages || crHelper.isShutdown()) {
          if (logger.isDebugEnabled()) {
            logger.debug("{} ignoring message of type {} from client {} due to shutdown.",
                getName(), message.getMessageType(), proxyId);
          }
          return;
        }

        if (message.getMessageType() != MessageType.PING) {
          // check for invalid number of message parts
          if (message.getNumberOfParts() <= 0) {
            failureCount++;
            if (failureCount > 3) {
              processMessages = false;
              return;
            }
            return;
          }
        }

        if (logger.isTraceEnabled()) {
          logger.trace("{} received {} with txid {}", getName(),
              message.getMessageType(), message.getTransactionId());
          if (message.getTransactionId() < -1) {
            message.setTransactionId(-1);
          }
        }

        if (message.getMessageType() != MessageType.PING) {
          // we have a real message (non-ping),
          // so let's call receivedPing to let the CHM know client is busy
          acceptor.getClientHealthMonitor().receivedPing(proxyId);
        }
        Command command = getCommand(message.getMessageType());
        if (command == null) {
          command = Default.getCommand();
        }

        // if a subject exists for this uniqueId, binds the subject to this thread so that we can do
        // authorization later
        threadState = bindSubject(command);
        command.execute(message, this, securityService);
      }
    } finally {
      suspendThreadMonitoring();
      // Keep track of the fact that a message is no longer being
      // processed.
      serverConnectionCollection.connectionsProcessing.decrementAndGet();
      setNotProcessingMessage();
      clearRequestMessage();
      if (threadState != null) {
        threadState.clear();
      }
    }
  }

  ThreadState bindSubject(Command command) {
    if (!securityService.isIntegratedSecurity()) {
      return null;
    }

    if (communicationMode.isWAN()) {
      return null;
    }

    if (command instanceof PutUserCredentials) {
      return null;
    }

