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

import static java.lang.String.format;
import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.internal.logging.LogWriterFactory.toSecurityLogWriter;
import static org.apache.geode.internal.statistics.StatisticsClockFactory.disabledClock;
import static org.apache.geode.logging.internal.spi.LogWriterLevel.ALL;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.URL;
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.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

import javax.naming.Context;

import io.micrometer.core.instrument.MeterRegistry;
import jakarta.transaction.TransactionManager;
import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelCriterion;
import org.apache.geode.GemFireIOException;
import org.apache.geode.LogWriter;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.CacheTransactionManager;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.CacheXmlException;
import org.apache.geode.cache.Declarable;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.DynamicRegionFactory;
import org.apache.geode.cache.GatewayException;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionExistsException;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.TransactionListener;
import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueFactoryImpl;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.internal.ClientMetadataService;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.query.CqAttributes;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.CqServiceStatistics;
import org.apache.geode.cache.query.Index;
import org.apache.geode.cache.query.IndexInvalidException;
import org.apache.geode.cache.query.IndexType;
import org.apache.geode.cache.query.Query;
import org.apache.geode.cache.query.QueryInvalidException;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.internal.InternalQueryService;
import org.apache.geode.cache.query.internal.QueryConfigurationService;
import org.apache.geode.cache.query.internal.QueryConfigurationServiceException;
import org.apache.geode.cache.query.internal.QueryConfigurationServiceImpl;
import org.apache.geode.cache.query.internal.QueryMonitor;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.xml.QueryConfigurationServiceCreation;
import org.apache.geode.cache.query.internal.xml.QueryMethodAuthorizerCreation;
import org.apache.geode.cache.query.security.MethodInvocationAuthorizer;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.snapshot.CacheSnapshotService;
import org.apache.geode.cache.util.GatewayConflictResolver;
import org.apache.geode.cache.wan.GatewayReceiver;
import org.apache.geode.cache.wan.GatewayReceiverFactory;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.cache.wan.GatewaySenderFactory;
import org.apache.geode.cache.wan.GatewayTransportFilter;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.ServerLauncherParameters;
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.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.i18n.LogWriterI18n;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.cache.CacheConfig;
import org.apache.geode.internal.cache.CachePerfStats;
import org.apache.geode.internal.cache.CacheServerImpl;
import org.apache.geode.internal.cache.CacheService;
import org.apache.geode.internal.cache.DiskStoreAttributes;
import org.apache.geode.internal.cache.DiskStoreFactoryImpl;
import org.apache.geode.internal.cache.DiskStoreImpl;
import org.apache.geode.internal.cache.DiskStoreMonitor;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.ExpirationScheduler;
import org.apache.geode.internal.cache.FilterProfile;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InitialImageOperation;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalCacheForClientAccess;
import org.apache.geode.internal.cache.InternalCacheServer;
import org.apache.geode.internal.cache.InternalRegion;
import org.apache.geode.internal.cache.InternalRegionArguments;
import org.apache.geode.internal.cache.InternalRegionFactory;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.PoolFactoryImpl;
import org.apache.geode.internal.cache.PoolManagerImpl;
import org.apache.geode.internal.cache.RegionListener;
import org.apache.geode.internal.cache.TXEntryStateFactory;
import org.apache.geode.internal.cache.TXManagerImpl;
import org.apache.geode.internal.cache.TombstoneService;
import org.apache.geode.internal.cache.backup.BackupService;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.control.ResourceAdvisor;
import org.apache.geode.internal.cache.event.EventTrackerExpiryTask;
import org.apache.geode.internal.cache.eviction.HeapEvictor;
import org.apache.geode.internal.cache.eviction.OffHeapEvictor;
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.ha.HARegionQueue;
import org.apache.geode.internal.cache.persistence.PersistentMemberManager;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
import org.apache.geode.internal.cache.wan.InternalGatewaySenderFactory;
import org.apache.geode.internal.cache.wan.WANServiceProvider;
import org.apache.geode.internal.jndi.JNDIInvoker;
import org.apache.geode.internal.lang.SystemProperty;
import org.apache.geode.internal.lang.SystemPropertyHelper;
import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.logging.LocalLogWriter;
import org.apache.geode.internal.offheap.MemoryAllocator;
import org.apache.geode.internal.security.SecurityService;
import org.apache.geode.internal.security.SecurityServiceFactory;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.management.internal.JmxManagerAdvisor;
import org.apache.geode.management.internal.RestAgent;
import org.apache.geode.pdx.JSONFormatter;
import org.apache.geode.pdx.PdxInstance;
import org.apache.geode.pdx.PdxInstanceFactory;
import org.apache.geode.pdx.PdxSerializer;
import org.apache.geode.pdx.internal.TypeRegistry;

