/*
 * 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.cache.Region.SEPARATOR;
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.logging.internal.spi.LoggingProvider.SECURITY_LOGGER_NAME;
import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;

import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.Socket;
import java.security.Principal;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Supplier;

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

import org.apache.geode.CancelException;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.CacheEvent;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.InterestRegistrationEvent;
import org.apache.geode.cache.InterestRegistrationListener;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionExistsException;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.client.internal.PoolImpl.PoolTask;
import org.apache.geode.cache.query.CqException;
import org.apache.geode.cache.query.Query;
import org.apache.geode.cache.query.internal.DefaultQuery;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.cq.ServerCQ;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.ClassLoadUtils;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.cache.CacheClientStatus;
import org.apache.geode.internal.cache.CacheDistributionAdvisor;
import org.apache.geode.internal.cache.CacheServerImpl;
import org.apache.geode.internal.cache.ClientRegionEventImpl;
import org.apache.geode.internal.cache.ClientServerObserver;
import org.apache.geode.internal.cache.ClientServerObserverHolder;
import org.apache.geode.internal.cache.Conflatable;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.EnumListenerEvent;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.FilterProfile;
import org.apache.geode.internal.cache.FilterRoutingInfo.FilterInfo;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalCacheEvent;
import org.apache.geode.internal.cache.InternalRegion;
import org.apache.geode.internal.cache.RegionEventImpl;
import org.apache.geode.internal.cache.ha.HAContainerMap;
import org.apache.geode.internal.cache.ha.HAContainerRegion;
import org.apache.geode.internal.cache.ha.HAContainerWrapper;
import org.apache.geode.internal.cache.ha.HARegionQueue;
import org.apache.geode.internal.cache.ha.ThreadIdentifier;
import org.apache.geode.internal.cache.tier.CommunicationMode;
import org.apache.geode.internal.cache.tier.InterestType;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.OverflowAttributes;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.net.SocketCloser;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.statistics.DummyStatisticsFactory;
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.security.AuthenticationExpiredException;
import org.apache.geode.security.AuthenticationFailedException;
import org.apache.geode.security.AuthenticationRequiredException;
import org.apache.geode.security.GemFireSecurityException;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Class {@code CacheClientNotifier} works on the server and manages client socket connections
 * to clients requesting notification of updates and notifies them when updates occur.
 *
 * @since GemFire 3.2
 */
public class CacheClientNotifier {
  private static final Logger logger = LogService.getLogger();
  private static final Logger secureLogger = LogService.getLogger(SECURITY_LOGGER_NAME);

  @MakeNotStatic
  private static volatile CacheClientNotifier ccnSingleton;

  private final SocketMessageWriter socketMessageWriter;
  private final ClientRegistrationEventQueueManager clientRegistrationEventQueueManager;
  private final CacheClientProxyFactory cacheClientProxyFactory;

  @VisibleForTesting
  static CacheClientNotifier getInstance(InternalCache cache,
      ClientRegistrationEventQueueManager clientRegistrationEventQueueManager,
      StatisticsClock statisticsClock,
      CacheServerStats acceptorStats,
      int maximumMessageCount,
      int messageTimeToLive,
      ConnectionListener listener,
      OverflowAttributes overflowAttributes,
      boolean isGatewayReceiver,
      SocketMessageWriter socketMessageWriter) {
    if (ccnSingleton == null) {
      ccnSingleton = new CacheClientNotifier(cache, clientRegistrationEventQueueManager,
          statisticsClock, acceptorStats, maximumMessageCount, messageTimeToLive, listener,
          isGatewayReceiver, new CacheClientProxyFactory(), socketMessageWriter);
    }

    if (!isGatewayReceiver && ccnSingleton.getHaContainer() == null) {
      // Gateway receiver might have create CCN instance without HaContainer
      // In this case, the HaContainer should be lazily created here
      ccnSingleton.initHaContainer(overflowAttributes);
    }
    return ccnSingleton;
  }

