package org.infinispan.configuration.parsing;

import static org.infinispan.configuration.parsing.ParseUtils.ignoreAttribute;
import static org.infinispan.configuration.parsing.Parser.NAMESPACE;
import static org.infinispan.factories.KnownComponentNames.ASYNC_NOTIFICATION_EXECUTOR;
import static org.infinispan.factories.KnownComponentNames.BLOCKING_EXECUTOR;
import static org.infinispan.factories.KnownComponentNames.EXPIRATION_SCHEDULED_EXECUTOR;
import static org.infinispan.factories.KnownComponentNames.NON_BLOCKING_EXECUTOR;
import static org.infinispan.factories.KnownComponentNames.shortened;
import static org.infinispan.util.logging.Log.CONFIG;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.net.URL;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Supplier;

import org.infinispan.commons.configuration.io.ConfigurationReader;
import org.infinispan.commons.configuration.io.ConfigurationResourceResolver;
import org.infinispan.commons.configuration.io.NamingStrategy;
import org.infinispan.commons.marshall.AdvancedExternalizer;
import org.infinispan.commons.util.FileLookupFactory;
import org.infinispan.commons.util.Util;
import org.infinispan.configuration.cache.AbstractStoreConfigurationBuilder;
import org.infinispan.configuration.cache.AsyncStoreConfigurationBuilder;
import org.infinispan.configuration.cache.StoreConfigurationBuilder;
import org.infinispan.configuration.global.AllowListConfigurationBuilder;
import org.infinispan.configuration.global.GlobalAuthorizationConfigurationBuilder;
import org.infinispan.configuration.global.GlobalConfigurationBuilder;
import org.infinispan.configuration.global.GlobalRoleConfigurationBuilder;
import org.infinispan.configuration.global.GlobalSecurityConfigurationBuilder;
import org.infinispan.configuration.global.GlobalStateConfigurationBuilder;
import org.infinispan.configuration.global.SerializationConfigurationBuilder;
import org.infinispan.configuration.global.ShutdownHookBehavior;
import org.infinispan.configuration.global.ThreadPoolBuilderAdapter;
import org.infinispan.configuration.global.ThreadPoolConfiguration;
import org.infinispan.configuration.global.ThreadsConfigurationBuilder;
import org.infinispan.configuration.global.TransportConfigurationBuilder;
import org.infinispan.factories.threads.DefaultThreadFactory;
import org.infinispan.globalstate.ConfigurationStorage;
import org.infinispan.globalstate.LocalConfigurationStorage;
import org.infinispan.remoting.transport.Transport;
import org.infinispan.remoting.transport.jgroups.BuiltinJGroupsChannelConfigurator;
import org.infinispan.remoting.transport.jgroups.EmbeddedJGroupsChannelConfigurator;
import org.infinispan.remoting.transport.jgroups.FileJGroupsChannelConfigurator;
import org.infinispan.remoting.transport.jgroups.JGroupsChannelConfigurator;
import org.infinispan.remoting.transport.jgroups.JGroupsTransport;
import org.infinispan.security.PrincipalRoleMapper;
import org.infinispan.security.mappers.ClusterRoleMapper;
import org.infinispan.security.mappers.CommonNameRoleMapper;
import org.infinispan.security.mappers.IdentityRoleMapper;
import org.jgroups.conf.ProtocolConfiguration;
import org.kohsuke.MetaInfServices;

/**
 * This class implements the parser for Infinispan/AS7/EAP/JDG schema files
 *
 * @author Tristan Tarrant
 * @author Galder Zamarreño
 * @since 9.0
 */
@MetaInfServices(ConfigurationParser.class)
@Namespace(root = "infinispan")
@Namespace(uri = NAMESPACE + "*", root = "infinispan")
public class Parser extends CacheParser {

   public static final String NAMESPACE = "urn:infinispan:config:";

   public Parser() {
   }

   @Override
   public void readElement(final ConfigurationReader reader, final ConfigurationBuilderHolder holder) {
      while (reader.inTag("infinispan")) {
         Element element = Element.forName(reader.getLocalName());
         switch (element) {
            case CACHE_CONTAINER: {
               parseContainer(reader, holder);
               break;
            }
            case JGROUPS: {
               // Preload some default JGroups stacks
               addJGroupsDefaultStacksIfNeeded(reader, holder);
               parseJGroups(reader, holder);
               break;
            }
            case THREADS: {
               parseThreads(reader, holder);
               break;
            }
            default: {
               reader.handleAny(holder);
               break;
            }
         }
      }
   }

   private void parseSerialization(final ConfigurationReader reader, final ConfigurationBuilderHolder holder) {
      GlobalConfigurationBuilder builder = holder.getGlobalConfigurationBuilder();
      for (int i = 0; i < reader.getAttributeCount(); i++) {
         ParseUtils.requireNoNamespaceAttribute(reader, i);
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));

