/*
 * 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.cache.query.cq.internal;

import static org.apache.geode.internal.cache.tier.MessageType.CLEAR_REGION;
import static org.apache.geode.internal.cache.tier.MessageType.DESTROY_REGION;
import static org.apache.geode.internal.cache.tier.MessageType.EXCEPTION;
import static org.apache.geode.internal.cache.tier.MessageType.INVALIDATE_REGION;
import static org.apache.geode.internal.cache.tier.MessageType.LOCAL_CREATE;
import static org.apache.geode.internal.cache.tier.MessageType.LOCAL_DESTROY;
import static org.apache.geode.internal.cache.tier.MessageType.LOCAL_UPDATE;

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.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

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

import org.apache.geode.InvalidDeltaException;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.SystemFailure;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheEvent;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.RegionEvent;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.internal.GetEventValueOp;
import org.apache.geode.cache.client.internal.InternalPool;
import org.apache.geode.cache.client.internal.QueueManager;
import org.apache.geode.cache.client.internal.UserAttributes;
import org.apache.geode.cache.query.CqAttributes;
import org.apache.geode.cache.query.CqClosedException;
import org.apache.geode.cache.query.CqException;
import org.apache.geode.cache.query.CqExistsException;
import org.apache.geode.cache.query.CqListener;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.CqServiceStatistics;
import org.apache.geode.cache.query.CqStatusListener;
import org.apache.geode.cache.query.QueryException;
import org.apache.geode.cache.query.QueryInvalidException;
import org.apache.geode.cache.query.RegionNotFoundException;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.cq.internal.ops.ServerCQProxyImpl;
import org.apache.geode.cache.query.internal.CompiledSelect;
import org.apache.geode.cache.query.internal.CqQueryVsdStats;
import org.apache.geode.cache.query.internal.CqStateImpl;
import org.apache.geode.cache.query.internal.DefaultQuery;
import org.apache.geode.cache.query.internal.ExecutionContext;
import org.apache.geode.cache.query.internal.cq.ClientCQ;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.cq.InternalCqQuery;
import org.apache.geode.cache.query.internal.cq.ServerCQ;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.FilterProfile;
import org.apache.geode.internal.cache.FilterRoutingInfo;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.CacheClientProxy;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.Part;
import org.apache.geode.internal.lang.utils.JavaWorkarounds;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Implements the CqService functionality.
 *
 * @since GemFire 5.5
 */
public class CqServiceImpl implements CqService {
  private static final Logger logger = LogService.getLogger();

  /**
   * System property to evaluate the query even though the initial results are not required when cq
   * is executed using the execute() method.
   */
  public static boolean EXECUTE_QUERY_DURING_INIT = Boolean.parseBoolean(System
      .getProperty(GeodeGlossary.GEMFIRE_PREFIX + "cq.EXECUTE_QUERY_DURING_INIT", "true"));


  private static final String CQ_NAME_PREFIX = "GfCq";

  private final InternalCache cache;

  /**
   * Manages cq pools to determine if a status of connect or disconnect needs to be sent out
   */
  private final HashMap<String, Boolean> cqPoolsConnected = new HashMap<>();

  /**
   * Manages CQ objects. uses serverCqName as key and CqQueryImpl as value
   *
   * GuardedBy cqQueryMapLock
   */
  private volatile HashMap<String, CqQueryImpl> cqQueryMap = new HashMap<>();

  private final Object cqQueryMapLock = new Object();

  private volatile boolean isRunning = false;

  /**
   * Used by client when multiuser-authentication is true.
   */
  private final HashMap<String, UserAttributes> cqNameToUserAttributesMap = new HashMap<>();

  // Map to manage the similar CQs (having same query - performance optimization).
  // With query as key and Set of CQs as values.
  private final ConcurrentHashMap<String, Set<String>> matchingCqMap;

  // CQ Service statistics
  private final CqServiceStatisticsImpl cqServiceStats;
  private final CqServiceVsdStats stats;

  // CQ identifier, also used in auto generated CQ names
  private volatile long cqId = 1;

  /* This is to manage region to CQs map, client side book keeping. */
  private final HashMap<String, ArrayList<String>> baseRegionToCqNameMap = new HashMap<>();