  /**
   * Factory method to construct a CacheClientNotifier {@code CacheClientNotifier} instance.
   *
   * @param cache The GemFire {@code InternalCache}
   * @param clientRegistrationEventQueueManager Manages temporary registration queues for clients
   * @return A {@code CacheClientNotifier} instance
   */
  public static synchronized CacheClientNotifier getInstance(InternalCache cache,
      ClientRegistrationEventQueueManager clientRegistrationEventQueueManager,
      StatisticsClock statisticsClock,
      CacheServerStats acceptorStats,
      int maximumMessageCount,
      int messageTimeToLive,
      ConnectionListener listener,
      OverflowAttributes overflowAttributes,
      boolean isGatewayReceiver) {
    return getInstance(cache, clientRegistrationEventQueueManager, statisticsClock,
        acceptorStats, maximumMessageCount, messageTimeToLive, listener, overflowAttributes,
        isGatewayReceiver, new SocketMessageWriter());
  }

  public static CacheClientNotifier getInstance() {
    return ccnSingleton;
  }

  @VisibleForTesting
  public static void resetInstance() {
    ccnSingleton = null;
  }

  /**
   * Registers a new client updater that wants to receive updates with this server.
   *
   * @param socket The socket over which the server communicates with the client.
   * @param isPrimary Whether server is the primary subscription end point for this client
   * @param acceptorId ID of the acceptor used to clean up the client connection
   * @param notifyBySubscription Whether the Server is running in NotifyBySubscription mode
   * @throws IOException Can occur if there are issues communicating over the socket
   */
  public void registerClient(final ClientRegistrationMetadata clientRegistrationMetadata,
      final Socket socket, final boolean isPrimary, final long acceptorId,
      final boolean notifyBySubscription) throws IOException {
    long startTime = statistics.startTime();

    ClientProxyMembershipID clientProxyMembershipID =
        clientRegistrationMetadata.getClientProxyMembershipID();
    DataOutputStream dataOutputStream = clientRegistrationMetadata.getDataOutputStream();
    KnownVersion clientVersion = clientRegistrationMetadata.getClientVersion();

    try {
      if (isClientPermitted(clientRegistrationMetadata, clientProxyMembershipID)) {
        ClientRegistrationEventQueueManager.ClientRegistrationEventQueue clientRegistrationEventQueue =
            clientRegistrationEventQueueManager.create(clientProxyMembershipID,
                new ConcurrentLinkedQueue<>(),
                new ReentrantReadWriteLock());

        try {
          registerClientInternal(clientRegistrationMetadata, socket, isPrimary, acceptorId,
              notifyBySubscription);
        } finally {
          clientRegistrationEventQueueManager.drain(clientRegistrationEventQueue, this);
        }
      }
    } catch (AuthenticationRequiredException ex) {
      handleAuthenticationException(clientProxyMembershipID, dataOutputStream, clientVersion, ex,
          Handshake.REPLY_EXCEPTION_AUTHENTICATION_REQUIRED);
    } catch (AuthenticationFailedException | AuthenticationExpiredException ex) {
      handleAuthenticationException(clientProxyMembershipID, dataOutputStream, clientVersion, ex,
          Handshake.REPLY_EXCEPTION_AUTHENTICATION_FAILED);
    } catch (CacheException e) {
      logger.warn(
          String.format("%s :registerClient: Exception encountered in registration %s", this, e),
          e);
      throw new IOException(String.format(
          "Exception occurred while trying to register interest due to : %s", e.getMessage()), e);
    } catch (Exception ex) {
      logger.warn(String.format("An exception was thrown for client [%s].",
          clientProxyMembershipID != null ? clientProxyMembershipID : "unknown"), ex);
      socketMessageWriter.writeException(dataOutputStream,
          CommunicationMode.UnsuccessfulServerToClient.getModeNumber(), ex, clientVersion);
    }

    statistics.endClientRegistration(startTime);
  }

