/*
 * 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;

import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.ANY_INIT;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;

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

import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.CacheEvent;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.SerializedCacheValue;
import org.apache.geode.cache.query.internal.CqStateImpl;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.cq.CqServiceProvider;
import org.apache.geode.cache.query.internal.cq.ServerCQ;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionAdvisee;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.distributed.internal.HighPriorityDistributionMessage;
import org.apache.geode.distributed.internal.MessageWithReply;
import org.apache.geode.distributed.internal.ReplyMessage;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.ClassLoadUtils;
import org.apache.geode.internal.CopyOnWriteHashSet;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
import org.apache.geode.internal.cache.DistributedPutAllOperation.PutAllEntryData;
import org.apache.geode.internal.cache.DistributedRemoveAllOperation.RemoveAllEntryData;
import org.apache.geode.internal.cache.FilterRoutingInfo.FilterInfo;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
import org.apache.geode.internal.cache.tier.InterestType;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.UnregisterAllInterest;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.offheap.annotations.Unretained;
import org.apache.geode.internal.serialization.DataSerializableFixedID;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.util.concurrent.CopyOnWriteHashMap;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * FilterProfile represents a distributed system member and is used for two purposes: processing
 * client-bound events, and providing information for profile exchanges.
 *
 * FilterProfiles represent client IDs, including durable Queue IDs, with long integers. This
 * reduces the size of routing information when sent over the network.
 *
 * @since GemFire 6.5
 */
public class FilterProfile implements DataSerializableFixedID {
  private static final Logger logger = LogService.getLogger();

  /** enumeration of distributed profile operations */
  enum operationType {
    REGISTER_KEY,
    REGISTER_KEYS,
    REGISTER_PATTERN,
    REGISTER_FILTER,
    UNREGISTER_KEY,
    UNREGISTER_KEYS,
    UNREGISTER_PATTERN,
    UNREGISTER_FILTER,
    CLEAR,
    HAS_CQ,
    REGISTER_CQ,
    CLOSE_CQ,
    STOP_CQ,
    SET_CQ_STATE
  }

  /**
   * these booleans tell whether the associated operationType pertains to CQs or not
   */
  @Immutable
  private static final boolean[] isCQOperation =
      {false, false, false, false, false, false, false, false, false, true, true, true, true, true};

  /**
   * types of interest a client may have<br>
   * NONE - not interested<br>
   * UPDATES - wants update messages<br>
   * INVALIDATES - wants invalidations instead of updates
   */
  public enum interestType {
    NONE, UPDATES, INVALIDATES
  }

  /**
   * The keys in which clients are interested. This is a map keyed on client id, with a HashSet of
   * the interested keys as the values.
   */
  private final CopyOnWriteHashMap<Object, Set> keysOfInterest = new CopyOnWriteHashMap<>();

  private final CopyOnWriteHashMap<Object, Set> keysOfInterestInv = new CopyOnWriteHashMap<>();

  /**
   * The patterns in which clients are interested. This is a map keyed on client id, with a HashMap
   * (key name to compiled pattern) as the values.
   */
  private final CopyOnWriteHashMap<Object, Map<Object, Pattern>> patternsOfInterest =
      new CopyOnWriteHashMap<>();

  private final CopyOnWriteHashMap<Object, Map<Object, Pattern>> patternsOfInterestInv =
      new CopyOnWriteHashMap<>();

  /**
   * The filtering classes in which clients are interested. This is a map keyed on client id, with a
   * HashMap (key name to {@link InterestFilter}) as the values.
   */
  private final CopyOnWriteHashMap<Object, Map> filtersOfInterest = new CopyOnWriteHashMap<>();

  private final CopyOnWriteHashMap<Object, Map> filtersOfInterestInv = new CopyOnWriteHashMap<>();

  /**
   * Set of clients that we have ALL_KEYS interest for and who want updates
   */
  private final CopyOnWriteHashSet<Long> allKeyClients = new CopyOnWriteHashSet<>();

  /**
   * Set of clients that we have ALL_KEYS interest for and who want invalidations
   */
  private final CopyOnWriteHashSet<Long> allKeyClientsInv = new CopyOnWriteHashSet<>();

  /**
   * The region associated with this profile
   */
  transient LocalRegion region;

  /**
   * Whether this is a local profile or one that describes another process
   */
  private transient boolean isLocalProfile;

