/*
 * 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.logging.internal.spi.LoggingProvider.SECURITY_LOGGER_NAME;

import java.io.IOException;
import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.logging.log4j.Logger;
import org.apache.shiro.subject.Subject;
import org.apache.shiro.util.ThreadState;
import org.jetbrains.annotations.NotNull;

import org.apache.geode.CancelException;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.ClientSession;
import org.apache.geode.cache.DynamicRegionFactory;
import org.apache.geode.cache.InterestResultPolicy;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.client.internal.RegisterInterestTracker;
import org.apache.geode.cache.operations.DestroyOperationContext;
import org.apache.geode.cache.operations.InvalidateOperationContext;
import org.apache.geode.cache.operations.OperationContext;
import org.apache.geode.cache.operations.PutOperationContext;
import org.apache.geode.cache.operations.RegionClearOperationContext;
import org.apache.geode.cache.operations.RegionCreateOperationContext;
import org.apache.geode.cache.operations.RegionDestroyOperationContext;
import org.apache.geode.cache.query.CqException;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.cq.InternalCqQuery;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.OperationExecutors;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.SystemTimer.SystemTimerTask;
import org.apache.geode.internal.cache.CacheDistributionAdvisee;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.InitialImageAdvice;
import org.apache.geode.internal.cache.Conflatable;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.EnumListenerEvent;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.StateFlushOperation;
import org.apache.geode.internal.cache.ha.HARegionQueue;
import org.apache.geode.internal.cache.tier.InterestType;
import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessageImpl.CqNameToOp;
import org.apache.geode.internal.cache.tier.sockets.command.Get70;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.logging.LogWriterImpl;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.security.AuthorizeRequestPP;
import org.apache.geode.internal.security.SecurityService;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.security.AccessControl;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Class <code>CacheClientProxy</code> represents the server side of the {@link CacheClientUpdater}.
 * It queues messages to be sent from the server to the client. It then reads those messages from
 * the queue and sends them to the client.
 *
 * @since GemFire 4.2
 */
@SuppressWarnings("synthetic-access")
public class CacheClientProxy implements ClientSession {
  private static final Logger logger = LogService.getLogger();
  private static final Logger secureLogger = LogService.getLogger(SECURITY_LOGGER_NAME);

  @Immutable
  @VisibleForTesting
  protected static final CacheClientProxyStatsFactory DEFAULT_CACHECLIENTPROXYSTATSFACTORY =
      (statisticsFactory, proxyId, remoteHostAddress) -> new CacheClientProxyStats(
          statisticsFactory,
          "id_" + proxyId.getDistributedMember().getId() + "_at_" + remoteHostAddress);
  @Immutable
  private static final MessageDispatcherFactory DEFAULT_MESSAGEDISPATCHERFACTORY =
      MessageDispatcher::new;

  /**
   * The socket between the server and the client
   */
  protected Socket _socket;

  private final AtomicBoolean _socketClosed = new AtomicBoolean();

  /**
   * A communication buffer used by each message we send to the client
   */
  protected ByteBuffer _commBuffer;

  /**
   * The remote host's IP address string (cached for convenience)
   */
  protected String _remoteHostAddress;

  /**
   * Concurrency: protected by synchronization of {@link #isMarkedForRemovalLock}
   */
  protected volatile boolean isMarkedForRemoval = false;

  /**
   * @see #isMarkedForRemoval
   */
  protected final Object isMarkedForRemovalLock = new Object();

  /**
   * The proxy id of the client represented by this proxy
   */
  protected ClientProxyMembershipID proxyID;

  /**
   * The GemFire cache
   */
  protected final InternalCache _cache;

  /**
   * The list of keys that the client represented by this proxy is interested in (stored by region)
   */
  protected final ClientInterestList[] cils = new ClientInterestList[2];

  /**
   * A thread that dispatches messages to the client
   */
  protected volatile MessageDispatcher _messageDispatcher;

  /**
   * The statistics for this proxy
   */
  protected final CacheClientProxyStats _statistics;

  protected final AtomicReference<SystemTimerTask> _durableExpirationTask = new AtomicReference<>();

  protected SystemTimer durableTimer;

  /**
   * Whether this dispatcher is paused
   */
  protected volatile boolean _isPaused = true;

