package org.apache.commons.jcs3.engine.control;

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

import java.io.IOException;
import java.io.InputStream;
import java.lang.management.ManagementFactory;
import java.security.AccessControlException;
import java.util.List;
import java.util.Objects;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import javax.management.MBeanServer;
import javax.management.ObjectName;

import org.apache.commons.jcs3.access.exception.CacheException;
import org.apache.commons.jcs3.admin.JCSAdminBean;
import org.apache.commons.jcs3.auxiliary.AuxiliaryCache;
import org.apache.commons.jcs3.auxiliary.AuxiliaryCacheAttributes;
import org.apache.commons.jcs3.auxiliary.AuxiliaryCacheFactory;
import org.apache.commons.jcs3.auxiliary.remote.behavior.IRemoteCacheConstants;
import org.apache.commons.jcs3.engine.CompositeCacheAttributes;
import org.apache.commons.jcs3.engine.ElementAttributes;
import org.apache.commons.jcs3.engine.behavior.ICache;
import org.apache.commons.jcs3.engine.behavior.ICacheType.CacheType;
import org.apache.commons.jcs3.engine.behavior.ICompositeCacheAttributes;
import org.apache.commons.jcs3.engine.behavior.ICompositeCacheManager;
import org.apache.commons.jcs3.engine.behavior.IElementAttributes;
import org.apache.commons.jcs3.engine.behavior.IProvideScheduler;
import org.apache.commons.jcs3.engine.behavior.IShutdownObserver;
import org.apache.commons.jcs3.engine.control.event.ElementEventQueue;
import org.apache.commons.jcs3.engine.control.event.behavior.IElementEventQueue;
import org.apache.commons.jcs3.engine.stats.CacheStats;
import org.apache.commons.jcs3.engine.stats.behavior.ICacheStats;
import org.apache.commons.jcs3.log.Log;
import org.apache.commons.jcs3.log.LogManager;
import org.apache.commons.jcs3.utils.config.OptionConverter;
import org.apache.commons.jcs3.utils.threadpool.DaemonThreadFactory;
import org.apache.commons.jcs3.utils.threadpool.ThreadPoolManager;
import org.apache.commons.jcs3.utils.timing.ElapsedTimer;

/**
 * Manages a composite cache. This provides access to caches and is the primary way to shutdown the
 * caching system as a whole.
 * <p>
 * The composite cache manager is responsible for creating / configuring cache regions. It serves as
 * a factory for the ComositeCache class. The CompositeCache is the core of JCS, the hub for various
 * auxiliaries.
 */