  /** Currently installed CQ count on the region */
  AtomicInteger cqCount;

  /** CQs that are registered on the remote node **/
  private final CopyOnWriteHashMap<String, ServerCQ> cqs = new CopyOnWriteHashMap<>();

  /* the ID of the member that this profile describes */
  private DistributedMember memberID;

  /**
   * since client identifiers can be long, we use a mapping for on-wire operations
   */
  IDMap clientMap;

  /**
   * since CQ identifiers can be long, we use a mapping for on-wire operations
   */
  IDMap cqMap;

  private final Object interestListLock = new Object();

  /**
   * Queues the Filter Profile messages that are received during profile exchange.
   */
  private final Map<InternalDistributedMember, LinkedList<OperationMessage>> filterProfileMsgQueue =
      new HashMap<>();

  public FilterProfile() {
    cqCount = new AtomicInteger();
  }

  /**
   * used for instantiation of a profile associated with a region and not describing region filters
   * in a different process. Do not use this method when instantiating profiles to store in
   * distribution advisor profiles.
   */
  public FilterProfile(LocalRegion r) {
    this(r, r.getMyId(), r.getGemFireCache().getCacheServers().size() > 0);
  }

  /**
   * used for instantiation of a profile associated with a region and not describing region filters
   * in a different process. Do not use this method when instantiating profiles to store in
   * distribution advisor profiles.
   */
  public FilterProfile(LocalRegion r, DistributedMember member, boolean hasCacheServer) {
    region = r;
    isLocalProfile = true;
    memberID = member;
    cqCount = new AtomicInteger();
    clientMap = new IDMap();
    cqMap = new IDMap();
    localProfile.hasCacheServer = hasCacheServer;
  }

  public static boolean isCqOp(operationType opType) {
    return isCQOperation[opType.ordinal()];
  }

  /** return a set containing all clients that have registered interest for values */
  private Set getAllClientsWithInterest() {
    Set clientsWithInterest = new HashSet(getAllKeyClients());
    clientsWithInterest.addAll(getPatternsOfInterest().keySet());
    clientsWithInterest.addAll(getFiltersOfInterest().keySet());
    clientsWithInterest.addAll(getKeysOfInterest().keySet());
    return clientsWithInterest;
  }

  /** return a set containing all clients that have registered interest for invalidations */
  private Set getAllClientsWithInterestInv() {
    Set clientsWithInterestInv = new HashSet(getAllKeyClientsInv());
    clientsWithInterestInv.addAll(getPatternsOfInterestInv().keySet());
    clientsWithInterestInv.addAll(getFiltersOfInterestInv().keySet());
    clientsWithInterestInv.addAll(getKeysOfInterestInv().keySet());
    return clientsWithInterestInv;
  }

  /**
   * Registers interest in the input region name and key
   *
   * @param inputClientID The identity of the interested client
   * @param interest The key in which to register interest
   * @param typeOfInterest the type of interest the client is registering
   * @param updatesAsInvalidates whether the client just wants invalidations
   * @return a set of the keys that were registered, which may be null
   */
  public Set registerClientInterest(Object inputClientID, Object interest,
      final @NotNull InterestType typeOfInterest,
      boolean updatesAsInvalidates) {
    Set keysRegistered = new HashSet();
    operationType opType = null;

    Long clientID = getClientIDForMaps(inputClientID);
    synchronized (interestListLock) {
      switch (typeOfInterest) {
        case KEY:
          opType = operationType.REGISTER_KEY;
          Map<Object, Set> koi =
              updatesAsInvalidates ? getKeysOfInterestInv() : getKeysOfInterest();
          registerKeyInMap(interest, keysRegistered, clientID, koi);
          break;
        case REGULAR_EXPRESSION:
          opType = operationType.REGISTER_PATTERN;
          if (interest.equals(".*")) {
            Set akc = updatesAsInvalidates ? getAllKeyClientsInv() : getAllKeyClients();
            if (akc.add(clientID)) {
              keysRegistered.add(interest);
            }
          } else {
            Map<Object, Map<Object, Pattern>> pats =
                updatesAsInvalidates ? getPatternsOfInterestInv() : getPatternsOfInterest();
            registerPatternInMap(interest, keysRegistered, clientID, pats);
          }
          break;
        case FILTER_CLASS: {
          opType = operationType.REGISTER_FILTER;
          Map<Object, Map> filts =
              updatesAsInvalidates ? getFiltersOfInterestInv() : getFiltersOfInterest();
          registerFilterClassInMap(interest, clientID, filts);
          break;
        }

        default:
          throw new InternalGemFireError(
              "Unknown interest type");
      } // switch
      if (isLocalProfile && opType != null) {
        sendProfileOperation(clientID, opType, interest, updatesAsInvalidates);
      }
    } // synchronized
    return keysRegistered;
  }

