/*
 * 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 java.io.BufferedInputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.EmptyStackException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.ServiceLoader;
import java.util.Set;
import java.util.Stack;
import java.util.StringTokenizer;

import javax.naming.NamingException;
import javax.xml.XMLConstants;
import javax.xml.parsers.SAXParser;
import javax.xml.parsers.SAXParserFactory;

import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.Logger;
import org.xml.sax.Attributes;
import org.xml.sax.ContentHandler;
import org.xml.sax.InputSource;
import org.xml.sax.Locator;
import org.xml.sax.SAXException;
import org.xml.sax.SAXParseException;
import org.xml.sax.ext.DefaultHandler2;

import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheWriter;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.CacheXmlException;
import org.apache.geode.cache.CustomExpiry;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.Declarable;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.DiskWriteAttributes;
import org.apache.geode.cache.DynamicRegionFactory;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.ExpirationAction;
import org.apache.geode.cache.ExpirationAttributes;
import org.apache.geode.cache.GatewayException;
import org.apache.geode.cache.InterestPolicy;
import org.apache.geode.cache.LossAction;
import org.apache.geode.cache.MembershipAttributes;
import org.apache.geode.cache.MirrorType;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.PartitionResolver;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionExistsException;
import org.apache.geode.cache.ResumptionAction;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.SubscriptionAttributes;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.TransactionListener;
import org.apache.geode.cache.TransactionWriter;
import org.apache.geode.cache.asyncqueue.AsyncEventListener;
import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory;
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.SocketFactory;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.partition.PartitionListener;
import org.apache.geode.cache.query.IndexType;
import org.apache.geode.cache.query.internal.index.IndexCreationData;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.server.ClientSubscriptionConfig;
import org.apache.geode.cache.server.ServerLoadProbe;
import org.apache.geode.cache.util.GatewayConflictResolver;
import org.apache.geode.cache.util.ObjectSizer;
import org.apache.geode.cache.wan.GatewayEventFilter;
import org.apache.geode.cache.wan.GatewayEventSubstitutionFilter;
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.compression.Compressor;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.DiskStoreAttributes;
import org.apache.geode.internal.cache.DiskWriteAttributesImpl;
import org.apache.geode.internal.cache.EvictionAttributesImpl;
import org.apache.geode.internal.cache.FixedPartitionAttributesImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.PartitionAttributesImpl;
import org.apache.geode.internal.cache.PartitionedRegionHelper;
import org.apache.geode.internal.classloader.ClassPathLoader;
import org.apache.geode.internal.datasource.ConfigProperty;
import org.apache.geode.internal.datasource.DataSourceCreateException;
import org.apache.geode.internal.jndi.JNDIInvoker;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.pdx.PdxSerializer;

/**
 * Parses an XML file and creates a {@link Cache}/{@link ClientCache} and {@link Region}s from it.
 * It works in two phases. The first phase parses the XML and instantiates {@link Declarable}s. If
 * any problems occur, a {@link CacheXmlException} is thrown. The second phase actually
 * {@linkplain CacheCreation#create creates} the {@link Cache}/{@link ClientCache},{@link Region}s,
 * etc.
 *
 *
 * @since GemFire 3.0
 */
@SuppressWarnings("deprecation")
public class CacheXmlParser extends CacheXml implements ContentHandler {

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

  /**
   * @since GemFire 8.1
   */
  private static final String BUFFER_SIZE = "http://apache.org/xml/properties/input-buffer-size";

  /**
   * @since GemFire 8.1
   */
  private static final String DISALLOW_DOCTYPE_DECL_FEATURE =
      "http://apache.org/xml/features/disallow-doctype-decl";

  /**
   * @since GemFire 8.1
   */
  private static final String JAXP_SCHEMA_LANGUAGE =
      "http://java.sun.com/xml/jaxp/properties/schemaLanguage";

  /** The cache to be created */
  private CacheCreation cache;
  /** The stack of intermediate values used while parsing */
  protected Stack<Object> stack = new Stack<>();

  /**
   * Delegate {@link XmlParser}s mapped by namespace URI.
   *
   * @since GemFire 8.1
   */
  private final HashMap<String, XmlParser> delegates = new HashMap<>();

