/*
 * JBoss, the OpenSource J2EE webOS
 *
 * Distributable under LGPL license.
 * See terms of license at gnu.org.
 */
package org.jboss.cache.pojo.impl;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.aop.Advised;
import org.jboss.aop.Advisor;
import org.jboss.aop.InstanceAdvisor;
import org.jboss.aop.advice.Interceptor;
import org.jboss.aop.proxy.ClassProxy;
import org.jboss.cache.Cache;
import org.jboss.cache.CacheException;
import org.jboss.cache.CacheSPI;
import org.jboss.cache.Fqn;
import org.jboss.cache.Node;
import org.jboss.cache.Region;
import org.jboss.cache.pojo.PojoCacheException;
import org.jboss.cache.pojo.collection.CollectionInterceptorUtil;
import org.jboss.cache.pojo.interceptors.dynamic.AbstractCollectionInterceptor;
import org.jboss.cache.pojo.interceptors.dynamic.BaseInterceptor;
import org.jboss.cache.pojo.memory.FieldPersistentReference;
import org.jboss.cache.pojo.util.AopUtil;

import java.lang.reflect.Field;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * Delegate class for PojoCache, the real implementation code happens here.
 *
 * @author Ben Wang
 */
public class PojoCacheDelegate
{
   private PojoCacheImpl pojoCache;
   private Cache<Object, Object> cache;
   private final static Log log = LogFactory.getLog(PojoCacheDelegate.class);
   private InternalHelper internal_;
   private AdvisedPojoHandler advisedHandler_;
   private ObjectGraphHandler graphHandler_;
   private CollectionClassHandler collectionHandler_;
   private SerializableObjectHandler serializableHandler_;
   // Use ThreadLocal to hold a boolean isBulkRemove
   private ThreadLocal<Boolean> bulkRemove_ = new ThreadLocal<Boolean>();
   private final String DETACH = "DETACH";
   private PojoUtil util_ = new PojoUtil();

   public PojoCacheDelegate(PojoCacheImpl cache)
   {
      pojoCache = cache;
      this.cache = pojoCache.getCache();
      internal_ = new InternalHelper(cache);
      graphHandler_ = new ObjectGraphHandler(pojoCache, internal_);
      collectionHandler_ = new CollectionClassHandler(pojoCache, internal_);
      serializableHandler_ = new SerializableObjectHandler(pojoCache, internal_);
      advisedHandler_ = new AdvisedPojoHandler(pojoCache, internal_, util_);
   }

   public void setBulkRemove(boolean bulk)
   {
      bulkRemove_.set(bulk);
   }

   private boolean getBulkRemove()
   {
      return bulkRemove_.get();
   }

   public Object getObject(Fqn fqn, String field) throws CacheException
   {
      // TODO Must we really to couple with BR? JBCACHE-669
      Object pojo = internal_.getPojo(fqn, field);
      if (pojo != null)
      {
         // we already have an advised instance
         if (log.isDebugEnabled())
         {
            log.debug("getObject(): id: " + fqn + " retrieved from existing instance directly. ");
         }
         return pojo;
      }

      // OK. So we are here meaning that this is a failover or passivation since the transient
      // pojo instance is not around. Let's also make sure the right classloader is used
      // as well.
      ClassLoader prevCL = Thread.currentThread().getContextClassLoader();
      try
      {
         Region region = cache.getRegion(fqn, false);
         if (region != null && region.getClassLoader() != null)
            Thread.currentThread().setContextClassLoader(region.getClassLoader());

         return getObjectInternal(fqn, field);
      }
      finally
      {
         Thread.currentThread().setContextClassLoader(prevCL);
      }
   }

   public Object putObjectI(Fqn fqn, Object obj, String field) throws CacheException
   {
      // Skip some un-necessary update if obj is the same class as the old one
      Object oldValue = internal_.getPojo(fqn, field);
      if (oldValue == obj && (obj instanceof Advised || obj instanceof ClassProxy))
      {
         if (log.isDebugEnabled())
         {
            log.debug("putObject(): id: " + fqn + " pojo is already in the cache. Return right away.");
         }
         return obj;
      }
      return null;
   }