  private void registerFilterClassInMap(Object interest, Long clientID, Map<Object, Map> filts) {
    // get instance of the filter
    Class filterClass;
    InterestFilter filter;
    try {
      filterClass = ClassLoadUtils.classFromName((String) interest);
      filter = (InterestFilter) filterClass.newInstance();
    } catch (ClassNotFoundException cnfe) {
      throw new RuntimeException(String.format("Class %s not found in classpath.",
          interest), cnfe);
    } catch (Exception e) {
      throw new RuntimeException(String.format("Class %s could not be instantiated.",
          interest), e);
    }
    Map interestMap = filts.get(clientID);
    if (interestMap == null) {
      interestMap = new CopyOnWriteHashMap();
      filts.put(clientID, interestMap);
    }
    interestMap.put(interest, filter);
  }

  private void registerPatternInMap(Object interest, Set keysRegistered, Long clientID,
      Map<Object, Map<Object, Pattern>> pats) {
    Pattern pattern = Pattern.compile((String) interest);
    Map<Object, Pattern> interestMap = pats.get(clientID);
    if (interestMap == null) {
      interestMap = new CopyOnWriteHashMap<>();
      pats.put(clientID, interestMap);
    }
    Pattern oldPattern = interestMap.put(interest, pattern);
    if (oldPattern == null) {
      // If the pattern didn't exist, add it to the set of keys to pass to any listeners.
      keysRegistered.add(interest);
    }
  }

  private void registerKeyInMap(Object interest, Set keysRegistered, Long clientID,
      Map<Object, Set> koi) {
    Set interestList = koi.get(clientID);
    if (interestList == null) {
      interestList = new CopyOnWriteHashSet();
      koi.put(clientID, interestList);
    }
    interestList.add(interest);
    keysRegistered.add(interest);
  }

  /**
   * Unregisters a client's interest
   *
   * @param inputClientID The identity of the client that is losing interest
   * @param interest The key in which to unregister interest
   * @param interestType the type of uninterest
   * @return the keys unregistered, which may be null
   */
  public Set unregisterClientInterest(Object inputClientID, Object interest,
      final @NotNull InterestType interestType) {
    Long clientID;
    if (inputClientID instanceof Long) {
      clientID = (Long) inputClientID;
    } else {
      Map<Object, Long> cids = clientMap.realIDs; // read
      clientID = cids.get(inputClientID);
      if (clientID == null) {
        if (logger.isDebugEnabled()) {
          logger.debug(
              "region profile unable to find '{}' for unregistration.  Probably means there is no durable queue.",
              inputClientID);
        }
        return null;
      }
    }
    Set keysUnregistered = new HashSet();
    operationType opType = null;
    synchronized (interestListLock) {
      switch (interestType) {
        case KEY: {
          opType = operationType.UNREGISTER_KEY;
          unregisterClientKeys(inputClientID, interest, clientID, keysUnregistered);
          break;
        }
        case REGULAR_EXPRESSION: {
          opType = operationType.UNREGISTER_PATTERN;
          unregisterClientPattern(interest, clientID, keysUnregistered);
          break;
        }
        case FILTER_CLASS: {
          opType = operationType.UNREGISTER_FILTER;
          unregisterClientFilterClass(interest, clientID);
          break;
        }
        default:
          throw new InternalGemFireError(
              "bad interest type");
      }
      if (region != null && isLocalProfile) {
        sendProfileOperation(clientID, opType, interest, false);
      }
    } // synchronized
    return keysUnregistered;
  }