         switch (attribute) {
            case MARSHALLER_CLASS: {
               builder.serialization().marshaller(Util.getInstance(value, holder.getClassLoader()));
               break;
            }
            case VERSION: {
               if (reader.getSchema().since(11, 0)) {
                  throw ParseUtils.attributeRemoved(reader, i);
               }
               ignoreAttribute(reader, i);
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }



      while (reader.hasNext()) {
         reader.nextElement();
         Element element = Element.forName(reader.getLocalName());
         switch (element) {
            case SERIALIZATION: {
               reader.require(ConfigurationReader.ElementType.END_ELEMENT);
               return;
            }
            case ADVANCED_EXTERNALIZERS:
            case SERIALIZATION_CONTEXT_INITIALIZERS: {
               // Empty elements for YAML/JSON
               break;
            }
            case ADVANCED_EXTERNALIZER: {
               CONFIG.advancedExternalizerDeprecated();
               parseAdvancedExternalizer(reader, holder.getClassLoader(), builder.serialization());
               break;
            }
            case SERIALIZATION_CONTEXT_INITIALIZER: {
               parseSerializationContextInitializer(reader, holder.getClassLoader(), builder.serialization());
               break;
            }
            case WHITE_LIST:
               if (reader.getSchema().since(12, 0)) {
                  throw ParseUtils.elementRemoved(reader, Element.ALLOW_LIST.getLocalName());
               }
               CONFIG.elementDeprecatedUseOther(Element.WHITE_LIST, Element.ALLOW_LIST);
               parseAllowList(reader, builder.serialization().allowList(), Element.WHITE_LIST);
               break;
            case ALLOW_LIST: {
               if (reader.getSchema().since(10, 0)) {
                  parseAllowList(reader, builder.serialization().allowList(), Element.ALLOW_LIST);
                  break;
               } else {
                  throw ParseUtils.unexpectedElement(reader);
               }
            }
            default: {
               throw ParseUtils.unexpectedElement(reader);
            }
         }
      }
   }

   private void parseSerializationContextInitializer(final ConfigurationReader reader, final ClassLoader classLoader,
                                                     final SerializationConfigurationBuilder builder) {
      int attributes = reader.getAttributeCount();
      ParseUtils.requireAttributes(reader, Attribute.CLASS.getLocalName());
      for (int i = 0; i < attributes; i++) {
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));
         switch (attribute) {
            case CLASS: {
               builder.addContextInitializer(Util.getInstance(value, classLoader));
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }
      ParseUtils.requireNoContent(reader);
   }

   private void parseAllowList(final ConfigurationReader reader, final AllowListConfigurationBuilder builder, Element outerElement) {
      while (reader.inTag(outerElement)) {
         Element element = Element.forName(reader.getLocalName());
         switch (element) {
            case CLASS: {
               builder.addClass(reader.getElementText());
               break;
            }
            case REGEX: {
               builder.addRegexp(reader.getElementText());
               break;
            }
            default: {
               throw ParseUtils.unexpectedElement(reader);
            }
         }
      }
   }

   private void parseAdvancedExternalizer(final ConfigurationReader reader, final ClassLoader classLoader,
                                          final SerializationConfigurationBuilder builder) {
      int attributes = reader.getAttributeCount();
      AdvancedExternalizer<?> advancedExternalizer = null;
      Integer id = null;
      ParseUtils.requireAttributes(reader, Attribute.CLASS.getLocalName());
      for (int i = 0; i < attributes; i++) {
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));
         switch (attribute) {
            case CLASS: {
               advancedExternalizer = Util.getInstance(value, classLoader);
               break;
            }
            case ID: {
               id = Integer.valueOf(value);
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }

      ParseUtils.requireNoContent(reader);

      if (id != null) {
         builder.addAdvancedExternalizer(id, advancedExternalizer);
      } else {
         builder.addAdvancedExternalizer(advancedExternalizer);
      }
   }

   private void parseThreads(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      while (reader.hasNext()) {
         reader.nextElement();
         Element element = Element.forName(reader.getLocalName());
         switch (element) {
            case THREADS: {
               reader.require(ConfigurationReader.ElementType.END_ELEMENT);
               return;
            }
            case THREAD_FACTORIES:
            case THREAD_POOLS: {
               // Wrapper elements for YAML/JSON parsers
               break;
            }
            case THREAD_FACTORY: {
               parseThreadFactory(reader, holder);
               break;
            }
            case CACHED_THREAD_POOL: {
               parseCachedThreadPool(reader, holder);
               break;
            }
            case SCHEDULED_THREAD_POOL: {
               parseScheduledThreadPool(reader, holder);
               break;
            }
            case BLOCKING_BOUNDED_QUEUE_THREAD_POOL: {
               parseBoundedQueueThreadPool(reader, holder, false);
               break;
            }
            case NON_BLOCKING_BOUNDED_QUEUE_THREAD_POOL: {
               parseBoundedQueueThreadPool(reader, holder, true);
               break;
            }
            default: {
               throw ParseUtils.unexpectedElement(reader);
            }
         }
      }
   }

   private void parseBoundedQueueThreadPool(ConfigurationReader reader, ConfigurationBuilderHolder holder,
         boolean isNonBlocking) {
      ThreadsConfigurationBuilder threadsBuilder = holder.getGlobalConfigurationBuilder().threads();

      String name = null;
      String threadFactoryName = null;
      int maxThreads = 0;
      int coreThreads = 0;
      int queueLength = 0;
      long keepAlive = 0;

      for (int i = 0; i < reader.getAttributeCount(); i++) {
         ParseUtils.requireNoNamespaceAttribute(reader, i);
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));

