package com.atlassian.cache.hazelcast;
   
   import java.util.Collection;
   
   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.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.MoreObjects;
  import com.google.common.base.Throwables;
  import com.hazelcast.core.EntryAdapter;
  import com.hazelcast.core.EntryEvent;
  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 CacheVersion cacheVersion;
  
      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<>(actualListenerSupport), true);
          }
  
          @Override
          protected void initValueless(final CacheEntryListenerSupport<K, OsgiSafe<V>> actualListenerSupport)
          {
              map.addEntryListener(new HazelcastCacheEntryListener<>(actualListenerSupport), false);
          }
      };
  
      public HazelcastCache(String name, IMap<K, OsgiSafe<V>> map, CacheLoader<K, V> cacheLoader,  CacheVersion cacheVersion, HazelcastCacheManager cacheManager)
      {
          super(name, cacheManager);
  
          this.map = map;
          this.cacheVersion = checkNotNull(cacheVersion);
          if (cacheLoader != null)
          {
              this.cacheLoader = new CacheVersionAwareCacheLoader<>(cacheLoader, this.cacheVersion);
          }
          else
          {
              this.cacheLoader = cacheLoader;
          }
      }
  
      @Override
      public void clear()
      {
          cleanupMap();
      }
  
      @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()
     {
         cleanupMap();
     }
 
     private void cleanupMap()
     {
       cacheVersion.incrementAndGet();
       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));
     }
 
     @Nonnull
     @Override
     protected String getHazelcastMapName()
     {
         return map.getName();
     }
 
     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<>(event));
         }
 
         @Override
         public void onEvict(@Nonnull CacheEntryEvent<K, OsgiSafe<V>> event)
         {
             delegate.onEvict(new OsgiSafeCacheEntryEvent<>(event));
         }
 
         @Override
         public void onRemove(@Nonnull CacheEntryEvent<K, OsgiSafe<V>> event)
         {
             delegate.onRemove(new OsgiSafeCacheEntryEvent<>(event));
         }
 
         @Override
         public void onUpdate(@Nonnull CacheEntryEvent<K, OsgiSafe<V>> event)
         {
             delegate.onUpdate(new OsgiSafeCacheEntryEvent<>(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;
             }
 
             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(MoreObjects.firstNonNull(current, value));
         }
         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> extends EntryAdapter<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());
         }
     }
 }