  /**
   * Document {@link Locator} used for {@link SAXParseException}.
   *
   * @since GemFire 8.2
   */
  protected Locator documentLocator;

  ////////////////////// Static Methods //////////////////////
  /**
   * Parses XML data and from it creates an instance of <code>CacheXmlParser</code> that can be used
   * to {@link #create}the {@link Cache}, etc.
   *
   * @param is the <code>InputStream</code> of XML to be parsed
   *
   * @return a <code>CacheXmlParser</code>, typically used to create a cache from the parsed XML
   *
   * @throws CacheXmlException Something went wrong while parsing the XML
   *
   * @since GemFire 4.0
   *
   */
  public static CacheXmlParser parse(InputStream is) {

    /*
     * The API doc http://java.sun.com/javase/6/docs/api/org/xml/sax/InputSource.html for the SAX
     * InputSource says: "... standard processing of both byte and character streams is to close
     * them on as part of end-of-parse cleanup, so applications should not attempt to re-use such
     * streams after they have been handed to a parser."
     *
     * In order to block the parser from closing the stream, we wrap the InputStream in a filter,
     * i.e., UnclosableInputStream, whose close() function does nothing.
     *
     */
    class UnclosableInputStream extends BufferedInputStream {
      public UnclosableInputStream(InputStream stream) {
        super(stream);
      }

      @Override
      public void close() {}
    }

    CacheXmlParser handler = new CacheXmlParser();
    try {
      SAXParserFactory factory = SAXParserFactory.newInstance();
      factory.setFeature(DISALLOW_DOCTYPE_DECL_FEATURE, true);
      factory.setValidating(true);
      factory.setNamespaceAware(true);
      UnclosableInputStream bis = new UnclosableInputStream(is);
      try {
        SAXParser parser = factory.newSAXParser();
        // Parser always reads one buffer plus a little extra worth before
        // determining that the DTD is there. Setting mark twice the parser
        // buffer size.
        bis.mark((Integer) parser.getProperty(BUFFER_SIZE) * 2);
        parser.setProperty(JAXP_SCHEMA_LANGUAGE, XMLConstants.W3C_XML_SCHEMA_NS_URI);
        parser.parse(bis, new DefaultHandlerDelegate(handler));
      } catch (CacheXmlException e) {
        if (null != e.getCause()
            && e.getCause().getMessage().contains(DISALLOW_DOCTYPE_DECL_FEATURE)) {
          // Not schema based document, try dtd.
          bis.reset();
          factory.setFeature(DISALLOW_DOCTYPE_DECL_FEATURE, false);
          SAXParser parser = factory.newSAXParser();
          parser.parse(bis, new DefaultHandlerDelegate(handler));
        } else {
          throw e;
        }
      }
      return handler;
    } catch (Exception ex) {
      if (ex instanceof CacheXmlException) {
        while (true /* ex instanceof CacheXmlException */) {
          Throwable cause = ex.getCause();
          if (!(cause instanceof CacheXmlException)) {
            break;
          } else {
            ex = (CacheXmlException) cause;
          }
        }
        throw (CacheXmlException) ex;
      } else if (ex instanceof SAXException) {
        // Silly JDK 1.4.2 XML parser wraps RunTime exceptions in a
        // SAXException. Pshaw!
        SAXException sax = (SAXException) ex;
        Exception cause = sax.getException();
        if (cause instanceof CacheXmlException) {
          while (true /* cause instanceof CacheXmlException */) {
            Throwable cause2 = cause.getCause();
            if (!(cause2 instanceof CacheXmlException)) {
              break;
            } else {
              cause = (CacheXmlException) cause2;
            }
          }
          throw (CacheXmlException) cause;
        }
      }
      throw new CacheXmlException(
          "While parsing XML", ex);
    }
  }

  /**
   * Helper method for parsing an integer
   *
   * @throws CacheXmlException If <code>s</code> is a malformed integer
   */
  private static int parseInt(String s) {
    try {
      return Integer.parseInt(s);
    } catch (NumberFormatException ex) {
      throw new CacheXmlException(
          String.format("Malformed integer %s", s), ex);
    }
  }

