/*
 * 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.commons.jcs3.jcache;

import static org.apache.commons.jcs3.jcache.Asserts.assertNotNull;
import static org.apache.commons.jcs3.jcache.serialization.Serializations.copy;

import java.io.Closeable;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.cache.Cache;
import javax.cache.CacheException;
import javax.cache.CacheManager;
import javax.cache.configuration.CacheEntryListenerConfiguration;
import javax.cache.configuration.Configuration;
import javax.cache.configuration.Factory;
import javax.cache.event.CacheEntryEvent;
import javax.cache.event.EventType;
import javax.cache.expiry.Duration;
import javax.cache.expiry.EternalExpiryPolicy;
import javax.cache.expiry.ExpiryPolicy;
import javax.cache.integration.CacheLoader;
import javax.cache.integration.CacheLoaderException;
import javax.cache.integration.CacheWriter;
import javax.cache.integration.CacheWriterException;
import javax.cache.integration.CompletionListener;
import javax.cache.processor.EntryProcessor;
import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.EntryProcessorResult;
import javax.management.ObjectName;

import org.apache.commons.jcs3.engine.CacheElement;
import org.apache.commons.jcs3.engine.ElementAttributes;
import org.apache.commons.jcs3.engine.behavior.ICacheElement;
import org.apache.commons.jcs3.engine.behavior.IElementAttributes;
import org.apache.commons.jcs3.engine.behavior.IElementSerializer;
import org.apache.commons.jcs3.jcache.jmx.JCSCacheMXBean;
import org.apache.commons.jcs3.jcache.jmx.JCSCacheStatisticsMXBean;
import org.apache.commons.jcs3.jcache.jmx.JMXs;
import org.apache.commons.jcs3.jcache.proxy.ExceptionWrapperHandler;
import org.apache.commons.jcs3.jcache.thread.DaemonThreadFactory;
import org.apache.commons.jcs3.utils.serialization.StandardSerializer;

// TODO: configure serializer
public class JCSCache<K, V> implements Cache<K, V>
{
    private final ExpiryAwareCache<K, V> delegate;
    private final JCSCachingManager manager;
    private final JCSConfiguration<K, V> config;
    private final CacheLoader<K, V> loader;
    private final CacheWriter<? super K, ? super V> writer;
    private final ExpiryPolicy expiryPolicy;
    private final ObjectName cacheConfigObjectName;
    private final ObjectName cacheStatsObjectName;
    private final String name;
    private volatile boolean closed;
    private final Map<CacheEntryListenerConfiguration<K, V>, JCSListener<K, V>> listeners = new ConcurrentHashMap<>();
    private final Statistics statistics = new Statistics();
    private final ExecutorService pool;
    private final IElementSerializer serializer; // using json/xml should work as well -> don't force Serializable


    public JCSCache(final ClassLoader classLoader, final JCSCachingManager mgr,
                    final String cacheName, final JCSConfiguration<K, V> configuration,
                    final Properties properties, final ExpiryAwareCache<K, V> cache)
    {
        manager = mgr;

        name = cacheName;

        delegate = cache;
        if (delegate.getElementAttributes() == null)
        {
            delegate.setElementAttributes(new ElementAttributes());
        }
        delegate.getElementAttributes().addElementEventHandler(new EvictionListener(statistics));

        config = configuration;

        final int poolSize = Integer.parseInt(property(properties, cacheName, "pool.size", "3"));
        final DaemonThreadFactory threadFactory = new DaemonThreadFactory("JCS-JCache-" + cacheName + "-");
        pool = poolSize > 0 ? Executors.newFixedThreadPool(poolSize, threadFactory) : Executors.newCachedThreadPool(threadFactory);

        try
        {
            serializer = IElementSerializer.class.cast(classLoader.loadClass(property(properties, "serializer", cacheName, StandardSerializer.class.getName())).newInstance());
        }
        catch (final Exception e)
        {
            throw new IllegalArgumentException(e);
        }

        final Factory<CacheLoader<K, V>> cacheLoaderFactory = configuration.getCacheLoaderFactory();
        if (cacheLoaderFactory == null)
        {
            loader = NoLoader.INSTANCE;
        }
        else
        {
            loader = ExceptionWrapperHandler
                    .newProxy(classLoader, cacheLoaderFactory.create(), CacheLoaderException.class, CacheLoader.class);
        }

        final Factory<CacheWriter<? super K, ? super V>> cacheWriterFactory = configuration.getCacheWriterFactory();
        if (cacheWriterFactory == null)
        {
            writer = NoWriter.INSTANCE;
        }
        else
        {
            writer = ExceptionWrapperHandler
                    .newProxy(classLoader, cacheWriterFactory.create(), CacheWriterException.class, CacheWriter.class);
        }

        final Factory<ExpiryPolicy> expiryPolicyFactory = configuration.getExpiryPolicyFactory();
        if (expiryPolicyFactory == null)
        {
            expiryPolicy = new EternalExpiryPolicy();
        }
        else
        {
            expiryPolicy = expiryPolicyFactory.create();
        }

        for (final CacheEntryListenerConfiguration<K, V> listener : config.getCacheEntryListenerConfigurations())
        {
            listeners.put(listener, new JCSListener<>(listener));
        }
        delegate.init(this, listeners);

        statistics.setActive(config.isStatisticsEnabled());

        final String mgrStr = manager.getURI().toString().replaceAll(",|:|=|\n", ".");
        final String cacheStr = name.replaceAll(",|:|=|\n", ".");
        try
        {
            cacheConfigObjectName = new ObjectName("javax.cache:type=CacheConfiguration,"
                    + "CacheManager=" + mgrStr + "," + "Cache=" + cacheStr);
            cacheStatsObjectName = new ObjectName("javax.cache:type=CacheStatistics,"
                    + "CacheManager=" + mgrStr + "," + "Cache=" + cacheStr);
        }
        catch (final Exception e)
        {
            throw new IllegalArgumentException(e);
        }
        if (config.isManagementEnabled())
        {
            JMXs.register(cacheConfigObjectName, new JCSCacheMXBean<>(this));
        }
        if (config.isStatisticsEnabled())
        {
            JMXs.register(cacheStatsObjectName, new JCSCacheStatisticsMXBean(statistics));
        }
    }

    private static String property(final Properties properties, final String cacheName, final String name, final String defaultValue)
    {
        return properties.getProperty(cacheName + "." + name, properties.getProperty(name, defaultValue));
    }

    private void assertNotClosed()
    {
        if (isClosed())
        {
            throw new IllegalStateException("cache closed");
        }
    }

    @Override
    public V get(final K key)
    {
        assertNotClosed();
        assertNotNull(key, "key");
        final long getStart = Times.now(false);
        return doGetControllingExpiry(getStart, key, true, false, false, true);
    }

    private V doLoad(final K key, final boolean update, final long now, final boolean propagateLoadException)
    {
        V v = null;
        try
        {
            v = loader.load(key);
        }
        catch (final CacheLoaderException e)
        {
            if (propagateLoadException)
            {
                throw e;
            }
        }
        if (v != null)
        {
            final Duration duration = update ? expiryPolicy.getExpiryForUpdate() : expiryPolicy.getExpiryForCreation();
            if (isNotZero(duration))
            {
                final IElementAttributes clone = delegate.getElementAttributes().clone();
                if (ElementAttributes.class.isInstance(clone))
                {
                    ElementAttributes.class.cast(clone).setCreateTime();
                }
                final ICacheElement<K, V> element = updateElement(key, v, duration, clone);
                try
                {
                    delegate.update(element);
                }
                catch (final IOException e)
                {
                    throw new CacheException(e);
                }
            }
        }
        return v;
    }

    private ICacheElement<K, V> updateElement(final K key, final V v, final Duration duration, final IElementAttributes attrs)
    {
        final ICacheElement<K, V> element = new CacheElement<>(name, key, v);
        if (duration != null)
        {
            attrs.setTimeFactorForMilliseconds(1);
            final boolean eternal = duration.isEternal();
            attrs.setIsEternal(eternal);
            if (!eternal)
            {
                attrs.setLastAccessTimeNow();
            }
            // MaxLife = -1 to use IdleTime excepted if jcache.ccf asked for something else
        }
        element.setElementAttributes(attrs);
        return element;
    }

    private void touch(final K key, final ICacheElement<K, V> element)
    {
        if (config.isStoreByValue())
        {
            final K copy = copy(serializer, manager.getClassLoader(), key);
            try
            {
                delegate.update(new CacheElement<>(name, copy, element.getVal(), element.getElementAttributes()));
            }
            catch (final IOException e)
            {
                throw new CacheException(e);
            }
        }
    }

    @Override
    public Map<K, V> getAll(final Set<? extends K> keys)
    {
        assertNotClosed();
        for (final K k : keys)
        {
            assertNotNull(k, "key");
        }

        final long now = Times.now(false);
        final Map<K, V> result = new HashMap<>();
        for (final K key : keys) {
            assertNotNull(key, "key");

            final ICacheElement<K, V> elt = delegate.get(key);
            V val = elt != null ? elt.getVal() : null;
            if (val == null && config.isReadThrough())
            {
                val = doLoad(key, false, now, false);
                if (val != null)
                {
                    result.put(key, val);
                }
            }
            else if (elt != null)
            {
                final Duration expiryForAccess = expiryPolicy.getExpiryForAccess();
                if (isNotZero(expiryForAccess))
                {
                    touch(key, elt);
                    result.put(key, val);
                }
                else
                {
                    forceExpires(key);
                }
            }
        }
        return result;
    }

    @Override
    public boolean containsKey(final K key)
    {
        assertNotClosed();
        assertNotNull(key, "key");
        return delegate.get(key) != null;
    }

    @Override
    public void put(final K key, final V rawValue)
    {
        assertNotClosed();
        assertNotNull(key, "key");
        assertNotNull(rawValue, "value");

        final ICacheElement<K, V> oldElt = delegate.get(key);
        final V old = oldElt != null ? oldElt.getVal() : null;

        final boolean storeByValue = config.isStoreByValue();
        final V value = storeByValue ? copy(serializer, manager.getClassLoader(), rawValue) : rawValue;

        final boolean created = old == null;
        final Duration duration = created ? expiryPolicy.getExpiryForCreation() : expiryPolicy.getExpiryForUpdate();
        if (isNotZero(duration))
        {
            final boolean statisticsEnabled = config.isStatisticsEnabled();
            final long start = Times.now(false);

            final K jcsKey = storeByValue ? copy(serializer, manager.getClassLoader(), key) : key;
            final ICacheElement<K, V> element = updateElement( // reuse it to create basic structure
                    jcsKey, value, created ? null : duration,
                    oldElt != null ? oldElt.getElementAttributes() : delegate.getElementAttributes().clone());
            if (created && duration != null) { // set maxLife
                final IElementAttributes copy = element.getElementAttributes();
                copy.setTimeFactorForMilliseconds(1);
                final boolean eternal = duration.isEternal();
                copy.setIsEternal(eternal);
                if (ElementAttributes.class.isInstance(copy)) {
                    ElementAttributes.class.cast(copy).setCreateTime();
                }
                if (!eternal)
                {
                    copy.setIsEternal(false);
                    if (duration == expiryPolicy.getExpiryForAccess())
                    {
                        element.getElementAttributes().setIdleTime(duration.getTimeUnit().toMillis(duration.getDurationAmount()));
                    }
                    else
                        {
                        element.getElementAttributes().setMaxLife(duration.getTimeUnit().toMillis(duration.getDurationAmount()));
                    }
                }
                element.setElementAttributes(copy);
            }
            writer.write(new JCSEntry<>(jcsKey, value));
            try
            {
                delegate.update(element);
            }
            catch (final IOException e)
            {
                throw new CacheException(e);
            }
            for (final JCSListener<K, V> listener : listeners.values())
            {
                if (created)
                {
                    listener.onCreated(Arrays.<CacheEntryEvent<? extends K, ? extends V>> asList(new JCSCacheEntryEvent<>(this,
                            EventType.CREATED, null, key, value)));
                }
                else
                {
                    listener.onUpdated(Arrays.<CacheEntryEvent<? extends K, ? extends V>> asList(new JCSCacheEntryEvent<>(this,
                            EventType.UPDATED, old, key, value)));
                }
            }

            if (statisticsEnabled)
            {
                statistics.increasePuts(1);
                statistics.addPutTime(System.currentTimeMillis() - start);
            }
        }
        else
        {
            if (!created)
            {
                forceExpires(key);
            }
        }
    }

    private static boolean isNotZero(final Duration duration)
    {
        return duration == null || !duration.isZero();
    }

    private void forceExpires(final K cacheKey)
    {
        final ICacheElement<K, V> elt = delegate.get(cacheKey);
        delegate.remove(cacheKey);
        for (final JCSListener<K, V> listener : listeners.values())
        {
            listener.onExpired(Arrays.<CacheEntryEvent<? extends K, ? extends V>> asList(new JCSCacheEntryEvent<>(this,
                    EventType.REMOVED, null, cacheKey, elt.getVal())));
        }
    }

    @Override
    public V getAndPut(final K key, final V value)
    {
        assertNotClosed();
        assertNotNull(key, "key");
        assertNotNull(value, "value");
        final long getStart = Times.now(false);
        final V v = doGetControllingExpiry(getStart, key, false, false, true, false);
        put(key, value);
        return v;
    }

    @Override
    public void putAll(final Map<? extends K, ? extends V> map)
    {
        assertNotClosed();
        final TempStateCacheView<K, V> view = new TempStateCacheView<>(this);
        for (final Map.Entry<? extends K, ? extends V> e : map.entrySet())
        {
            view.put(e.getKey(), e.getValue());
        }
        view.merge();
    }

    @Override
    public boolean putIfAbsent(final K key, final V value)
    {
        if (!containsKey(key))
        {
            put(key, value);
            return true;
        }
        return false;
    }

    @Override
    public boolean remove(final K key)
    {
        assertNotClosed();
        assertNotNull(key, "key");

        final boolean statisticsEnabled = config.isStatisticsEnabled();
        final long start = Times.now(!statisticsEnabled);

        writer.delete(key);
        final K cacheKey = key;

        final ICacheElement<K, V> v = delegate.get(cacheKey);
        delegate.remove(cacheKey);

        final V value = v != null && v.getVal() != null ? v.getVal() : null;
        final boolean remove = v != null;
        for (final JCSListener<K, V> listener : listeners.values())
        {
            listener.onRemoved(Arrays.<CacheEntryEvent<? extends K, ? extends V>> asList(new JCSCacheEntryEvent<>(this,
                    EventType.REMOVED, null, key, value)));
        }
        if (remove && statisticsEnabled)
        {
            statistics.increaseRemovals(1);
            statistics.addRemoveTime(Times.now(false) - start);
        }
        return remove;
    }

    @Override
    public boolean remove(final K key, final V oldValue)
    {
        assertNotClosed();
        assertNotNull(key, "key");
        assertNotNull(oldValue, "oldValue");
        final long getStart = Times.now(false);
        final V v = doGetControllingExpiry(getStart, key, false, false, false, false);
        if (oldValue.equals(v))
        {
            remove(key);
            return true;
        }
        else if (v != null)
        {
            // weird but just for stats to be right (org.jsr107.tck.expiry.CacheExpiryTest.removeSpecifiedEntryShouldNotCallExpiryPolicyMethods())
            expiryPolicy.getExpiryForAccess();
        }
        return false;
    }

    @Override
    public V getAndRemove(final K key)
    {
        assertNotClosed();
        assertNotNull(key, "key");
        final long getStart = Times.now(false);
        final V v = doGetControllingExpiry(getStart, key, false, false, true, false);
        remove(key);
        return v;
    }

    private V doGetControllingExpiry(final long getStart, final K key, final boolean updateAcess, final boolean forceDoLoad, final boolean skipLoad,
            final boolean propagateLoadException)
    {
        final boolean statisticsEnabled = config.isStatisticsEnabled();
        final ICacheElement<K, V> elt = delegate.get(key);
        V v = elt != null ? elt.getVal() : null;
        if (v == null && (config.isReadThrough() || forceDoLoad))
        {
            if (!skipLoad)
            {
                v = doLoad(key, false, getStart, propagateLoadException);
            }
        }
        else if (statisticsEnabled)
        {
            if (v != null)
            {
                statistics.increaseHits(1);
            }
            else
            {
                statistics.increaseMisses(1);
            }
        }

        if (updateAcess && elt != null)
        {
            final Duration expiryForAccess = expiryPolicy.getExpiryForAccess();
            if (!isNotZero(expiryForAccess))
            {
                forceExpires(key);
            }
            else if (expiryForAccess != null && (!elt.getElementAttributes().getIsEternal() || !expiryForAccess.isEternal()))
            {
                try
                {
                    delegate.update(updateElement(key, elt.getVal(), expiryForAccess, elt.getElementAttributes()));
                }
                catch (final IOException e)
                {
                    throw new CacheException(e);
                }
            }
        }
        if (statisticsEnabled && v != null)
        {
            statistics.addGetTime(Times.now(false) - getStart);
        }
        return v;
    }

    @Override
    public boolean replace(final K key, final V oldValue, final V newValue)
    {
        assertNotClosed();
        assertNotNull(key, "key");
        assertNotNull(oldValue, "oldValue");
        assertNotNull(newValue, "newValue");
        final boolean statisticsEnabled = config.isStatisticsEnabled();
        final ICacheElement<K, V> elt = delegate.get(key);
        if (elt != null)
        {
            V value = elt.getVal();
            if (value != null && statisticsEnabled)
            {
                statistics.increaseHits(1);
            }
            if (value == null && config.isReadThrough())
            {
                value = doLoad(key, false, Times.now(false), false);
            }
            if (value != null && value.equals(oldValue))
            {
                put(key, newValue);
                return true;
            }
            else if (value != null)
            {
                final Duration expiryForAccess = expiryPolicy.getExpiryForAccess();
                if (expiryForAccess != null && (!elt.getElementAttributes().getIsEternal() || !expiryForAccess.isEternal()))
                {
                    try
                    {
                        delegate.update(updateElement(key, elt.getVal(), expiryForAccess, elt.getElementAttributes()));
                    }
                    catch (final IOException e)
                    {
                        throw new CacheException(e);
                    }
                }
            }
        }
        else if (statisticsEnabled)
        {
            statistics.increaseMisses(1);
        }
        return false;
    }

    @Override
    public boolean replace(final K key, final V value)
    {
        assertNotClosed();
        assertNotNull(key, "key");
        assertNotNull(value, "value");
        final boolean statisticsEnabled = config.isStatisticsEnabled();
        if (containsKey(key))
        {
            if (statisticsEnabled)
            {
                statistics.increaseHits(1);
            }
            put(key, value);
            return true;
        }
        else if (statisticsEnabled)
        {
            statistics.increaseMisses(1);
        }
        return false;
    }

    @Override
    public V getAndReplace(final K key, final V value)
    {
        assertNotClosed();
        assertNotNull(key, "key");
        assertNotNull(value, "value");

        final boolean statisticsEnabled = config.isStatisticsEnabled();

        final ICacheElement<K, V> elt = delegate.get(key);
        if (elt != null)
        {
            V oldValue = elt.getVal();
            if (oldValue == null && config.isReadThrough())
            {
                oldValue = doLoad(key, false, Times.now(false), false);
            }
            else if (statisticsEnabled)
            {
                statistics.increaseHits(1);
            }
            put(key, value);
            return oldValue;
        }
        else if (statisticsEnabled)
        {
            statistics.increaseMisses(1);
        }
        return null;
    }

    @Override
    public void removeAll(final Set<? extends K> keys)
    {
        assertNotClosed();
        assertNotNull(keys, "keys");
        for (final K k : keys)
        {
            remove(k);
        }
    }

    @Override
    public void removeAll()
    {
        assertNotClosed();
        for (final K k : delegate.getKeySet())
        {
            remove(k);
        }
    }

    @Override
    public void clear()
    {
        assertNotClosed();
        try
        {
            delegate.removeAll();
        }
        catch (final IOException e)
        {
            throw new CacheException(e);
        }
    }

    @Override
    public <C2 extends Configuration<K, V>> C2 getConfiguration(final Class<C2> clazz)
    {
        assertNotClosed();
        return clazz.cast(config);
    }

    @Override
    public void loadAll(final Set<? extends K> keys, final boolean replaceExistingValues, final CompletionListener completionListener)
    {
        assertNotClosed();
        assertNotNull(keys, "keys");
        for (final K k : keys)
        {
            assertNotNull(k, "a key");
        }
        pool.submit(() -> doLoadAll(keys, replaceExistingValues, completionListener));
    }

    private void doLoadAll(final Set<? extends K> keys, final boolean replaceExistingValues, final CompletionListener completionListener)
    {
        try
        {
            final long now = Times.now(false);
            for (final K k : keys)
            {
                if (replaceExistingValues)
                {
                    doLoad(k, containsKey(k), now, completionListener != null);
                    continue;
                }
                else if (containsKey(k))
                {
                    continue;
                }
                doGetControllingExpiry(now, k, true, true, false, completionListener != null);
            }
        }
        catch (final RuntimeException e)
        {
            if (completionListener != null)
            {
                completionListener.onException(e);
                return;
            }
        }
        if (completionListener != null)
        {
            completionListener.onCompletion();
        }
    }

    @Override
    public <T> T invoke(final K key, final EntryProcessor<K, V, T> entryProcessor, final Object... arguments) throws EntryProcessorException
    {
        final TempStateCacheView<K, V> view = new TempStateCacheView<>(this);
        final T t = doInvoke(view, key, entryProcessor, arguments);
        view.merge();
        return t;
    }

    private <T> T doInvoke(final TempStateCacheView<K, V> view, final K key, final EntryProcessor<K, V, T> entryProcessor,
            final Object... arguments)
    {
        assertNotClosed();
        assertNotNull(entryProcessor, "entryProcessor");
        assertNotNull(key, "key");
        try
        {
            if (config.isStatisticsEnabled())
            {
                if (containsKey(key))
                {
                    statistics.increaseHits(1);
                }
                else
                {
                    statistics.increaseMisses(1);
                }
            }
            return entryProcessor.process(new JCSMutableEntry<>(view, key), arguments);
        }
        catch (final Exception ex)
        {
            return throwEntryProcessorException(ex);
        }
    }

    private static <T> T throwEntryProcessorException(final Exception ex)
    {
        if (EntryProcessorException.class.isInstance(ex))
        {
            throw EntryProcessorException.class.cast(ex);
        }
        throw new EntryProcessorException(ex);
    }

    @Override
    public <T> Map<K, EntryProcessorResult<T>> invokeAll(final Set<? extends K> keys, final EntryProcessor<K, V, T> entryProcessor,
            final Object... arguments)
    {
        assertNotClosed();
        assertNotNull(entryProcessor, "entryProcessor");
        final Map<K, EntryProcessorResult<T>> results = new HashMap<>();
        for (final K k : keys)
        {
            try
            {
                final T invoke = invoke(k, entryProcessor, arguments);
                if (invoke != null)
                {
                    results.put(k, () -> invoke);
                }
            }
            catch (final Exception e)
            {
                results.put(k, () -> throwEntryProcessorException(e));
            }
        }
        return results;
    }

    @Override
    public void registerCacheEntryListener(final CacheEntryListenerConfiguration<K, V> cacheEntryListenerConfiguration)
    {
        assertNotClosed();
        if (listeners.containsKey(cacheEntryListenerConfiguration))
        {
            throw new IllegalArgumentException(cacheEntryListenerConfiguration + " already registered");
        }
        listeners.put(cacheEntryListenerConfiguration, new JCSListener<>(cacheEntryListenerConfiguration));
        config.addListener(cacheEntryListenerConfiguration);
    }

    @Override
    public void deregisterCacheEntryListener(final CacheEntryListenerConfiguration<K, V> cacheEntryListenerConfiguration)
    {
        assertNotClosed();
        listeners.remove(cacheEntryListenerConfiguration);
        config.removeListener(cacheEntryListenerConfiguration);
    }

    @Override
    public Iterator<Entry<K, V>> iterator()
    {
        assertNotClosed();
        final Iterator<K> keys = new HashSet<>(delegate.getKeySet()).iterator();
        return new Iterator<Entry<K, V>>()
        {
            private K lastKey;

            @Override
            public boolean hasNext()
            {
                return keys.hasNext();
            }

            @Override
            public Entry<K, V> next()
            {
                lastKey = keys.next();
                return new JCSEntry<>(lastKey, get(lastKey));
            }

            @Override
            public void remove()
            {
                if (isClosed() || lastKey == null)
                {
                    throw new IllegalStateException(isClosed() ? "cache closed" : "call next() before remove()");
                }
                JCSCache.this.remove(lastKey);
            }
        };
    }

    @Override
    public String getName()
    {
        assertNotClosed();
        return name;
    }

    @Override
    public CacheManager getCacheManager()
    {
        assertNotClosed();
        return manager;
    }

    @Override
    public synchronized void close()
    {
        if (isClosed())
        {
            return;
        }

        for (final Runnable task : pool.shutdownNow()) {
            task.run();
        }

        manager.release(getName());
        closed = true;
        close(loader);
        close(writer);
        close(expiryPolicy);
        for (final JCSListener<K, V> listener : listeners.values())
        {
            close(listener);
        }
        listeners.clear();
        JMXs.unregister(cacheConfigObjectName);
        JMXs.unregister(cacheStatsObjectName);
        try
        {
            delegate.removeAll();
        }
        catch (final IOException e)
        {
            throw new CacheException(e);
        }
    }

    private static void close(final Object potentiallyCloseable)
    {
        if (Closeable.class.isInstance(potentiallyCloseable))
        {
            Closeable.class.cast(potentiallyCloseable);
        }
    }

    @Override
    public boolean isClosed()
    {
        return closed;
    }

    @Override
    public <T> T unwrap(final Class<T> clazz)
    {
        assertNotClosed();
        if (clazz.isInstance(this))
        {
            return clazz.cast(this);
        }
        if (clazz.isAssignableFrom(Map.class) || clazz.isAssignableFrom(ConcurrentMap.class))
        {
            return clazz.cast(delegate);
        }
        throw new IllegalArgumentException(clazz.getName() + " not supported in unwrap");
    }

    public Statistics getStatistics()
    {
        return statistics;
    }

    public void enableManagement()
    {
        config.managementEnabled();
        JMXs.register(cacheConfigObjectName, new JCSCacheMXBean<>(this));
    }

    public void disableManagement()
    {
        config.managementDisabled();
        JMXs.unregister(cacheConfigObjectName);
    }

    public void enableStatistics()
    {
        config.statisticsEnabled();
        statistics.setActive(true);
        JMXs.register(cacheStatsObjectName, new JCSCacheStatisticsMXBean(statistics));
    }

    public void disableStatistics()
    {
        config.statisticsDisabled();
        statistics.setActive(false);
        JMXs.unregister(cacheStatsObjectName);
    }
}