package com.atlassian.cache.hazelcast;
   
   import javax.annotation.Nonnull;
   import javax.annotation.Nullable;
   
   import com.atlassian.cache.CacheException;
   import com.atlassian.cache.CachedReference;
   import com.atlassian.cache.CachedReferenceEvent;
   import com.atlassian.cache.CachedReferenceListener;
  import com.atlassian.cache.Supplier;
  import com.atlassian.cache.impl.CachedReferenceListenerSupport;
  import com.atlassian.cache.impl.DefaultCachedReferenceEvent;
  import com.atlassian.cache.impl.ValueCachedReferenceListenerSupport;
  import com.atlassian.hazelcast.serialization.OsgiSafe;
  
  import com.google.common.base.Objects;
  import com.google.common.base.Throwables;
  import com.hazelcast.core.EntryAdapter;
  import com.hazelcast.core.EntryEvent;
  import com.hazelcast.core.IMap;
  
  import static com.atlassian.cache.hazelcast.OsgiSafeUtils.unwrap;
  import static com.atlassian.cache.hazelcast.OsgiSafeUtils.wrap;
  import static com.google.common.base.Preconditions.checkNotNull;
  
  /**
   * A {@link CachedReference} implementation to satisfy the requirements of {@link com.atlassian.cache.CacheFactory}.
   *
   * @since 2.4.0
   */
  public class HazelcastCachedReference<V> extends ManagedCacheSupport implements CachedReference<V>
  {
      private final IMap<String, OsgiSafe<V>> hazelcastMap;
      private final Supplier<V> supplier;
  
      private final CachedReferenceListenerSupport<OsgiSafe<V>> listenerSupport = new ValueCachedReferenceListenerSupport<OsgiSafe<V>>()
      {
          @Override
          protected void initValue(final CachedReferenceListenerSupport<OsgiSafe<V>> actualListenerSupport)
          {
              hazelcastMap.addEntryListener(new HazelcastCachedReferenceListener<OsgiSafe<V>>(actualListenerSupport), true);
          }
  
          @Override
          protected void initValueless(final CachedReferenceListenerSupport<OsgiSafe<V>> actualListenerSupport)
          {
              hazelcastMap.addEntryListener(new HazelcastCachedReferenceListener<OsgiSafe<V>>(actualListenerSupport), false);
          }
      };
  
      private static final String REFERENCE_KEY = "ReferenceKey";
  
      @SuppressWarnings ("AssignmentToCollectionOrArrayFieldFromParameter")
          // The map is a delegate
      HazelcastCachedReference(String name, IMap<String, OsgiSafe<V>> hazelcastMap, Supplier<V> supplier, HazelcastCacheManager cacheManager)
      {
          super(name, cacheManager);
  
          this.hazelcastMap = hazelcastMap;
          this.supplier = supplier;
      }
  
      @Override
      public void clear()
      {
          hazelcastMap.remove(REFERENCE_KEY);
      }
  
      @Nonnull
      @SuppressWarnings ("unchecked")
      @Override
      public V get()
      {
          try
          {
              OsgiSafe<V> value = hazelcastMap.get(REFERENCE_KEY);
              if (value == null)
              {
                  V newValue = supplier.get();
                  if (newValue == null)
                  {
                      throw new CacheException("The provided supplier returned null. Null values are not supported.");
                  }
                  value = wrap(newValue);
                  OsgiSafe<V> current = hazelcastMap.putIfAbsent(REFERENCE_KEY, value);
                  return unwrap(Objects.firstNonNull(current, value));
              }
              return unwrap(value);
          }
          catch (RuntimeException e)
          {
              Throwables.propagateIfInstanceOf(e, CacheException.class);
              throw new CacheException(e);
          }
      }
  
      @Override
      public void reset()
      {
         try
         {
             hazelcastMap.remove(REFERENCE_KEY);
         }
         catch (RuntimeException e)
         {
             throw new CacheException(e);
         }
     }
 
     @Override
     public boolean equals(@Nullable final Object o)
     {
         if (this == o)
         {
             return true;
         }
         if (o == null || getClass() != o.getClass())
         {
             return false;
         }
         HazelcastCachedReference<?> other = (HazelcastCachedReference<?>) o;
         return hazelcastMap.equals(other.hazelcastMap);
     }
 
     @Override
     public int hashCode()
     {
         return 3 + hazelcastMap.hashCode();
     }
 
     @Override
     public void addListener(@Nonnull CachedReferenceListener<V> listener, boolean includeValues)
     {
         listenerSupport.add(new OsgiSafeCachedReferenceListener<V>(listener), includeValues);
     }
 
     @Override
     public void removeListener(@Nonnull CachedReferenceListener<V> listener)
     {
         listenerSupport.remove(new OsgiSafeCachedReferenceListener<V>(listener));
     }
 
     @Override
     protected String getHazelcastMapName()
     {
         return hazelcastMap.getName();
     }
 
     private static class OsgiSafeCachedReferenceListener<V> implements CachedReferenceListener<OsgiSafe<V>>
     {
         private final CachedReferenceListener<V> delegate;
 
         private OsgiSafeCachedReferenceListener(CachedReferenceListener<V> listener)
         {
             this.delegate = checkNotNull(listener, "listener");
         }
 
         @Override
         public void onEvict(@Nonnull CachedReferenceEvent<OsgiSafe<V>> event)
         {
             delegate.onEvict(new DefaultCachedReferenceEvent<V>(unwrap(event.getValue())));
         }
 
         @Override
         public void onSet(@Nonnull CachedReferenceEvent<OsgiSafe<V>> event)
         {
             delegate.onSet(new DefaultCachedReferenceEvent<V>(unwrap(event.getValue())));
         }
 
         @Override
         public void onReset(@Nonnull CachedReferenceEvent<OsgiSafe<V>> event)
         {
             delegate.onReset(new DefaultCachedReferenceEvent<V>(unwrap(event.getValue())));
         }
 
         @Override
         public boolean equals(Object o)
         {
             if (this == o)
             {
                 return true;
             }
             if (o == null || getClass() != o.getClass())
             {
                 return false;
             }
 
             OsgiSafeCachedReferenceListener that = (OsgiSafeCachedReferenceListener) o;
             return delegate.equals(that.delegate);
         }
 
         @Override
         public int hashCode()
         {
             return delegate.hashCode();
         }
     }
 
     @SuppressWarnings ("unchecked")
     private static class HazelcastCachedReferenceListener<V> extends EntryAdapter<String, V>
     {
         private final CachedReferenceListenerSupport listenerSupport;
 
         private HazelcastCachedReferenceListener(final CachedReferenceListenerSupport listenerSupport)
         {
             this.listenerSupport = listenerSupport;
         }
 
         @Override
         public void entryAdded(EntryEvent<String, V> event)
         {
             listenerSupport.notifySet(event.getValue());
         }
 
         @Override
         public void entryRemoved(EntryEvent<String, V> event)
         {
             listenerSupport.notifyReset(event.getOldValue());
         }
 
         @Override
         public void entryUpdated(EntryEvent<String, V> event)
         {
             listenerSupport.notifySet(event.getValue());
         }
 
         @Override
         public void entryEvicted(EntryEvent<String, V> event)
         {
             listenerSupport.notifyEvict(event.getOldValue());
         }
     }
 }