package com.trendmicro.grid.acl.ds.cache;
   
   import org.infinispan.Cache;
   import org.infinispan.util.concurrent.NotifyingFuture;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   
   import java.util.*;
   import java.util.concurrent.ExecutionException;
  
  /**
   * Adapts a cache to CacheSource and CacheDestination with optional key and value translation.
   *
   * @author juergen_kellerer, 2011-03-24
   * @version 1.0
   */
  public class CacheAdapter<K, V> implements CacheSource<K, V>, CacheDestination<K, V> {
  
  	private static final Logger log = LoggerFactory.getLogger(CacheAdapter.class);
  
  	protected boolean sync;
  	protected Cache<K, V> delegate;
  
  	/**
  	 * Constructs a new adapter for the specified delegate.
  	 *
  	 * @param delegate the cache to adapt.
  	 * @param sync     specifies whether all cache operations must be synchronous.
  	 */
  	public CacheAdapter(Cache<K, V> delegate, boolean sync) {
  		this.delegate = delegate;
  		this.sync = sync;
  	}
  
  	/**
  	 * {@inheritDoc}
  	 */
  	@Override
  	@SuppressWarnings("unchecked")
  	public List<V> getAll(Collection<K> keys) {
  		if (sync) {
  			final List<V> result = new ArrayList<V>(keys.size());
  			for (K key : keys)
  				result.add(key == null ? null : delegate.get(key));
  
  			return result;
  		} else {
  			final NotifyingFuture<V>[] futures = new NotifyingFuture[keys.size()];
  			int i = 0;
  			for (K key : keys)
  				futures[i++] = key == null ? null : delegate.getAsync(key);
  
  			return waitFor(Arrays.asList(futures));
  		}
  	}
  
  	/**
  	 * {@inheritDoc}
  	 */
  	@Override
  	public boolean containsKey(K key) {
  		return key != null && delegate.containsKey(key);
  	}
  
  	/**
  	 * {@inheritDoc}
  	 */
  	@Override
  	public int size() {
  		return delegate.size();
  	}
  
  	/**
  	 * {@inheritDoc}
  	 */
  	@Override
  	public boolean isEmpty() {
  		return delegate.isEmpty();
  	}
  
  	/**
  	 * Waits on all notifying futures and returns their results.
  	 *
  	 * @param futures the futures to wait on.
  	 * @return the results of the futures.
  	 */
  	protected <V> List<V> waitFor(Collection<NotifyingFuture<V>> futures) {
  		final List<V> results = new ArrayList<V>(futures.size());
  		try {
  			for (NotifyingFuture<V> future : futures)
  				results.add(future == null ? null : future.get());
  
  			return results;
  
  		} catch (InterruptedException e) {
  			Thread.currentThread().interrupt();
  			throw new RuntimeException(e);
  		} catch (ExecutionException e) {
  			throw new RuntimeException(e);
 		}
 	}
 
 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public void removeAll(Collection<K> keys) {
 		final Collection<NotifyingFuture<V>> futures = new ArrayList<NotifyingFuture<V>>(keys.size());
 		for (K key : keys) {
 			if (!skipKey(key, null)) futures.add(delegate.removeAsync(key));
 		}
 
 		if (sync) waitFor(futures);
 	}
 
 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public void clear() {
 		if (sync)
 			delegate.clear();
 		else
 			delegate.clearAsync();
 	}
 
 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public boolean supportsBatching() {
 		return delegate.getConfiguration().isInvocationBatchingEnabled();
 	}
 
 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public boolean startBatch() {
 		return delegate.startBatch();
 	}
 
 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public void endBatch(boolean success) {
 		delegate.endBatch(success);
 	}
 
 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	@SuppressWarnings("unchecked")
 	public void putAll(Map<? extends K, ? extends V> values, Mode mode) {
 		switch (mode) {
 			case PutIfAbsent:
 				putIfAbsent((Map<K, V>) values);
 				break;
 
 			case PutIfValueDiffers:
 				putIfValueDiffers((Map<K, V>) values);
 				break;
 
 			case ReplaceValue:
 				replaceValue((Map<K, V>) values);
 				break;
 
 			default:
 				Map<? extends K, ? extends V> valuesToPut = null;
 				for (Map.Entry<? extends K, ? extends V> entry : values.entrySet()) {
 					if (skipKey(entry.getKey(), entry.getValue())) {
 						if (valuesToPut == null) valuesToPut = new LinkedHashMap<K, V>(values);
 
 						valuesToPut.remove(entry.getKey());
 					}
 				}
 
 				if (valuesToPut == null) valuesToPut = values;
 
 				if (sync)
 					delegate.putAll(valuesToPut);
 				else
 					delegate.putAllAsync(valuesToPut);
 		}
 
 		logCacheWrites((Map<K, V>) values, mode);
 	}
 
 	/**
 	 * Implements the write mode "putIfAbsent".
 	 *
 	 * @param values the values to write to the cache.
 	 */
 	protected void putIfAbsent(Map<K, V> values) {
 		for (Map.Entry<K, V> entry : values.entrySet()) {
 			final K key = entry.getKey();
 			final V value = entry.getValue();
 
 			if (skipKey(key, value)) continue;
 
 			if (sync)
 				delegate.putIfAbsent(key, value);
 			else
 				delegate.putForExternalRead(key, value);
 		}
 	}
 
 	/**
 	 * Implements the write mode "replaceValue".
 	 *
 	 * @param values the values to write to the cache.
 	 */
 	protected void replaceValue(Map<K, V> values) {
 		final Collection<NotifyingFuture<V>> futures = new ArrayList<NotifyingFuture<V>>(values.size());
 		for (Map.Entry<K, V> entry : values.entrySet()) {
 			final K key = entry.getKey();
 			final V value = entry.getValue();
 
 			if (skipKey(key, value)) continue;
 
 			futures.add(delegate.replaceAsync(key, value));
 		}
 
 		if (sync) waitFor(futures);
 	}
 
 	/**
 	 * Implements the write mode "putIfValueDiffers".
 	 *
 	 * @param values the values to write to the cache.
 	 */
 	@SuppressWarnings("unchecked")
 	protected void putIfValueDiffers(Map<K, V> values) {
 		final Collection<NotifyingFuture<Object>> futures = new ArrayList<NotifyingFuture<Object>>(values.size());
 		for (Map.Entry<K, V> entry : values.entrySet()) {
 			final K key = entry.getKey();
 			final V value = entry.getValue();
 
 			if (skipKey(key, value)) continue;
 
 			final V oldValue = delegate.get(key);
 			if (oldValue == null)
 				futures.add((NotifyingFuture<Object>) delegate.putAsync(key, value));
 			else if (!oldValue.equals(value))
 				futures.add((NotifyingFuture) delegate.replaceAsync(key, oldValue, value));
 		}
 
 		if (sync) waitFor(futures);
 	}
 
 	/**
 	 * Returns true if the specified key should be skipped during a put operation.
 	 *
 	 * @param key   the key to evaluate.
 	 * @param value the value of the key.
 	 * @return true if the key should not be considered when writing data to the cache.
 	 */
 	protected boolean skipKey(K key, V value) {
 		if (key == null) {
 			if (log.isDebugEnabled()) log.debug("Skipping request to map the value '{}' against key 'null'. This is not supported!", value);
 			return true;
 		}
 		return false;
 	}
 
 	private void logCacheWrites(Map<K, V> values, Mode mode) {
 		if (log.isTraceEnabled()) {
 			for (Map.Entry<K, V> entry : values.entrySet()) {
 				final K key = entry.getKey();
 				final V value = entry.getValue();
 
 				if (skipKey(key, value)) continue;
 
 				String v = value == null ? "<null>" : value.getClass().getSimpleName();
 				log.trace("Cached '{}' for key '{}' (mode '{}')", new Object[]{v, key, mode});
 			}
 		}
 	}
 
 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public String toString() {
 		return "CacheAdapter{" +
 				"sync=" + sync +
 				", delegate=" + delegate +
 				'}';
 	}
 }