/*
 * 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 static org.apache.geode.internal.statistics.StatisticsClockFactory.disabledClock;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.logging.log4j.Logger;

import org.apache.geode.DataSerializer;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.CacheCallback;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.CacheStatistics;
import org.apache.geode.cache.CacheWriter;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.CustomExpiry;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskWriteAttributes;
import org.apache.geode.cache.EntryExistsException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.EvictionAttributesMutator;
import org.apache.geode.cache.ExpirationAction;
import org.apache.geode.cache.ExpirationAttributes;
import org.apache.geode.cache.MembershipAttributes;
import org.apache.geode.cache.MirrorType;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAccessException;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionMembershipListener;
import org.apache.geode.cache.RegionService;
import org.apache.geode.cache.RoleException;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.StatisticsDisabledException;
import org.apache.geode.cache.SubscriptionAttributes;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.query.FunctionDomainException;
import org.apache.geode.cache.query.NameResolutionException;
import org.apache.geode.cache.query.QueryInvocationTargetException;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.TypeMismatchException;
import org.apache.geode.cache.query.internal.index.IndexManager;
import org.apache.geode.cache.snapshot.RegionSnapshotService;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.compression.Compressor;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DistributionAdvisor;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
import org.apache.geode.internal.cache.extension.Extensible;
import org.apache.geode.internal.cache.extension.ExtensionPoint;
import org.apache.geode.internal.cache.extension.SimpleExtensionPoint;
import org.apache.geode.internal.cache.snapshot.RegionSnapshotServiceImpl;
import org.apache.geode.internal.serialization.DataSerializableFixedID;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.util.ArrayUtils;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.logging.internal.log4j.api.LogWithToString;
import org.apache.geode.pdx.internal.PeerTypeRegistration;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Takes care of RegionAttributes, AttributesMutator, and some no-brainer method implementations.
 */