   /**
    * Note that caller of this method will take care of synchronization within the <code>fqn</code> sub-tree.
    */
   public Object putObjectII(Fqn fqn, Object obj, String field) throws CacheException
   {
      // Skip some un-necessary update if obj is the same class as the old one
      Object oldValue = internal_.getPojo(fqn, field);
      if (oldValue == obj)
      {
         if (log.isDebugEnabled())
         {
            log.debug("putObject(): id: " + fqn + " pojo is already in the cache. Return right away.");
         }
         return obj;
      }

      // remove old value before overwriting it. This is necessary to detach any interceptor.
      // TODO Or can we simply walk thru that somewhere? Well, there is also implication of Collection though
      pojoCache.detach(fqn, field);

      if (obj == null)
      {
         return oldValue;// we are done
      }

      // creates the internal node first without going thru the interceptor.
      // This way we don't block on __JBossInternal__ node.
      //createChildNodeFirstWithoutLocking(internalFqn);

      if ((obj instanceof Advised || obj instanceof ClassProxy) && isMultipleReferencedPut(obj))
      {
         // we pass in the originating fqn intentionaly
         graphHandler_.put(fqn, obj, field);
      }
      else
      {
         Fqn internalFqn = createInternalFqn(fqn, obj);
         if (log.isDebugEnabled())
         {
            log.debug("putObject(): id: " + fqn + " will store the pojo in the internal area: "
                      + internalFqn);
         }

         if (obj instanceof Advised)
         {
            advisedHandler_.put(internalFqn, fqn, obj);
         }
         else if (isCollection(obj))
         {
            collectionHandler_.put(internalFqn, fqn, obj);
            //
         }
         else
         {
            // must be Serializable, including primitive types
            serializableHandler_.put(internalFqn, obj);
         }

         // Used by notification sub-system
         cache.put(internalFqn, InternalConstant.POJOCACHE_STATUS, "ATTACHED");

         setPojoReference(fqn, obj, field, internalFqn);
      }

      return oldValue;
   }

   Fqn createInternalFqn(Fqn fqn, Object obj) throws CacheException
   {
      // Create an internal Fqn name
      return AopUtil.createInternalFqn(fqn, cache);
   }

   Fqn setPojoReference(Fqn fqn, Object obj, String field, Fqn internalFqn) throws CacheException
   {
      // Create PojoReference
      CachedType type = pojoCache.getCachedType(obj.getClass());
      PojoReference pojoReference = new PojoReference();
      pojoReference.setPojoClass(type.getType());

      // store PojoReference
      pojoReference.setFqn(internalFqn);
      internal_.putPojoReference(fqn, pojoReference, field);
      if (log.isDebugEnabled())
      {
         log.debug("put(): inserting PojoReference with id: " + fqn);
      }
      // store obj in the internal fqn
      return internalFqn;
   }

   private void createChildNodeFirstWithoutLocking(Fqn internalFqn)
   {
      int size = internalFqn.size();
      Fqn f = internalFqn.getSubFqn(0, size - 1);
      Fqn child = internalFqn.getSubFqn(size - 1, size);

      Node base = cache.getRoot().getChild(f);
      if (base == null)
      {
         log.debug("The node retrieved is null from fqn: " + f);
         return;
      }
      base.addChild(child);
   }

   /**
    * Note that caller of this method will take care of synchronization within the <code>fqn</code> sub-tree.
    *
    * @param fqn
    * @return
    * @throws CacheException
    */
   public Object removeObject(Fqn fqn, String field) throws CacheException
   {
      // the class attribute is implicitly stored as an immutable read-only attribute
      PojoReference pojoReference = internal_.getPojoReference(fqn, field);
      if (pojoReference == null)
      {
         //  clazz and pojoReference can be not null if this node is the replicated brother node.
         if (log.isTraceEnabled())
         {
            log.trace("removeObject(): clazz is null. id: " + fqn + " No need to remove.");
         }
         return null;
      }

      Class clazz = pojoReference.getPojoClass();
      Fqn internalFqn = pojoReference.getFqn();

      if (log.isDebugEnabled())
      {
         log.debug("removeObject(): removing object from id: " + fqn
                   + " with the corresponding internal id: " + internalFqn);
      }

      Object result = pojoCache.getObject(internalFqn);
      if (result == null)
      {
         return null;
      }

      if (graphHandler_.isMultipleReferenced(internalFqn))
      {
         graphHandler_.remove(fqn, internalFqn, result);
      }
      else
      {
         cache.put(internalFqn, InternalConstant.POJOCACHE_STATUS, "DETACHING");
         if (Advised.class.isAssignableFrom(clazz))
         {
            advisedHandler_.remove(internalFqn, result, clazz);
            internal_.cleanUp(internalFqn, null);
         }
         else if (isCollectionGet(clazz))
         {
            // We need to return the original reference
            result = collectionHandler_.remove(internalFqn, result);
            internal_.cleanUp(internalFqn, null);
         }
         else
         {// Just Serializable objects. Do a brute force remove is ok.
            serializableHandler_.remove();
            internal_.cleanUp(internalFqn, null);
         }
      }

      internal_.cleanUp(fqn, field);
      // remove the interceptor as well.
      return result;
   }

   public Map findObjects(Fqn fqn) throws CacheException
   {

      // Traverse from fqn to do getObject, if it return a pojo we then stop.
      Map map = new HashMap();
      Object pojo = getObject(fqn, null);
      if (pojo != null)
      {
         map.put(fqn, pojo);// we are done!
         return map;
      }

      findChildObjects(fqn, map);
      if (log.isDebugEnabled())
      {
         log.debug("_findObjects(): id: " + fqn + " size of pojos found: " + map.size());
      }
      return map;
   }