  /**
   * Access and modification to the contents of this map do not necessarily need to be lock
   * protected. This is just used to optimize construction of a server side cq name. Missing values
   * in this cache will mean a look up for a specific proxy id and cq name will miss and reconstruct
   * the string before adding it back to the cache
   */
  private static final ConcurrentHashMap<String, ConcurrentHashMap<ClientProxyMembershipID, String>> serverCqNameCache =
      new ConcurrentHashMap<>();

  /**
   * Constructor.
   *
   * @param cache The cache used for the service
   */
  public CqServiceImpl(final InternalCache cache) {
    if (cache == null) {
      throw new IllegalStateException("cache is null");
    }
    cache.getCancelCriterion().checkCancelInProgress(null);

    this.cache = cache;

    // Initialize the Map which maintains the matching cqs.
    matchingCqMap = new ConcurrentHashMap<>();

    // Initialize the VSD statistics
    StatisticsFactory factory = this.cache.getDistributedSystem();
    stats = new CqServiceVsdStats(factory);
    cqServiceStats = new CqServiceStatisticsImpl(this);
  }

  /**
   * Returns the cache associated with the cqService.
   */
  public Cache getCache() {
    return cache;
  }

  public InternalCache getInternalCache() {
    return cache;
  }

  public CqServiceVsdStats stats() {
    return stats;
  }

  @Override
  public synchronized ClientCQ newCq(String cqName, String queryString, CqAttributes cqAttributes,
      InternalPool pool, boolean isDurable)
      throws QueryInvalidException, CqExistsException, CqException {
    if (queryString == null) {
      throw new IllegalArgumentException(
          String.format("Null argument %s", "queryString"));

    } else if (cqAttributes == null) {
      throw new IllegalArgumentException(
          String.format("Null argument %s", "cqAttribute"));
    }

    if (isServer()) {
      throw new IllegalStateException(
          "client side newCq() method invocation on server.");
    }

    // Check if the given cq already exists.
    if (cqName != null && isCqExists(cqName)) {
      throw new CqExistsException(
          String.format("CQ with the given name already exists. CqName : %s",
              cqName));
    }


    ServerCQProxyImpl serverProxy = pool == null ? null : new ServerCQProxyImpl(pool);
    ClientCQImpl cQuery =
        new ClientCQImpl(this, cqName, queryString, cqAttributes, serverProxy, isDurable);
    cQuery.updateCqCreateStats();

    // cQuery.initCq();

    // Check if query is valid.
    cQuery.validateCq();

    // Add cq into meta region.
    // Check if Name needs to be generated.
    if (cqName == null) {
      // in the case of cqname internally generated, the CqExistsException needs
      // to be taken care internally.
      while (true) {
        cQuery.setName(generateCqName());
        try {
          addToCqMap(cQuery);
        } catch (CqExistsException ex) {
          if (logger.isDebugEnabled()) {
            logger.debug("Got CqExistsException while intializing cq : {} Error : {}",
                cQuery.getName(), ex.getMessage());
          }
          continue;
        }
        break;
      }
    } else {
      addToCqMap(cQuery);
    }

    addToBaseRegionToCqNameMap(cQuery.getBaseRegionName(), cQuery.getServerCqName());

    return cQuery;
  }

  /**
   * Executes the given CqQuery, if the CqQuery for that name is not there it registers the one and
   * executes. This is called on the Server.
   *
   * @param manageEmptyRegions whether to update the 6.1 emptyRegions map held in the CCN
   * @param regionDataPolicy the data policy of the region associated with the query. This is only
   *        needed if manageEmptyRegions is true.
   * @param emptyRegionsMap map of empty regions.
   * @throws IllegalStateException if this is called at client side.
   */
  @Override
  public synchronized ServerCQ executeCq(final @NotNull String cqName,
      final @NotNull String queryString, final int cqState,
      final @NotNull ClientProxyMembershipID clientProxyId,
      final @Nullable CacheClientNotifier ccn,
      final boolean isDurable,
      final boolean manageEmptyRegions,
      @Nullable DataPolicy regionDataPolicy,
      final @NotNull Map<String, Integer> emptyRegionsMap)
      throws CqException, RegionNotFoundException, CqClosedException {
    if (!isServer()) {
      throw new IllegalStateException(
          String.format("Server side executeCq method is called on client. CqName : %s",
              cqName));
    }

    final String serverCqName = constructServerCqName(cqName, clientProxyId);
    final ServerCQImpl cQuery;

    // If this CQ is not yet registered in Server, register CQ.
    if (!isCqExists(serverCqName)) {
      cQuery = new ServerCQImpl(this, cqName, queryString, isDurable,
          constructServerCqName(cqName, clientProxyId));

      try {
        cQuery.registerCq(clientProxyId, ccn, cqState);
        if (manageEmptyRegions) {
          if (emptyRegionsMap.containsKey(cQuery.getBaseRegionName())) {
            regionDataPolicy = DataPolicy.EMPTY;
          }

          final CacheClientProxy proxy = getCacheClientProxy(clientProxyId, ccn);
          ccn.updateMapOfEmptyRegions(
              proxy.getRegionsWithEmptyDataPolicy(),
              cQuery.getBaseRegionName(), regionDataPolicy);
        }
      } catch (CqException cqe) {
        logger.info("Exception while registering CQ on server. CqName : {}",
            cQuery.getName());
        throw cqe;
      }

    } else {
      cQuery = (ServerCQImpl) getCq(serverCqName);
      resumeCQ(cqState, cQuery);
    }

    if (logger.isDebugEnabled()) {
      logger.debug("Successfully created CQ on the server. CqName : {}", cQuery.getName());
    }

    return cQuery;
  }