/**
 * Represents a {@link Cache} that is created declaratively. Notice that it implements the
 * {@link Cache} interface so that this class must be updated when {@link Cache} is modified. This
 * class is public for testing purposes.
 *
 * @since GemFire 3.0
 */
public class CacheCreation implements InternalCache {

  private static final Logger logger = LogService.getLogger();

  @Immutable
  private static final RegionAttributes defaults = new AttributesFactory().create();

  @Immutable
  private static final Optional<Boolean> parallelDiskStoreRecovery = SystemProperty
      .getProductBooleanProperty(SystemPropertyHelper.PARALLEL_DISK_STORE_RECOVERY);

  /**
   * Store the current CacheCreation that is doing a create. Used from PoolManager to defer to
   * CacheCreation as a manager of pools.
   *
   * @since GemFire 5.7
   */
  private static final ThreadLocal<PoolManagerImpl> createInProgress = new ThreadLocal<>();

  /**
   * The amount of time to wait for a distributed lock
   */
  private int lockTimeout = GemFireCacheImpl.DEFAULT_LOCK_TIMEOUT;
  private boolean hasLockTimeout;

  /**
   * The duration of a lease on a distributed lock
   */
  private int lockLease = GemFireCacheImpl.DEFAULT_LOCK_LEASE;
  private boolean hasLockLease;

  /**
   * The amount of time to wait for a {@code netSearch}
   */
  private int searchTimeout = GemFireCacheImpl.DEFAULT_SEARCH_TIMEOUT;
  private boolean hasSearchTimeout;

  private boolean hasMessageSyncInterval;

  /**
   * This cache's roots keyed on name
   */
  protected final Map<String, Region<?, ?>> roots = new LinkedHashMap<>();

  /**
   * Are dynamic regions enabled in this cache?
   */
  private DynamicRegionFactory.Config dynamicRegionFactoryConfig;
  private boolean hasDynamicRegionFactory;

  /**
   * Is this a cache server?
   */
  private boolean isServer;
  private boolean hasServer;

  /**
   * The cache servers configured for this cache
   */
  private final List<CacheServer> bridgeServers = new ArrayList<>();

  // Stores the properties used to initialize declarables.
  private final Map<Declarable, Properties> declarablePropertiesMap = new HashMap<>();
  private final List<DeclarableAndProperties> declarablePropertiesList = new ArrayList<>();

  private final Set<GatewaySender> gatewaySenders = new HashSet<>();

  private final Set<GatewayReceiver> gatewayReceivers = new HashSet<>();

  private final Set<AsyncEventQueue> asyncEventQueues = new HashSet<>();

  private GatewayConflictResolver gatewayConflictResolver;

  /**
   * The copyOnRead attribute
   */
  private boolean copyOnRead = GemFireCacheImpl.DEFAULT_COPY_ON_READ;
  private boolean hasCopyOnRead;

  /**
   * The CacheTransactionManager representative for this Cache
   */
  private CacheTransactionManagerCreation cacheTransactionManagerCreation;

  /**
   * The named region attributes associated with this cache
   */
  private final Map<String, RegionAttributes<?, ?>> namedRegionAttributes = new HashMap<>();

  /**
   * The names of the region attributes in the order in which they were added. Keeping track of this
   * ensures that named region attributes are processed in the correct order. That is, "parent"
   * named region attributes will be processed before "children" named region attributes.
   */
  private final List<String> regionAttributesNames = new ArrayList<>();

