package com.atlassian.cache.vcache;
   
   import java.util.Optional;
   import javax.annotation.Nonnull;
   
   import com.atlassian.cache.CacheSettings;
   import com.atlassian.cache.CachedReference;
   import com.atlassian.cache.CachedReferenceListener;
   import com.atlassian.cache.Supplier;
  import com.atlassian.vcache.JvmCache;
  import com.atlassian.vcache.PutPolicy;
  import com.atlassian.vcache.StableReadExternalCache;
  
  import static com.atlassian.vcache.VCacheUtils.join;
  import static java.util.Objects.requireNonNull;
  
  class HybridCachedReference<V>
          extends ManagedCacheSupport
          implements CachedReference<V>
  {
      private static final String REFERENCE_KEY = "ReferenceKey";
  
      private final JvmCache<String, Versioned<V>> localVersioned;
      private final StableReadExternalCache<String> globalVersions;
      private final Supplier<V> supplier;
  
      HybridCachedReference(String name,
                            JvmCache<String, Versioned<V>> localVersioned,
                            StableReadExternalCache<String> globalVersions,
                            Supplier<V> supplier,
                            CacheSettings settings)
      {
          super(name, settings);
          this.localVersioned = requireNonNull(localVersioned);
          this.globalVersions = requireNonNull(globalVersions);
          this.supplier = requireNonNull(supplier);
      }
  
      @Nonnull
      @Override
      public V get()
      {
          final Optional<Versioned<V>> localVersion = localVersioned.get(REFERENCE_KEY);
          final Optional<String> globalVersion = join(globalVersions.get(REFERENCE_KEY));
  
          if (!localVersion.isPresent()
                  || !globalVersion.isPresent()
                  || !globalVersion.get().equals(localVersion.get().getVersion()))
          {
              // Either missing a version, or a version mismatch
              localVersioned.removeAll();
              final Versioned<V> newVersioned =
                      localVersioned.get(REFERENCE_KEY, () -> new Versioned<>(Utils.uniqueId(), supplier.get()));
              join(globalVersions.put(REFERENCE_KEY, newVersioned.getVersion(), PutPolicy.PUT_ALWAYS));
  
              return newVersioned.getValue();
          }
  
          return localVersion.get().getValue();
      }
  
      @Override
      public void reset()
      {
          join(globalVersions.put(REFERENCE_KEY, Utils.uniqueId(), PutPolicy.PUT_ALWAYS));
          localVersioned.removeAll();
      }
  
      @Override
      public void addListener(@Nonnull CachedReferenceListener<V> listener, boolean includeValues)
      {
          throw new UnsupportedOperationException("Unsupported when using the VCache implementation");
      }
  
      @Override
      public void removeListener(@Nonnull CachedReferenceListener<V> listener)
      {
          throw new UnsupportedOperationException("Unsupported when using the VCache implementation");
      }
  
      @Override
      public void clear()
      {
          if (isFlushable())
          {
              reset();
          }
      }
  
      @Override
      public boolean isLocal()
      {
          return false;
      }
  }