  public CacheClientProxy getCacheClientProxy(ClientProxyMembershipID clientProxyId,
      CacheClientNotifier ccn) throws CqException {
    CacheClientProxy proxy = ccn.getClientProxy(clientProxyId, true);
    if (proxy == null) {
      throw new CqException("No Cache Client Proxy found while executing CQ.");
    }
    return proxy;
  }

  @Override
  public void resumeCQ(int cqState, ServerCQ cQuery) {
    // Initialize the state of CQ.
    if (((CqStateImpl) cQuery.getState()).getState() != cqState) {
      cQuery.setCqState(cqState);
      // addToCqEventKeysMap(cQuery);
      // Send state change info to peers.
      cQuery.getCqBaseRegion().getFilterProfile().setCqState(cQuery);
    }
    // If we are going to set the state to running, we need to check to see if it matches any other
    // cq
    if (cqState == CqStateImpl.RUNNING) {
      // Add to the matchedCqMap.
      addToMatchingCqMap((CqQueryImpl) cQuery);
    }
  }

  /**
   * Adds the given CQ and cqQuery object into the CQ map.
   */
  void addToCqMap(CqQueryImpl cq) throws CqExistsException, CqException {
    // On server side cqName will be server side cqName.
    String sCqName = cq.getServerCqName();
    if (logger.isDebugEnabled()) {
      logger.debug("Adding to CQ Repository. CqName : {} ServerCqName : {}", cq.getName(), sCqName);
    }
    HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
    if (cqMap.containsKey(sCqName)) {
      throw new CqExistsException(
          String.format("A CQ with the given name %s already exists.",
              sCqName));
    }
    synchronized (cqQueryMapLock) {
      HashMap<String, CqQueryImpl> tmpCqQueryMap = new HashMap<>(cqQueryMap);
      try {
        tmpCqQueryMap.put(sCqName, cq);
      } catch (Exception ex) {
        String errMsg =
            "Failed to store Continuous Query in the repository. CqName: %s %s";
        Object[] errMsgArgs = new Object[] {sCqName, ex.getLocalizedMessage()};
        String s = String.format(errMsg, errMsgArgs);
        logger.error(s);
        throw new CqException(s, ex);
      }
      UserAttributes attributes = UserAttributes.userAttributes.get();
      if (attributes != null) {
        cqNameToUserAttributesMap.put(cq.getName(), attributes);
      }
      cqQueryMap = tmpCqQueryMap;
    }
  }

  /**
   * Removes given CQ from the cqMap..
   */
  void removeCq(String cqName) {
    // On server side cqName will be server side cqName.
    synchronized (cqQueryMapLock) {
      HashMap<String, CqQueryImpl> tmpCqQueryMap = new HashMap<>(cqQueryMap);
      tmpCqQueryMap.remove(cqName);
      cqNameToUserAttributesMap.remove(cqName);
      cqQueryMap = tmpCqQueryMap;
    }
  }

  @Override
  public CqQuery getClientCqFromServer(ClientProxyMembershipID clientProxyId, String clientCqName) {
    // On server side cqName will be server side cqName.
    HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
    return cqMap.get(constructServerCqName(clientCqName, clientProxyId));
  }