  private void unregisterClientFilterClass(Object interest, Long clientID) {
    if (interest == UnregisterAllInterest.singleton()) {
      if (getFiltersOfInterest().get(clientID) != null) {
        getFiltersOfInterest().remove(clientID);
      }
      if (getFiltersOfInterestInv().get(clientID) != null) {
        getFiltersOfInterestInv().remove(clientID);
      }
      return;
    }
    Map interestMap = getFiltersOfInterest().get(clientID);
    if (interestMap != null) {
      interestMap.remove(interest);
      if (interestMap.isEmpty()) {
        getFiltersOfInterest().remove(clientID);
      }
    }
    interestMap = getFiltersOfInterestInv().get(clientID);
    if (interestMap != null) {
      interestMap.remove(interest);
      if (interestMap.isEmpty()) {
        getFiltersOfInterestInv().remove(clientID);
      }
    }
  }

  private void unregisterClientPattern(Object interest, Long clientID, Set keysUnregistered) {
    if (interest instanceof String && interest.equals(".*")) { // ALL_KEYS
      unregisterAllKeys(interest, clientID, keysUnregistered);
      return;
    }
    if (interest == UnregisterAllInterest.singleton()) {
      unregisterClientIDFromMap(clientID, getPatternsOfInterest(), keysUnregistered);
      unregisterClientIDFromMap(clientID, getPatternsOfInterestInv(), keysUnregistered);
      if (getAllKeyClients().remove(clientID)) {
        keysUnregistered.add(".*");
      }
      if (getAllKeyClientsInv().remove(clientID)) {
        keysUnregistered.add(".*");
      }
    } else {
      unregisterPatternFromMap(getPatternsOfInterest(), interest, clientID, keysUnregistered);
      unregisterPatternFromMap(getPatternsOfInterestInv(), interest, clientID, keysUnregistered);
    }
  }

  private void unregisterPatternFromMap(Map<Object, Map<Object, Pattern>> map, Object interest,
      Long clientID, Set keysUnregistered) {
    Map interestMap = map.get(clientID);
    if (interestMap != null) {
      Object obj = interestMap.remove(interest);
      if (obj != null) {
        keysUnregistered.add(interest);
      }
      if (interestMap.isEmpty()) {
        map.remove(clientID);
      }
    }
  }

  private void unregisterClientIDFromMap(Long clientID, Map interestMap, Set keysUnregistered) {
    if (interestMap.get(clientID) != null) {
      Map removed = (Map) interestMap.remove(clientID);
      if (removed != null) {
        keysUnregistered.addAll(removed.keySet());
      }
    }
  }

  private void unregisterAllKeys(Object interest, Long clientID, Set keysUnregistered) {
    if (getAllKeyClients().remove(clientID)) {
      keysUnregistered.add(interest);
    }
    if (getAllKeyClientsInv().remove(clientID)) {
      keysUnregistered.add(interest);
    }
  }

  private void unregisterClientKeys(Object inputClientID, Object interest, Long clientID,
      Set keysUnregistered) {
    if (interest == UnregisterAllInterest.singleton()) {
      clearInterestFor(inputClientID);
      return;
    }
    unregisterKeyFromMap(getKeysOfInterest(), interest, clientID, keysUnregistered);
    unregisterKeyFromMap(getKeysOfInterestInv(), interest, clientID, keysUnregistered);
    return;
  }

  private void unregisterKeyFromMap(Map<Object, Set> map, Object interest, Long clientID,
      Set keysUnregistered) {
    Set interestList = map.get(clientID);
    if (interestList != null) {
      boolean removed = interestList.remove(interest);
      if (removed) {
        keysUnregistered.add(interest);
      }
      if (interestList.isEmpty()) {
        map.remove(clientID);
      }
    }
  }

  /**
   * Registers interest in a set of keys for a client
   *
   * @param keys The list of keys in which to register interest
   * @param updatesAsInvalidates whether to send invalidations instead of updates
   * @return the registered keys
   */
  public Set registerClientInterestList(Object inputClientID, List keys,
      boolean updatesAsInvalidates) {
    Long clientID = getClientIDForMaps(inputClientID);
    Set keysRegistered = new HashSet(keys);
    synchronized (interestListLock) {
      Map<Object, Set> koi = updatesAsInvalidates ? getKeysOfInterestInv() : getKeysOfInterest();
      CopyOnWriteHashSet interestList = (CopyOnWriteHashSet) koi.get(clientID);
      if (interestList == null) {
        interestList = new CopyOnWriteHashSet();
        koi.put(clientID, interestList);
      } else {
        // Get the list of keys that will be registered new, not already registered.
        keysRegistered.removeAll(interestList.getSnapshot());
      }
      interestList.addAll(keys);

      if (region != null && isLocalProfile) {
        sendProfileOperation(clientID, operationType.REGISTER_KEYS, keys, updatesAsInvalidates);
      }
    } // synchronized
    return keysRegistered;
  }