  /**
   * The named disk store attributes associated with this cache. Made this linked so its iteration
   * would be in insert order. This is important for unit testing 44914.
   */
  private final Map<String, DiskStore> diskStores = new LinkedHashMap<>();

  private final List<File> backups = new ArrayList<>();

  private final CacheConfig cacheConfig = new CacheConfig();

  /**
   * A logger that is used in debugging
   */
  private final InternalLogWriter logWriter = new LocalLogWriter(ALL.intLevel(), System.out);

  private final InternalLogWriter securityLogWriter = toSecurityLogWriter(logWriter);

  /**
   * {@link ExtensionPoint} support.
   *
   * @since GemFire 8.1
   */
  private final SimpleExtensionPoint<Cache> extensionPoint = new SimpleExtensionPoint<>(this, this);

  private final PoolManagerImpl poolManager = new PoolManagerImpl(false);

  private volatile FunctionServiceCreation functionServiceCreation;

  private volatile boolean hasFunctionService;

  private volatile boolean hasResourceManager;

  private volatile ResourceManagerCreation resourceManagerCreation;

  private volatile SerializerCreation serializerCreation;

  private Declarable initializer;

  private Properties initializerProps;

  private final InternalQueryService queryService = createInternalQueryService();

  private QueryConfigurationServiceCreation queryConfigurationServiceCreation;

  /**
   * Creates a new {@code CacheCreation} with no root regions
   */
  public CacheCreation() {
    this(false);
  }

  /**
   * @param forParsing if true then this creation is used for parsing xml; if false then it is used
   *        for generating xml.
   *
   * @since GemFire 5.7
   */
  public CacheCreation(boolean forParsing) {
    initializeRegionShortcuts();
    if (!forParsing) {
      createInProgress.set(poolManager);
    }
  }

  /**
   * clear thread locals that may have been set by previous uses of CacheCreation
   */
  public static void clearThreadLocals() {
    createInProgress.remove();
  }

  /**
   * @since GemFire 5.7
   */
  void startingGenerate() {
    createInProgress.set(null);
  }

  RegionAttributes getDefaultAttributes() {
    return defaults;
  }

  protected void initializeRegionShortcuts() {
    GemFireCacheImpl.initializeRegionShortcuts(this);
  }

  /**
   * Sets the attributes of the root region
   *
   * @throws RegionExistsException If this cache already contains a region with the same name as
   *         {@code root}.
   */
  void addRootRegion(RegionCreation root) throws RegionExistsException {
    String name = root.getName();
    RegionCreation existing = (RegionCreation) roots.get(name);

    if (existing != null) {
      throw new RegionExistsException(existing);
    }

    roots.put(root.getName(), root);
  }

  @Override
  public int getLockTimeout() {
    return lockTimeout;
  }

  @Override
  public void setLockTimeout(int seconds) {
    lockTimeout = seconds;
    hasLockTimeout = true;
  }

  boolean hasLockTimeout() {
    return hasLockTimeout;
  }

  @Override
  public int getLockLease() {
    return lockLease;
  }

  @Override
  public void setLockLease(int seconds) {
    lockLease = seconds;
    hasLockLease = true;
  }

  boolean hasLockLease() {
    return hasLockLease;
  }

  @Override
  public int getSearchTimeout() {
    return searchTimeout;
  }

  @Override
  public void setSearchTimeout(int seconds) {
    searchTimeout = seconds;
    hasSearchTimeout = true;
  }

  boolean hasSearchTimeout() {
    return hasSearchTimeout;
  }

  @Override
  public int getMessageSyncInterval() {
    return HARegionQueue.getMessageSyncInterval();
  }

  @Override
  public void setMessageSyncInterval(int seconds) {
    if (seconds < 0) {
      throw new IllegalArgumentException(
          "The 'message-sync-interval' property for cache cannot be negative");
    }
    HARegionQueue.setMessageSyncInterval(seconds);
    hasMessageSyncInterval = true;
  }

  boolean hasMessageSyncInterval() {
    return hasMessageSyncInterval;
  }

  @Override
  public Set<Region<?, ?>> rootRegions() {
    return Collections.unmodifiableSet(new LinkedHashSet<>(roots.values()));
  }