         switch (attribute) {
            case NAME: {
               name = value;
               break;
            }
            case THREAD_FACTORY: {
               threadFactoryName = value;
               break;
            }
            case CORE_THREADS: {
               coreThreads = Integer.parseInt(value);
               break;
            }
            case MAX_THREADS: {
               maxThreads = Integer.parseInt(value);
               break;
            }
            case QUEUE_LENGTH: {
               queueLength = Integer.parseInt(value);
               break;
            }
            case KEEP_ALIVE_TIME: {
               keepAlive = Long.parseLong(value);
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }

      threadsBuilder.addBoundedThreadPool(name).threadFactory(threadFactoryName).coreThreads(coreThreads)
            .maxThreads(maxThreads).queueLength(queueLength).keepAliveTime(keepAlive).nonBlocking(isNonBlocking);
      ParseUtils.requireNoContent(reader);
   }

   private void parseScheduledThreadPool(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      ThreadsConfigurationBuilder threadsBuilder = holder.getGlobalConfigurationBuilder().threads();
      String name = null;
      String threadFactoryName = null;
      for (int i = 0; i < reader.getAttributeCount(); i++) {
         ParseUtils.requireNoNamespaceAttribute(reader, i);
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));

         switch (attribute) {
            case NAME: {
               name = value;
               break;
            }
            case THREAD_FACTORY: {
               threadFactoryName = value;
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }

      threadsBuilder.addScheduledThreadPool(name).threadFactory(threadFactoryName);

      ParseUtils.requireNoContent(reader);
   }

   private void parseCachedThreadPool(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      ThreadsConfigurationBuilder threadsBuilder = holder.getGlobalConfigurationBuilder().threads();
      String name = null;
      String threadFactoryName = null;
      for (int i = 0; i < reader.getAttributeCount(); i++) {
         ParseUtils.requireNoNamespaceAttribute(reader, i);
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));

         switch (attribute) {
            case NAME: {
               name = value;
               break;
            }
            case THREAD_FACTORY: {
               threadFactoryName = value;
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }

      threadsBuilder.addCachedThreadPool(name).threadFactory(threadFactoryName);

      ParseUtils.requireNoContent(reader);
   }

   private void parseThreadFactory(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      String name = null;
      String threadGroupName = null;
      String threadNamePattern = null;
      int priority = 1; // minimum priority

      for (int i = 0; i < reader.getAttributeCount(); i++) {
         ParseUtils.requireNoNamespaceAttribute(reader, i);
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));

         switch (attribute) {
            case NAME: {
               name = value;
               break;
            }
            case GROUP_NAME: {
               threadGroupName = value;
               break;
            }
            case THREAD_NAME_PATTERN: {
               threadNamePattern = value;
               break;
            }
            case PRIORITY: {
               priority = Integer.parseInt(value);
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }

      holder.getGlobalConfigurationBuilder().threads().addThreadFactory(name).groupName(threadGroupName).priority(priority).threadNamePattern(threadNamePattern);
      ParseUtils.requireNoContent(reader);
   }

   private void parseJGroups(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      Transport transport = null;
      for (int i = 0; i < reader.getAttributeCount(); i++) {
            if (!ParseUtils.isNoNamespaceAttribute(reader, i))
            continue;
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));

         switch (attribute) {
            case TRANSPORT:
               transport = Util.getInstance(value, holder.getClassLoader());
               break;
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }

      if (transport == null) {
         // Set up default transport
         holder.getGlobalConfigurationBuilder().transport().defaultTransport();
      } else {
         holder.getGlobalConfigurationBuilder().transport().transport(transport);
      }