  @Override
  public InternalCqQuery getCq(String cqName) {
    // On server side cqName will be server side cqName.
    return cqQueryMap.get(cqName);
  }

  @Override
  public Collection<? extends InternalCqQuery> getAllCqs() {
    return cqQueryMap.values();
  }

  @Override
  public Collection<? extends InternalCqQuery> getAllCqs(final String regionName)
      throws CqException {
    if (regionName == null) {
      throw new IllegalArgumentException(
          String.format("Null argument %s", "regionName"));
    }

    String[] cqNames;

    synchronized (baseRegionToCqNameMap) {
      ArrayList<String> cqs = baseRegionToCqNameMap.get(regionName);
      if (cqs == null) {
        return null;
      }
      cqNames = new String[cqs.size()];
      cqs.toArray(cqNames);
    }

    ArrayList<InternalCqQuery> cQueryList = new ArrayList<>();
    for (String cqName : cqNames) {
      InternalCqQuery cq = getCq(cqName);
      if (cq != null) {
        cQueryList.add(cq);
      }
    }

    return cQueryList;
  }

  @Override
  public synchronized void executeAllClientCqs() throws CqException {
    executeCqs(getAllCqs());
  }

  @Override
  public synchronized void executeAllRegionCqs(final String regionName) throws CqException {
    executeCqs(getAllCqs(regionName));
  }

  @Override
  public synchronized void executeCqs(Collection<? extends InternalCqQuery> cqs)
      throws CqException {
    if (cqs == null) {
      return;
    }
    String cqName = null;
    for (CqQuery cq : cqs) {
      if (!cq.isClosed() && cq.isStopped()) {
        try {
          cqName = cq.getName();
          cq.execute();
        } catch (QueryException | CqClosedException e) {
          if (logger.isDebugEnabled()) {
            logger.debug("Failed to execute the CQ, CqName : {} Error : {}", cqName,
                e.getMessage());
          }
        }
      }
    }
  }

  @Override
  public synchronized void stopAllClientCqs() throws CqException {
    stopCqs(getAllCqs());
  }

  @Override
  public synchronized void stopAllRegionCqs(final String regionName) throws CqException {
    stopCqs(getAllCqs(regionName));
  }

  @Override
  public synchronized void stopCqs(Collection<? extends InternalCqQuery> cqs) throws CqException {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    if (isDebugEnabled) {
      if (cqs == null) {
        logger.debug("CqService.stopCqs cqs : null");
      } else {
        logger.debug("CqService.stopCqs cqs : ({} queries)", cqs.size());
      }
    }

    if (cqs == null) {
      return;
    }

    String cqName = null;
    for (CqQuery cq : cqs) {
      if (!cq.isClosed() && cq.isRunning()) {
        try {
          cqName = cq.getName();
          cq.stop();
        } catch (QueryException | CqClosedException e) {
          if (isDebugEnabled) {
            logger.debug("Failed to stop the CQ, CqName : {} Error : {}", cqName, e.getMessage());
          }
        }
      }
    }
  }

  @Override
  public void closeCqs(final String regionName) throws CqException {
    Collection<? extends InternalCqQuery> cqs = getAllCqs(regionName);
    if (cqs != null) {
      String cqName = null;
      for (InternalCqQuery cq : cqs) {
        try {
          cqName = cq.getName();

          if (isServer()) {
            // invoked on the server
            cq.close(false);
          } else {
            // TODO: grid: if regionName has a pool check its keepAlive
            boolean keepAlive = cache.keepDurableSubscriptionsAlive();
            if (cq.isDurable() && keepAlive) {
              logger.warn("Not sending CQ close to the server as it is a durable CQ");
              cq.close(false);
            } else {
              cq.close(true);
            }
          }

        } catch (QueryException | CqClosedException e) {
          if (logger.isDebugEnabled()) {
            logger.debug("Failed to close the CQ, CqName : {} Error : {}", cqName, e.getMessage());
          }
        }
      }
    }
  }