  /**
   * create diskstore factory
   *
   * @since GemFire prPersistSprint2
   */
  @Override
  public DiskStoreFactory createDiskStoreFactory() {
    return new DiskStoreFactoryImpl(this);
  }

  /**
   * Returns null if the current thread is not doing a CacheCreation create. Otherwise returns the
   * PoolManagerImpl of the CacheCreation of the create being invoked.
   *
   * @since GemFire 5.7
   */
  public static PoolManagerImpl getCurrentPoolManager() {
    return createInProgress.get();
  }

  /**
   * Fills in the contents of a {@link Cache} based on this creation object's state.
   */
  void create(InternalCache cache)
      throws TimeoutException, CacheWriterException, GatewayException, RegionExistsException,
      QueryConfigurationServiceException {
    extensionPoint.beforeCreate(cache);

    cache.setDeclarativeCacheConfig(cacheConfig);

    if (cache.isClient()) {
      throw new IllegalStateException(
          "You must use client-cache in the cache.xml when ClientCacheFactory is used.");
    }

    initializeDeclarablesMap(cache);

    if (hasFunctionService()) {
      getFunctionServiceCreation().create();
    }

    if (hasLockLease()) {
      cache.setLockLease(lockLease);
    }
    if (hasLockTimeout()) {
      cache.setLockTimeout(lockTimeout);
    }
    if (hasSearchTimeout()) {
      cache.setSearchTimeout(searchTimeout);
    }
    if (hasMessageSyncInterval()) {
      cache.setMessageSyncInterval(getMessageSyncInterval());
    }
    if (gatewayConflictResolver != null) {
      cache.setGatewayConflictResolver(gatewayConflictResolver);
    }

    // create connection pools
    Map<String, Pool> pools = getPools();
    if (!pools.isEmpty()) {
      for (Pool pool : pools.values()) {
        PoolFactoryImpl poolFactory = (PoolFactoryImpl) PoolManager.createFactory();
        poolFactory.init(pool);
        poolFactory.create(pool.getName());
      }
    }

    if (hasResourceManager()) {
      getResourceManager().configure(cache.getResourceManager());
    }

    DiskStoreAttributesCreation pdxRegDSC = initializePdxDiskStore(cache);

    cache.initializePdxRegistry();

    createDiskStores(cache, pdxRegDSC);

    if (hasDynamicRegionFactory()) {
      DynamicRegionFactory.get().open(getDynamicRegionFactoryConfig());
    }
    if (hasServer()) {
      cache.setIsServer(isServer);
    }
    if (hasCopyOnRead()) {
      cache.setCopyOnRead(copyOnRead);
    }

    if (cacheTransactionManagerCreation != null
        && cacheTransactionManagerCreation.getListeners().length > 0
        && cache.getCacheTransactionManager() != null) {
      cache.getCacheTransactionManager()
          .initListeners(cacheTransactionManagerCreation.getListeners());
    }

    if (cacheTransactionManagerCreation != null && cache.getCacheTransactionManager() != null) {
      cache.getCacheTransactionManager().setWriter(cacheTransactionManagerCreation.getWriter());
    }

    for (GatewaySender senderCreation : getGatewaySenders()) {
      GatewaySenderFactory factory = cache.createGatewaySenderFactory();
      ((InternalGatewaySenderFactory) factory).configureGatewaySender(senderCreation);
      GatewaySender gatewaySender =
          factory.create(senderCreation.getId(), senderCreation.getRemoteDSId());
      // Start the sender if it is not set to manually start
      if (gatewaySender.isManualStart()) {
        logger.info("{} is not being started since it is configured for manual start",
            gatewaySender);
      }
    }

    for (AsyncEventQueue asyncEventQueueCreation : getAsyncEventQueues()) {
      AsyncEventQueueFactoryImpl asyncQueueFactory =
          (AsyncEventQueueFactoryImpl) cache.createAsyncEventQueueFactory();
      asyncQueueFactory.configureAsyncEventQueue(asyncEventQueueCreation);
      if (asyncEventQueueCreation.isDispatchingPaused()) {
        asyncQueueFactory.pauseEventDispatching();
      }

      AsyncEventQueue asyncEventQueue = cache.getAsyncEventQueue(asyncEventQueueCreation.getId());
      if (asyncEventQueue == null) {
        asyncQueueFactory.create(asyncEventQueueCreation.getId(),
            asyncEventQueueCreation.getAsyncEventListener());
      }
    }

    cache.initializePdxRegistry();

    for (String id : regionAttributesNames) {
      RegionAttributesCreation creation = (RegionAttributesCreation) getRegionAttributes(id);
      creation.inheritAttributes(cache, false);

      // Don't let the RegionAttributesCreation escape to the user
      AttributesFactory<?, ?> factory = new AttributesFactory<>(creation);
      RegionAttributes<?, ?> attrs = factory.create();

      cache.setRegionAttributes(id, attrs);
    }

    initializeRegions(roots, cache);

    cache.readyDynamicRegionFactory();

    // Create and start the CacheServers after the gateways have been initialized
    startCacheServers(getCacheServers(), cache, ServerLauncherParameters.INSTANCE);

    for (GatewayReceiver receiverCreation : getGatewayReceivers()) {
      GatewayReceiverFactory factory = cache.createGatewayReceiverFactory();
      factory.setBindAddress(receiverCreation.getBindAddress());
      factory.setMaximumTimeBetweenPings(receiverCreation.getMaximumTimeBetweenPings());
      factory.setStartPort(receiverCreation.getStartPort());
      factory.setEndPort(receiverCreation.getEndPort());
      factory.setSocketBufferSize(receiverCreation.getSocketBufferSize());
      factory.setManualStart(receiverCreation.isManualStart());
      for (GatewayTransportFilter filter : receiverCreation.getGatewayTransportFilters()) {
        factory.addGatewayTransportFilter(filter);
      }
      factory.setHostnameForSenders(receiverCreation.getHostnameForSenders());
      GatewayReceiver receiver = factory.create();
      if (receiver.isManualStart()) {
        logger.info("{} is not being started since it is configured for manual start", receiver);
      }
    }

    if (queryConfigurationServiceCreation != null) {
      QueryConfigurationServiceImpl queryConfigService =
          (QueryConfigurationServiceImpl) cache.getService(QueryConfigurationService.class);
      if (queryConfigService != null) {
        QueryMethodAuthorizerCreation authorizerCreation =
            queryConfigurationServiceCreation.getMethodAuthorizerCreation();
        if (authorizerCreation != null) {
          queryConfigService.updateMethodAuthorizer(cache, true, authorizerCreation);
        }
      }
    }

    cache.setBackupFiles(backups);

    runInitializer(cache);
    cache.setInitializer(getInitializer(), getInitializerProps());

    // Create all extensions
    extensionPoint.fireCreate(cache);
  }

