package com.atlassian.cache.ehcache;
   
   import java.io.Serializable;
   
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   
   import net.sf.ehcache.CacheException;
   import net.sf.ehcache.Ehcache;
  import net.sf.ehcache.Element;
  
  /**
   * An {@link RMISynchronousCacheReplicator} that works around https://jira.terracotta.org/jira/browse/EHC-683.
   *
   * @since 2.0
   */
  public class RMISynchronousCacheReplicator extends net.sf.ehcache.distribution.RMISynchronousCacheReplicator
  {
      // Log as the superclass to keep all replication messages within the same logger name
      private static final Logger LOG =
              LoggerFactory.getLogger(net.sf.ehcache.distribution.RMISynchronousCacheReplicator.class);
  
      /**
       * Constructor
       *
       * @param replicatePuts whether to replicate puts
       * @param replicatePutsViaCopy whether to replicate puts by copying the serialized value
       * @param replicateUpdates whether to replicate updates
       * @param replicateUpdatesViaCopy whether to replicate updates by copying the serialized value
       * @param replicateRemovals whether to replicate removals
       */
      public RMISynchronousCacheReplicator(final boolean replicatePuts, final boolean replicatePutsViaCopy,
              final boolean replicateUpdates, final boolean replicateUpdatesViaCopy, final boolean replicateRemovals)
      {
          super(replicatePuts, replicatePutsViaCopy, replicateUpdates, replicateUpdatesViaCopy, replicateRemovals);
      }
  
      // Overridden to fix https://jira.terracotta.org/jira/browse/EHC-683
      @Override
      public void notifyElementPut(final Ehcache cache, final Element element) throws CacheException {
          if (notAlive() || !replicatePuts) {
              return;
          }
  
          if (replicatePutsViaCopy) {
              replicateViaCopy(cache, element);
          }
          else {
              replicateViaKeyInvalidation(cache, element);
          }
      }
  
      private void replicateViaCopy(final Ehcache cache, final Element element)
      {
          if (element.isSerializable()) {
              replicatePutNotification(cache, element);
              return;
          }
          if (!element.isKeySerializable()) {
              logUnserializableKey(element);
          }
          if (LOG.isWarnEnabled() && !(element.getObjectValue() instanceof Serializable)) {
              LOG.error("Value class {} is not Serializable => cannot be replicated",
                      element.getObjectValue().getClass().getName());
          }
      }
  
      private void replicateViaKeyInvalidation(final Ehcache cache, final Element element)
      {
          if (element.isKeySerializable()) {
              replicateRemovalNotification(cache, (Serializable) element.getObjectKey());
              return;
          }
          logUnserializableKey(element);
      }
  
      private void logUnserializableKey(final Element element)
      {
          if (LOG.isWarnEnabled()) {
              LOG.error("Key class {} is not Serializable => cannot be replicated",
                      element.getObjectKey().getClass().getName());
          }
      }
  }