  /**
   * Continues the registration of a new client that wants to receive updates with this server.
   *
   * @param clientRegistrationMetadata Contains registration info pertaining to the client
   * @param socket The socket over which the server communicates with the client.
   * @param isPrimary Whether server is the primary subscription end point for this client
   * @param acceptorId ID of the acceptor used to clean up the client connection
   * @param notifyBySubscription Whether the Server is running in NotifyBySubscription mode
   * @throws IOException Can occur if there are issues communicating over the socket
   * @throws CacheException A generic exception, which indicates a cache error has occurred.
   * @throws ClassNotFoundException Thrown when the ClientProxyMembershipID class is not found
   * @throws NoSuchMethodException Potentially thrown by performPostAuthorization
   * @throws InvocationTargetException Potentially thrown by performPostAuthorization
   * @throws IllegalAccessException Potentially thrown by performPostAuthorization
   */
  void registerClientInternal(final ClientRegistrationMetadata clientRegistrationMetadata,
      final Socket socket,
      final boolean isPrimary,
      final long acceptorId, final boolean notifyBySubscription)
      throws IOException, CacheException, ClassNotFoundException, NoSuchMethodException,
      InvocationTargetException, IllegalAccessException {
    ClientProxyMembershipID clientProxyMembershipID =
        clientRegistrationMetadata.getClientProxyMembershipID();
    byte clientConflation = clientRegistrationMetadata.getClientConflation();
    KnownVersion clientVersion = clientRegistrationMetadata.getClientVersion();

    CacheClientProxy cacheClientProxy = getClientProxy(clientProxyMembershipID);
    DistributedMember member = clientProxyMembershipID.getDistributedMember();
    DistributedSystem system = getCache().getDistributedSystem();
    Properties sysProps = system.getProperties();
    String authenticator = sysProps.getProperty(SECURITY_CLIENT_AUTHENTICATOR);
    // in multi-user case, this would return null
    Object subjectOrPrincipal =
        getSubjectOrPrincipal(clientRegistrationMetadata, member, system, authenticator);
    Subject subject = subjectOrPrincipal instanceof Subject ? (Subject) subjectOrPrincipal : null;

    // Initialize the socket
    socket.setTcpNoDelay(true);
    socket.setSendBufferSize(CacheClientNotifier.socketBufferSize);
    socket.setReceiveBufferSize(CacheClientNotifier.socketBufferSize);


    if (logger.isDebugEnabled()) {
      logger.debug(
          "CacheClientNotifier: Initialized server-to-client socket with send buffer size: {} bytes and receive buffer size: {} bytes",
          socket.getSendBufferSize(), socket.getReceiveBufferSize());
    }

    // Determine whether the client is durable or not.
    byte responseByte = CommunicationMode.SuccessfulServerToClient.getModeNumber();
    boolean successful = true;
    boolean clientIsDurable = clientProxyMembershipID.isDurable();
    if (logger.isDebugEnabled()) {
      if (clientIsDurable) {
        logger.debug("CacheClientNotifier: Attempting to register durable client: {}",
            clientProxyMembershipID.getDurableId());
      } else {
        logger.debug("CacheClientNotifier: Attempting to register non-durable client");
      }
    }

    byte endpointType = 0x00;
    int queueSize = 0;
    String unsuccessfulMsg = null;
    if (clientIsDurable) {
      if (cacheClientProxy == null) {
        if (isTimedOut(clientProxyMembershipID)) {
          queueSize = PoolImpl.PRIMARY_QUEUE_TIMED_OUT;
        } else {
          queueSize = PoolImpl.PRIMARY_QUEUE_NOT_AVAILABLE;
        }
        // No proxy exists for this durable client. It must be created.
        if (logger.isDebugEnabled()) {
          logger.debug(
              "CacheClientNotifier: No proxy exists for durable client with id {}. It must be created.",
              clientProxyMembershipID.getDurableId());
        }
        cacheClientProxy =
            cacheClientProxyFactory.create(this, socket, clientProxyMembershipID, isPrimary,
                clientConflation, clientVersion, acceptorId, notifyBySubscription,
                cache.getSecurityService(), subject, statisticsClock);
        successful = initializeProxy(cacheClientProxy);
      } else {
        cacheClientProxy.setSubject(subject);
        if (cacheClientProxy.isPrimary()) {
          endpointType = (byte) 2;
        } else {
          endpointType = (byte) 1;
        }
        queueSize = cacheClientProxy.getQueueSize();
        // A proxy exists for this durable client. It must be reinitialized.
        if (cacheClientProxy.isPaused()) {
          if (CacheClientProxy.testHook != null) {
            CacheClientProxy.testHook.doTestHook("CLIENT_PRE_RECONNECT");
          }
          if (cacheClientProxy.lockDrain()) {
            try {
              if (logger.isDebugEnabled()) {
                logger.debug(
                    "CacheClientNotifier: A proxy exists for durable client with id {}. This proxy will be reinitialized: {}",
                    clientProxyMembershipID.getDurableId(), cacheClientProxy);
              }
              statistics.incDurableReconnectionCount();
              cacheClientProxy.getProxyID()
                  .updateDurableTimeout(clientProxyMembershipID.getDurableTimeout());
              cacheClientProxy.reinitialize(socket, clientProxyMembershipID,
                  isPrimary, clientConflation,
                  clientVersion);
              cacheClientProxy.setMarkerEnqueued(true);
              if (CacheClientProxy.testHook != null) {
                CacheClientProxy.testHook.doTestHook("CLIENT_RECONNECTED");
              }
            } finally {
              cacheClientProxy.unlockDrain();
            }
          } else {
            unsuccessfulMsg =
                "CacheClientNotifier: Connection refused due to cq queue being drained from admin command, please wait...";
            logger.warn(unsuccessfulMsg);
            responseByte = Handshake.REPLY_REFUSED;
            if (CacheClientProxy.testHook != null) {
              CacheClientProxy.testHook.doTestHook("CLIENT_REJECTED_DUE_TO_CQ_BEING_DRAINED");
            }
          }
        } else {
          // The existing proxy is already running (which means that another
          // client is already using this durable id.
          unsuccessfulMsg =
              String.format(
                  "The requested durable client has the same identifier ( %s ) as an existing durable client ( %s ). Duplicate durable clients are not allowed.",
                  clientProxyMembershipID.getDurableId(), cacheClientProxy);
          logger.warn(unsuccessfulMsg);
          // Set the unsuccessful response byte.
          responseByte = Handshake.REPLY_EXCEPTION_DUPLICATE_DURABLE_CLIENT;
        }
      }
    } else {
      CacheClientProxy staleClientProxy = getClientProxy(clientProxyMembershipID);
      if (staleClientProxy != null) {
        // A proxy exists for this non-durable client. It must be closed.
        if (logger.isDebugEnabled()) {
          logger.debug(
              "CacheClientNotifier: A proxy exists for this non-durable client. It must be closed.");
        }
        if (staleClientProxy.startRemoval()) {
          staleClientProxy.waitRemoval();
        } else {
          staleClientProxy.close(false, false); // do not check for queue, just close it
          removeClientProxy(staleClientProxy); // remove old proxy from proxy set
        }
      } // non-null stale proxy

      // Create the new proxy for this non-durable client
      cacheClientProxy =
          new CacheClientProxy(this, socket, clientProxyMembershipID, isPrimary, clientConflation,
              clientVersion, acceptorId, notifyBySubscription, cache.getSecurityService(), subject,
              statisticsClock);
      successful = initializeProxy(cacheClientProxy);
    }

    if (!successful) {
      cacheClientProxy = null;
      responseByte = Handshake.REPLY_REFUSED;
      unsuccessfulMsg =
          String.format(
              "A previous connection attempt from this client is still being processed: %s",
              clientProxyMembershipID);
      logger.warn(unsuccessfulMsg);
    } else if (subject != null) {
      secureLogger.debug("CacheClientProxy {} using subject {} {} ", cacheClientProxy,
          subject.getPrincipal(), subject);
    }

    // Tell the client that the proxy has been registered using the response
    // byte. This byte will be read on the client by the CacheClientUpdater to
    // determine whether the registration was successful. The times when
    // registration is unsuccessful currently are if a duplicate durable client
    // is attempted to be registered or authentication fails.
    try {
      DataOutputStream dos =
          new DataOutputStream(new BufferedOutputStream(socket.getOutputStream()));
      // write the message type, message length and the error message (if any)
      socketMessageWriter.writeHandshakeMessage(dos, responseByte, unsuccessfulMsg, clientVersion,
          endpointType, queueSize);
    } catch (IOException ioe) {// remove the added proxy if we get IOException.
      if (cacheClientProxy != null) {
        // do not check for queue, just close it
        boolean keepProxy = cacheClientProxy.close(false, false);
        if (!keepProxy) {
          removeClientProxy(cacheClientProxy);
        }
      }
      throw ioe;
    }

    if (unsuccessfulMsg != null && logger.isDebugEnabled()) {
      logger.debug(unsuccessfulMsg);
    }

    // If the client is not durable, start its message processor
    // Starting it here (instead of in the CacheClientProxy constructor)
    // will ensure that the response byte is sent to the client before
    // the marker message. If the client is durable, the message processor
    // is not started until the clientReady message is received.
    if (!clientIsDurable && cacheClientProxy != null
        && responseByte == CommunicationMode.SuccessfulServerToClient.getModeNumber()) {
      // The startOrResumeMessageDispatcher tests if the proxy is a primary.
      // If this is a secondary proxy, the dispatcher is not started.
      // The false parameter signifies that a marker message has not already been
      // processed. This will generate and send one.
      cacheClientProxy.startOrResumeMessageDispatcher(false);
    }

    if (responseByte == CommunicationMode.SuccessfulServerToClient.getModeNumber()) {
      if (logger.isDebugEnabled()) {
        logger.debug("CacheClientNotifier: Successfully registered {}", cacheClientProxy);
      }
      performPostAuthorization(cacheClientProxy, clientProxyMembershipID, member,
          sysProps,
          subjectOrPrincipal);
    } else {
      try {
        // prevent leak by closing socket
        socket.close();
      } catch (IOException ignore) {
      }
      logger.warn(
          "CacheClientNotifier: Unsuccessfully registered client with identifier {} and response code {}",
          new Object[] {clientProxyMembershipID, responseByte});
    }
  }