  /**
   * Unregisters interest in given keys for the given client
   *
   * @param inputClientID The fully-qualified name of the region in which to unregister interest
   * @param keys The list of keys in which to unregister interest
   * @return the unregistered keys
   */
  public Set unregisterClientInterestList(Object inputClientID, List keys) {
    Long clientID = getClientIDForMaps(inputClientID);
    Set keysUnregistered = new HashSet(keys);
    Set keysNotUnregistered = new HashSet(keys);
    synchronized (interestListLock) {
      CopyOnWriteHashSet interestList = (CopyOnWriteHashSet) getKeysOfInterest().get(clientID);
      if (interestList != null) {
        // Get the list of keys that are not registered but in unregister set.
        keysNotUnregistered.removeAll(interestList.getSnapshot());
        interestList.removeAll(keys);

        if (interestList.isEmpty()) {
          getKeysOfInterest().remove(clientID);
        }
      }
      interestList = (CopyOnWriteHashSet) getKeysOfInterestInv().get(clientID);
      if (interestList != null) {
        keysNotUnregistered.removeAll(interestList.getSnapshot());
        interestList.removeAll(keys);

        if (interestList.isEmpty()) {
          getKeysOfInterestInv().remove(clientID);
        }
      }

      if (region != null && isLocalProfile) {
        sendProfileOperation(clientID, operationType.UNREGISTER_KEYS, keys, false);
      }
    } // synchronized
    // Get the keys that are not unregistered.
    keysUnregistered.removeAll(keysNotUnregistered);
    return keysUnregistered;
  }

  public Set getKeysOfInterestFor(Object inputClientID) {
    Long clientID = getClientIDForMaps(inputClientID);
    Set keys1 = getKeysOfInterest().get(clientID);
    Set keys2 = getKeysOfInterestInv().get(clientID);
    if (keys1 == null) {
      if (keys2 == null) {
        return null;
      }
      return Collections.unmodifiableSet(keys2);
    } else if (keys2 == null) {
      return Collections.unmodifiableSet(keys1);
    } else {
      Set result = new HashSet(keys1);
      result.addAll(keys2);
      return Collections.unmodifiableSet(result);
    }
  }

  public Map<String, Pattern> getPatternsOfInterestFor(Object inputClientID) {
    Long clientID = getClientIDForMaps(inputClientID);
    Map patterns1 = getPatternsOfInterest().get(clientID);
    Map patterns2 = getPatternsOfInterestInv().get(clientID);
    if (patterns1 == null) {
      if (patterns2 == null) {
        return null;
      }
      return Collections.unmodifiableMap(patterns2);
    } else if (patterns2 == null) {
      return Collections.unmodifiableMap(patterns1);
    } else {
      Map<String, Pattern> result = new HashMap(patterns1);
      result.putAll(patterns2);
      return Collections.unmodifiableMap(result);
    }
  }

  public boolean hasKeysOfInterestFor(Object inputClientID, boolean wantInvalidations) {
    Long clientID = getClientIDForMaps(inputClientID);
    if (wantInvalidations) {
      return getKeysOfInterestInv().containsKey(clientID);
    }
    return getKeysOfInterest().containsKey(clientID);
  }

  public boolean hasAllKeysInterestFor(Object inputClientID) {
    Long clientID = getClientIDForMaps(inputClientID);
    return hasAllKeysInterestFor(clientID, false) || hasAllKeysInterestFor(clientID, true);
  }

  public boolean hasAllKeysInterestFor(Object inputClientID, boolean wantInvalidations) {
    Long clientID = getClientIDForMaps(inputClientID);
    if (wantInvalidations) {
      return getAllKeyClientsInv().contains(clientID);
    }
    return getAllKeyClients().contains(clientID);
  }

  public boolean hasRegexInterestFor(Object inputClientID, boolean wantInvalidations) {
    Long clientID = getClientIDForMaps(inputClientID);
    if (wantInvalidations) {
      return (getPatternsOfInterestInv().containsKey(clientID));
    }
    return (getPatternsOfInterest().containsKey(clientID));
  }