  /**
   * Called directly on server side.
   */
  @Override
  public void stopCq(String cqName, ClientProxyMembershipID clientId) throws CqException {
    String serverCqName = cqName;
    if (clientId != null) {
      serverCqName = constructServerCqName(cqName, clientId);
      removeFromCacheForServerToConstructedCQName(cqName, clientId);
    }

    ServerCQImpl cQuery = null;
    String errMsg = null;
    Exception ex = null;

    try {
      HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
      if (!cqMap.containsKey(serverCqName)) {
        /*
         * gregp 052808: We should silently fail here instead of throwing error. This is to deal
         * with races in recovery
         */
        return;
      }
      cQuery = (ServerCQImpl) getCq(serverCqName);

    } catch (CacheLoaderException e1) {
      errMsg = "CQ not found in the cq meta region, CqName: %s";
      ex = e1;
    } catch (TimeoutException e2) {
      errMsg = "Timeout while trying to get CQ from meta region, CqName: %s";
      ex = e2;
    } finally {
      if (ex != null) {
        String s = String.format(errMsg, cqName);
        if (logger.isDebugEnabled()) {
          logger.debug(s);
        }
        throw new CqException(s, ex);
      }
    }

    try {
      if (!cQuery.isStopped()) {
        cQuery.stop();
      }
    } catch (CqClosedException cce) {
      throw new CqException(cce.getMessage());
    } finally {
      // If this CQ is stopped, disable caching event keys for this CQ.
      // this.removeCQFromCaching(cQuery.getServerCqName());
      removeFromMatchingCqMap(cQuery);
    }
    // Send stop message to peers.
    cQuery.getCqBaseRegion().getFilterProfile().stopCq(cQuery);
  }

  @Override
  public void closeCq(String cqName, ClientProxyMembershipID clientProxyId) throws CqException {
    String serverCqName = cqName;
    if (clientProxyId != null) {
      serverCqName = constructServerCqName(cqName, clientProxyId);
      removeFromCacheForServerToConstructedCQName(cqName, clientProxyId);
    }

    ServerCQImpl cQuery = null;
    String errMsg = null;
    Exception ex = null;

    try {
      HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
      if (!cqMap.containsKey(serverCqName)) {
        /*
         * gregp 052808: We should silently fail here instead of throwing error. This is to deal
         * with races in recovery
         */
        return;
      }
      cQuery = (ServerCQImpl) cqMap.get(serverCqName);

    } catch (CacheLoaderException e1) {
      errMsg = "CQ not found in the cq meta region, CqName: %s";
      ex = e1;
    } catch (TimeoutException e2) {
      errMsg = "Timeout while trying to get CQ from meta region, CqName: %s";
      ex = e2;
    } finally {
      if (ex != null) {
        String s = String.format(errMsg, cqName);
        if (logger.isDebugEnabled()) {
          logger.debug(s);
        }
        throw new CqException(s, ex);
      }
    }

    try {
      cQuery.close(false);

      // Repository Region.
      // If CQ event caching is enabled, remove this CQs event cache reference.
      // removeCQFromCaching(serverCqName);

      // CqBaseRegion
      try {
        LocalRegion baseRegion = cQuery.getCqBaseRegion();
        if (baseRegion != null && !baseRegion.isDestroyed()) {
          // Server specific clean up.
          if (isServer()) {
            FilterProfile fp = baseRegion.getFilterProfile();
            if (fp != null) {
              fp.closeCq(cQuery);
            }
            CacheClientProxy clientProxy =
                cQuery.getCacheClientNotifier().getClientProxy(clientProxyId);
            clientProxy.decCqCount();
            if (clientProxy.hasNoCq()) {
              stats.decClientsWithCqs();
            }
          }
        }
      } catch (Exception e) {
        // May be cache is being shutdown
        if (logger.isDebugEnabled()) {
          logger.debug("Failed to remove CQ from the base region. CqName : {}", cqName);
        }
      }

      if (isServer()) {
        removeFromBaseRegionToCqNameMap(cQuery.getRegionName(), serverCqName);
      }

      LocalRegion baseRegion = cQuery.getCqBaseRegion();
      if (baseRegion.getFilterProfile().getCqCount() <= 0) {
        if (logger.isDebugEnabled()) {
          logger.debug(
              "Should update the profile for this partitioned region {} for not requiring old value",
              baseRegion);
        }
      }
    } catch (CqClosedException cce) {
      throw new CqException(cce.getMessage());
    } finally {
      removeFromMatchingCqMap(cQuery);
    }
  }

  @Override
  public void closeAllCqs(boolean clientInitiated) {
    closeAllCqs(clientInitiated, getAllCqs());
  }