  private void handleAuthenticationException(final ClientProxyMembershipID clientProxyMembershipID,
      final DataOutputStream dataOutputStream, final KnownVersion clientVersion,
      final GemFireSecurityException ex, final byte replyExceptionAuthenticationFailed)
      throws IOException {
    securityLogWriter.warning(
        String.format("An exception was thrown for client [%s]. %s",
            clientProxyMembershipID, ex));
    socketMessageWriter.writeException(dataOutputStream, replyExceptionAuthenticationFailed, ex,
        clientVersion);
  }

  private boolean initializeProxy(CacheClientProxy l_proxy) throws CacheException {
    if (!isProxyInInitializationMode(l_proxy)) {
      if (logger.isDebugEnabled()) {
        logger.debug("Initializing proxy: {}", l_proxy);
      }
      try {
        // Add client proxy to initialization list. This has to be done before
        // the queue is created so that events can be buffered here for delivery
        // to the queue once it's initialized (bug #41681 and others)
        addClientInitProxy(l_proxy);
        l_proxy.initializeMessageDispatcher();
        // Initialization success. Add to client proxy list.
        addClientProxy(l_proxy);
        return true;
      } catch (RegionExistsException ree) {
        if (logger.isDebugEnabled()) {
          String name = ree.getRegion() != null ? ree.getRegion().getFullPath() : "null region";
          logger.debug("Found RegionExistsException while initializing proxy. Region name: {}",
              name);
        }
        // This will return false;
      } finally {
        removeClientInitProxy(l_proxy);
      }
    }
    return false;
  }