public class CompositeCacheManager
    implements IRemoteCacheConstants, ICompositeCacheManager, IProvideScheduler
{
    /** The logger */
    private static final Log log = LogManager.getLog( CompositeCacheManager.class );

    /** JMX object name */
    public static final String JMX_OBJECT_NAME = "org.apache.commons.jcs3:type=JCSAdminBean";

    /** This is the name of the config file that we will look for by default. */
    private static final String DEFAULT_CONFIG = "/cache.ccf";

    /** default region prefix */
    private static final String DEFAULT_REGION = "jcs.default";

    /** Should we use system property substitutions. */
    private static final boolean DEFAULT_USE_SYSTEM_PROPERTIES = true;

    /** Once configured, you can force a reconfiguration of sorts. */
    private static final boolean DEFAULT_FORCE_RECONFIGURATION = false;

    /** Caches managed by this cache manager */
    private final ConcurrentMap<String, ICache<?, ?>> caches = new ConcurrentHashMap<>();

    /** Number of clients accessing this cache manager */
    private final AtomicInteger clients = new AtomicInteger(0);

    /** Default cache attributes for this cache manager */
    private ICompositeCacheAttributes defaultCacheAttr = new CompositeCacheAttributes();

    /** Default element attributes for this cache manager */
    private IElementAttributes defaultElementAttr = new ElementAttributes();

    /** Used to keep track of configured auxiliaries */
    private final ConcurrentMap<String, AuxiliaryCacheFactory> auxiliaryFactoryRegistry =
        new ConcurrentHashMap<>( );

    /** Used to keep track of attributes for auxiliaries. */
    private final ConcurrentMap<String, AuxiliaryCacheAttributes> auxiliaryAttributeRegistry =
        new ConcurrentHashMap<>( );

    /** Used to keep track of configured auxiliaries */
    private final ConcurrentMap<String, AuxiliaryCache<?, ?>> auxiliaryCaches =
        new ConcurrentHashMap<>( );

    /** Properties with which this manager was configured. This is exposed for other managers. */
    private Properties configurationProperties;

    /** The default auxiliary caches to be used if not preconfigured */
    private String defaultAuxValues;

    /** The Singleton Instance */
    private static CompositeCacheManager instance;

    /** Stack for those waiting for notification of a shutdown. */
    private final LinkedBlockingDeque<IShutdownObserver> shutdownObservers = new LinkedBlockingDeque<>();

    /** The central background scheduler. */
    private ScheduledExecutorService scheduledExecutor;

    /** The central event queue. */
    private IElementEventQueue elementEventQueue;

    /** Shutdown hook thread instance */
    private Thread shutdownHook;

    /** Indicates whether the instance has been initialized. */
    private boolean isInitialized;

    /** Indicates whether configure has been called. */
    private boolean isConfigured;

    /** Indicates whether JMX bean has been registered. */
    private boolean isJMXRegistered;

    private String jmxName = JMX_OBJECT_NAME;

    /**
     * Gets the CacheHub instance. For backward compatibility, if this creates the instance it will
     * attempt to configure it with the default configuration. If you want to configure from your
     * own source, use {@link #getUnconfiguredInstance}and then call {@link #configure}
     * <p>
     * @return CompositeCacheManager
     * @throws CacheException if the configuration cannot be loaded
     */
    public static synchronized CompositeCacheManager getInstance() throws CacheException
    {
        return getInstance( DEFAULT_CONFIG );
    }

    /**
     * Initializes the cache manager using the props file for the given name.
     * <p>
     * @param propsFilename
     * @return CompositeCacheManager configured from the give propsFileName
     * @throws CacheException if the configuration cannot be loaded
     */
    public static synchronized CompositeCacheManager getInstance( final String propsFilename ) throws CacheException
    {
        if ( instance == null )
        {
            log.info( "Instance is null, creating with config [{0}]", propsFilename );
            instance = createInstance();
        }

        if (!instance.isInitialized())
        {
            instance.initialize();
        }

        if (!instance.isConfigured())
        {
            instance.configure( propsFilename );
        }

        instance.clients.incrementAndGet();

        return instance;
    }

    /**
     * Get a CacheHub instance which is not configured. If an instance already exists, it will be
     * returned.
     *<p>
     * @return CompositeCacheManager
     */
    public static synchronized CompositeCacheManager getUnconfiguredInstance()
    {
        if ( instance == null )
        {
            log.info( "Instance is null, returning unconfigured instance" );
            instance = createInstance();
        }

        if (!instance.isInitialized())
        {
            instance.initialize();
        }

        instance.clients.incrementAndGet();

        return instance;
    }

    /**
     * Simple factory method, must override in subclasses so getInstance creates / returns the
     * correct object.
     * <p>
     * @return CompositeCacheManager
     */
    protected static CompositeCacheManager createInstance()
    {
        return new CompositeCacheManager();
    }

    /**
     * Default constructor
     */
    protected CompositeCacheManager()
    {
        // empty
    }

    /** Creates a shutdown hook and starts the scheduler service */
    protected synchronized void initialize()
    {
        if (!isInitialized)
        {
            this.shutdownHook = new Thread(() -> {
                if ( isInitialized() )
                {
                    log.info("Shutdown hook activated. Shutdown was not called. Shutting down JCS.");
                    shutDown();
                }
            });
            try
            {
                Runtime.getRuntime().addShutdownHook( shutdownHook );
            }
            catch ( final AccessControlException e )
            {
                log.error( "Could not register shutdown hook.", e );
            }

            this.scheduledExecutor = Executors.newScheduledThreadPool(4,
                    new DaemonThreadFactory("JCS-Scheduler-", Thread.MIN_PRIORITY));

            // Register JMX bean
            if (!isJMXRegistered && jmxName != null)
            {
                final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
                final JCSAdminBean adminBean = new JCSAdminBean(this);
                try
                {
                    final ObjectName jmxObjectName = new ObjectName(jmxName);
                    mbs.registerMBean(adminBean, jmxObjectName);
                    isJMXRegistered = true;
                }
                catch (final Exception e)
                {
                    log.warn( "Could not register JMX bean.", e );
                }
            }

            isInitialized = true;
        }
    }

    /**
     * Get the element event queue
     *
     * @return the elementEventQueue
     */
    public IElementEventQueue getElementEventQueue()
    {
        return elementEventQueue;
    }

    /**
     * Get the scheduler service
     *
     * @return the scheduledExecutor
     */
    @Override
    public ScheduledExecutorService getScheduledExecutorService()
    {
        return scheduledExecutor;
    }

    /**
     * Configure with default properties file
     * @throws CacheException if the configuration cannot be loaded
     */
    public void configure() throws CacheException
    {
        configure( DEFAULT_CONFIG );
    }

    /**
     * Configure from specific properties file.
     * <p>
     * @param propFile Path <u>within classpath </u> to load configuration from
     * @throws CacheException if the configuration cannot be loaded
     */
    public void configure( final String propFile ) throws CacheException
    {
        log.info( "Creating cache manager from config file: {0}", propFile );

        final Properties props = new Properties();

        try (InputStream is = getClass().getResourceAsStream( propFile ))
        {
            props.load( is );
            log.debug( "File [{0}] contained {1} properties", () -> propFile, props::size);
        }
        catch ( final IOException ex )
        {
            throw new CacheException("Failed to load properties for name [" + propFile + "]", ex);
        }

        configure( props );
    }

    /**
     * Configure from properties object.
     * <p>
     * This method will call configure, instructing it to use system properties as a default.
     * @param props
     */
    public void configure( final Properties props )
    {
        configure( props, DEFAULT_USE_SYSTEM_PROPERTIES );
    }

    /**
     * Configure from properties object, overriding with values from the system properties if
     * instructed.
     * <p>
     * You can override a specific value by passing in a system property:
     * <p>
     * For example, you could override this value in the cache.ccf file by starting up your program
     * with the argument: -Djcs.auxiliary.LTCP.attributes.TcpListenerPort=1111
     * <p>
     * @param props
     * @param useSystemProperties -- if true, values starting with jcs will be put into the props
     *            file prior to configuring the cache.
     */
    public void configure( final Properties props, final boolean useSystemProperties )
    {
        configure( props, useSystemProperties, DEFAULT_FORCE_RECONFIGURATION );
    }

    /**
     * Configure from properties object, overriding with values from the system properties if
     * instructed.
     * <p>
     * You can override a specific value by passing in a system property:
     * <p>
     * For example, you could override this value in the cache.ccf file by starting up your program
     * with the argument: -Djcs.auxiliary.LTCP.attributes.TcpListenerPort=1111
     * <p>
     * @param props
     * @param useSystemProperties -- if true, values starting with jcs will be put into the props
     *            file prior to configuring the cache.
     * @param forceReconfiguration - if the manager is already configured, we will try again. This
     *            may not work properly.
     */
    public synchronized void configure( final Properties props, final boolean useSystemProperties, final boolean forceReconfiguration )
    {
        if ( props == null )
        {
            log.error( "No properties found. Please configure the cache correctly." );
            return;
        }

        if ( isConfigured )
        {
            if ( !forceReconfiguration )
            {
                log.debug( "Configure called after the manager has been configured.  "
                         + "Force reconfiguration is false. Doing nothing" );
                return;
            }
            log.info( "Configure called after the manager has been configured.  "
                    + "Force reconfiguration is true. Reconfiguring as best we can." );
        }
        if ( useSystemProperties )
        {
            CompositeCacheConfigurator.overrideWithSystemProperties( props );
        }
        doConfigure( props );
    }

    /**
     * Configure the cache using the supplied properties.
     * <p>
     * @param properties assumed not null
     */
    private synchronized void doConfigure( final Properties properties )
    {
        // We will expose this for managers that need raw properties.
        this.configurationProperties = properties;

        // set the props value and then configure the ThreadPoolManager
        ThreadPoolManager.setProps( properties );
        final ThreadPoolManager poolMgr = ThreadPoolManager.getInstance();
        log.debug( "ThreadPoolManager = {0}", poolMgr);

        // Create event queue
        this.elementEventQueue = new ElementEventQueue();

        // configure the cache
        final CompositeCacheConfigurator configurator = newConfigurator();

        final ElapsedTimer timer = new ElapsedTimer();

        // set default value list
        this.defaultAuxValues = OptionConverter.findAndSubst( CompositeCacheManager.DEFAULT_REGION,
                properties );

        log.info( "Setting default auxiliaries to \"{0}\"", this.defaultAuxValues );

        // set default cache attr
        this.defaultCacheAttr = configurator.parseCompositeCacheAttributes( properties, "",
                new CompositeCacheAttributes(), DEFAULT_REGION );

        log.info( "setting defaultCompositeCacheAttributes to {0}", this.defaultCacheAttr );

        // set default element attr
        this.defaultElementAttr = configurator.parseElementAttributes( properties, "",
                new ElementAttributes(), DEFAULT_REGION );

        log.info( "setting defaultElementAttributes to {0}", this.defaultElementAttr );

        // set up system caches to be used by non system caches
        // need to make sure there is no circularity of reference
        configurator.parseSystemRegions( properties, this );

        // setup preconfigured caches
        configurator.parseRegions( properties, this );

        log.info( "Finished configuration in {0} ms.", timer::getElapsedTime);

        isConfigured = true;
    }

    /**
     * Gets the defaultCacheAttributes attribute of the CacheHub object
     * <p>
     * @return The defaultCacheAttributes value
     */
    public ICompositeCacheAttributes getDefaultCacheAttributes()
    {
        return this.defaultCacheAttr.clone();
    }

    /**
     * Gets the defaultElementAttributes attribute of the CacheHub object
     * <p>
     * @return The defaultElementAttributes value
     */
    public IElementAttributes getDefaultElementAttributes()
    {
        return this.defaultElementAttr.clone();
    }

    /**
     * Gets the cache attribute of the CacheHub object
     * <p>
     * @param cacheName
     * @return CompositeCache -- the cache region controller
     */
    @Override
    public <K, V> CompositeCache<K, V> getCache( final String cacheName )
    {
        return getCache( cacheName, getDefaultCacheAttributes() );
    }

    /**
     * Gets the cache attribute of the CacheHub object
     * <p>
     * @param cacheName
     * @param cattr
     * @return CompositeCache
     */
    public <K, V> CompositeCache<K, V> getCache( final String cacheName, final ICompositeCacheAttributes cattr )
    {
        cattr.setCacheName( cacheName );
        return getCache( cattr, getDefaultElementAttributes() );
    }

    /**
     * Gets the cache attribute of the CacheHub object
     * <p>
     * @param cacheName
     * @param cattr
     * @param attr
     * @return CompositeCache
     */
    public <K, V> CompositeCache<K, V>  getCache( final String cacheName, final ICompositeCacheAttributes cattr, final IElementAttributes attr )
    {
        cattr.setCacheName( cacheName );
        return getCache( cattr, attr );
    }

    /**
     * Gets the cache attribute of the CacheHub object
     * <p>
     * @param cattr
     * @return CompositeCache
     */
    public <K, V> CompositeCache<K, V>  getCache( final ICompositeCacheAttributes cattr )
    {
        return getCache( cattr, getDefaultElementAttributes() );
    }

    /**
     * If the cache has already been created, then the CacheAttributes and the element Attributes
     * will be ignored. Currently there is no overriding the CacheAttributes once it is set up. You
     * can change the default ElementAttributes for a region later.
     * <p>
     * Overriding the default elemental attributes will require changing the way the attributes are
     * assigned to elements. Get cache creates a cache with defaults if none are specified. We might
     * want to create separate method for creating/getting. . .
     * <p>
     * @param cattr
     * @param attr
     * @return CompositeCache
     */
    @SuppressWarnings("unchecked") // Need to cast because of common map for all caches
    public <K, V> CompositeCache<K, V>  getCache( final ICompositeCacheAttributes cattr, final IElementAttributes attr )
    {
        log.debug( "attr = {0}", attr );

        return (CompositeCache<K, V>) caches.computeIfAbsent(cattr.getCacheName(),
                cacheName -> {
            final CompositeCacheConfigurator configurator = newConfigurator();
            return configurator.parseRegion( this.getConfigurationProperties(), this, cacheName,
                                              this.defaultAuxValues, cattr );
        });
    }

    protected CompositeCacheConfigurator newConfigurator() {
        return new CompositeCacheConfigurator();
    }

    /**
     * @param name
     */
    public void freeCache( final String name )
    {
        freeCache( name, false );
    }

    /**
     * @param name
     * @param fromRemote
     */
    public void freeCache( final String name, final boolean fromRemote )
    {
        final CompositeCache<?, ?> cache = (CompositeCache<?, ?>) caches.remove( name );

        if ( cache != null )
        {
            cache.dispose( fromRemote );
        }
    }

    /**
     * Calls freeCache on all regions
     */
    public synchronized void shutDown()
    {
        // shutdown element event queue
        if (this.elementEventQueue != null)
        {
            this.elementEventQueue.dispose();
        }

        // notify any observers
        IShutdownObserver observer = null;
        while ((observer = shutdownObservers.poll()) != null)
        {
            observer.shutdown();
        }

        // Unregister JMX bean
        if (isJMXRegistered)
        {
            final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
            try
            {
                final ObjectName jmxObjectName = new ObjectName(jmxName);
                mbs.unregisterMBean(jmxObjectName);
            }
            catch (final Exception e)
            {
                log.warn( "Could not unregister JMX bean.", e );
            }

            isJMXRegistered = false;
        }

        // do the traditional shutdown of the regions.
        getCacheNames().forEach(this::freeCache);

        // shut down auxiliaries
        for (final String key : auxiliaryCaches.keySet())
        {
            try
            {
                freeAuxiliaryCache(key);
            }
            catch (final IOException e)
            {
                log.warn("Auxiliary cache {0} failed to shut down", key, e);
            }
        }

        // shut down factories
        auxiliaryFactoryRegistry.values().forEach(AuxiliaryCacheFactory::dispose);

        auxiliaryAttributeRegistry.clear();
        auxiliaryFactoryRegistry.clear();

        // shutdown all scheduled jobs
        this.scheduledExecutor.shutdownNow();

        // shutdown all thread pools
        ThreadPoolManager.dispose();

        if (shutdownHook != null)
        {
            try
            {
                Runtime.getRuntime().removeShutdownHook(shutdownHook);
            }
            catch (final IllegalStateException e)
            {
                // May fail if the JVM is already shutting down
            }

            this.shutdownHook = null;
        }

        isConfigured = false;
        isInitialized = false;
    }

    /** */
    public void release()
    {
        release( false );
    }

    /**
     * @param fromRemote
     */
    private void release( final boolean fromRemote )
    {
        synchronized ( CompositeCacheManager.class )
        {
            // Wait until called by the last client
            if ( clients.decrementAndGet() > 0 )
            {
                log.debug( "Release called, but {0} remain", clients);
                return;
            }

            log.debug( "Last client called release. There are {0} caches which will be disposed",
                    caches::size);

            caches.values().stream()
                .filter(Objects::nonNull)
                .forEach(cache -> ((CompositeCache<?, ?>)cache).dispose( fromRemote ));
        }
    }

    /**
     * Returns a list of the current cache names.
     * @return Set<String>
     */
    public Set<String> getCacheNames()
    {
        return caches.keySet();
    }

    /**
     * @return ICacheType.CACHE_HUB
     */
    public CacheType getCacheType()
    {
        return CacheType.CACHE_HUB;
    }

    /**
     * @param auxFac
     */
    public void registryFacPut( final AuxiliaryCacheFactory auxFac )
    {
        auxiliaryFactoryRegistry.put( auxFac.getName(), auxFac );
    }

    /**
     * @param name
     * @return AuxiliaryCacheFactory
     */
    public AuxiliaryCacheFactory registryFacGet( final String name )
    {
        return auxiliaryFactoryRegistry.get( name );
    }

    /**
     * @param auxAttr
     */
    public void registryAttrPut( final AuxiliaryCacheAttributes auxAttr )
    {
        auxiliaryAttributeRegistry.put( auxAttr.getName(), auxAttr );
    }

    /**
     * @param name
     * @return AuxiliaryCacheAttributes
     */
    public AuxiliaryCacheAttributes registryAttrGet( final String name )
    {
        return auxiliaryAttributeRegistry.get( name );
    }

    /**
     * Add a cache to the map of registered caches
     *
     * @param cacheName the region name
     * @param cache the cache instance
     */
    public void addCache(final String cacheName, final ICache<?, ?> cache)
    {
        caches.put(cacheName, cache);
    }

    /**
     * Add a cache to the map of registered auxiliary caches
     *
     * @param auxName the auxiliary name
     * @param cacheName the region name
     * @param cache the cache instance
     */
    public void addAuxiliaryCache(final String auxName, final String cacheName, final AuxiliaryCache<?, ?> cache)
    {
        final String key = String.format("aux.%s.region.%s", auxName, cacheName);
        auxiliaryCaches.put(key, cache);
    }

    /**
     * Get a cache from the map of registered auxiliary caches
     *
     * @param auxName the auxiliary name
     * @param cacheName the region name
     *
     * @return the cache instance
     */
    @Override
    @SuppressWarnings("unchecked") // because of common map for all auxiliary caches
    public <K, V> AuxiliaryCache<K, V> getAuxiliaryCache(final String auxName, final String cacheName)
    {
        final String key = String.format("aux.%s.region.%s", auxName, cacheName);
        return (AuxiliaryCache<K, V>) auxiliaryCaches.get(key);
    }

    /**
     * Dispose a cache and remove it from the map of registered auxiliary caches
     *
     * @param auxName the auxiliary name
     * @param cacheName the region name
     * @throws IOException if disposing of the cache fails
     */
    public void freeAuxiliaryCache(final String auxName, final String cacheName) throws IOException
    {
        final String key = String.format("aux.%s.region.%s", auxName, cacheName);
        freeAuxiliaryCache(key);
    }

    /**
     * Dispose a cache and remove it from the map of registered auxiliary caches
     *
     * @param key the key into the map of auxiliaries
     * @throws IOException if disposing of the cache fails
     */
    public void freeAuxiliaryCache(final String key) throws IOException
    {
        final AuxiliaryCache<?, ?> aux = auxiliaryCaches.remove( key );

        if ( aux != null )
        {
            aux.dispose();
        }
    }

    /**
     * Gets stats for debugging. This calls gets statistics and then puts all the results in a
     * string. This returns data for all regions.
     * <p>
     * @return String
     */
    @Override
    public String getStats()
    {
        final ICacheStats[] stats = getStatistics();
        if ( stats == null )
        {
            return "NONE";
        }

        // force the array elements into a string.
        final StringBuilder buf = new StringBuilder();
        Stream.of(stats).forEach(stat -> {
            buf.append( "\n---------------------------\n" );
            buf.append( stat );
        });
        return buf.toString();
    }

    /**
     * This returns data gathered for all regions and all the auxiliaries they currently uses.
     * <p>
     * @return ICacheStats[]
     */
    public ICacheStats[] getStatistics()
    {
        final List<ICacheStats> cacheStats = caches.values().stream()
            .filter(Objects::nonNull)
            .map(cache -> ((CompositeCache<?, ?>)cache).getStatistics() )
            .collect(Collectors.toList());

        return cacheStats.toArray( new CacheStats[0] );
    }

    /**
     * Perhaps the composite cache itself should be the observable object. It doesn't make much of a
     * difference. There are some problems with region by region shutdown. Some auxiliaries are
     * global. They will need to track when every region has shutdown before doing things like
     * closing the socket with a lateral.
     * <p>
     * @param observer
     */
    @Override
    public void registerShutdownObserver( final IShutdownObserver observer )
    {
        if (!shutdownObservers.contains(observer))
        {
                shutdownObservers.push( observer );
        }
        else
        {
                log.warn("Shutdown observer added twice {0}", observer);
        }
    }

    /**
     * @param observer
     */
    @Override
    public void deregisterShutdownObserver( final IShutdownObserver observer )
    {
        shutdownObservers.remove( observer );
    }

    /**
     * This is exposed so other manager can get access to the props.
     * <p>
     * @return the configurationProperties
     */
    @Override
    public Properties getConfigurationProperties()
    {
        return configurationProperties;
    }

    /**
     * @return the isInitialized
     */
    public boolean isInitialized()
    {
        return isInitialized;
    }

    /**
     * @return the isConfigured
     */
    public boolean isConfigured()
    {
        return isConfigured;
    }

    public void setJmxName(final String name)
    {
        if (isJMXRegistered)
        {
            throw new IllegalStateException("Too late, MBean registration is done");
        }
        jmxName = name;
    }
}