  /**
   * Close all CQs executing in this VM, and release resources associated with executing CQs.
   * CqQuerys created by other VMs are unaffected.
   */
  private void closeAllCqs(boolean clientInitiated, Collection<? extends InternalCqQuery> cqs) {
    closeAllCqs(clientInitiated, cqs, cache.keepDurableSubscriptionsAlive());
  }

  @Override
  public void closeAllCqs(boolean clientInitiated, Collection<? extends InternalCqQuery> cqs,
      boolean keepAlive) {

    if (cqs != null) {
      String cqName = null;
      if (logger.isDebugEnabled()) {
        logger.debug("Closing all CQs, number of CQ to be closed : {}", cqs.size());
      }
      for (InternalCqQuery cQuery : cqs) {
        try {
          cqName = cQuery.getName();

          if (isServer()) {
            cQuery.close(false);
          } else {
            if (clientInitiated) {
              cQuery.close(true);
            } else {
              if (!isServer() && cQuery.isDurable() && keepAlive) {
                logger.warn("Not sending CQ close to the server as it is a durable CQ");
                cQuery.close(false);
              } else {
                cQuery.close(true);
              }
            }
          }
        } catch (QueryException | CqClosedException e) {
          if (!isRunning()) {
            // Not cache shutdown
            logger
                .warn("Failed to close CQ {} : {}",
                    cqName, e.getMessage());
          }
          if (logger.isDebugEnabled()) {
            logger.debug(e.getMessage(), e);
          }
        }
      }
    }
  }

  @Override
  public CqServiceStatistics getCqStatistics() {
    return cqServiceStats;
  }

  @Override
  public void closeClientCqs(ClientProxyMembershipID clientProxyId) throws CqException {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    if (isDebugEnabled) {
      logger.debug("Closing Client CQs for the client: {}", clientProxyId);
    }
    List<ServerCQ> cqs = getAllClientCqs(clientProxyId);
    for (ServerCQ cq : cqs) {
      CqQueryImpl cQuery = (CqQueryImpl) cq;
      try {
        cQuery.close(false);
      } catch (QueryException | CqClosedException e) {
        if (isDebugEnabled) {
          logger.debug("Failed to close the CQ, CqName : {} Error : {}", cQuery.getName(),
              e.getMessage());
        }
      }
    }
  }

  @Override
  public List<ServerCQ> getAllClientCqs(ClientProxyMembershipID clientProxyId) {
    Collection<? extends InternalCqQuery> cqs = getAllCqs();
    ArrayList<ServerCQ> clientCqs = new ArrayList<>();

    for (InternalCqQuery cq : cqs) {
      ServerCQImpl cQuery = (ServerCQImpl) cq;
      ClientProxyMembershipID id = cQuery.getClientProxyId();
      if (id != null && id.equals(clientProxyId)) {
        clientCqs.add(cQuery);
      }
    }
    return clientCqs;
  }

  @Override
  public List<String> getAllDurableClientCqs(ClientProxyMembershipID clientProxyId)
      throws CqException {
    if (clientProxyId == null) {
      throw new CqException(
          String.format("Unable to retrieve durable CQs for client proxy id %s", clientProxyId));
    }
    List<ServerCQ> cqs = getAllClientCqs(clientProxyId);
    ArrayList<String> durableClientCqs = new ArrayList<>();

    for (ServerCQ cq : cqs) {
      ServerCQImpl cQuery = (ServerCQImpl) cq;
      if (cQuery != null && cQuery.isDurable()) {
        ClientProxyMembershipID id = cQuery.getClientProxyId();
        if (id != null && id.equals(clientProxyId)) {
          durableClientCqs.add(cQuery.getName());
        }
      }
    }
    return durableClientCqs;
  }

  /**
   * Server side method. Closes non-durable CQs for the given client proxy id.
   */
  @Override
  public void closeNonDurableClientCqs(ClientProxyMembershipID clientProxyId) throws CqException {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    if (isDebugEnabled) {
      logger.debug("Closing Client CQs for the client: {}", clientProxyId);
    }
    List<ServerCQ> cqs = getAllClientCqs(clientProxyId);
    for (ServerCQ cq : cqs) {
      ServerCQImpl cQuery = (ServerCQImpl) cq;
      try {
        if (!cQuery.isDurable()) {
          cQuery.close(false);
        }
      } catch (QueryException | CqClosedException e) {
        if (isDebugEnabled) {
          logger.debug("Failed to close the CQ, CqName : {} Error : {}", cQuery.getName(),
              e.getMessage());
        }
      }
    }
  }