  /**
   * Makes Primary to this CacheClientProxy and start the dispatcher of the CacheClientProxy
   *
   * @param isClientReady Whether the marker has already been processed. This value helps determine
   *        whether to start the dispatcher.
   */
  public void makePrimary(ClientProxyMembershipID proxyId, boolean isClientReady) {
    CacheClientProxy proxy = getClientProxy(proxyId);
    if (proxy == null) {
      throw new InternalGemFireError("No cache client proxy on this node for proxyId " + proxyId);
    }
    proxy.setPrimary(true);

    /*
     * If the client represented by this proxy has: - already processed the marker message
     * (meaning the client is failing over to this server as its primary) <or> - is not durable
     * (meaning the marker message is being processed automatically
     *
     * Then, start or resume the dispatcher. Otherwise, let the clientReady message start the
     * dispatcher. See CacheClientProxy.startOrResumeMessageDispatcher if
     * (!proxy._messageDispatcher.isAlive()) {
     */
    if (!proxy.isDurable()) {
      logger.debug("CacheClientNotifier: Notifying non-durable proxy to start dispatcher for: {}",
          proxy);
      proxy.startOrResumeMessageDispatcher(false);
    } else if (isClientReady) {
      logger.debug("CacheClientNotifier: Notifying durable proxy to start dispatcher for: {}",
          proxy);
      // if durable client is failing over to this server as its primary, the marker message may or
      // may not have been delivered to the client yet. If we put the marker on the queue,
      // the marker may be pushed all the way at the end of the queue resulting in the update events
      // being not processed normally by the client. So we will need to send the marker directly
      // to the client instead of enqueuing it.
      proxy.startOrResumeMessageDispatcher(true);
    }
  }