  /**
   * Helper method for parsing a long
   *
   * @throws CacheXmlException If <code>s</code> is a malformed long
   */
  private static long parseLong(String s) {
    try {
      return Long.parseLong(s);
    } catch (NumberFormatException ex) {
      throw new CacheXmlException(
          String.format("Malformed integer %s", s), ex);
    }
  }

  /**
   * Helper method for parsing a boolean
   *
   */
  private static boolean parseBoolean(String s) {
    return Boolean.parseBoolean(s);
  }

  /**
   * Helper method for parsing an float
   *
   * @throws CacheXmlException If <code>s</code> is a malformed float
   */
  private static float parseFloat(String s) {
    try {
      return Float.parseFloat(s);
    } catch (NumberFormatException ex) {
      throw new CacheXmlException(
          String.format("Malformed float %s", s), ex);
    }
  }

  /**
   * Returns the {@link CacheCreation} generated by this parser.
   */
  public CacheCreation getCacheCreation() {
    return cache;
  }

  /**
   * Creates cache artifacts ({@link Cache}s, etc.) based upon the XML parsed by this parser.
   */
  public void create(InternalCache cache)
      throws TimeoutException, GatewayException, CacheWriterException, RegionExistsException {
    if (this.cache == null) {
      String s = "A cache or client-cache element is required";
      throw new CacheXmlException(
          "No cache element specified.");
    }
    this.cache.create(cache);
  }

  /**
   * When a <code>cache</code> element is first encountered, we create a {@link CacheCreation}and
   * fill it in appropriately
   */
  private void startCache(Attributes atts) {
    if (cache != null) {
      throw new CacheXmlException("Only a single cache or client-cache element is allowed");
    }
    cache = new CacheCreation(true);
    String lockLease = atts.getValue(LOCK_LEASE);
    if (lockLease != null) {
      cache.setLockLease(parseInt(lockLease));
    }
    String lockTimeout = atts.getValue(LOCK_TIMEOUT);
    if (lockTimeout != null) {
      cache.setLockTimeout(parseInt(lockTimeout));
    }
    String searchTimeout = atts.getValue(SEARCH_TIMEOUT);
    if (searchTimeout != null) {
      cache.setSearchTimeout(parseInt(searchTimeout));
    }
    String messageSyncInterval = atts.getValue(MESSAGE_SYNC_INTERVAL);
    if (messageSyncInterval != null) {
      cache.setMessageSyncInterval(parseInt(messageSyncInterval));
    }
    String isServer = atts.getValue(IS_SERVER);
    if (isServer != null) {
      boolean b = Boolean.parseBoolean(isServer);
      cache.setIsServer(b);
    }
    String copyOnRead = atts.getValue(COPY_ON_READ);
    if (copyOnRead != null) {
      cache.setCopyOnRead(Boolean.parseBoolean(copyOnRead));
    }
    stack.push(cache);
  }

  /**
   * When a <code>client-cache</code> element is first encountered, we create a
   * {@link ClientCacheCreation}and fill it in appropriately
   */
  private void startClientCache(Attributes atts) {
    if (cache != null) {
      throw new CacheXmlException("Only a single cache or client-cache element is allowed");
    }
    cache = new ClientCacheCreation(true);
    String copyOnRead = atts.getValue(COPY_ON_READ);
    if (copyOnRead != null) {
      cache.setCopyOnRead(Boolean.parseBoolean(copyOnRead));
    }
    stack.push(cache);
  }