  /**
   * Is the CQ service in a cache server environment
   *
   * @return true if cache server, false otherwise
   */
  public boolean isServer() {
    return !cache.getCacheServers().isEmpty();
  }

  /**
   * Cleans up the CqService.
   */
  @Override
  public void close() {
    if (logger.isDebugEnabled()) {
      logger.debug("Closing CqService. {}", this);
    }
    // Close All the CQs.
    // Need to take care when Clients are still connected...
    closeAllCqs(false);
    isRunning = false;
  }

  @Override
  public boolean isRunning() {
    return isRunning;
  }

  @Override
  public void start() {
    isRunning = true;
  }

  /**
   * @return Returns the serverCqName.
   */
  @Override
  public String constructServerCqName(String cqName, ClientProxyMembershipID clientProxyId) {
    ConcurrentHashMap<ClientProxyMembershipID, String> cache =
        JavaWorkarounds.computeIfAbsent(serverCqNameCache, cqName,
            key -> new ConcurrentHashMap<>());

    String cName = cache.get(clientProxyId);
    if (null == cName) {
      final StringBuilder sb = new StringBuilder(cqName).append("__");
      if (clientProxyId.isDurable()) {
        sb.append(clientProxyId.getDurableId());
      } else {
        sb.append(clientProxyId.getDSMembership());
      }
      cName = sb.toString();
      cache.put(clientProxyId, cName);
    }

    return cName;
  }

  private void removeFromCacheForServerToConstructedCQName(final String cqName,
      ClientProxyMembershipID clientProxyMembershipID) {
    ConcurrentHashMap<ClientProxyMembershipID, String> cache = serverCqNameCache.get(cqName);
    if (cache != null) {
      cache.remove(clientProxyMembershipID);
      if (cache.size() == 0) {
        serverCqNameCache.remove(cqName);
      }
    }
  }

  /**
   * Checks if CQ with the given name already exists.
   *
   * @param cqName name of the CQ.
   *
   * @return true if exists else false.
   */
  private synchronized boolean isCqExists(String cqName) {
    HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
    return cqMap.containsKey(cqName);
  }

  /**
   * Generates a name for CQ. Checks if CQ with that name already exists if so generates a new
   * cqName.
   */
  private synchronized String generateCqName() {
    while (true) {
      String cqName = CQ_NAME_PREFIX + (cqId++);
      if (!isCqExists(cqName)) {
        return cqName;
      }
    }
  }

  @Override
  public void dispatchCqListeners(HashMap<String, MessageType> cqs, MessageType messageType,
      Object key,
      Object value, byte[] delta, QueueManager qManager, EventID eventId) {
    Object[] fullValue = new Object[1];
    Iterator<Map.Entry<String, MessageType>> iter = cqs.entrySet().iterator();
    String cqName = null;
    final boolean isDebugEnabled = logger.isDebugEnabled();
    while (iter.hasNext()) {
      try {
        Map.Entry<String, MessageType> entry = iter.next();
        cqName = entry.getKey();
        ClientCQImpl cQuery = (ClientCQImpl) getCq(cqName);

        if (cQuery == null || (!cQuery.isRunning() && cQuery.getQueuedEvents() == null)) {
          if (isDebugEnabled) {
            logger.debug("Unable to invoke CqListener, {}, CqName : {}",
                ((cQuery == null) ? "CQ not found" : " CQ is Not running"), cqName);
          }
          continue;
        }

        MessageType cqOp = entry.getValue();

        // If Region destroy event, close the cq.
        if (cqOp == DESTROY_REGION) {
          // The close will also invoke the listeners close().
          try {
            cQuery.close(false);
          } catch (Exception ex) {
            // handle?
          }
          continue;
        }

        // Construct CqEvent.
        CqEventImpl cqEvent = new CqEventImpl(cQuery, getOperation(messageType), getOperation(cqOp),
            key, value, delta, qManager, eventId);

        // Update statistics
        cQuery.updateStats(cqEvent);

        // Check if CQ Event needs to be queued.
        if (cQuery.getQueuedEvents() != null) {
          synchronized (cQuery.queuedEventsSynchObject) {
            // Get latest value.
            ConcurrentLinkedQueue<CqEventImpl> queuedEvents = cQuery.getQueuedEvents();