  /**
   * Adds or updates entry in the dispatched message map when client sends an ack.
   */
  public boolean processDispatchedMessage(ClientProxyMembershipID proxyId, EventID eid) {
    boolean success = false;
    CacheClientProxy proxy = getClientProxy(proxyId);
    if (proxy != null) {
      HARegionQueue harq = proxy.getHARegionQueue();
      harq.addDispatchedMessage(new ThreadIdentifier(eid.getMembershipID(), eid.getThreadID()),
          eid.getSequenceID());
      success = true;
    }
    return success;
  }

  /**
   * Sets keepalive on the proxy of the given membershipID
   *
   * @param membershipID Uniquely identifies the client pool
   * @since GemFire 5.7
   */
  public void setKeepAlive(ClientProxyMembershipID membershipID, boolean keepAlive) {
    if (logger.isDebugEnabled()) {
      logger.debug("CacheClientNotifier: setKeepAlive client: {}", membershipID);
    }
    CacheClientProxy proxy = getClientProxy(membershipID);
    if (proxy != null) {
      proxy.setKeepAlive(keepAlive);
    }
  }

  /**
   * Unregisters an existing client from this server.
   *
   * @param memberId Uniquely identifies the client
   */
  void unregisterClient(ClientProxyMembershipID memberId, boolean normalShutdown) {
    if (logger.isDebugEnabled()) {
      logger.debug("CacheClientNotifier: Unregistering all clients with member id: {}", memberId);
    }
    CacheClientProxy proxy = getClientProxy(memberId);
    if (proxy != null) {
      final boolean isTraceEnabled = logger.isTraceEnabled();
      if (isTraceEnabled) {
        logger.trace("CacheClientNotifier: Potential client: {}", proxy);
      }

      // If the proxy's member id is the same as the input member id, add
      // it to the set of dead proxies.
      if (!proxy.startRemoval()) {
        if (isTraceEnabled) {
          logger.trace("CacheClientNotifier: Potential client: {} matches {}", proxy, memberId);
        }
        closeDeadProxies(Collections.singletonList(proxy), normalShutdown);
      }
    }
  }