  public boolean hasFilterInterestFor(Object inputClientID, boolean wantInvalidations) {
    Long clientID = getClientIDForMaps(inputClientID);
    if (wantInvalidations) {
      return getFiltersOfInterestInv().containsKey(clientID);
    }
    return getFiltersOfInterest().containsKey(clientID);
  }


  /** determines whether there is any remaining interest for the given identifier */
  public boolean hasInterestFor(Object inputClientID) {
    Long clientID;
    if (inputClientID instanceof Long) {
      clientID = (Long) inputClientID;
    } else {
      Map<Object, Long> cids = clientMap.realIDs; // read
      clientID = cids.get(inputClientID);
      if (clientID == null) {
        return false;
      }
    }
    return hasAllKeysInterestFor(clientID, true) || hasKeysOfInterestFor(clientID, true)
        || hasRegexInterestFor(clientID, true) || hasFilterInterestFor(clientID, true)
        || hasKeysOfInterestFor(clientID, false) || hasRegexInterestFor(clientID, false)
        || hasAllKeysInterestFor(clientID, false)
        || hasFilterInterestFor(clientID, false);
  }

  /*
   * Returns whether this interest list has any keys, patterns or filters of interest. It answers
   * the question: Are any clients being notified because of this interest list? @return whether
   * this interest list has any keys, patterns or filters of interest
   */
  public boolean hasInterest() {
    return (!getAllKeyClients().isEmpty()) || (!getAllKeyClientsInv().isEmpty())
        || (!getKeysOfInterest().isEmpty()) || (!getPatternsOfInterest().isEmpty())
        || (!getFiltersOfInterest().isEmpty()) || (!getKeysOfInterestInv().isEmpty())
        || (!getPatternsOfInterestInv().isEmpty())
        || (!getFiltersOfInterestInv().isEmpty());
  }

  /** removes all interest for the given identifier */
  public void clearInterestFor(Object inputClientID) {
    Long clientID;
    if (inputClientID instanceof Long) {
      clientID = (Long) inputClientID;
    } else {
      Map<Object, Long> cids = clientMap.realIDs;
      clientID = cids.get(inputClientID);
      if (clientID == null) {
        // haven't seen this client yet
        return;
      }
    }
    synchronized (interestListLock) {
      {
        Set akc = getAllKeyClients();
        akc.remove(clientID);
      }
      {
        Set akci = getAllKeyClientsInv();
        akci.remove(clientID);
      }
      {
        Map<Object, Set> keys = getKeysOfInterest();
        keys.remove(clientID);
      }
      {
        Map<Object, Set> keys = getKeysOfInterestInv();
        keys.remove(clientID);
      }
      {
        Map<Object, Map<Object, Pattern>> pats = getPatternsOfInterest();
        pats.remove(clientID);
      }
      {
        Map<Object, Map<Object, Pattern>> pats = getPatternsOfInterestInv();
        pats.remove(clientID);
      }
      {
        Map<Object, Map> filters = getFiltersOfInterest();
        filters.remove(clientID);
      }
      {
        Map<Object, Map> filters = getFiltersOfInterestInv();
        filters.remove(clientID);
      }
      if (clientMap != null) {
        clientMap.removeIDMapping(clientID);
      }
      if (region != null && isLocalProfile) {
        sendProfileOperation(clientID, operationType.CLEAR, null, false);
      }
    }
  }

  /**
   * Obtains the number of CQs registered on the region. Assumption: CQs are not duplicated among
   * clients.
   *
   * @return int currently registered CQs of the region
   */
  public int getCqCount() {
    return cqCount.get();
  }

  public void incCqCount() {
    cqCount.getAndIncrement();
  }

  public void decCqCount() {
    cqCount.decrementAndGet();
  }

  /**
   * Returns the CQs registered on this region.
   */
  public Map getCqMap() {
    return cqs;
  }

  /**
   * does this profile contain any continuous queries?
   */
  public boolean hasCQs() {
    return cqCount.get() > 0;
  }

  public ServerCQ getCq(String cqName) {
    return cqs.get(cqName);
  }