  /**
   * @since GemFire 5.7
   */
  private void startPool(Attributes atts) {
    PoolFactory f = cache.createPoolFactory();
    String name = atts.getValue(NAME).trim();
    stack.push(name);
    stack.push(f);
    String v;
    v = atts.getValue(SUBSCRIPTION_TIMEOUT_MULTIPLIER);
    if (v != null) {
      f.setSubscriptionTimeoutMultiplier(parseInt(v));
    }
    v = atts.getValue(SOCKET_CONNECT_TIMEOUT);
    if (v != null) {
      f.setSocketConnectTimeout(parseInt(v));
    }
    v = atts.getValue(FREE_CONNECTION_TIMEOUT);
    if (v != null) {
      f.setFreeConnectionTimeout(parseInt(v));
    }
    v = atts.getValue(SERVER_CONNECTION_TIMEOUT);
    if (v != null) {
      f.setServerConnectionTimeout(parseInt(v));
    }
    v = atts.getValue(LOAD_CONDITIONING_INTERVAL);
    if (v != null) {
      f.setLoadConditioningInterval(parseInt(v));
    }
    v = atts.getValue(MIN_CONNECTIONS);
    if (v != null) {
      f.setMinConnections(parseInt(v));
    }
    v = atts.getValue(MAX_CONNECTIONS);
    if (v != null) {
      f.setMaxConnections(parseInt(v));
    }
    v = atts.getValue(RETRY_ATTEMPTS);
    if (v != null) {
      f.setRetryAttempts(parseInt(v));
    }
    v = atts.getValue(IDLE_TIMEOUT);
    if (v != null) {
      f.setIdleTimeout(parseLong(v));
    }
    v = atts.getValue(PING_INTERVAL);
    if (v != null) {
      f.setPingInterval(parseLong(v));
    }
    v = atts.getValue(SUBSCRIPTION_ENABLED);
    if (v != null) {
      f.setSubscriptionEnabled(parseBoolean(v));
    }
    v = atts.getValue(PR_SINGLE_HOP_ENABLED);
    if (v != null) {
      f.setPRSingleHopEnabled(parseBoolean(v));
    }
    v = atts.getValue(SUBSCRIPTION_MESSAGE_TRACKING_TIMEOUT);
    if (v != null) {
      f.setSubscriptionMessageTrackingTimeout(parseInt(v));
    }
    v = atts.getValue(SUBSCRIPTION_ACK_INTERVAL);
    if (v != null) {
      f.setSubscriptionAckInterval(parseInt(v));
    }
    v = atts.getValue(SUBSCRIPTION_REDUNDANCY);
    if (v != null) {
      f.setSubscriptionRedundancy(parseInt(v));
    }
    v = atts.getValue(READ_TIMEOUT);
    if (v != null) {
      f.setReadTimeout(parseInt(v));
    }
    v = atts.getValue(SERVER_GROUP);
    if (v != null) {
      f.setServerGroup(v.trim());
    }
    v = atts.getValue(SOCKET_BUFFER_SIZE);
    if (v != null) {
      f.setSocketBufferSize(parseInt(v));
    }
    v = atts.getValue(STATISTIC_INTERVAL);
    if (v != null) {
      f.setStatisticInterval(parseInt(v));
    }
    v = atts.getValue(THREAD_LOCAL_CONNECTIONS);
    if (v != null) {
      f.setThreadLocalConnections(parseBoolean(v));
    }
    v = atts.getValue(MULTIUSER_SECURE_MODE_ENABLED);
    if (v != null) {
      f.setMultiuserAuthentication(parseBoolean(v));
    }
  }

  /**
   * @since GemFire 5.7
   */
  private void endPool() {
    PoolFactory f = (PoolFactory) stack.pop();
    String name = (String) stack.pop();
    f.create(name);
  }

  /**
   * @since GemFire 5.7
   */
  private void doLocator(Attributes atts) {
    PoolFactory f = (PoolFactory) stack.peek();
    String host = atts.getValue(HOST).trim();
    int port = parseInt(atts.getValue(PORT));
    f.addLocator(host, port);
  }

  /**
   * @since GemFire 5.7
   */
  private void doServer(Attributes atts) {
    PoolFactory f = (PoolFactory) stack.peek();
    String host = atts.getValue(HOST).trim();
    int port = parseInt(atts.getValue(PORT));
    f.addServer(host, port);
  }