  private void createDiskStores(InternalCache cache, DiskStoreAttributesCreation pdxRegDSC) {
    Stream<DiskStore> diskStoreStream;
    if (parallelDiskStoreRecovery.orElse(true)) {
      diskStoreStream = diskStores.values().parallelStream();
    } else {
      diskStoreStream = diskStores.values().stream();
    }
    diskStoreStream.forEach(diskStore -> {
      DiskStoreAttributesCreation creation = (DiskStoreAttributesCreation) diskStore;
      if (creation != pdxRegDSC) {
        createDiskStore(creation, cache);
      }
    });
  }

  public void initializeDeclarablesMap(InternalCache cache) {
    for (DeclarableAndProperties struct : declarablePropertiesList) {
      Declarable declarable = struct.getDeclarable();
      Properties properties = struct.getProperties();
      try {
        declarable.initialize(cache, properties);
        // for backwards compatibility
        declarable.init(properties);
      } catch (Exception ex) {
        throw new CacheXmlException(
            "Exception while initializing an instance of " + declarable.getClass().getName(), ex);
      }
      declarablePropertiesMap.put(declarable, properties);
    }
    cache.addDeclarableProperties(declarablePropertiesMap);
  }

  void initializeRegions(Map<String, Region<?, ?>> declarativeRegions, Cache cache) {
    for (Region region : declarativeRegions.values()) {
      RegionCreation regionCreation = (RegionCreation) region;
      regionCreation.createRoot(cache);
    }
  }