      while (reader.inTag(Element.JGROUPS)) {
         Element element = Element.forName(reader.getLocalName());
         switch (element) {
            case STACKS: {
               // Wrapper element for YAML/JSON
               break;
            }
            case STACK_FILE:
               parseStackFile(reader, holder);
               break;
            case STACK:
               if (!reader.getSchema().since(10, 0)) {
                  throw ParseUtils.unexpectedElement(reader);
               }
               parseJGroupsStack(reader, holder);
               break;
            default: {
               throw ParseUtils.unexpectedElement(reader);
            }
         }
      }
   }

   private void addJGroupsStackFile(ConfigurationBuilderHolder holder, String name, String path, Properties properties, ConfigurationResourceResolver resourceResolver) {
      URL url = FileLookupFactory.newInstance().lookupFileLocation(path, holder.getClassLoader());
      try (InputStream xml = (url != null ? url : resourceResolver.resolveResource(path)).openStream()) {
         holder.addJGroupsStack(new FileJGroupsChannelConfigurator(name, path, xml, properties));
      } catch (FileNotFoundException e) {
         throw CONFIG.jgroupsConfigurationNotFound(path);
      } catch (IOException e) {
         throw CONFIG.unableToAddJGroupsStack(name, e);
      }
   }

   private void parseJGroupsStack(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      String stackName = ParseUtils.requireAttributes(reader, Attribute.NAME)[0];
      EmbeddedJGroupsChannelConfigurator stackConfigurator = new EmbeddedJGroupsChannelConfigurator(stackName);
      String extend = null;
      for (int i = 0; i < reader.getAttributeCount(); i++) {
         ParseUtils.requireNoNamespaceAttribute(reader, i);
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));
         switch (attribute) {
            case NAME:
               break;
            case EXTENDS:
               extend = value;
               break;
            default:
               throw ParseUtils.unexpectedAttribute(reader, i);
         }
      }

      List<ProtocolConfiguration> stack = stackConfigurator.getProtocolStack();
      while (reader.inTag()) {
         Element element = Element.forName(reader.getLocalName());
         switch (element) {
            case REMOTE_SITES:
               parseJGroupsRelay(reader, holder, stackConfigurator);
               break;
            default:
               // It should be an actual JGroups protocol
               String protocolName = reader.getLocalName(NamingStrategy.IDENTITY);
               Map<String, String> protocolAttributes = new HashMap<>();
               for (int i = 0; i < reader.getAttributeCount(); i++) {
                  protocolAttributes.put(reader.getAttributeName(i, NamingStrategy.SNAKE_CASE), reader.getAttributeValue(i));
               }
               ParseUtils.requireNoContent(reader);
               stack.add(new ProtocolConfiguration(protocolName, protocolAttributes));
               break;
         }
      }
      holder.addJGroupsStack(stackConfigurator, extend);
   }

   private void parseJGroupsRelay(ConfigurationReader reader, ConfigurationBuilderHolder holder, EmbeddedJGroupsChannelConfigurator stackConfigurator) {
      String defaultStack = ParseUtils.requireSingleAttribute(reader, Attribute.DEFAULT_STACK);
      if (holder.getJGroupsStack(defaultStack) == null) {
         throw CONFIG.missingJGroupsStack(defaultStack);
      }
      while (reader.inTag()) {
         Element element = Element.forName(reader.getLocalName());
         switch (element) {
            case REMOTE_SITE:
               String remoteSite = ParseUtils.requireAttributes(reader, Attribute.NAME)[0];
               String stack = defaultStack;
               for (int i = 0; i < reader.getAttributeCount(); i++) {
                  Attribute attribute = Attribute.forName(reader.getAttributeName(i));
                  switch (attribute) {
                     case NAME:
                        break;
                     case STACK:
                        stack = reader.getAttributeValue(i);
                        if (holder.getJGroupsStack(stack) == null) {
                           throw CONFIG.missingJGroupsStack(stack);
                        }
                        break;
                     default:
                        throw ParseUtils.unexpectedAttribute(reader, i);
                  }
               }
               ParseUtils.requireNoContent(reader);
               stackConfigurator.addRemoteSite(remoteSite, holder.getJGroupsStack(stack));
               break;
            default:
               throw ParseUtils.unexpectedElement(reader);
         }
      }
   }

   private void parseStackFile(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      String[] attributes = ParseUtils.requireAttributes(reader, Attribute.NAME, Attribute.PATH);
      ParseUtils.requireNoContent(reader);

      addJGroupsStackFile(holder, attributes[0], attributes[1], reader.getProperties(), reader.getResourceResolver());
   }

   private void parseContainer(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      holder.pushScope(ParserScope.CACHE_CONTAINER);
      GlobalConfigurationBuilder builder = holder.getGlobalConfigurationBuilder();
      for (int i = 0; i < reader.getAttributeCount(); i++) {
         if (!ParseUtils.isNoNamespaceAttribute(reader, i))
            continue;
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));
         switch (attribute) {
            case NAME: {
               builder.cacheManagerName(value);
               break;
            }
            case DEFAULT_CACHE: {
               builder.defaultCacheName(value);
               break;
            }
            case ALIASES:
            case JNDI_NAME:
            case MODULE:
            case START:
            case ASYNC_EXECUTOR:
            case PERSISTENCE_EXECUTOR:
            case STATE_TRANSFER_EXECUTOR: {
               if (reader.getSchema().since(11, 0)) {
                  throw ParseUtils.attributeRemoved(reader, i);
               } else {
                  ignoreAttribute(reader, i);
               }
               break;
            }
            case LISTENER_EXECUTOR: {
               builder.listenerThreadPoolName(value);
               builder.listenerThreadPool().read(createThreadPoolConfiguration(value, ASYNC_NOTIFICATION_EXECUTOR, holder));
               break;
            }
            case EVICTION_EXECUTOR:
               if (reader.getSchema().since(11, 0)) {
                  throw ParseUtils.attributeRemoved(reader, i);
               } else {
                  CONFIG.evictionExecutorDeprecated();
               }
               // fallthrough
            case EXPIRATION_EXECUTOR: {
               builder.expirationThreadPoolName(value);
               builder.expirationThreadPool().read(createThreadPoolConfiguration(value, EXPIRATION_SCHEDULED_EXECUTOR, holder));
               break;
            }
            case REPLICATION_QUEUE_EXECUTOR: {
               if (reader.getSchema().since(9, 0)) {
                  throw ParseUtils.attributeRemoved(reader, i);
               } else {
                  CONFIG.ignoredReplicationQueueAttribute(attribute.getLocalName(), reader.getLocation().getLineNumber());
               }
               break;
            }
            case NON_BLOCKING_EXECUTOR: {
               builder.nonBlockingThreadPoolName(value);
               builder.nonBlockingThreadPool().read(createThreadPoolConfiguration(value, NON_BLOCKING_EXECUTOR, holder));
               break;
            }
            case BLOCKING_EXECUTOR: {
               builder.blockingThreadPoolName(value);
               builder.blockingThreadPool().read(createThreadPoolConfiguration(value, BLOCKING_EXECUTOR, holder));
               break;
            }
            case STATISTICS: {
               boolean statistics = Boolean.parseBoolean(value);
               builder.cacheContainer().statistics(statistics);
               if (!reader.getSchema().since(10, 1)) {
                  builder.jmx().enabled(statistics);
               }
               break;
            }
            case SHUTDOWN_HOOK: {
               builder.shutdown().hookBehavior(ShutdownHookBehavior.valueOf(value));
               break;
            }
            case ZERO_CAPACITY_NODE: {
               builder.zeroCapacityNode(Boolean.parseBoolean(value));
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }

      while (reader.inTag(Element.CACHE_CONTAINER)) {
         Element element = Element.forName(reader.getLocalName());
         switch (element) {
            case CACHES: {
               // Wrapper element for YAML/JSON
               break;
            }
            case TRANSPORT: {
               parseTransport(reader, holder);
               break;
            }
            case LOCAL_CACHE: {
               parseLocalCache(reader, holder, false);
               break;
            }
            case LOCAL_CACHE_CONFIGURATION: {
               parseLocalCache(reader, holder, true);
               break;
            }
            case INVALIDATION_CACHE: {
               parseInvalidationCache(reader, holder, false);
               break;
            }
            case INVALIDATION_CACHE_CONFIGURATION: {
               parseInvalidationCache(reader, holder, true);
               break;
            }
            case REPLICATED_CACHE: {
               parseReplicatedCache(reader, holder, false);
               break;
            }
            case REPLICATED_CACHE_CONFIGURATION: {
               parseReplicatedCache(reader, holder, true);
               break;
            }
            case DISTRIBUTED_CACHE: {
               parseDistributedCache(reader, holder, false);
               break;
            }
            case DISTRIBUTED_CACHE_CONFIGURATION: {
               parseDistributedCache(reader, holder, true);
               break;
            }
            case SCATTERED_CACHE: {
               if (reader.getSchema().since(9, 1)) {
                  parseScatteredCache(reader, holder, false);
               } else {
                  throw ParseUtils.unexpectedElement(reader);
               }
               break;
            }
            case SCATTERED_CACHE_CONFIGURATION: {
               if (reader.getSchema().since(9, 1)) {
                  parseScatteredCache(reader, holder, true);
               } else {
                  throw ParseUtils.unexpectedElement(reader);
               }
               break;
            }
            case SERIALIZATION: {
               parseSerialization(reader, holder);
               break;
            }
            case MODULES: {
               if (reader.getSchema().since(9, 0)) {
                  throw ParseUtils.elementRemoved(reader);
               } else {
                  parseModules(reader, holder);
               }
               break;
            }
            case METRICS: {
               parseMetrics(reader, holder);
               break;
            }
            case JMX: {
               parseJmx(reader, holder);
               break;
            }
            case SECURITY: {
               parseGlobalSecurity(reader, holder);
               break;
            }
            case GLOBAL_STATE: {
               if (reader.getSchema().since(8, 1)) {
                  parseGlobalState(reader, holder);
               } else {
                  throw ParseUtils.unexpectedElement(reader);
               }
               break;
            }
            default: {
               reader.handleAny(holder);
            }
         }
      }
      holder.popScope();
   }

   private void parseGlobalSecurity(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      GlobalSecurityConfigurationBuilder security = holder.getGlobalConfigurationBuilder().security();
      for (int i = 0; i < reader.getAttributeCount(); i++) {
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));
         switch (attribute) {
            case CACHE_SIZE: {
               security.securityCacheSize(Integer.parseInt(value));
               break;
            }
            case CACHE_TIMEOUT: {
               security.securityCacheTimeout(Long.parseLong(value), TimeUnit.MILLISECONDS);
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }

      while (reader.inTag()) {
         Element element = Element.forName(reader.getLocalName());
         switch (element) {
            case AUTHORIZATION: {
               parseGlobalAuthorization(reader, holder);
               break;
            }
            default: {
               throw ParseUtils.unexpectedElement(reader);
            }
         }
      }
   }

   private void parseGlobalAuthorization(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      GlobalAuthorizationConfigurationBuilder builder = holder.getGlobalConfigurationBuilder().security().authorization().enable();
      for (int i = 0; i < reader.getAttributeCount(); i++) {
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));
         switch (attribute) {
            case AUDIT_LOGGER: {
               builder.auditLogger(Util.getInstance(value, holder.getClassLoader()));
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }
      PrincipalRoleMapper roleMapper = null;
      while (reader.hasNext()) {
         reader.nextElement();
         Element element = Element.forName(reader.getLocalName());
         switch (element) {
            case AUTHORIZATION: {
               reader.require(ConfigurationReader.ElementType.END_ELEMENT);
               if (roleMapper != null) {
                  builder.principalRoleMapper(roleMapper);
               }
               return;
            }
            case IDENTITY_ROLE_MAPPER:
               if (roleMapper != null) {
                  throw ParseUtils.unexpectedElement(reader);
               }
               ParseUtils.requireNoAttributes(reader);
               ParseUtils.requireNoContent(reader);
               roleMapper = new IdentityRoleMapper();
               break;
            case COMMON_NAME_ROLE_MAPPER:
               if (roleMapper != null) {
                  throw ParseUtils.unexpectedElement(reader);
               }
               ParseUtils.requireNoAttributes(reader);
               ParseUtils.requireNoContent(reader);
               roleMapper = new CommonNameRoleMapper();
               break;
            case CLUSTER_ROLE_MAPPER:
               if (roleMapper != null) {
                  throw ParseUtils.unexpectedElement(reader);
               }
               ParseUtils.requireNoAttributes(reader);
               ParseUtils.requireNoContent(reader);
               roleMapper = new ClusterRoleMapper();
               break;
            case CUSTOM_ROLE_MAPPER:
               if (roleMapper != null) {
                  throw ParseUtils.unexpectedElement(reader);
               }
               roleMapper = parseCustomMapper(reader, holder);
               break;
            case ROLES: {
               // Wrapper element for YAML/JSON
               break;
            }
            case ROLE: {
               parseGlobalRole(reader, builder);
               break;
            }
            default: {
               throw ParseUtils.unexpectedElement(reader);
            }
         }
      }
   }

   private PrincipalRoleMapper parseCustomMapper(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      String mapperClass = ParseUtils.requireSingleAttribute(reader, Attribute.CLASS.getLocalName());
      ParseUtils.requireNoContent(reader);
      return Util.getInstance(mapperClass, holder.getClassLoader());
   }

   private void parseGlobalRole(ConfigurationReader reader, GlobalAuthorizationConfigurationBuilder builder) {
      String[] attributes = ParseUtils.requireAttributes(reader, Attribute.NAME.getLocalName(), Attribute.PERMISSIONS.getLocalName());
      GlobalRoleConfigurationBuilder role = builder.role(attributes[0]);
      for(String permission : attributes[1].split("\\s+")) {
         role.permission(permission);
      }
      for (int i = 0; i < reader.getAttributeCount(); i++) {
         ParseUtils.requireNoNamespaceAttribute(reader, i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));
         switch (attribute) {
            case NAME:
            case PERMISSIONS: {
               // Already handled
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }
      ParseUtils.requireNoContent(reader);
   }

   private void parseMetrics(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      GlobalConfigurationBuilder builder = holder.getGlobalConfigurationBuilder();
      for (int i = 0; i < reader.getAttributeCount(); i++) {
         ParseUtils.requireNoNamespaceAttribute(reader, i);
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));
         switch (attribute) {
            case GAUGES: {
               builder.metrics().gauges(Boolean.parseBoolean(value));
               break;
            }
            case HISTOGRAMS: {
               builder.metrics().histograms(Boolean.parseBoolean(value));
               break;
            }
            case PREFIX: {
               builder.metrics().prefix(value);
               break;
            }
            case NAMES_AS_TAGS: {
               builder.metrics().namesAsTags(Boolean.parseBoolean(value));
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }

      ParseUtils.requireNoContent(reader);
   }

   private void parseJmx(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      GlobalConfigurationBuilder builder = holder.getGlobalConfigurationBuilder();
      for (int i = 0; i < reader.getAttributeCount(); i++) {
         ParseUtils.requireNoNamespaceAttribute(reader, i);
         String value = reader.getAttributeValue(i);
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));
         switch (attribute) {
            case ENABLED: {
               builder.jmx().enabled(Boolean.parseBoolean(value));
               break;
            }
            case DOMAIN: {
               builder.jmx().domain(value);
               break;
            }
            case MBEAN_SERVER_LOOKUP: {
               builder.jmx().mBeanServerLookup(Util.getInstance(value, holder.getClassLoader()));
               break;
            }
            case ALLOW_DUPLICATE_DOMAINS: {
               if (!reader.getSchema().since(11, 0)) {
                  ignoreAttribute(reader, i);
                  break;
               } else {
                  throw ParseUtils.attributeRemoved(reader, i);
               }
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }

      Properties properties = parseProperties(reader, Element.JMX);
      builder.jmx().withProperties(properties);
   }

   private void parseModules(ConfigurationReader reader, ConfigurationBuilderHolder holder)
         {
      while (reader.inTag()) {
         reader.handleAny(holder);
      }
   }

   private void parseTransport(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      if (holder.getGlobalConfigurationBuilder().transport().getTransport() == null) {
         holder.getGlobalConfigurationBuilder().transport().defaultTransport();
      }
      TransportConfigurationBuilder transport = holder.getGlobalConfigurationBuilder().transport();
      for (int i = 0; i < reader.getAttributeCount(); i++) {
         String value = reader.getAttributeValue(i);
         if (ParseUtils.isNoNamespaceAttribute(reader, i)) {
            Attribute attribute = Attribute.forName(reader.getAttributeName(i));
            switch (attribute) {
               case STACK: {
                  // Make sure we have the default stacks added
                  addJGroupsDefaultStacksIfNeeded(reader, holder);
                  JGroupsChannelConfigurator jGroupsStack = holder.getJGroupsStack(value);
                  if (jGroupsStack == null) {
                     throw CONFIG.missingJGroupsStack(value);
                  }
                  Properties p = new Properties();
                  p.put(JGroupsTransport.CHANNEL_CONFIGURATOR, jGroupsStack);
                  p.put("stack", value);
                  transport.withProperties(p);
                  break;
               }
               case CLUSTER: {
                  transport.clusterName(value);
                  break;
               }
               case EXECUTOR: {
                  if (reader.getSchema().since(11, 0)) {
                     throw ParseUtils.attributeRemoved(reader, i);
                  } else {
                     CONFIG.ignoredAttribute("transport executor", "11.0", attribute.getLocalName(), reader.getLocation().getLineNumber());
                  }
                  break;
               }
               case TOTAL_ORDER_EXECUTOR: {
                  if (reader.getSchema().since(9, 0)) {
                     throw ParseUtils.attributeRemoved(reader, i);
                  } else {
                     CONFIG.ignoredAttribute("total order executor", "9.0", attribute.getLocalName(), reader.getLocation().getLineNumber());
                  }
                  break;
               }
               case REMOTE_COMMAND_EXECUTOR: {
                  if (reader.getSchema().since(11, 0)) {
                     throw ParseUtils.attributeRemoved(reader, i);
                  } else {
                     CONFIG.ignoredAttribute("remote command executor", "11.0", attribute.getLocalName(), reader.getLocation().getLineNumber());
                  }
                  break;
               }
               case LOCK_TIMEOUT: {
                  transport.distributedSyncTimeout(Long.parseLong(value));
                  break;
               }
               case NODE_NAME: {
                  transport.nodeName(value);
                  holder.getGlobalConfigurationBuilder().threads().nodeName(value);
                  break;
               }
               case LOCKING:
                  break;
               case MACHINE_ID: {
                  transport.machineId(value);
                  break;
               }
               case RACK_ID: {
                  transport.rackId(value);
                  break;
               }
               case SITE: {
                  transport.siteId(value);
                  break;
               }
               case INITIAL_CLUSTER_SIZE: {
                  if (reader.getSchema().since(8, 2)) {
                     transport.initialClusterSize(Integer.parseInt(value));
                  } else {
                     throw ParseUtils.unexpectedAttribute(reader, i);
                  }
                  break;
               }
               case INITIAL_CLUSTER_TIMEOUT: {
                  if (reader.getSchema().since(8, 2)) {
                     transport.initialClusterTimeout(Long.parseLong(value), TimeUnit.MILLISECONDS);
                  } else {
                     throw ParseUtils.unexpectedAttribute(reader, i);
                  }
                  break;
               }
               default: {
                  throw ParseUtils.unexpectedAttribute(reader, i);
               }
            }
         } else {
            reader.handleAttribute(holder, i);
         }
      }
      Properties properties = parseProperties(reader, Element.TRANSPORT);
      for (Map.Entry<Object, Object> propertyEntry : properties.entrySet()) {
        transport.addProperty((String) propertyEntry.getKey(), (String) propertyEntry.getValue());
      }
   }

   private void parseGlobalState(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      ParseUtils.requireNoAttributes(reader);
      GlobalStateConfigurationBuilder builder = holder.getGlobalConfigurationBuilder().globalState().enable();
      ConfigurationStorage storage = null;
      while (reader.inTag()) {
         Element element = Element.forName(reader.getLocalName());
         switch (element) {
            case PERSISTENT_LOCATION: {
               parseGlobalStatePath(reader, builder::persistentLocation);
               break;
            }
            case SHARED_PERSISTENT_LOCATION: {
               parseGlobalStatePath(reader, builder::sharedPersistentLocation);
               break;
            }
            case TEMPORARY_LOCATION: {
               parseGlobalStatePath(reader, builder::temporaryLocation);
               break;
            }
            case IMMUTABLE_CONFIGURATION_STORAGE: {
               if (storage != null) {
                  throw ParseUtils.unexpectedElement(reader);
               }
               storage = ConfigurationStorage.IMMUTABLE;
               break;
            }
            case VOLATILE_CONFIGURATION_STORAGE: {
               if (storage != null) {
                  throw ParseUtils.unexpectedElement(reader);
               }
               ParseUtils.requireNoAttributes(reader);
               ParseUtils.requireNoContent(reader);
               storage = ConfigurationStorage.VOLATILE;
               break;
            }
            case OVERLAY_CONFIGURATION_STORAGE: {
               if (storage != null) {
                  throw ParseUtils.unexpectedElement(reader);
               }
               ParseUtils.requireNoAttributes(reader);
               ParseUtils.requireNoContent(reader);
               storage = ConfigurationStorage.OVERLAY;
               break;
            }
            case MANAGED_CONFIGURATION_STORAGE: {
               if (storage != null) {
                  throw ParseUtils.unexpectedElement(reader);
               } else {
                  throw CONFIG.managerConfigurationStorageUnavailable();
               }
            }
            case CUSTOM_CONFIGURATION_STORAGE: {
               if (storage != null) {
                  throw ParseUtils.unexpectedElement(reader);
               }
               storage = ConfigurationStorage.CUSTOM;
               builder.configurationStorageSupplier(parseCustomConfigurationStorage(reader, holder));
               break;
            }
            default: {
               throw ParseUtils.unexpectedElement(reader);
            }
         }
      }
      if (storage != null) {
         builder.configurationStorage(storage);
      }
   }

   private void parseGlobalStatePath(ConfigurationReader reader, BiConsumer<String, String> pathItems) {
      String path = ParseUtils.requireAttributes(reader, Attribute.PATH.getLocalName())[0];
      String relativeTo = null;
      for (int i = 0; i < reader.getAttributeCount(); i++) {
         Attribute attribute = Attribute.forName(reader.getAttributeName(i));
         switch (attribute) {
            case RELATIVE_TO: {
               relativeTo = ParseUtils.requireAttributeProperty(reader, i);
               break;
            }
            case PATH: {
               // Handled above
               break;
            }
            default: {
               throw ParseUtils.unexpectedAttribute(reader, i);
            }
         }
      }
      ParseUtils.requireNoContent(reader);
      pathItems.accept(path, relativeTo);
   }

   private Supplier<? extends LocalConfigurationStorage> parseCustomConfigurationStorage(ConfigurationReader reader, ConfigurationBuilderHolder holder) {
      String storageClass = ParseUtils.requireSingleAttribute(reader, Attribute.CLASS.getLocalName());
      ParseUtils.requireNoContent(reader);
      return Util.getInstanceSupplier(storageClass, holder.getClassLoader());
   }

   private ThreadPoolConfiguration createThreadPoolConfiguration(String threadPoolName, String componentName, ConfigurationBuilderHolder holder) {
      ThreadsConfigurationBuilder threads = holder.getGlobalConfigurationBuilder().threads();
      ThreadPoolBuilderAdapter threadPool = threads.getThreadPool(threadPoolName);

      if (threadPool == null)
         throw CONFIG.undefinedThreadPoolName(threadPoolName);

      ThreadPoolConfiguration threadPoolConfiguration = threadPool.asThreadPoolConfigurationBuilder();
      boolean isNonBlocking = threadPoolConfiguration.threadPoolFactory().createsNonBlockingThreads();
      if (NON_BLOCKING_EXECUTOR.equals(componentName) && !isNonBlocking) {
         throw CONFIG.threadPoolFactoryIsBlocking(threadPoolName, componentName);
      }
      DefaultThreadFactory threadFactory = threadPoolConfiguration.threadFactory();
      if (threadFactory != null) {
         threadFactory.setComponent(shortened(componentName));
      }
      return threadPoolConfiguration;
   }

   /**
    * @deprecated use {@link CacheParser#parseStoreAttribute(ConfigurationReader, int, AbstractStoreConfigurationBuilder)}
    */
   @Deprecated
   public static void parseStoreAttribute(ConfigurationReader reader, int index, AbstractStoreConfigurationBuilder<?, ?> storeBuilder) {
      CacheParser.parseStoreAttribute(reader, index, storeBuilder);
   }

   /**
    * @deprecated use {@link CacheParser#parseStoreElement(ConfigurationReader, StoreConfigurationBuilder)}
    */
   @Deprecated
   public static void parseStoreElement(ConfigurationReader reader, StoreConfigurationBuilder<?, ?> storeBuilder) {
      CacheParser.parseStoreElement(reader, storeBuilder);
   }

   /**
    * @deprecated use {@link CacheParser#parseStoreWriteBehind(ConfigurationReader, AsyncStoreConfigurationBuilder)}
    */
   @Deprecated
   public static void parseStoreWriteBehind(ConfigurationReader reader, AsyncStoreConfigurationBuilder<?> storeBuilder) {
      CacheParser.parseStoreWriteBehind(reader, storeBuilder);
   }

   /**
    * @deprecated use {@link CacheParser#parseStoreProperty(ConfigurationReader, StoreConfigurationBuilder)}
    */
   @Deprecated
   public static void parseStoreProperty(ConfigurationReader reader, StoreConfigurationBuilder<?, ?> storeBuilder) {
      String property = ParseUtils.requireSingleAttribute(reader, Attribute.NAME.getLocalName());
      String value = reader.getElementText();
      storeBuilder.addProperty(property, value);
   }

   private void addJGroupsDefaultStacksIfNeeded(final ConfigurationReader reader, final ConfigurationBuilderHolder holder) {
      if (holder.getJGroupsStack(BuiltinJGroupsChannelConfigurator.TCP_STACK_NAME) == null) {
         holder.addJGroupsStack(BuiltinJGroupsChannelConfigurator.TCP(reader.getProperties()));
         holder.addJGroupsStack(BuiltinJGroupsChannelConfigurator.UDP(reader.getProperties()));
         holder.addJGroupsStack(BuiltinJGroupsChannelConfigurator.KUBERNETES(reader.getProperties()));
         holder.addJGroupsStack(BuiltinJGroupsChannelConfigurator.EC2(reader.getProperties()));
         holder.addJGroupsStack(BuiltinJGroupsChannelConfigurator.GOOGLE(reader.getProperties()));
         holder.addJGroupsStack(BuiltinJGroupsChannelConfigurator.AZURE(reader.getProperties()));
      }
   }

   @Override
   public Namespace[] getNamespaces() {
      return ParseUtils.getNamespaceAnnotations(getClass());
   }

}