   private Object getObjectInternal(Fqn fqn, String field) throws CacheException
   {
      Fqn internalFqn = fqn;
      PojoReference pojoReference = internal_.getPojoReference(fqn, field);
      if (pojoReference != null)
      {
         internalFqn = pojoReference.getFqn();
      }
      else if (field != null)
      {
         return null;
      }

      if (log.isDebugEnabled())
         log.debug("getObject(): id: " + fqn + " with a corresponding internal id: " + internalFqn);

      /**
       * Reconstruct the managed POJO
       */
      Object obj;

      PojoInstance pojoInstance = internal_.getPojoInstance(internalFqn);

      if (pojoInstance == null)
         return null;
         //throw new PojoCacheException("PojoCacheDelegate.getObjectInternal(): null PojoInstance for fqn: " + internalFqn);

      Class clazz = pojoInstance.getPojoClass();

      // Check for both Advised and Collection classes for object graph.
      // Note: no need to worry about multiple referencing here. If there is a graph, we won't come this far.
      if (Advised.class.isAssignableFrom(clazz))
      {
         obj = advisedHandler_.get(internalFqn, clazz, pojoInstance);
      }
      else if (isCollectionGet(clazz))
      {// Must be Collection classes. We will use aop.ClassProxy instance instead.
         obj = collectionHandler_.get(internalFqn, clazz, pojoInstance);
      }
      else
      {
         // Maybe it is just a serialized object.
         obj = serializableHandler_.get(internalFqn, clazz, pojoInstance);
      }

      InternalHelper.setPojo(pojoInstance, obj);
      return obj;
   }

   private boolean isCollectionGet(Class clazz)
   {
      if (Map.class.isAssignableFrom(clazz) || Collection.class.isAssignableFrom(clazz))
      {
         return true;
      }

      return false;
   }


   private boolean isMultipleReferencedPut(Object obj)
   {
      Interceptor interceptor = null;
      if (obj instanceof Advised)
      {
         InstanceAdvisor advisor = ((Advised) obj)._getInstanceAdvisor();
         if (advisor == null)
         {
            throw new PojoCacheException("_putObject(): InstanceAdvisor is null for: " + obj);
         }

         // Step Check for cross references
         interceptor = AopUtil.findCacheInterceptor(advisor);
      }
      else
      {
         interceptor = CollectionInterceptorUtil.getInterceptor((ClassProxy) obj);
      }
      if (interceptor == null) return false;

      Fqn originalFqn = null;

      // ah, found something. So this will be multiple referenced.
      originalFqn = ((BaseInterceptor) interceptor).getFqn();

      return originalFqn != null;

   }

   private boolean isCollection(Object obj)
   {
      return obj instanceof Collection || obj instanceof Map;

   }

   private void detachInterceptor(InstanceAdvisor advisor, Interceptor interceptor,
                                  boolean detachOnly, Map undoMap)
   {
      if (!detachOnly)
      {
         util_.detachInterceptor(advisor, interceptor);
         undoMap.put(advisor, interceptor);
      }
      else
      {
         undoMap.put(DETACH, interceptor);
      }
   }

   private static void undoInterceptorDetach(Map undoMap)
   {
      for (Iterator it = undoMap.keySet().iterator(); it.hasNext();)
      {
         Object obj = it.next();

         if (obj instanceof InstanceAdvisor)
         {
            InstanceAdvisor advisor = (InstanceAdvisor) obj;
            BaseInterceptor interceptor = (BaseInterceptor) undoMap.get(advisor);

            if (interceptor == null)
            {
               throw new IllegalStateException("PojoCacheDelegate.undoInterceptorDetach(): null interceptor");
            }

            advisor.appendInterceptor(interceptor);
         }
         else
         {
            BaseInterceptor interceptor = (BaseInterceptor) undoMap.get(obj);
            boolean copyToCache = false;
            ((AbstractCollectionInterceptor) interceptor).attach(null, copyToCache);
         }
      }
   }

   private void findChildObjects(Fqn fqn, Map map) throws CacheException
   {
      // We need to traverse then
      Node root = cache.getRoot();
      Node current = root.getChild(fqn);

      if (current == null) return;

      Collection<Node> col = current.getChildren();
      if (col == null) return;
      for (Node n : col)
      {
         Fqn newFqn = n.getFqn();
         if (InternalHelper.isInternalNode(newFqn)) continue;// skip

         Object pojo = getObject(newFqn, null);
         if (pojo != null)
         {
            map.put(newFqn, pojo);
         }
         else
         {
            findChildObjects(newFqn, map);
         }
      }
   }

   public boolean exists(Fqn<?> id)
   {
      return internal_.getPojoReference(id, null) != null || internal_.getPojoInstance(id) != null;
   }
}