  /**
   * Reconfigures the server using the specified parameters.
   *
   * @param serverImpl CacheServer implementation.
   * @param parameters Parameters to reconfigure the server.
   */
  private void reconfigureServer(CacheServerImpl serverImpl, ServerLauncherParameters parameters) {
    if (parameters == null) {
      return;
    }

    if (parameters.getPort() != null
        && parameters.getPort() != CacheServer.DEFAULT_PORT) {
      serverImpl.setPort(parameters.getPort());
    }
    if (parameters.getMaxThreads() != null
        && parameters.getMaxThreads() != CacheServer.DEFAULT_MAX_THREADS) {
      serverImpl.setMaxThreads(parameters.getMaxThreads());
    }
    if (parameters.getMaxConnections() != null
        && parameters.getMaxConnections() != CacheServer.DEFAULT_MAX_CONNECTIONS) {
      serverImpl.setMaxConnections(parameters.getMaxConnections());
    }
    if (parameters.getMaxMessageCount() != null
        && parameters.getMaxMessageCount() != CacheServer.DEFAULT_MAXIMUM_MESSAGE_COUNT) {
      serverImpl.setMaximumMessageCount(parameters.getMaxMessageCount());
    }
    if (parameters.getSocketBufferSize() != null
        && parameters.getSocketBufferSize() != CacheServer.DEFAULT_SOCKET_BUFFER_SIZE) {
      serverImpl.setSocketBufferSize(parameters.getSocketBufferSize());
    }
    if (parameters.getBindAddress() != null
        && parameters.getBindAddress() != CacheServer.DEFAULT_BIND_ADDRESS) {
      serverImpl.setBindAddress(parameters.getBindAddress().trim());
    }
    if (parameters.getMessageTimeToLive() != null
        && parameters.getMessageTimeToLive() != CacheServer.DEFAULT_MESSAGE_TIME_TO_LIVE) {
      serverImpl.setMessageTimeToLive(parameters.getMessageTimeToLive());
    }
    if (parameters.getHostnameForClients() != null
        && parameters.getHostnameForClients() != CacheServer.DEFAULT_HOSTNAME_FOR_CLIENTS) {
      serverImpl.setHostnameForClients(parameters.getHostnameForClients());
    }
  }

  /**
   * starts declarative cache servers if a server is not running on the port already. Also adds a
   * default server to the param declarativeCacheServers if a serverPort is specified.
   */
  void startCacheServers(List<CacheServer> declarativeCacheServers, Cache cache,
      ServerLauncherParameters parameters) {
    Integer serverPort = null;
    String serverBindAdd = null;
    Boolean disableDefaultServer = null;

    if (parameters != null) {
      serverPort = parameters.getPort();
      serverBindAdd = parameters.getBindAddress();
      disableDefaultServer = parameters.isDisableDefaultServer();
    }

    if (declarativeCacheServers.size() > 1 && (serverPort != null || serverBindAdd != null)) {
      throw new RuntimeException(
          "When using -server-port or -server-bind-address arguments, the cache-xml-file can not have more than one cache-server defined.");
    }

    CacheServerCreation defaultServer = null;
    boolean hasServerPortOrBindAddress = serverPort != null || serverBindAdd != null;
    boolean isDefaultServerDisabled = disableDefaultServer == null || !disableDefaultServer;

    if (declarativeCacheServers.isEmpty() && hasServerPortOrBindAddress
        && isDefaultServerDisabled) {
      boolean existingCacheServer = false;

      List<CacheServer> cacheServers = cache.getCacheServers();
      if (cacheServers != null) {
        for (CacheServer cacheServer : cacheServers) {
          if (serverPort == cacheServer.getPort()) {
            existingCacheServer = true;
          }
        }
      }

      if (!existingCacheServer) {
        defaultServer = new CacheServerCreation((InternalCache) cache, false);
        declarativeCacheServers.add(defaultServer);
      }
    }

    for (CacheServer declarativeCacheServer : declarativeCacheServers) {
      CacheServerCreation declaredCacheServer = (CacheServerCreation) declarativeCacheServer;

      boolean startServer = true;
      List<CacheServer> cacheServers = cache.getCacheServers();
      if (cacheServers != null) {
        for (CacheServer cacheServer : cacheServers) {
          if (declaredCacheServer.getPort() == cacheServer.getPort()) {
            startServer = false;
          }
        }
      }

      if (!startServer) {
        continue;
      }

      CacheServerImpl impl = (CacheServerImpl) cache.addCacheServer();
      impl.configureFrom(declaredCacheServer);

      if (declaredCacheServer == defaultServer) {
        impl.setIsDefaultServer();
      }

      reconfigureServer(impl, parameters);

      try {
        if (!impl.isRunning()) {
          impl.start();
        }

      } catch (IOException ex) {
        throw new GemFireIOException(format("While starting cache server %s", impl), ex);
      }
    }
  }