  /**
   * When a <code>cache-server</code> element is first encountered, we create a new
   * {@link CacheCreation#addCacheServer() CacheServer} in the cache.
   *
   * @since GemFire 4.0
   */
  private void startCacheServer(Attributes atts) {
    CacheServer bridge = cache.addCacheServer();
    String port = atts.getValue(PORT);
    if (port != null) {
      bridge.setPort(parseInt(port));
    }
    String bindAddress = atts.getValue(BIND_ADDRESS);
    if (bindAddress != null) {
      bridge.setBindAddress(bindAddress.trim());
    }
    String hostnameForClients = atts.getValue(HOSTNAME_FOR_CLIENTS);
    if (hostnameForClients != null) {
      bridge.setHostnameForClients(hostnameForClients.trim());
    }
    String maxConnections = atts.getValue(MAX_CONNECTIONS);
    if (maxConnections != null) {
      bridge.setMaxConnections(parseInt(maxConnections));
    }
    String maxThreads = atts.getValue(MAX_THREADS);
    if (maxThreads != null) {
      bridge.setMaxThreads(parseInt(maxThreads));
    }
    String notifyBySubscription = atts.getValue(NOTIFY_BY_SUBSCRIPTION);
    if (notifyBySubscription != null) {
      boolean b = Boolean.parseBoolean(notifyBySubscription);
      bridge.setNotifyBySubscription(b);
    }
    String socketBufferSize = atts.getValue(SOCKET_BUFFER_SIZE);
    if (socketBufferSize != null) {
      bridge.setSocketBufferSize(Integer.parseInt(socketBufferSize));
    }
    String tcpDelay = atts.getValue(TCP_NO_DELAY);
    if (tcpDelay != null) {
      bridge.setTcpNoDelay(Boolean.parseBoolean(tcpDelay));
    }
    String maximumTimeBetweenPings = atts.getValue(MAXIMUM_TIME_BETWEEN_PINGS);
    if (maximumTimeBetweenPings != null) {
      bridge.setMaximumTimeBetweenPings(Integer.parseInt(maximumTimeBetweenPings));
    }
    String maximumMessageCount = atts.getValue(MAXIMUM_MESSAGE_COUNT);
    if (maximumMessageCount != null) {
      bridge.setMaximumMessageCount(Integer.parseInt(maximumMessageCount));
    }
    String messageTimeToLive = atts.getValue(MESSAGE_TIME_TO_LIVE);
    if (messageTimeToLive != null) {
      bridge.setMessageTimeToLive(Integer.parseInt(messageTimeToLive));
    }
    String loadPollInterval = atts.getValue(LOAD_POLL_INTERVAL);
    if (loadPollInterval != null) {
      bridge.setLoadPollInterval(Long.parseLong(loadPollInterval));
    }
    stack.push(bridge);
  }

