package com.atlassian.cache.hazelcast;
   
   import java.util.Collection;
   import java.util.concurrent.TimeUnit;
   
   import javax.annotation.Nonnull;
   
   import com.atlassian.cache.Cache;
   import com.atlassian.cache.CacheEntryEvent;
  import com.atlassian.cache.CacheEntryListener;
  import com.atlassian.cache.CacheException;
  import com.atlassian.cache.CacheLoader;
  import com.atlassian.cache.CacheSettings;
  import com.atlassian.cache.ManagedCache;
  import com.atlassian.cache.Supplier;
  import com.atlassian.cache.impl.CacheEntryListenerSupport;
  import com.atlassian.cache.impl.CacheLoaderSupplier;
  import com.atlassian.cache.impl.DefaultCacheEntryEvent;
  import com.atlassian.cache.impl.ValueCacheEntryListenerSupport;
  import com.atlassian.hazelcast.serialization.OsgiSafe;
  
  import com.google.common.base.Objects;
  import com.google.common.base.Throwables;
  import com.hazelcast.config.MapConfig;
  import com.hazelcast.core.EntryEvent;
  import com.hazelcast.core.EntryListener;
  import com.hazelcast.core.IMap;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import static com.atlassian.cache.hazelcast.OsgiSafeUtils.unwrap;
  import static com.atlassian.cache.hazelcast.OsgiSafeUtils.wrap;
  import static com.google.common.base.Preconditions.checkNotNull;
  
  /**
   * Hazelcast implementation of the {@link Cache} and {@link ManagedCache} interfaces
   *
   * @since 2.4.0
   */
  public class HazelcastCache<K, V> extends ManagedCacheSupport implements Cache<K, V>
  {
      private static final Logger log = LoggerFactory.getLogger(HazelcastCache.class);
  
      private final CacheLoader<K, V> cacheLoader;
      private final IMap<K, OsgiSafe<V>> map;
      private final CacheEntryListenerSupport<K, OsgiSafe<V>> listenerSupport = new ValueCacheEntryListenerSupport<K, OsgiSafe<V>>()
      {
          @Override
          protected void initValue(final CacheEntryListenerSupport<K, OsgiSafe<V>> actualListenerSupport)
          {
              map.addEntryListener(new HazelcastCacheEntryListener<K, OsgiSafe<V>>(actualListenerSupport), true);
          }
  
          @Override
          protected void initValueless(final CacheEntryListenerSupport<K, OsgiSafe<V>> actualListenerSupport)
          {
              map.addEntryListener(new HazelcastCacheEntryListener<K, OsgiSafe<V>>(actualListenerSupport), false);
          }
      };
  
      public HazelcastCache(String name, IMap<K, OsgiSafe<V>> map, MapConfig config, CacheLoader<K, V> cacheLoader,
              CacheSettings settings)
      {
          super(name, config, settings);
  
          this.map = map;
          this.cacheLoader = cacheLoader;
      }
  
      @Override
      public void clear()
      {
          if (isFlushable())
          {
              map.clear();
          }
          else
          {
              log.debug("Not clearing cache {} because it's configured as not flushable.", getName());
          }
      }
  
      @Override
      public boolean containsKey(@Nonnull K k)
      {
          return map.containsKey(k);
      }
  
      @SuppressWarnings ("unchecked")
      @Override
      public V get(@Nonnull final K key)
      {
          return getOrLoad(key, cacheLoader == null ? null : new CacheLoaderSupplier<K, V>(key, cacheLoader));
      }
  
      @Nonnull
      @Override
      public V get(@Nonnull final K key, @Nonnull final Supplier<? extends V> valueSupplier)
     {
         return getOrLoad(key, valueSupplier);
     }
 
     @Nonnull
     @Override
     public Collection<K> getKeys()
     {
         return map.keySet();
     }
 
     @Override
     public void put(@Nonnull K key, @Nonnull V value)
     {
         map.put(checkNotNull(key, "key"), wrap(checkNotNull(value, "value")));
     }
 
     @Override
     public V putIfAbsent(@Nonnull K key, @Nonnull V value)
     {
         return unwrap(map.putIfAbsent(checkNotNull(key, "key"), wrap(checkNotNull(value, "value"))));
     }
 
     @Override
     public void remove(@Nonnull K key)
     {
         map.remove(checkNotNull(key, "key"));
     }
 
     @Override
     public boolean remove(@Nonnull K key, @Nonnull V value)
     {
         return map.remove(checkNotNull(key, "key"), wrap(checkNotNull(value, "value")));
     }
 
     @Override
     public void removeAll()
     {
         map.clear();
     }
 
     @Override
     public boolean replace(@Nonnull K key, @Nonnull V oldValue, @Nonnull V newValue)
     {
         return map.replace(checkNotNull(key, "key"),
                 wrap(checkNotNull(oldValue, "oldValue")), wrap(checkNotNull(newValue, "newValue")));
     }
 
     @Override
     public void addListener(@Nonnull CacheEntryListener<K, V> listener, boolean includeValues)
     {
         listenerSupport.add(new OsgiSafeCacheEntryListener<K, V>(listener), includeValues);
     }
 
     @Override
     public void removeListener(@Nonnull CacheEntryListener<K, V> listener)
     {
         listenerSupport.remove(new OsgiSafeCacheEntryListener<K, V>(listener));
     }
 
     private static class OsgiSafeCacheEntryEvent<K, V> extends DefaultCacheEntryEvent<K, V>
     {
         public OsgiSafeCacheEntryEvent(CacheEntryEvent<K, OsgiSafe<V>> event)
         {
             super(event.getKey(), unwrap(event.getValue()), unwrap(event.getOldValue()));
         }
     }
 
     private static class OsgiSafeCacheEntryListener<K, V> implements CacheEntryListener<K, OsgiSafe<V>>
     {
         private final CacheEntryListener<K, V> delegate;
 
         private OsgiSafeCacheEntryListener(CacheEntryListener<K, V> listener)
         {
             this.delegate = checkNotNull(listener, "listener");
         }
 
         @Override
         public void onAdd(@Nonnull CacheEntryEvent<K, OsgiSafe<V>> event)
         {
             delegate.onAdd(new OsgiSafeCacheEntryEvent<K, V>(event));
         }
 
         @Override
         public void onEvict(@Nonnull CacheEntryEvent<K, OsgiSafe<V>> event)
         {
             delegate.onEvict(new OsgiSafeCacheEntryEvent<K, V>(event));
         }
 
         @Override
         public void onRemove(@Nonnull CacheEntryEvent<K, OsgiSafe<V>> event)
         {
             delegate.onRemove(new OsgiSafeCacheEntryEvent<K, V>(event));
         }
 
         @Override
         public void onUpdate(@Nonnull CacheEntryEvent<K, OsgiSafe<V>> event)
         {
             delegate.onUpdate(new OsgiSafeCacheEntryEvent<K, V>(event));
         }
 
         @Override
         public boolean equals(Object o)
         {
             if (this == o)
             {
                 return true;
             }
             if (o == null || getClass() != o.getClass())
             {
                 return false;
             }
 
             OsgiSafeCacheEntryListener that = (OsgiSafeCacheEntryListener) o;
             return delegate.equals(that.delegate);
         }
 
         @Override
         public int hashCode()
         {
             return delegate.hashCode();
         }
     }
 
 
     private V getOrLoad(final K key, final Supplier<? extends V> valueSupplier)
     {
         try
         {
             OsgiSafe<V> value = map.get(checkNotNull(key, "key"));
             if (value != null)
             {
                 return value.getValue();
             }
             else if (valueSupplier == null)
             {
                 return null;
             }
 
             // note: we cannot use an EntryProcessor here because it would need to be passed the cacheLoader, which is
             // not Serializable. Take out a lock on the entry to ensure that we don't store a stale value in the cache
             // because the cache is invalidated between cache value calculation and adding it to the map.
             map.lock(key, 5, TimeUnit.MINUTES);
             try
             {
                 V newValue = valueSupplier.get();
                 //noinspection ConstantConditions
                 if (newValue == null)
                 {
                     throw new CacheException("The provided cacheLoader returned null. Null values are not supported.");
                 }
                 value = wrap(newValue);
                 OsgiSafe<V> current = map.putIfAbsent(key, value);
 
                 return unwrap(Objects.firstNonNull(current, value));
             }
             finally
             {
                 map.unlock(key);
             }
         }
         catch (RuntimeException e)
         {
             Throwables.propagateIfInstanceOf(e, CacheException.class);
             throw new CacheException("Problem retrieving a value from cache " + getName(), e);
         }
     }
 
     private static class HazelcastCacheEntryListener<K, V> implements EntryListener<K, V>
     {
         private final CacheEntryListenerSupport<K, V> listenerSupport;
 
         private HazelcastCacheEntryListener(final CacheEntryListenerSupport<K, V> listenerSupport)
         {
             this.listenerSupport = checkNotNull(listenerSupport, "listenerSupport");
         }
 
         @Override
         public void entryAdded(EntryEvent<K, V> event)
         {
             listenerSupport.notifyAdd(event.getKey(), event.getValue());
         }
 
         @Override
         public void entryRemoved(EntryEvent<K, V> event)
         {
             listenerSupport.notifyRemove(event.getKey(), event.getOldValue());
         }
 
         @Override
         public void entryUpdated(EntryEvent<K, V> event)
         {
             listenerSupport.notifyUpdate(event.getKey(), event.getValue(), event.getOldValue());
         }
 
         @Override
         public void entryEvicted(EntryEvent<K, V> event)
         {
             listenerSupport.notifyEvict(event.getKey(), event.getOldValue());
         }
     }
 }