  /**
   * Returns a description of the disk store used by the pdx registry.
   */
  DiskStoreAttributesCreation initializePdxDiskStore(InternalCache cache) {
    // Create the disk store used by the pdx registry first.
    // If it is using the default disk store we need to create it now.
    // If the cache has a pool then no need to create disk store.
    DiskStoreAttributesCreation pdxRegDSC = null;
    if (TypeRegistry.mayNeedDiskStore(cache)) {
      String pdxRegDsName = cache.getPdxDiskStore();
      if (pdxRegDsName == null) {
        pdxRegDsName = DiskStoreFactory.DEFAULT_DISK_STORE_NAME;
      }
      pdxRegDSC = (DiskStoreAttributesCreation) diskStores.get(pdxRegDsName);
      if (pdxRegDSC == null) {
        if (pdxRegDsName.equals(DiskStoreFactory.DEFAULT_DISK_STORE_NAME)) {
          // need to create default disk store
          cache.getOrCreateDefaultDiskStore();
        }
      } else {
        createDiskStore(pdxRegDSC, cache);
      }
    }
    return pdxRegDSC;
  }

  protected void createDiskStore(DiskStoreAttributesCreation creation, InternalCache cache) {
    // Don't let the DiskStoreAttributesCreation escape to the user
    DiskStoreFactory factory = cache.createDiskStoreFactory(creation);
    factory.create(creation.getName());
  }

  /**
   * Returns whether or not this {@code CacheCreation} is equivalent to another {@code Cache}.
   */
  public boolean sameAs(Cache other) {
    boolean sameConfig = other.getLockLease() == getLockLease()
        && other.getLockTimeout() == getLockTimeout()
        && other.getSearchTimeout() == getSearchTimeout()
        && other.getMessageSyncInterval() == getMessageSyncInterval()
        && other.getCopyOnRead() == getCopyOnRead() && other.isServer() == isServer();

    if (!sameConfig) {
      throw new RuntimeException("!sameConfig");
    }

    DynamicRegionFactory.Config drc1 = getDynamicRegionFactoryConfig();
    if (drc1 != null) {
      // we have a dynamic region factory
      DynamicRegionFactory.Config drc2;
      if (other instanceof CacheCreation) {
        drc2 = ((CacheCreation) other).getDynamicRegionFactoryConfig();
      } else {
        drc2 = DynamicRegionFactory.get().getConfig();
      }
      if (drc2 == null) {
        return false;
      }
      if (!drc1.equals(drc2)) {
        return false;
      }

    } else {
      // we have no dynamic region factory; how about other?
      if (other instanceof CacheCreation) {
        if (((CacheCreation) other).getDynamicRegionFactoryConfig() != null) {
          return false;
        }
      } else {
        // other must be real cache in which case we compare to DynamicRegionFactory
        if (DynamicRegionFactory.get().isOpen()) {
          return false;
        }
      }
    }

    Collection<CacheServer> myBridges = getCacheServers();
    Collection<CacheServer> otherBridges = other.getCacheServers();
    if (myBridges.size() != otherBridges.size()) {
      throw new RuntimeException("cacheServers size");
    }

    for (CacheServer myBridge1 : myBridges) {