@SuppressWarnings("deprecation")
public abstract class AbstractRegion implements InternalRegion, AttributesMutator, CacheStatistics,
    DataSerializableFixedID, Extensible<Region<?, ?>>, EvictableRegion, LogWithToString {

  private static final Logger logger = LogService.getLogger();
  private final ReentrantReadWriteLock readWriteLockForCacheLoader = new ReentrantReadWriteLock();
  private final ReentrantReadWriteLock readWriteLockForCacheWriter = new ReentrantReadWriteLock();
  protected final ConcurrentHashMap<RegionEntry, EntryExpiryTask> entryExpiryTasks =
      new ConcurrentHashMap<>();
  /**
   * Identifies the static order in which this region was created in relation to other regions or
   * other instances of this region during the life of this JVM.
   */
  private final int serialNumber;

  /**
   * Used to synchronize WRITES to cacheListeners. Always do copy on write.
   */
  private final Object clSync = new Object();

  // Used to synchronize creation of IndexManager
  private final Object imSync = new Object();

  /**
   * NOTE: care must be taken to never modify the array of cacheListeners in place. Instead allocate
   * a new array and modify it. This field is volatile so that it can be read w/o getting clSync.
   */
  private volatile CacheListener[] cacheListeners;

  /**
   * All access to cacheLoader must be protected by readWriteLockForCacheLoader
   */
  private volatile CacheLoader cacheLoader;

  /**
   * All access to cacheWriter must be protected by readWriteLockForCacheWriter
   */
  private volatile CacheWriter cacheWriter;

  protected int entryIdleTimeout;

  private ExpirationAction entryIdleTimeoutExpirationAction;

  protected CustomExpiry customEntryIdleTimeout;

  protected int entryTimeToLive;

  ExpirationAction entryTimeToLiveExpirationAction;

  protected CustomExpiry customEntryTimeToLive;

  protected int initialCapacity;

  protected Class keyConstraint;

  protected Class valueConstraint;

  protected float loadFactor;

  private DataPolicy dataPolicy;

  protected int regionIdleTimeout;

  private ExpirationAction regionIdleTimeoutExpirationAction;

  protected int regionTimeToLive;

  private ExpirationAction regionTimeToLiveExpirationAction;

  @Immutable
  public static final Scope DEFAULT_SCOPE = Scope.DISTRIBUTED_NO_ACK;

  protected Scope scope = DEFAULT_SCOPE;

  protected boolean statisticsEnabled;

  protected boolean isLockGrantor;

  private boolean mcastEnabled;

  protected int concurrencyLevel;

  private volatile boolean concurrencyChecksEnabled;

  protected boolean earlyAck;

  private final boolean isPdxTypesRegion;

  protected Set<String> gatewaySenderIds;

  protected Set<String> asyncEventQueueIds;

  private Set<String> visibleAsyncEventQueueIds;

  /**
   * This set is always unmodifiable.
   */
  private Set<String> allGatewaySenderIds;

  protected boolean enableSubscriptionConflation;

  protected boolean publisher;

  protected boolean enableAsyncConflation;

  /**
   * True if this region uses off-heap memory; otherwise false (default)
   *
   * @since Geode 1.0
   */
  protected boolean offHeap;

  private boolean cloningEnable = false;

  private DiskWriteAttributes diskWriteAttributes;

  protected File[] diskDirs;
  protected int[] diskSizes;
  protected String diskStoreName;
  protected boolean isDiskSynchronous;
  private boolean indexMaintenanceSynchronous = false;

  protected volatile IndexManager indexManager = null;

  // The ThreadLocal is used to identify if the thread is an
  // index creation thread. This identification helps skip the synchronization
  // block
  // if the value is "REMOVED" token. This prevents the dead lock , in case the
  // lock
  // over the entry is held by any Index Update Thread.
  // This is used to fix Bug # 33336.
  private static final ThreadLocal<Boolean> isIndexCreator = new ThreadLocal<>();

  /** Attributes that define this Region as a PartitionedRegion */
  protected PartitionAttributes partitionAttributes;

  private final EvictionAttributesImpl evictionAttributes;

  /** The membership attributes defining required roles functionality */
  protected MembershipAttributes membershipAttributes;

  /** The subscription attributes defining required roles functionality */
  protected SubscriptionAttributes subscriptionAttributes;

  /** should this region ignore in-progress JTA transactions? */
  protected boolean ignoreJTA;

  private final AtomicLong lastAccessedTime;

  private final AtomicLong lastModifiedTime;

  private static final boolean trackHits =
      !Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "ignoreHits");

  private static final boolean trackMisses =
      !Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "ignoreMisses");

  private final AtomicLong hitCount = new AtomicLong();

  private final AtomicLong missCount = new AtomicLong();

  protected String poolName;

  protected Compressor compressor;

  /**
   * @see #getExtensionPoint()
   * @since GemFire 8.1
   */
  private final ExtensionPoint<Region<?, ?>> extensionPoint =
      new SimpleExtensionPoint<Region<?, ?>>(this, this);

  protected final InternalCache cache;

  private final PoolFinder poolFinder;

  private final StatisticsClock statisticsClock;

  /** Creates a new instance of AbstractRegion */
  protected AbstractRegion(InternalCache cache, RegionAttributes<?, ?> attrs, String regionName,
      InternalRegionArguments internalRegionArgs, PoolFinder poolFinder,
      StatisticsClock statisticsClock) {
    this.poolFinder = poolFinder;
    this.statisticsClock = statisticsClock;

    this.cache = cache;
    serialNumber = DistributionAdvisor.createSerialNumber();
    isPdxTypesRegion = PeerTypeRegistration.REGION_NAME.equals(regionName);
    lastAccessedTime = new AtomicLong(cacheTimeMillis());
    lastModifiedTime = new AtomicLong(lastAccessedTime.get());
    dataPolicy = attrs.getDataPolicy(); // do this one first
    scope = attrs.getScope();

    offHeap = attrs.getOffHeap();

    // fix bug #52033 by invoking setOffHeap now (localMaxMemory may now be the temporary
    // placeholder for off-heap until DistributedSystem is created
    // found non-null PartitionAttributes and offHeap is true so let's setOffHeap on PA now
    PartitionAttributes<?, ?> partitionAttributes1 = attrs.getPartitionAttributes();
    if (offHeap && partitionAttributes1 != null) {
      PartitionAttributesImpl impl = (PartitionAttributesImpl) partitionAttributes1;
      impl.setOffHeap(true);
    }

    evictionAttributes = new EvictionAttributesImpl(attrs.getEvictionAttributes());
    if (attrs.getPartitionAttributes() != null && evictionAttributes.getAlgorithm()
        .isLRUMemory() && attrs.getPartitionAttributes().getLocalMaxMemory() != 0
        && evictionAttributes
            .getMaximum() != attrs.getPartitionAttributes().getLocalMaxMemory()) {
      logger.warn(
          "For region {} with data policy PARTITION, memory LRU eviction attribute maximum has been reset from {}MB to local-max-memory {}MB",
          new Object[] {regionName, evictionAttributes.getMaximum(),
              attrs.getPartitionAttributes().getLocalMaxMemory()});
      evictionAttributes.setMaximum(attrs.getPartitionAttributes().getLocalMaxMemory());
    }

    storeCacheListenersField(attrs.getCacheListeners());
    assignCacheLoader(attrs.getCacheLoader());
    assignCacheWriter(attrs.getCacheWriter());
    regionTimeToLive = attrs.getRegionTimeToLive().getTimeout();
    regionTimeToLiveExpirationAction = attrs.getRegionTimeToLive().getAction();
    setRegionTimeToLiveAtts();
    regionIdleTimeout = attrs.getRegionIdleTimeout().getTimeout();
    regionIdleTimeoutExpirationAction = attrs.getRegionIdleTimeout().getAction();
    setRegionIdleTimeoutAttributes();
    entryTimeToLive = attrs.getEntryTimeToLive().getTimeout();
    entryTimeToLiveExpirationAction = attrs.getEntryTimeToLive().getAction();
    setEntryTimeToLiveAttributes();
    customEntryTimeToLive = attrs.getCustomEntryTimeToLive();
    entryIdleTimeout = attrs.getEntryIdleTimeout().getTimeout();
    entryIdleTimeoutExpirationAction = attrs.getEntryIdleTimeout().getAction();
    setEntryIdleTimeoutAttributes();
    customEntryIdleTimeout = attrs.getCustomEntryIdleTimeout();
    updateEntryExpiryPossible();
    statisticsEnabled = attrs.getStatisticsEnabled();
    ignoreJTA = attrs.getIgnoreJTA();
    isLockGrantor = attrs.isLockGrantor();
    keyConstraint = attrs.getKeyConstraint();
    valueConstraint = attrs.getValueConstraint();
    initialCapacity = attrs.getInitialCapacity();
    loadFactor = attrs.getLoadFactor();
    concurrencyLevel = attrs.getConcurrencyLevel();
    setConcurrencyChecksEnabled(
        attrs.getConcurrencyChecksEnabled() && supportsConcurrencyChecks());
    earlyAck = attrs.getEarlyAck();
    gatewaySenderIds = attrs.getGatewaySenderIds();
    asyncEventQueueIds = attrs.getAsyncEventQueueIds();
    initializeVisibleAsyncEventQueueIds(internalRegionArgs);
    setAllGatewaySenderIds();
    enableSubscriptionConflation = attrs.getEnableSubscriptionConflation();
    publisher = attrs.getPublisher();
    enableAsyncConflation = attrs.getEnableAsyncConflation();
    indexMaintenanceSynchronous = attrs.getIndexMaintenanceSynchronous();
    mcastEnabled = attrs.getMulticastEnabled();
    partitionAttributes = attrs.getPartitionAttributes();
    membershipAttributes = attrs.getMembershipAttributes();
    subscriptionAttributes = attrs.getSubscriptionAttributes();
    cloningEnable = attrs.getCloningEnabled();
    poolName = attrs.getPoolName();
    if (poolName != null) {
      PoolImpl cp = getPool();
      if (cp == null) {
        throw new IllegalStateException(
            String.format("The connection pool %s has not been created",
                poolName));
      }
      cp.attach();
      if (cp.getMultiuserAuthentication() && !getDataPolicy().isEmpty()) {
        throw new IllegalStateException(
            "Region must have empty data-policy " + "when multiuser-authentication is true.");
      }
    }

    diskStoreName = attrs.getDiskStoreName();
    isDiskSynchronous = attrs.isDiskSynchronous();
    if (diskStoreName == null) {
      diskWriteAttributes = attrs.getDiskWriteAttributes();
      isDiskSynchronous = diskWriteAttributes.isSynchronous(); // fixes bug 41313
      diskDirs = attrs.getDiskDirs();
      diskSizes = attrs.getDiskDirSizes();
    }

    compressor = attrs.getCompressor();
    // enable concurrency checks for persistent regions
    if (!attrs.getConcurrencyChecksEnabled() && attrs.getDataPolicy().withPersistence()
        && supportsConcurrencyChecks()) {
      throw new IllegalStateException(
          "Concurrency checks cannot be disabled for regions that use persistence");
    }
  }

  @VisibleForTesting
  AbstractRegion() {
    statisticsClock = disabledClock();
    cache = null;
    serialNumber = 0;
    isPdxTypesRegion = false;
    lastAccessedTime = new AtomicLong(0);
    lastModifiedTime = new AtomicLong(0);
    evictionAttributes = new EvictionAttributesImpl();
    poolFinder = (a) -> null;
  }

  /**
   * configure this region to ignore or not ignore in-progress JTA transactions. Setting this to
   * true will cause cache operations to no longer notice JTA transactions. The default setting is
   * false
   *
   * @deprecated in 5.0 and later releases, use the region attribute ignoreJTA to configure this
   */
  @Deprecated
  public void setIgnoreJTA(boolean ignore) {
    ignoreJTA = ignore;
  }

  @SuppressWarnings("unchecked")
  @Override
  public void create(Object key, Object value)
      throws TimeoutException, EntryExistsException, CacheWriterException {
    create(key, value, null);
  }

  @Override
  public Object destroy(Object key)
      throws TimeoutException, EntryNotFoundException, CacheWriterException {
    return destroy(key, null);
  }

  @Override
  public Object get(Object key) throws CacheLoaderException, TimeoutException {
    return get(key, null, true, null);
  }

  @SuppressWarnings("unchecked")
  @Override
  public Object put(Object key, Object value) throws TimeoutException, CacheWriterException {
    return put(key, value, null);
  }

  @Override
  public Object get(Object key, Object aCallbackArgument)
      throws CacheLoaderException, TimeoutException {
    return get(key, aCallbackArgument, true, null);
  }

  @Override
  public void localDestroyRegion() {
    localDestroyRegion(null);
  }

  /**
   * @param key the key to find
   * @param aCallbackArgument argument for callbacks
   * @param generateCallbacks whether callbacks should be invoked
   * @param clientEvent client-side event, if any (used to pass back version information)
   * @return the value associated with the key
   */
  abstract Object get(Object key, Object aCallbackArgument, boolean generateCallbacks,
      EntryEventImpl clientEvent) throws TimeoutException, CacheLoaderException;

  @Override
  public void localDestroy(Object key) throws EntryNotFoundException {
    localDestroy(key, null);
  }

  @Override
  public void destroyRegion() throws CacheWriterException, TimeoutException {
    destroyRegion(null);
  }

  @Override
  public void invalidate(Object key) throws TimeoutException, EntryNotFoundException {
    invalidate(key, null);
  }

  @Override
  public void localInvalidate(Object key) throws EntryNotFoundException {
    localInvalidate(key, null);
  }

  @Override
  public void localInvalidateRegion() {
    localInvalidateRegion(null);
  }

  @Override
  public void invalidateRegion() throws TimeoutException {
    invalidateRegion(null);
  }

  abstract void basicClear(RegionEventImpl regionEvent);

  @Override
  public void clear() {
    checkReadiness();
    checkForLimitedOrNoAccess();
    RegionEventImpl regionEvent = new RegionEventImpl(this, Operation.REGION_CLEAR, null, false,
        getMyId(), generateEventID());
    basicClear(regionEvent);
  }

  abstract void basicLocalClear(RegionEventImpl rEvent);

  @Override
  public void localClear() {
    checkReadiness();
    checkForNoAccess();
    RegionEventImpl event = new RegionEventImpl(this, Operation.REGION_LOCAL_CLEAR, null, false,
        getMyId(), generateEventID()/* generate EventID */);
    basicLocalClear(event);
  }

  @Override
  public Map getAll(Collection keys) {
    return getAll(keys, null);
  }

  @Override
  public Map getAll(Collection keys, Object aCallbackArgument) {
    if (keys == null) {
      throw new NullPointerException("The collection of keys for getAll cannot be null");
    }
    checkReadiness();
    checkForLimitedOrNoAccess();
    return keys.isEmpty() ? new HashMap() : basicGetAll(keys, aCallbackArgument);
  }

  abstract Map basicGetAll(Collection keys, Object callback);

  protected StringBuilder getStringBuilder() {
    StringBuilder buf = new StringBuilder();
    buf.append(getClass().getName());
    buf.append("[path='").append(getFullPath()).append("';scope=").append(getScope())
        .append("';dataPolicy=").append(getDataPolicy());
    if (getConcurrencyChecksEnabled()) {
      buf.append("; concurrencyChecksEnabled");
    }
    return buf;
  }

  @Override
  public String toString() {
    return getStringBuilder().append(']').toString();
  }

  @Override
  public CacheLoader getCacheLoader() {
    readWriteLockForCacheLoader.readLock().lock();
    try {
      return cacheLoader;
    } finally {
      readWriteLockForCacheLoader.readLock().unlock();
    }
  }

  @Override
  public CacheWriter getCacheWriter() {
    readWriteLockForCacheWriter.readLock().lock();
    try {
      return cacheWriter;
    } finally {
      readWriteLockForCacheWriter.readLock().unlock();
    }
  }

  /**
   * Return a cache loader if this region has one. Note if region's loader is used to implement
   * bridge then null is returned.
   *
   * @since GemFire 5.7
   */
  CacheLoader basicGetLoader() {
    return cacheLoader;
  }

  /**
   * Return a cache writer if this region has one. Note if region's writer is used to implement
   * bridge then null is returned.
   *
   * @since GemFire 5.7
   */
  @Override
  public CacheWriter basicGetWriter() {
    return cacheWriter;
  }

  @Override
  public Class getKeyConstraint() {
    return keyConstraint;
  }

  @Override
  public Class getValueConstraint() {
    return valueConstraint;
  }

  private volatile ExpirationAttributes regionTimeToLiveAtts;

  private void setRegionTimeToLiveAtts() {
    regionTimeToLiveAtts =
        new ExpirationAttributes(regionTimeToLive, regionTimeToLiveExpirationAction);
  }

  @Override
  public ExpirationAttributes getRegionTimeToLive() {
    return regionTimeToLiveAtts;
  }

  private volatile ExpirationAttributes regionIdleTimeoutAttributes;

  private void setRegionIdleTimeoutAttributes() {
    regionIdleTimeoutAttributes =
        new ExpirationAttributes(regionIdleTimeout, regionIdleTimeoutExpirationAction);
  }

  @Override
  public ExpirationAttributes getRegionIdleTimeout() {
    return regionIdleTimeoutAttributes;
  }

  private volatile ExpirationAttributes entryTimeToLiveAtts;

  void setEntryTimeToLiveAttributes() {
    entryTimeToLiveAtts =
        new ExpirationAttributes(entryTimeToLive, entryTimeToLiveExpirationAction);
  }

  @Override
  public ExpirationAttributes getEntryTimeToLive() {
    return entryTimeToLiveAtts;
  }

  @Override
  public CustomExpiry getCustomEntryTimeToLive() {
    return customEntryTimeToLive;
  }

  private volatile ExpirationAttributes entryIdleTimeoutAttributes;

  private void setEntryIdleTimeoutAttributes() {
    entryIdleTimeoutAttributes =
        new ExpirationAttributes(entryIdleTimeout, entryIdleTimeoutExpirationAction);
  }

  @Override
  public ExpirationAttributes getEntryIdleTimeout() {
    return entryIdleTimeoutAttributes;
  }

  @Override
  public CustomExpiry getCustomEntryIdleTimeout() {
    return customEntryIdleTimeout;
  }

  @Override
  public MirrorType getMirrorType() {
    if (getDataPolicy().isNormal() || getDataPolicy().isPreloaded()
        || getDataPolicy().isEmpty() || getDataPolicy().withPartitioning()) {
      return MirrorType.NONE;
    } else if (getDataPolicy().withReplication()) {
      return MirrorType.KEYS_VALUES;
    } else {
      throw new IllegalStateException(
          String.format("No mirror type corresponds to data policy %s",
              getDataPolicy()));
    }
  }

  @Override
  public String getPoolName() {
    return poolName;
  }

  @Override
  public DataPolicy getDataPolicy() {
    return dataPolicy;
  }

  @Override
  public Scope getScope() {
    return scope;
  }

  @Override
  public CacheListener getCacheListener() {
    CacheListener[] listeners = fetchCacheListenersField();
    if (listeners == null || listeners.length == 0) {
      return null;
    }
    if (listeners.length == 1) {
      return listeners[0];
    } else {
      throw new IllegalStateException(
          "More than one cache listener exists.");
    }
  }

  public boolean isPdxTypesRegion() {
    return isPdxTypesRegion;
  }

  @Override
  public Set<String> getGatewaySenderIds() {
    return gatewaySenderIds;
  }

  @Override
  public Set<String> getAsyncEventQueueIds() {
    return asyncEventQueueIds;
  }

  @Override
  public Set<String> getVisibleAsyncEventQueueIds() {
    return visibleAsyncEventQueueIds;
  }

  @Override
  public Set<String> getAllGatewaySenderIds() {
    return allGatewaySenderIds;
  }

  /**
   * Return the remote DS IDs that need to receive events for this region.
   *
   * @param allGatewaySenderIds the set of gateway sender IDs to consider
   */
  List<Integer> getRemoteDsIds(Set<String> allGatewaySenderIds) throws IllegalStateException {
    int sz = allGatewaySenderIds.size();
    Set<GatewaySender> allGatewaySenders = cache.getAllGatewaySenders();
    if ((sz > 0 || isPdxTypesRegion) && !allGatewaySenders.isEmpty()) {
      List<Integer> allRemoteDSIds = new ArrayList<>(sz);
      for (GatewaySender sender : allGatewaySenders) {
        // This is for all regions except pdx Region
        if (!isPdxTypesRegion) {
          // Make sure we are distributing to only those senders whose id
          // is available on this region and whose state is running
          if (hasRunningGatewaySender(allGatewaySenders, sender)) {
            allRemoteDSIds.add(sender.getRemoteDSId());
          }
        } else { // this else is for PDX region
          allRemoteDSIds.add(sender.getRemoteDSId());
        }
      }
      return allRemoteDSIds;
    }
    return null;
  }

  @Immutable
  private static final CacheListener[] EMPTY_LISTENERS = new CacheListener[0];

  @Override
  public CacheListener[] getCacheListeners() {
    CacheListener[] listeners = fetchCacheListenersField();
    if (listeners == null || listeners.length == 0) {
      return EMPTY_LISTENERS;
    } else {
      CacheListener[] result = new CacheListener[listeners.length];
      System.arraycopy(listeners, 0, result, 0, listeners.length);
      return result;
    }
  }

  /**
   * Sets the cacheListeners field.
   */
  private void storeCacheListenersField(CacheListener[] value) {
    synchronized (clSync) {
      if (value != null && value.length != 0) {
        CacheListener[] cacheListeners = new CacheListener[value.length];
        System.arraycopy(value, 0, cacheListeners, 0, cacheListeners.length);
        value = cacheListeners;
      } else {
        value = EMPTY_LISTENERS;
      }
      cacheListeners = value;
    }
  }

  /**
   * Fetches the value in the cacheListeners field. NOTE: callers should not modify the contents of
   * the returned array.
   */
  CacheListener[] fetchCacheListenersField() {
    return cacheListeners;
  }

  @Override
  public int getInitialCapacity() {
    return initialCapacity;
  }

  @Override
  public float getLoadFactor() {
    return loadFactor;
  }

  abstract boolean isCurrentlyLockGrantor();

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

  /**
   * RegionAttributes implementation. Returns true if multicast can be used by the cache for this
   * region
   */
  @Override
  public boolean getMulticastEnabled() {
    return mcastEnabled;
  }

  @Override
  public boolean getStatisticsEnabled() {
    return statisticsEnabled;
  }

  @Override
  public boolean getIgnoreJTA() {
    return ignoreJTA;
  }

  @Override
  public int getConcurrencyLevel() {
    return concurrencyLevel;
  }

  @Override
  public boolean getConcurrencyChecksEnabled() {
    return concurrencyChecksEnabled;
  }

  public void setConcurrencyChecksEnabled(boolean concurrencyChecksEnabled) {
    this.concurrencyChecksEnabled = concurrencyChecksEnabled;
  }

  @Override
  public boolean getPersistBackup() {
    return getDataPolicy().withPersistence();
  }

  @Override
  public boolean getEarlyAck() {
    return earlyAck;
  }

  /*
   * @deprecated as of prPersistSprint1
   */
  @Deprecated
  @Override
  public boolean getPublisher() {
    return publisher;
  }

  @Override
  public boolean getEnableConflation() { // deprecated in 5.0
    return getEnableSubscriptionConflation();
  }

  @Override
  public boolean getEnableBridgeConflation() {// deprecated in 5.7
    return getEnableSubscriptionConflation();
  }

  @Override
  public boolean getEnableSubscriptionConflation() {
    return enableSubscriptionConflation;
  }

  @Override
  public boolean getEnableAsyncConflation() {
    return enableAsyncConflation;
  }

  /*
   * @deprecated as of prPersistSprint2
   */
  @Deprecated
  @Override
  public DiskWriteAttributes getDiskWriteAttributes() {
    return diskWriteAttributes;
  }

  @Override
  public abstract File[] getDiskDirs();

  @Override
  public String getDiskStoreName() {
    return diskStoreName;
  }

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

  @Override
  public boolean getIndexMaintenanceSynchronous() {
    return indexMaintenanceSynchronous;
  }

  @Override
  public PartitionAttributes getPartitionAttributes() {
    return partitionAttributes;
  }

  @Override
  public MembershipAttributes getMembershipAttributes() {
    return membershipAttributes;
  }

  @Override
  public SubscriptionAttributes getSubscriptionAttributes() {
    return subscriptionAttributes;
  }

  /**
   * Get IndexManger for region
   */
  @Override
  public IndexManager getIndexManager() {
    return indexManager;
  }

  /**
   * This method call is guarded by imSync lock created for each region. Set IndexManger for region.
   */
  @Override
  public void setIndexManager(IndexManager indexManager) {
    checkReadiness();
    this.indexManager = indexManager;
  }

  /**
   * Use ONLY imSync for IndexManager get and set.
   *
   * @return {@link IndexManager} lock.
   */
  @Override
  public Object getIMSync() {
    return imSync;
  }

  // The ThreadLocal is used to identify if the thread is an
  // index creation thread. This is used to fix Bug # 33336. The value
  // is set from IndexManager ,if the thread happens to be an IndexCreation
  // Thread.
  // Once the thread has created the Index , it will unset the value in the
  // ThreadLocal Object
  public void setFlagForIndexCreationThread(boolean value) {
    isIndexCreator.set(value ? Boolean.TRUE : null);
  }

  // The boolean is used in AbstractRegionEntry to skip the synchronized
  // block
  // in case the value of the entry is "REMOVED" token. This prevents dead lock
  // caused by the Bug # 33336
  @Override
  public boolean isIndexCreationThread() {
    Boolean value = isIndexCreator.get();
    return value != null ? value : false;
  }

  @Override
  public Region getRegion() {
    return this;
  }

  @Override
  public void addGatewaySenderId(String gatewaySenderId) {
    getGatewaySenderIds().add(gatewaySenderId);
    setAllGatewaySenderIds();
  }

  @Override
  public void removeGatewaySenderId(String gatewaySenderId) {
    getGatewaySenderIds().remove(gatewaySenderId);
    setAllGatewaySenderIds();
  }

  @Override
  public void addAsyncEventQueueId(String asyncEventQueueId) {
    addAsyncEventQueueId(asyncEventQueueId, false);
  }

  public void addAsyncEventQueueId(String asyncEventQueueId, boolean isInternal) {
    getAsyncEventQueueIds().add(asyncEventQueueId);
    if (!isInternal) {
      getVisibleAsyncEventQueueIds().add(asyncEventQueueId);
    }
    setAllGatewaySenderIds();
  }

  @Override
  public void removeAsyncEventQueueId(String asyncEventQueueId) {
    getAsyncEventQueueIds().remove(asyncEventQueueId);
    getVisibleAsyncEventQueueIds().remove(asyncEventQueueId);
    setAllGatewaySenderIds();
  }

  private void setAllGatewaySenderIds() {
    if (getGatewaySenderIds().isEmpty() && getAsyncEventQueueIds().isEmpty()) {
      allGatewaySenderIds = Collections.emptySet(); // fix for bug 45774
    }
    Set<String> tmp = new HashSet<>(getGatewaySenderIds());
    for (String asyncQueueId : getAsyncEventQueueIds()) {
      tmp.add(AsyncEventQueueImpl.getSenderIdFromAsyncEventQueueId(asyncQueueId));
    }
    allGatewaySenderIds = Collections.unmodifiableSet(tmp);
  }

  private void initializeVisibleAsyncEventQueueIds(InternalRegionArguments internalRegionArgs) {
    Set<String> visibleAsyncEventQueueIds = new CopyOnWriteArraySet<>(getAsyncEventQueueIds());
    // Remove all internal aeqIds from internal region args if necessary
    if (internalRegionArgs.getInternalAsyncEventQueueIds() != null) {
      visibleAsyncEventQueueIds.removeAll(internalRegionArgs.getInternalAsyncEventQueueIds());
    }