  /**
   * The client represented by the proxyId is ready to receive updates.
   */
  public void readyForEvents(ClientProxyMembershipID proxyId) {
    CacheClientProxy proxy = getClientProxy(proxyId);
    if (proxy != null) {
      // False signifies that a marker message has not already been processed.
      // Generate and send one.
      proxy.startOrResumeMessageDispatcher(false);
    }
  }

  ClientUpdateMessageImpl constructClientMessage(InternalCacheEvent event) {
    ClientUpdateMessageImpl clientMessage = null;
    EnumListenerEvent operation = event.getEventType();

    try {
      clientMessage = initializeMessage(operation, event);
    } catch (Exception e) {
      logger.fatal(String.format(
          "CacheClientNotifier: Cannot notify clients to perform operation %s on event %s",
          operation, event),
          e);
    }
    return clientMessage;
  }

  /**
   * notify interested clients of the given cache event. The event should have routing information
   * in it that determines which clients will receive the event.
   */
  public static void notifyClients(InternalCacheEvent event) {
    CacheClientNotifier instance = ccnSingleton;
    if (instance != null) {
      instance.singletonNotifyClients(event, null);
    }
  }

  public static boolean singletonHasClientProxies() {
    CacheClientNotifier instance = ccnSingleton;
    if (instance != null) {
      return instance.hasClientProxies();
    }
    return false;
  }

  private boolean hasClientProxies() {
    return !_initClientProxies.isEmpty() || !_clientProxies.isEmpty();
  }

  /**
   * notify interested clients of the given cache event using the given update message. The event
   * should have routing information in it that determines which clients will receive the event.
   */
  public static void notifyClients(InternalCacheEvent event, ClientUpdateMessage cmsg) {
    CacheClientNotifier instance = ccnSingleton;
    if (instance != null) {
      instance.singletonNotifyClients(event, cmsg);
    }
  }

  private void singletonNotifyClients(InternalCacheEvent event, ClientUpdateMessage cmsg) {
    if (!hasClientProxies()) {
      return;
    }

    FilterInfo filterInfo = event.getLocalFilterInfo();

    if (filterInfo != null) {
      // if the routing was made using an old profile we need to recompute it
      if (logger.isTraceEnabled()) {
        logger.trace("Event isOriginRemote={}", event.isOriginRemote());
      }
    }

    if (filterInfo == null
        || filterInfo.getCQs() == null && filterInfo.getInterestedClients() == null
            && filterInfo.getInterestedClientsInv() == null) {
      return;
    }

    long startTime = statistics.startTime();

    ClientUpdateMessageImpl clientMessage;
    if (cmsg == null) {
      clientMessage = constructClientMessage(event);
    } else {
      clientMessage = (ClientUpdateMessageImpl) cmsg;
    }
    if (clientMessage == null) {
      return;
    }

    FilterProfile regionProfile = ((InternalRegion) event.getRegion()).getFilterProfile();

    Set<ClientProxyMembershipID> filterClients =
        getFilterClientIDs(event, regionProfile, filterInfo, clientMessage);

    final Conflatable conflatable;
    if (clientMessage instanceof ClientTombstoneMessage) {
      // HAEventWrapper deserialization can't handle subclasses of ClientUpdateMessageImpl, so don't
      // wrap them
      conflatable = clientMessage;
    } else {
      HAEventWrapper wrapper = new HAEventWrapper(clientMessage);
      wrapper.incrementPutInProgressCounter("notify clients");
      conflatable = wrapper;

      // include new value in event if the entry is not a tombstone and there are clients
      if (!filterClients.isEmpty() && event.getOperation().isEntry()) {
        EntryEventImpl entryEvent = (EntryEventImpl) event;
        entryEvent.exportNewValue(clientMessage);
      }
    }

    // add event to temporary queue for clients in process of registering (if any)
    clientRegistrationEventQueueManager.add(event, clientMessage, conflatable, filterClients, this);