  public void registerCq(ServerCQ cq) {
    ensureCqID(cq);
    if (logger.isDebugEnabled()) {
      logger.debug("Adding CQ {} to this members FilterProfile.", cq.getServerCqName());
    }
    cqs.put(cq.getServerCqName(), cq);
    incCqCount();

    // cq.setFilterID(cqMap.getWireID(cq.getServerCqName()));
    sendCQProfileOperation(operationType.REGISTER_CQ, cq);
  }

  public void stopCq(ServerCQ cq) {
    ensureCqID(cq);
    if (logger.isDebugEnabled()) {
      logger.debug("Stopping CQ {} on this members FilterProfile.", cq.getServerCqName());
    }
    sendCQProfileOperation(operationType.STOP_CQ, cq);
  }

  public String generateCqName(String serverCqName) {
    // Set unique CQ Name with respect to the senders filterProfile.
    // To avoid any conflict with the CQ names while maintained in
    // the CqService.
    // The CQ will be identified in the remote node using its
    // serverCqName.
    return (serverCqName + hashCode());
  }

  void processRegisterCq(String serverCqName, ServerCQ ServerCQ, boolean addToCqMap) {
    processRegisterCq(serverCqName, ServerCQ, addToCqMap, GemFireCacheImpl.getInstance());
  }


  /**
   * adds a new CQ to this profile during a delta operation or deserialization
   *
   * @param serverCqName the query objects' name
   * @param ServerCQ the new query object
   * @param addToCqMap whether to add the query to this.cqs
   */
  void processRegisterCq(String serverCqName, ServerCQ ServerCQ, boolean addToCqMap,
      GemFireCacheImpl cache) {
    if (cache == null) {
      logger.info("Error while initializing the CQs with FilterProfile for CQ " + serverCqName
          + ", Error : Cache has been closed.");
      return;
    }
    ServerCQ cq = ServerCQ;
    try {
      CqService cqService = cache.getCqService();
      cqService.start();
      cq.setCqService(cqService);
      CqStateImpl cqState = (CqStateImpl) cq.getState();
      cq.setName(generateCqName(serverCqName));
      cq.registerCq(null, null, cqState.getState());
    } catch (Exception ex) {
      logger.info("Error while initializing the CQs with FilterProfile for CQ {}, Error : {}",
          serverCqName, ex.getMessage(), ex);

    }
    if (logger.isDebugEnabled()) {
      logger.debug("Adding CQ to remote members FilterProfile using name: {}", serverCqName);
    }

    // The region's FilterProfile is accessed through CQ reference as the
    // region is not set on the FilterProfile created for the peer nodes.
    if (cq.getCqBaseRegion() != null) {
      if (addToCqMap) {
        cqs.put(serverCqName, cq);
      }

      FilterProfile pf = cq.getCqBaseRegion().getFilterProfile();
      if (pf != null) {
        pf.incCqCount();
      }
    }
  }

  public void processCloseCq(String serverCqName) {
    ServerCQ cq = cqs.get(serverCqName);
    if (cq != null) {
      try {
        cq.close(false);
      } catch (Exception ex) {
        if (logger.isDebugEnabled()) {
          logger.debug(
              "Unable to close the CQ with the filterProfile, on region {} for CQ {}, Error : {}",
              region.getFullPath(), serverCqName, ex.getMessage(), ex);
        }
      }
      cqs.remove(serverCqName);
      cq.getCqBaseRegion().getFilterProfile().decCqCount();
    }
  }

  public void processSetCqState(String serverCqName, ServerCQ ServerCQ) {
    ServerCQ cq = cqs.get(serverCqName);
    if (cq != null) {
      CqStateImpl cqState = (CqStateImpl) ServerCQ.getState();
      cq.setCqState(cqState.getState());
    }
  }

  public void processStopCq(String serverCqName) {
    ServerCQ cq = cqs.get(serverCqName);
    if (cq != null) {
      try {
        cq.stop();
      } catch (Exception ex) {
        if (logger.isDebugEnabled()) {
          logger.debug(
              "Unable to stop the CQ with the filterProfile, on region {} for CQ {}, Error : {}",
              region.getFullPath(), serverCqName, ex.getMessage(), ex);
        }
      }
    }
  }

  public void setCqState(ServerCQ cq) {
    // This could be when CQ is stopped and restarted.
    ensureCqID(cq);
    if (logger.isDebugEnabled()) {
      logger.debug("Stopping CQ {} on this members FilterProfile.", cq.getServerCqName());