  /**
   * True if we are connected to a client.
   */
  private volatile boolean connected = false;

  /**
   * True if a marker message is still in the ha queue.
   */
  private boolean markerEnqueued = false;

  /**
   * The number of times to peek on shutdown before giving up and shutting down
   */
  protected static final int MAXIMUM_SHUTDOWN_PEEKS = Integer
      .getInteger(GeodeGlossary.GEMFIRE_PREFIX + "MAXIMUM_SHUTDOWN_PEEKS", 50);

  /*
   * The default maximum message queue size
   */
  // protected static final int MESSAGE_QUEUE_SIZE_DEFAULT = 230000;

  /** The message queue size */
  protected final int _maximumMessageCount;

  /**
   * The time (in seconds ) after which a message in the client queue will expire.
   */
  protected final int _messageTimeToLive;

  /**
   * The <code>CacheClientNotifier</code> registering this proxy.
   */
  protected final CacheClientNotifier _cacheClientNotifier;

  /**
   * for testing purposes, delays the start of the dispatcher thread
   */
  @MutableForTesting
  public static boolean isSlowStartForTesting = false;

  private boolean isPrimary;

  /** @since GemFire 5.7 */
  protected byte clientConflation = Handshake.CONFLATION_DEFAULT;

  /**
   * Flag to indicate whether to keep a durable client's queue alive
   */
  boolean keepalive = false;

  /**
   * for single user environment
   */
  private AccessControl postAuthzCallback;
  private Subject subject;

  /**
   * used for cq name to subject/auth mapping, always initialized in single/multi user cases
   */
  private ClientUserAuths clientUserAuths;

  private final Object clientUserAuthsLock = new Object();

  /**
   * The version of the client
   */
  private KnownVersion clientVersion;

  /**
   * A map of region name as key and integer as its value. Basically, it stores the names of the
   * regions with <code>DataPolicy</code> as EMPTY. If an event's region name is present in this
   * map, it's full value (and not delta) is sent to the client represented by this proxy.
   *
   * @since GemFire 6.1
   */
  private final Map<String, Integer> regionsWithEmptyDataPolicy = new HashMap<>();

  /**
   * A debug flag used for testing Backward compatibility
   */
  @MutableForTesting
  public static boolean AFTER_MESSAGE_CREATION_FLAG = false;

