/**
 * Copyright (c) 2000-2017 Liferay, Inc. All rights reserved.
 *
 * The contents of this file are subject to the terms of the Liferay Enterprise
 * pSubscription License ("License"). You may not use this file except in
 * compliance with the License. You can obtain a copy of the License by
 * contacting Liferay, Inc. See the License for the specific language governing
 * permissions and limitations under the License, including but not limited to
 * distribution rights of the Software.
 *
 *
 *
 */

package com.liferay.portal.kernel.cache.index;

import com.liferay.portal.kernel.cache.CacheListener;
import com.liferay.portal.kernel.cache.PortalCache;
import com.liferay.portal.kernel.concurrent.ConcurrentHashSet;

import java.io.Serializable;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * @author Shuyang Zhou
 */
public class PortalCacheIndexer<I, K extends Serializable, V> {

	public PortalCacheIndexer(
		IndexEncoder<I, K> indexEncoder, PortalCache<K, V> portalCache) {

		_indexEncoder = indexEncoder;

		_portalCache = portalCache;

		_portalCache.registerCacheListener(new IndexerCacheListener());

		for (K indexedCacheKey : _portalCache.getKeys()) {
			_addIndexedCacheKey(indexedCacheKey);
		}
	}

	public Set<K> getKeys(I index) {
		Set<K> keys = _indexedCacheKeys.get(index);

		if (keys == null) {
			return Collections.emptySet();
		}

		return new HashSet<K>(keys);
	}

	public void removeKeys(I index) {
		Set<K> keys = _indexedCacheKeys.remove(index);

		if (keys == null) {
			return;
		}

		for (K key : keys) {
			_portalCache.remove(key);
		}
	}

	private void _addIndexedCacheKey(K key) {
		I index = _indexEncoder.encode(key);

		Set<K> keys = _indexedCacheKeys.get(index);

		if (keys == null) {
			Set<K> newKeys = new ConcurrentHashSet<K>();

			newKeys.add(key);

			keys = _indexedCacheKeys.putIfAbsent(index, newKeys);

			if (keys == null) {
				return;
			}
		}

		keys.add(key);
	}

	private void _removeIndexedCacheKey(K key) {
		I index = _indexEncoder.encode(key);

		Set<K> keys = _indexedCacheKeys.get(index);

		if (keys == null) {
			return;
		}

		keys.remove(key);

		if (keys.isEmpty() && _indexedCacheKeys.remove(index, keys)) {
			for (K victimIndexedCacheKey : keys) {
				_addIndexedCacheKey(victimIndexedCacheKey);
			}
		}
	}

	private final ConcurrentMap<I, Set<K>> _indexedCacheKeys =
		new ConcurrentHashMap<I, Set<K>>();
	private final IndexEncoder<I, K> _indexEncoder;
	private final PortalCache<K, V> _portalCache;

	private class IndexerCacheListener implements CacheListener<K, V> {

		@Override
		public void notifyEntryEvicted(
			PortalCache<K, V> portalCache, K key, V value) {

			_removeIndexedCacheKey(key);
		}

		@Override
		public void notifyEntryExpired(
			PortalCache<K, V> portalCache, K key, V value) {

			_removeIndexedCacheKey(key);
		}

		@Override
		public void notifyEntryPut(
			PortalCache<K, V> portalCache, K key, V value) {

			_addIndexedCacheKey(key);
		}

		@Override
		public void notifyEntryRemoved(
			PortalCache<K, V> portalCache, K key, V value) {

			_removeIndexedCacheKey(key);
		}

		@Override
		public void notifyEntryUpdated(
			PortalCache<K, V> portalCache, K key, V value) {
		}

		@Override
		public void notifyRemoveAll(PortalCache<K, V> portalCache) {
			_indexedCacheKeys.clear();
		}

	}

}