  private void startGatewaySender(Attributes atts) {
    GatewaySenderFactory gatewaySenderFactory = cache.createGatewaySenderFactory();

    String parallel = atts.getValue(PARALLEL);
    if (parallel == null) {
      gatewaySenderFactory.setParallel(GatewaySender.DEFAULT_IS_PARALLEL);
    } else {
      gatewaySenderFactory.setParallel(Boolean.parseBoolean(parallel));
    }

    // manual-start
    String manualStart = atts.getValue(MANUAL_START);
    if (manualStart == null) {
      gatewaySenderFactory.setManualStart(GatewaySender.DEFAULT_MANUAL_START);
    } else {
      gatewaySenderFactory.setManualStart(Boolean.parseBoolean(manualStart));
    }

    // socket-buffer-size
    String socketBufferSize = atts.getValue(SOCKET_BUFFER_SIZE);
    if (socketBufferSize == null) {
      gatewaySenderFactory.setSocketBufferSize(GatewaySender.DEFAULT_SOCKET_BUFFER_SIZE);
    } else {
      gatewaySenderFactory.setSocketBufferSize(Integer.parseInt(socketBufferSize));
    }

    // socket-read-timeout
    String socketReadTimeout = atts.getValue(SOCKET_READ_TIMEOUT);
    if (socketReadTimeout == null) {
      gatewaySenderFactory.setSocketReadTimeout(GatewaySender.DEFAULT_SOCKET_READ_TIMEOUT);
    } else {
      gatewaySenderFactory.setSocketReadTimeout(Integer.parseInt(socketReadTimeout));
    }

    // batch-conflation
    String batchConflation = atts.getValue(ENABLE_BATCH_CONFLATION);
    if (batchConflation == null) {
      gatewaySenderFactory.setBatchConflationEnabled(GatewaySender.DEFAULT_BATCH_CONFLATION);
    } else {
      gatewaySenderFactory.setBatchConflationEnabled(Boolean.parseBoolean(batchConflation));
    }

    // batch-size
    String batchSize = atts.getValue(BATCH_SIZE);
    if (batchSize == null) {
      gatewaySenderFactory.setBatchSize(GatewaySender.DEFAULT_BATCH_SIZE);
    } else {
      gatewaySenderFactory.setBatchSize(Integer.parseInt(batchSize));
    }

    // batch-time-interval
    String batchTimeInterval = atts.getValue(BATCH_TIME_INTERVAL);
    if (batchTimeInterval == null) {
      gatewaySenderFactory.setBatchTimeInterval(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
    } else {
      gatewaySenderFactory.setBatchTimeInterval(Integer.parseInt(batchTimeInterval));
    }

    // enable-persistence
    String enablePersistence = atts.getValue(ENABLE_PERSISTENCE);
    if (enablePersistence == null) {
      gatewaySenderFactory.setPersistenceEnabled(GatewaySender.DEFAULT_PERSISTENCE_ENABLED);
    } else {
      gatewaySenderFactory.setPersistenceEnabled(Boolean.parseBoolean(enablePersistence));
    }

    String diskStoreName = atts.getValue(DISK_STORE_NAME);
    gatewaySenderFactory.setDiskStoreName(diskStoreName);

    String diskSynchronous = atts.getValue(DISK_SYNCHRONOUS);
    if (diskSynchronous == null) {
      gatewaySenderFactory.setDiskSynchronous(GatewaySender.DEFAULT_DISK_SYNCHRONOUS);
    } else {
      gatewaySenderFactory.setDiskSynchronous(Boolean.parseBoolean(diskSynchronous));
    }

    // maximum-queue-memory
    String maxQueueMemory = atts.getValue(MAXIMUM_QUEUE_MEMORY);
    if (maxQueueMemory == null) {
      gatewaySenderFactory.setMaximumQueueMemory(GatewaySender.DEFAULT_MAXIMUM_QUEUE_MEMORY);
    } else {
      gatewaySenderFactory.setMaximumQueueMemory(Integer.parseInt(maxQueueMemory));
    }


    String alertThreshold = atts.getValue(ALERT_THRESHOLD);
    if (alertThreshold == null) {
      gatewaySenderFactory.setAlertThreshold(GatewaySender.DEFAULT_ALERT_THRESHOLD);
    } else {
      gatewaySenderFactory.setAlertThreshold(Integer.parseInt(alertThreshold));
    }

    String dispatcherThreads = atts.getValue(DISPATCHER_THREADS);
    if (dispatcherThreads == null) {
      gatewaySenderFactory.setDispatcherThreads(GatewaySender.DEFAULT_DISPATCHER_THREADS);
    } else {
      gatewaySenderFactory.setDispatcherThreads(Integer.parseInt(dispatcherThreads));
    }

    String id = atts.getValue(ID);
    String orderPolicy = atts.getValue(ORDER_POLICY);
    if (orderPolicy != null) {
      try {
        gatewaySenderFactory
            .setOrderPolicy(GatewaySender.OrderPolicy.valueOf(orderPolicy.toUpperCase()));
      } catch (IllegalArgumentException e) {
        throw new InternalGemFireException(
            String.format("An invalid order-policy value (%s) was configured for gateway sender %s",
                id, orderPolicy));
      }
    }

    String remoteDS = atts.getValue(REMOTE_DISTRIBUTED_SYSTEM_ID);
    stack.push(id);
    stack.push(remoteDS);
    stack.push(gatewaySenderFactory);
    // GatewaySender sender = gatewaySenderFactory.create(id, Integer.parseInt(remoteDS));
    // stack.push(sender);

    String groupTransactionEvents = atts.getValue(GROUP_TRANSACTION_EVENTS);
    if (groupTransactionEvents == null) {
      gatewaySenderFactory
          .setGroupTransactionEvents(GatewaySender.DEFAULT_MUST_GROUP_TRANSACTION_EVENTS);
    } else {
      gatewaySenderFactory
          .setGroupTransactionEvents(Boolean.parseBoolean(groupTransactionEvents));
    }

    String enforceThreadsConnectSameReceiver = atts.getValue(ENFORCE_THREADS_CONNECT_SAME_RECEIVER);
    if (enforceThreadsConnectSameReceiver == null) {
      gatewaySenderFactory
          .setEnforceThreadsConnectSameReceiver(
              GatewaySender.DEFAULT_ENFORCE_THREADS_CONNECT_SAME_RECEIVER);
    } else {
      gatewaySenderFactory
          .setEnforceThreadsConnectSameReceiver(
              Boolean.parseBoolean(enforceThreadsConnectSameReceiver));
    }
  }

  private void startGatewayReceiver(Attributes atts) {
    GatewayReceiverFactory receiverFactory = cache.createGatewayReceiverFactory();

    // port
    String startPort = atts.getValue(START_PORT);
    if (startPort == null) {
      receiverFactory.setStartPort(GatewayReceiver.DEFAULT_START_PORT);
    } else {
      receiverFactory.setStartPort(Integer.parseInt(startPort));
    }

    String endPort = atts.getValue(END_PORT);
    if (endPort == null) {
      receiverFactory.setEndPort(GatewayReceiver.DEFAULT_END_PORT);
    } else {
      receiverFactory.setEndPort(Integer.parseInt(endPort));
    }

    String bindAddress = atts.getValue(BIND_ADDRESS);
    if (bindAddress == null) {
      receiverFactory.setBindAddress(GatewayReceiver.DEFAULT_BIND_ADDRESS);
    } else {
      receiverFactory.setBindAddress(bindAddress);
    }

    // maximum-time-between-pings
    String maxTimeBetweenPings = atts.getValue(MAXIMUM_TIME_BETWEEN_PINGS);
    if (maxTimeBetweenPings == null) {
      receiverFactory
          .setMaximumTimeBetweenPings(GatewayReceiver.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS);
    } else {
      receiverFactory.setMaximumTimeBetweenPings(Integer.parseInt(maxTimeBetweenPings));
    }

    // socket-buffer-size
    String socketBufferSize = atts.getValue(SOCKET_BUFFER_SIZE);
    if (socketBufferSize == null) {
      receiverFactory.setSocketBufferSize(GatewayReceiver.DEFAULT_SOCKET_BUFFER_SIZE);
    } else {
      receiverFactory.setSocketBufferSize(Integer.parseInt(socketBufferSize));
    }

    // manual-start
    String manualStart = atts.getValue(MANUAL_START);
    if (manualStart == null) {
      receiverFactory.setManualStart(GatewayReceiver.DEFAULT_MANUAL_START);
    } else {
      receiverFactory.setManualStart(Boolean.parseBoolean(manualStart));
    }

    // hostname-for-senders
    String hostnameForSenders = atts.getValue(HOSTNAME_FOR_SENDERS);
    if (hostnameForSenders == null) {
      receiverFactory.setHostnameForSenders(GatewayReceiver.DEFAULT_HOSTNAME_FOR_SENDERS);
    } else {
      receiverFactory.setHostnameForSenders(hostnameForSenders);
    }

    stack.push(receiverFactory);
  }


  /**
   * set attributes from clientHaQueue when we finish a cache server
   */
  private void endCacheServer() {
    List groups = new ArrayList();
    ServerLoadProbe probe = null;
    ClientHaQueueCreation haCreation = null;

    if (stack.peek() instanceof ServerLoadProbe) {
      probe = (ServerLoadProbe) stack.pop();
    }



    if (stack.peek() instanceof ClientHaQueueCreation) {
      haCreation = (ClientHaQueueCreation) stack.pop();
    }

    while (stack.peek() instanceof String) {
      groups.add(stack.pop());
    }
    CacheServer bs = (CacheServer) stack.pop();
    if (groups.size() > 0) {
      Collections.reverse(groups);
      String[] groupArray = new String[groups.size()];
      groups.toArray(groupArray);
      bs.setGroups(groupArray);
    }
    if (probe != null) {
      bs.setLoadProbe(probe);
    }
    if (haCreation != null) {
      ClientSubscriptionConfig csc = bs.getClientSubscriptionConfig();
      String diskStoreName = haCreation.getDiskStoreName();
      if (diskStoreName != null) {
        csc.setDiskStoreName(diskStoreName);
      } else {
        csc.setOverflowDirectory(haCreation.getOverflowDirectory() == null
            ? ClientSubscriptionConfig.DEFAULT_OVERFLOW_DIRECTORY
            : haCreation.getOverflowDirectory());
      }
      csc.setCapacity(haCreation.getCapacity());
      csc.setEvictionPolicy(haCreation.getEvictionPolicy());
    }
  }

  /**
   * When a <code>load-probe</code> element is encountered, create a new probe for the current
   * <code>CacheServer</code>.
   *
   * @since GemFire 5.7
   */
  private void endLoadProbe() {
    Declarable d = createDeclarable();
    if (!(d instanceof ServerLoadProbe)) {
      throw new CacheXmlException(String.format("A %s is not an instance of a %s",
          d.getClass().getName(), "BridgeLoadProbe"));
    }
    stack.push(d);
  }

  private void endSerialGatewaySender() {
    GatewaySenderFactory senderFactory = (GatewaySenderFactory) stack.pop();
    String remoteDSString = (String) stack.pop();
    String id = (String) stack.pop();
    senderFactory.create(id, Integer.parseInt(remoteDSString));
  }

  private void endGatewayReceiver() {
    GatewayReceiverFactory receiverFactory = (GatewayReceiverFactory) stack.pop();
    receiverFactory.create();
  }

  private void startDynamicRegionFactory(Attributes atts) {
    // push attributes onto the stack for processing in endDynamicRegionFactory
    String disablePersist = atts.getValue(DISABLE_PERSIST_BACKUP);
    if (disablePersist == null) {
      stack.push("false");
    } else {
      stack.push(disablePersist);
    }
    String disableRegisterInterest = atts.getValue(DISABLE_REGISTER_INTEREST);
    if (disableRegisterInterest == null) {
      stack.push("false");
    } else {
      stack.push(disableRegisterInterest);
    }
    String poolName = atts.getValue(POOL_NAME);
    stack.push(poolName);

    // hi-jack RegionAttributesCreation for the disk-dirs, loader, writer and compressor
    RegionAttributesCreation attrs = new RegionAttributesCreation(cache);
    stack.push(attrs);
  }


  private void endDynamicRegionFactory() {
    File dir = null;
    RegionAttributesCreation attrs;
    {
      Object o = stack.pop();
      if (o instanceof File) {
        dir = (File) o;
        stack.pop(); // dir size to be popped out. being used by persistent directories
        attrs = (RegionAttributesCreation) stack.pop();
      } else {
        attrs = (RegionAttributesCreation) o;
      }
    }
    String poolName = (String) stack.pop();
    String disableRegisterInterest = (String) stack.pop();
    String disablePersistBackup = (String) stack.pop();
    DynamicRegionFactory.Config cfg;
    cfg = new DynamicRegionFactory.Config(dir, poolName,
        !Boolean.parseBoolean(disablePersistBackup),
        !Boolean.parseBoolean(disableRegisterInterest));
    CacheCreation cache = (CacheCreation) stack.peek();
    cache.setDynamicRegionFactoryConfig(cfg);
  }

  /**
   * When a <code>gateway-conflict-resolver</code> element is encountered, create a new listener for
   * the <code>Cache</code>.
   */
  private void endGatewayConflictResolver() {
    Declarable d = createDeclarable();
    if (!(d instanceof GatewayConflictResolver)) {
      throw new CacheXmlException(
          String.format("A %s is not an instance of a GatewayConflictResolver.",
              d.getClass().getName()));
    }
    CacheCreation c = (CacheCreation) stack.peek();
    c.setGatewayConflictResolver((GatewayConflictResolver) d);
  }

  /**
   * When a <code>region</code> element is first encountered, we create a {@link RegionCreation}and
   * push it on the stack.
   */
  private void startRegion(Attributes atts) {
    String name = atts.getValue(NAME);
    String refid = atts.getValue(REFID);
    Assert.assertTrue(name != null);