  /**
   * Notify the region when a client interest registration occurs. This tells the region to update
   * access time when an update is to be pushed to a client. It is enabled only for
   * <code>PartitionedRegion</code>s currently.
   */
  protected static final boolean NOTIFY_REGION_ON_INTEREST =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "updateAccessTimeOnClientInterest");

  /**
   * The AcceptorImpl identifier to which the proxy is connected.
   */
  private final long _acceptorId;

  /** acceptor's setting for notifyBySubscription */
  private final boolean notifyBySubscription;

  /**
   * A counter that keeps track of how many task iterations that have occurred since the last ping
   * or message. The {@linkplain CacheClientNotifier#scheduleClientPingTask ping task} increments
   * it. Normal messages sent to the client reset it. If the counter reaches 3, a ping is sent.
   */
  private final AtomicInteger pingCounter = new AtomicInteger();


  /** Date on which this instances was created */
  private Date creationDate;

  /**
   * true when the durable client associated with this proxy is being restarted and prevents cqs
   * from being closed and drained
   **/
  private boolean drainLocked = false;
  private final Object drainLock = new Object();

  /** number of cq drains that are currently in progress **/
  private int numDrainsInProgress = 0;
  private final Object drainsInProgressLock = new Object();

  private final SecurityService securityService;
  private final StatisticsClock statisticsClock;

  private final MessageDispatcherFactory messageDispatcherFactory;

  /**
   * Constructor.
   *
   * @param ccn The <code>CacheClientNotifier</code> registering this proxy
   * @param socket The socket between the server and the client
   * @param proxyID representing the Connection Proxy of the clien
   * @param isPrimary The boolean stating whether this prozxy is primary
   * @throws CacheException {
   */
  protected CacheClientProxy(CacheClientNotifier ccn, Socket socket,
      ClientProxyMembershipID proxyID, boolean isPrimary, byte clientConflation,
      KnownVersion clientVersion, long acceptorId, boolean notifyBySubscription,
      SecurityService securityService, Subject subject, StatisticsClock statisticsClock)
      throws CacheException {
    this(ccn.getCache(), ccn, socket, proxyID, isPrimary, clientConflation, clientVersion,
        acceptorId, notifyBySubscription, securityService, subject, statisticsClock,
        ccn.getCache().getInternalDistributedSystem().getStatisticsManager(),
        DEFAULT_CACHECLIENTPROXYSTATSFACTORY,
        DEFAULT_MESSAGEDISPATCHERFACTORY, ServerConnection.getClientUserAuths(proxyID));
  }

  @VisibleForTesting
  protected CacheClientProxy(InternalCache cache, CacheClientNotifier ccn, Socket socket,
      ClientProxyMembershipID proxyID, boolean isPrimary, byte clientConflation,
      KnownVersion clientVersion, long acceptorId, boolean notifyBySubscription,
      SecurityService securityService, Subject subject, StatisticsClock statisticsClock,
      StatisticsFactory statisticsFactory,
      CacheClientProxyStatsFactory cacheClientProxyStatsFactory,
      MessageDispatcherFactory messageDispatcherFactory, ClientUserAuths clientUserAuths)
      throws CacheException {
    initializeTransientFields(socket, proxyID, isPrimary, clientConflation, clientVersion);
    _cacheClientNotifier = ccn;
    _cache = cache;
    this.securityService = securityService;
    _maximumMessageCount = ccn.getMaximumMessageCount();
    _messageTimeToLive = ccn.getMessageTimeToLive();
    _acceptorId = acceptorId;
    this.notifyBySubscription = notifyBySubscription;
    this.statisticsClock = statisticsClock;
    _statistics =
        cacheClientProxyStatsFactory.create(statisticsFactory, proxyID, _remoteHostAddress);
    this.subject = subject;

    // Create the interest list
    cils[RegisterInterestTracker.interestListIndex] =
        new ClientInterestList(this, this.proxyID);
    // Create the durable interest list
    cils[RegisterInterestTracker.durableInterestListIndex] =
        new ClientInterestList(this, getDurableId());
    postAuthzCallback = null;
    _cacheClientNotifier.getAcceptorStats().incCurrentQueueConnections();
    creationDate = new Date();
    this.messageDispatcherFactory = messageDispatcherFactory;
    this.clientUserAuths = clientUserAuths;
  }

  private void reinitializeClientAuths() {
    synchronized (clientUserAuthsLock) {
      ClientUserAuths newClientAuth = ServerConnection.getClientUserAuths(proxyID);
      newClientAuth.fillPreviousCQAuth(clientUserAuths);
      clientUserAuths = newClientAuth;
    }
  }

  public void setPostAuthzCallback(AccessControl authzCallback) {
    // TODO:hitesh synchronization
    synchronized (clientUserAuthsLock) {
      if (postAuthzCallback != null) {
        postAuthzCallback.close();
      }
      postAuthzCallback = authzCallback;
    }
  }

  public void setSubject(Subject subject) {
    // if we are replacing a subject here, the old subject's logout should be handled
    // by the ClientUserAuths already
    synchronized (clientUserAuthsLock) {
      this.subject = subject;
    }
  }

  protected Subject getSubject(String cqName) {
    synchronized (clientUserAuthsLock) {
      return clientUserAuths.getSubject(cqName);
    }
  }

  public void setCQVsUserAuth(String cqName, long uniqueId, boolean isDurable) {
    clientUserAuths.setUserAuthAttributesForCq(cqName, uniqueId, isDurable);
  }

  private void initializeTransientFields(Socket socket, ClientProxyMembershipID pid, boolean ip,
      byte cc, KnownVersion vers) {
    _socket = socket;
    proxyID = pid;
    connected = true;
    {
      int bufSize = 1024;
      try {
        bufSize = _socket.getSendBufferSize();
        if (bufSize < 1024) {
          bufSize = 1024;
        }
      } catch (SocketException ignore) {
      }
      _commBuffer = ServerConnection.allocateCommBuffer(bufSize, socket);
    }
    _remoteHostAddress = socket.getInetAddress().getHostAddress() + ":" + socket.getPort();
    isPrimary = ip;
    clientConflation = cc;
    clientVersion = vers;
  }

  public boolean isMarkerEnqueued() {
    return markerEnqueued;
  }

  public void setMarkerEnqueued(boolean bool) {
    markerEnqueued = bool;
  }

  public long getAcceptorId() {
    return _acceptorId;
  }

  /**
   * @return the notifyBySubscription
   */
  public boolean isNotifyBySubscription() {
    return notifyBySubscription;
  }


  /**
   * Returns the DistributedMember represented by this proxy
   */
  public ClientProxyMembershipID getProxyID() {
    return proxyID;
  }

  protected boolean isMember(ClientProxyMembershipID memberId) {
    return proxyID.equals(memberId);
  }

  /**
   * Set the queue keepalive option
   *
   * @param option whether to keep the durable client's queue alive
   */
  protected void setKeepAlive(boolean option) {
    keepalive = option;
  }

  /**
   * Returns the socket between the server and the client
   *
   * @return the socket between the server and the client
   */
  protected Socket getSocket() {
    return _socket;
  }

  public String getSocketHost() {
    return _socket.getInetAddress().getHostAddress();
  }

  protected ByteBuffer getCommBuffer() {
    return _commBuffer;
  }

  /**
   * Returns the remote host's IP address string
   *
   * @return the remote host's IP address string
   */
  protected String getRemoteHostAddress() {
    return _remoteHostAddress;
  }

  /**
   * Returns the remote host's port
   *
   * @return the remote host's port
   */
  public int getRemotePort() {
    return _socket.getPort();
  }

  /**
   * Returns whether the proxy is connected to a remote client
   *
   * @return whether the proxy is connected to a remote client
   */
  public boolean isConnected() {
    return connected;
  }

  /**
   * Mark the receiver as needing removal
   *
   * @return true if it was already marked for removal
   */
  protected boolean startRemoval() {
    boolean result;
    synchronized (isMarkedForRemovalLock) {
      result = isMarkedForRemoval;
      isMarkedForRemoval = true;
    }
    return result;
  }

  /**
   * Wait until the receiver's removal has completed before returning.
   *
   * @return true if the proxy was initially marked for removal
   */
  protected boolean waitRemoval() {
    boolean result;
    synchronized (isMarkedForRemovalLock) {
      result = isMarkedForRemoval;
      boolean interrupted = false;
      try {
        while (isMarkedForRemoval) {
          if (logger.isDebugEnabled()) {
            logger.debug("Waiting for CacheClientProxy removal: {}", this);
          }
          try {
            isMarkedForRemovalLock.wait();
          } catch (InterruptedException e) {
            interrupted = true;
            _cache.getCancelCriterion().checkCancelInProgress(e);
          }
        } // while
      } finally {
        if (interrupted) {
          Thread.currentThread().interrupt();
        }
      }
    } // synchronized
    return result;
  }

  /**
   * Indicate that removal has completed on this instance
   */
  protected void notifyRemoval() {
    synchronized (isMarkedForRemovalLock) {
      isMarkedForRemoval = false;
      isMarkedForRemovalLock.notifyAll();
    }
  }

  /**
   * Returns the GemFire cache
   *
   * @return the GemFire cache
   */
  public InternalCache getCache() {
    return _cache;
  }

  public Set<String> getInterestRegisteredRegions() {
    HashSet<String> regions = new HashSet<>();
    for (final ClientInterestList cil : cils) {
      if (!cil.regions.isEmpty()) {
        regions.addAll(cil.regions);
      }
    }
    return regions;
  }

  /**
   * Returns the proxy's statistics
   *
   * @return the proxy's statistics
   */
  public CacheClientProxyStats getStatistics() {
    return _statistics;
  }

  /**
   * Returns this proxy's <code>CacheClientNotifier</code>.
   *
   * @return this proxy's <code>CacheClientNotifier</code>
   */
  protected CacheClientNotifier getCacheClientNotifier() {
    return _cacheClientNotifier;
  }

  /**
   * Returns the size of the queue for heuristic purposes. This size may be changing concurrently if
   * puts/gets are occurring at the same time.
   */
  public int getQueueSize() {
    return _messageDispatcher == null ? 0 : _messageDispatcher.getQueueSize();
  }

  /**
   * returns the queue size calculated through stats
   */
  public int getQueueSizeStat() {
    return _messageDispatcher == null ? 0 : _messageDispatcher.getQueueSizeStat();
  }


  public boolean drainInProgress() {
    synchronized (drainsInProgressLock) {
      return numDrainsInProgress > 0;
    }
  }

  // Called from CacheClientNotifier when attempting to restart paused proxy
  // locking the drain lock requires that no drains are in progress
  // when the lock was acquired.
  public boolean lockDrain() {
    synchronized (drainsInProgressLock) {
      if (!drainInProgress()) {
        synchronized (drainLock) {
          if (testHook != null) {
            testHook.doTestHook("PRE_ACQUIRE_DRAIN_LOCK_UNDER_SYNC");
          }
          // prevent multiple lockings of drain lock
          if (!drainLocked) {
            drainLocked = true;
            return true;
          }
        }
      }
    }
    return false;
  }

  // Called from CacheClientNotifier when completed restart of proxy
  public void unlockDrain() {
    if (testHook != null) {
      testHook.doTestHook("PRE_RELEASE_DRAIN_LOCK");
    }
    synchronized (drainLock) {
      drainLocked = false;
    }
  }

  // Only close the client cq if it is paused and no one is attempting to restart the proxy
  public boolean closeClientCq(String clientCQName) throws CqException {
    if (testHook != null) {
      testHook.doTestHook("PRE_DRAIN_IN_PROGRESS");
    }
    synchronized (drainsInProgressLock) {
      numDrainsInProgress++;
    }
    if (testHook != null) {
      testHook.doTestHook("DRAIN_IN_PROGRESS_BEFORE_DRAIN_LOCK_CHECK");
    }
    try {
      // If the drain lock was acquired, the other thread did so before we could bump up
      // the numDrainsInProgress. That means we need to stop.
      if (drainLocked) {
        // someone is trying to restart a paused proxy
        String msg =
            String.format(
                "CacheClientProxy: Could not drain cq %s due to client proxy id %s reconnecting.",
                clientCQName, proxyID.getDurableId());
        logger.info(msg);
        throw new CqException(msg);
      }
      // isConnected is to protect against the case where a durable client has reconnected
      // but has not yet sent a ready for events message
      // we can probably remove the isPaused check
      if (isPaused() && !isConnected()) {
        CqService cqService = getCache().getCqService();
        if (cqService != null) {
          InternalCqQuery cqToClose =
              cqService.getCq(cqService.constructServerCqName(clientCQName, proxyID));
          // close and drain
          if (cqToClose != null) {
            cqService.closeCq(clientCQName, proxyID);
            _messageDispatcher.drainClientCqEvents(proxyID, cqToClose);
          } else {
            String msg = String.format("CQ Not found, Failed to close the specified CQ %s",
                clientCQName);
            logger.info(msg);
            throw new CqException(msg);
          }
        }
      } else {
        String msg =
            String.format(
                "CacheClientProxy: Could not drain cq %s because client proxy id %s is connected.",
                clientCQName, proxyID.getDurableId());
        logger.info(msg);
        throw new CqException(msg);
      }
    } finally {
      synchronized (drainsInProgressLock) {
        numDrainsInProgress--;
      }
      if (testHook != null) {
        testHook.doTestHook("DRAIN_COMPLETE");
      }

    }
    return true;
  }


  /**
   * Returns whether the proxy is alive. It is alive if its message dispatcher is processing
   * messages.
   *
   * @return whether the proxy is alive
   */
  protected boolean isAlive() {
    if (_messageDispatcher == null) {
      return false;
    }
    return !_messageDispatcher.isStopped();
  }


  public boolean isWaitingForReAuthentication() {
    if (_messageDispatcher == null) {
      return false;
    }
    return _messageDispatcher.isWaitingForReAuthentication();
  }

  public void notifyReAuthentication() {
    if (_messageDispatcher == null) {
      return;
    }

    // use another thread to do the notification so that the server operation won't be blocked
    ExecutorService threadPool =
        _cache.getDistributionManager().getExecutors().getWaitingThreadPool();
    threadPool.submit(() -> _messageDispatcher.notifyReAuthentication());
  }

  /**
   * Returns whether the proxy is paused. It is paused if its message dispatcher is paused. This
   * only applies to durable clients.
   *
   * @return whether the proxy is paused
   *
   * @since GemFire 5.5
   */
  public boolean isPaused() {
    return _isPaused;
  }

  protected void setPaused(boolean isPaused) {
    _isPaused = isPaused;
  }

  /**
   * Closes the proxy. This method checks the message queue for any unprocessed messages and
   * processes them for MAXIMUM_SHUTDOWN_PEEKS.
   *
   * @see CacheClientProxy#MAXIMUM_SHUTDOWN_PEEKS
   */
  protected void close() {
    close(true, false);
  }

  /**
   * Set to true once this proxy starts being closed. Remains true for the rest of its existence.
   */
  private final AtomicBoolean closing = new AtomicBoolean(false);

  /**
   * Close the <code>CacheClientProxy</code>.
   *
   * @param checkQueue Whether to message check the queue and process any contained messages (up to
   *        MAXIMUM_SHUTDOWN_PEEKS).
   * @param stoppedNormally Whether client stopped normally
   *
   * @return whether to keep this <code>CacheClientProxy</code>
   * @see CacheClientProxy#MAXIMUM_SHUTDOWN_PEEKS
   */
  protected boolean close(boolean checkQueue, boolean stoppedNormally) {
    // If the client is durable and either (a) it hasn't stopped normally or (b) it
    // has stopped normally but it is configured to be kept alive, set pauseDurable
    // to true
    final boolean pauseDurable =
        isDurable() && (!stoppedNormally || (getDurableKeepAlive() && stoppedNormally));

    boolean keepProxy = false;
    if (pauseDurable) {
      pauseDispatching();
      keepProxy = true;
    } else {
      terminateDispatching(checkQueue);
      closeTransientFields();
    }

    connected = false;

    // Close the Authorization callback or subject if we are not keeping the proxy
    try {
      if (!pauseDurable) {
        cleanClientAuths();
      }
    } catch (Exception ex) {
      logger.warn("{}", this, ex);
    }
    // Notify the caller whether to keep this proxy. If the proxy is durable
    // and should be paused, then return true; otherwise return false.
    return keepProxy;
  }

  // this needs to synchronized to avoid NPE between null check and operations
  private void cleanClientAuths() {
    synchronized (clientUserAuthsLock) {
      // single user case -- old security
      if (postAuthzCallback != null) {
        postAuthzCallback.close();
        postAuthzCallback = null;
      }
      // single user case -- integrated security
      // connection is closed, so we can log out this subject
      else if (subject != null) {
        secureLogger.debug("CacheClientProxy.close, logging out {}. ", subject.getPrincipal());
        subject.logout();
        subject = null;
      }
      // for multiUser case, in non-durable case, we are closing the connection
      else if (clientUserAuths != null) {
        secureLogger.debug("CacheClientProxy.close, cleanup all client subjects. ");
        clientUserAuths.cleanup(true);
        clientUserAuths = null;
      }
    }
  }

  protected void pauseDispatching() {
    if (_messageDispatcher == null) {
      return;
    }

    // If this is the primary, pause the dispatcher (which closes its transient
    // fields. Otherwise, just close the transient fields.
    if (logger.isDebugEnabled()) {
      logger.debug("{}: Pausing processing", this);
    }
    // BUGFIX for BUG#38234
    if (!testAndSetPaused(true) && isPrimary) {
      if (_messageDispatcher != Thread.currentThread()) {
        // don't interrupt ourself to fix bug 40611
        _messageDispatcher.interrupt();
      }
    }

    try {
      // Close transient fields
      closeTransientFields();
    } finally {
      // make sure this gets called if closeTransientFields throws; see bug 40611
      // Start timer
      scheduleDurableExpirationTask();
    }
  }

  private boolean testAndSetPaused(boolean newValue) {

    synchronized (_messageDispatcher._pausedLock) {
      if (_isPaused != newValue) {
        _isPaused = newValue;
        _messageDispatcher._pausedLock.notifyAll();
        return !_isPaused;
      } else {
        _messageDispatcher._pausedLock.notifyAll();
        return _isPaused;
      }
    }
  }

  protected void terminateDispatching(boolean checkQueue) {
    if (_messageDispatcher == null) {
      return;
    }

    try {
      if (logger.isDebugEnabled()) {
        logger.debug("{}: Terminating processing", this);
      }
      if (_messageDispatcher == Thread.currentThread()) {
        // I'm not even sure this is possible but if the dispatcher
        // calls us then at least call stopDispatching
        // the old code did this (I'm not even sure it is safe to do).
        // This needs to be done without testing OR setting "closing".
        _messageDispatcher.stopDispatching(checkQueue);
        cils[RegisterInterestTracker.interestListIndex].clearClientInterestList();
        cils[RegisterInterestTracker.durableInterestListIndex].clearClientInterestList();
        // VJR: bug 37487 fix
        destroyRQ();
        return;
      }

      if (!closing.compareAndSet(false, true)) {
        // must already be closing so just return
        // this is part of the fix for 37684
        return;
      }
      // Unregister interest in all interests (if necessary)
      cils[RegisterInterestTracker.interestListIndex].clearClientInterestList();
      cils[RegisterInterestTracker.durableInterestListIndex].clearClientInterestList();

      // If the message dispatcher is paused, unpause it. The next bit of
      // code will interrupt the waiter.
      if (testAndSetPaused(false)) {
        if (logger.isDebugEnabled()) {
          logger.debug("{}: Paused but terminating processing", this);
        }
        // Cancel the expiration task
        cancelDurableExpirationTask(false);
      }

      boolean alreadyDestroyed = false;
      boolean gotInterrupt = Thread.interrupted(); // clears the flag
      try {
        // Stop the message dispatcher
        _messageDispatcher.stopDispatching(checkQueue);

        gotInterrupt |= Thread.interrupted(); // clears the flag

        // to fix bug 37684
        // 1. check to see if dispatcher is still alive
        if (_messageDispatcher.isAlive()) {
          closeSocket();
          destroyRQ();
          alreadyDestroyed = true;
          _messageDispatcher.interrupt();
          if (_messageDispatcher.isAlive()) {
            try {
              _messageDispatcher.join(1000);
            } catch (InterruptedException ex) {
              gotInterrupt = true;
            }
            // if it is still alive then warn and move on
            if (_messageDispatcher.isAlive()) {
              // org.apache.geode.logging.internal.OSProcess.printStacks(org.apache.geode.logging.internal.OSProcess.getId());
              logger.warn("{}: Could not stop message dispatcher thread.",
                  this);
            }
          }
        }
      } finally {
        if (gotInterrupt) {
          Thread.currentThread().interrupt();
        }
        if (!alreadyDestroyed) {
          destroyRQ();
        }
      }
    } finally {
      // Close the statistics
      _statistics.close(); // fix for bug 40105
      closeTransientFields(); // make sure this happens
    }
  }

  private boolean closeSocket() {
    String remoteHostAddress = _remoteHostAddress;
    if (_socketClosed.compareAndSet(false, true) && remoteHostAddress != null) {
      // Only one thread is expected to close the socket
      _cacheClientNotifier.getSocketCloser().asyncClose(_socket, remoteHostAddress,
          () -> {
          });
      getCacheClientNotifier().getAcceptorStats().decCurrentQueueConnections();
      return true;
    }
    return false;
  }

  @VisibleForTesting
  protected void closeTransientFields() {
    if (!closeSocket()) {
      // The thread who closed the socket will be responsible to
      // releaseResourcesForAddress and clearClientInterestList
      return;
    }

    // Null out comm buffer, host address, ports and proxy id. All will be
    // replaced when the client reconnects.
    releaseCommBuffer();
    {
      String remoteHostAddress = _remoteHostAddress;
      if (remoteHostAddress != null) {
        _cacheClientNotifier.getSocketCloser().releaseResourcesForAddress(remoteHostAddress);
        _remoteHostAddress = null;
      }
    }
    try {
      cils[RegisterInterestTracker.interestListIndex].clearClientInterestList();
    } catch (CancelException e) {
      // ignore if cache is shutting down
    }
    // Commented to fix bug 40259
