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

import org.jboss.logging.Logger;
import org.jboss.remoting.invocation.InternalInvocation;

import javax.management.MBeanServer;
import javax.management.MBeanServerInvocationHandler;
import javax.management.ObjectName;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


/**
 * Responsible for all callbacks in remoting at invoker level (on the server side).
 *
 * @author <a href="mailto:telrod@e2technologies.net">Tom Elrod</a>
 */
public class ServerInvokerCallbackHandler implements InvokerCallbackHandler
{
   private InvocationRequest invocation;
   private Client callBackClient;
   private ArrayList callbacks = new ArrayList();
   private String sessionId;
   private InvokerLocator serverLocator;

   private SerializableStore callbackStore = null;

   /**
    * The map key to use when looking up any callback store that
    * should be used.  This key should be used when setting up
    * config in the invoker.
    */
   public static final String CALLBACK_STORE_KEY = "callbackStore";

   /**
    * The map key to use when looking up the percentage of free memory
    * available before tiggering persistence.
    */
   public static final String CALLBACK_MEM_CEILING = "callbackMemCeiling";

   /**
    * The percentage number of used memory before should persist messages.
    * For example, if 64MB available and only 30MB free mem and memPercentCeiling
    * is 50, then would trigger persisting of messages.
    */
   private double memPercentCeiling = 20; // 20% by default

   private static final Logger log = Logger.getLogger(ServerInvokerCallbackHandler.class);


   public ServerInvokerCallbackHandler(InvocationRequest invocation, InvokerLocator serverLocator, ServerInvoker owner) throws Exception
   {
      if(invocation == null)
      {
         throw new Exception("Can not construct ServerInvokerCallbackHandler with null InvocationRequest.");
      }
      this.invocation = invocation;
      this.serverLocator = serverLocator;
      init(invocation, owner);
   }

   private void init(InvocationRequest invocation, ServerInvoker owner) throws Exception
   {
      sessionId = invocation.getSessionId();
      if(invocation.getLocator() != null)
      {
         callBackClient = new Client(invocation.getLocator(), invocation.getSubsystem());
         callBackClient.connect();
      }
      else
      {
         createCallbackStore(owner, sessionId);
      }

      if(log.isDebugEnabled())
      {
         log.debug("Session id for callback handler is " + sessionId);
      }
   }


   public void setMemPercentCeiling(Double ceiling)
   {
      if(ceiling != null)
      {
         memPercentCeiling = ceiling.doubleValue();
      }
   }

   public Double getMemPercentCeiling()
   {
      return new Double(memPercentCeiling);
   }

   private void createCallbackStore(ServerInvoker owner, String sessionId) throws Exception
   {
      Map config = owner.getConfiguration();
      if(config != null)
      {
         // should either be a fully qualified class name or a mbean object name
         String storeName = (String) config.get(CALLBACK_STORE_KEY);
         if(storeName != null)
         {
            // will first try as a MBean
            try
            {
               MBeanServer server = owner.getMBeanServer();
               ObjectName storeObjectName = new ObjectName(storeName);
               if(server != null)
               {
                  callbackStore = (SerializableStore)
                        MBeanServerInvocationHandler.newProxyInstance(server,
                                                                      storeObjectName,
                                                                      SerializableStore.class,
                                                                      false);
               }
            }
            catch(Exception ex)
            {
               log.debug("Could not create callback store from the configration value given (" + storeName + ") as an MBean.");
               if(log.isTraceEnabled())
               {
                  log.trace("Error is: " + ex.getMessage(), ex);
               }
               callbackStore = null;
            }

            // now try by class name
            if(callbackStore == null)
            {
               try
               {
                  Class storeClass = Class.forName(storeName);
                  callbackStore = (SerializableStore) storeClass.newInstance();
               }
               catch(Exception e)
               {
                  log.debug("Could not create callback store from the configuration value given (" + storeName + ") as a fully qualified class name.");
                  if(log.isTraceEnabled())
                  {
                     log.trace("Error is: " + e.getMessage(), e);
                  }
               }
            }
         }
      }

      // if still null, then just use default
      if(callbackStore == null)
      {
         callbackStore = new NullCallbackStore();
      }
      else
      {
         // need to modify configuration to include session id for the callback client.
         Map storeConfig = new HashMap();
         storeConfig.putAll(owner.getConfiguration());

         String newFilePath = null;

         String filePath = (String) storeConfig.get(CallbackStore.FILE_PATH_KEY);
         if(filePath == null)
         {
            newFilePath = System.getProperty("jboss.server.data.dir", "data");
         }
         newFilePath = newFilePath + System.getProperty("file.separator") + "remoting" +
                       System.getProperty("file.separator") + sessionId;

         storeConfig.put(CallbackStore.FILE_PATH_KEY, newFilePath);

         callbackStore.setConfig(storeConfig);
      }

      callbackStore.create();
      callbackStore.start();

      configureMemCeiling(owner.getConfiguration());
   }

   private void configureMemCeiling(Map configuration)
   {
      if(configuration != null)
      {
         String ceiling = (String) configuration.get(CALLBACK_MEM_CEILING);
         if(ceiling != null)
         {
            try
            {
               double newCeiling = Double.parseDouble(ceiling);
               setMemPercentCeiling(new Double(newCeiling));
            }
            catch(NumberFormatException e)
            {
               log.warn("Found new store memory ceiling seting (" + ceiling + "), but can not convert to type double.", e);
            }
         }
      }
   }

   public Client getCallbackClient()
   {
      return callBackClient;
   }


   /**
    * Returns an id that can be used to identify this particular
    * callback handler, which should be representative of the
    * client invoker it will make callbacks to.  Currently, this
    * is the session id associated with the invocation request.
    *
    * @return
    */
   public static String getId(InvocationRequest invocation)
   {
      String sessionId = invocation.getSessionId();
      return sessionId;
   }

   /**
    * Returns an id that can be used to identify this particular
    * callback handler, which should be representative of the
    * client invoker it will make callbacks to.
    *
    * @return
    */
   public String getId()
   {
      return getId(invocation);
   }

   public List getCallbacks()
   {
      List callbackList = null;
      synchronized(callbacks)
      {
         callbackList = (List) callbacks.clone();
         callbacks.clear();
      }

      // get as many persisted callbacks as possible without over run on memory
      List persistedCallbacks = null;
      try
      {
         persistedCallbacks = getPersistedCallbacks();
      }
      catch(IOException e)
      {
         log.error("Can not get persisted callbacks.", e);
         throw new RuntimeException("Error getting callbacks", e);
      }
      callbackList.addAll(persistedCallbacks);

      return callbackList;
   }

   private List getPersistedCallbacks() throws IOException
   {
      List callbacks = new ArrayList();

      int size = callbackStore.size();
      for(int x = 0; x < size; x++)
      {
         callbacks.add(callbackStore.getNext());
         // check the amount of mem in use as get callbacks out so
         // don't load so many callbacks from store, that run out of memory.
         if(isMemLow())
         {
            new Thread()
            {
               public void run()
               {
                  System.gc();
               }
            }.start();
            break;
         }
      }

      return callbacks;
   }

   public boolean isPullCallbackHandler()
   {
      return (callBackClient == null);
   }

   /**
    * Will take the callback message and send back to client.
    * If client locator is null, will store them till client polls to get them.
    *
    * @param callback
    * @throws HandleCallbackException
    */
   public void handleCallback(InvocationRequest callback)
         throws HandleCallbackException
   {
      try
      {
         if(callBackClient == null)
         {
            // need to check if shoudl persist callback instead of keeping in memory
            if(shouldPersist())
            {
               try
               {
                  persistCallback(callback);
                  callback = null;
                  // try to help out with the amount of memory usuage
                  new Thread()
                  {
                     public void run()
                     {
                        System.gc();
                     }
                  }.start();
               }
               catch(IOException e)
               {
                  log.error("Unable to persist callback.", e);
                  throw new HandleCallbackException("Unable to persist callback and will not be able to deliver.", e);
               }
            }
            else
            {
               synchronized(callbacks)
               {
                  if(log.isDebugEnabled())
                  {
                     log.debug("pull callback.  adding to callback list");
                  }
                  callbacks.add(callback);
               }
            }
         }
         else
         {
            try
            {
               // make sure connected
               if(!callBackClient.isConnected())
               {
                  callBackClient.connect();
               }
               if(callBackClient.isConnected())
               {
                  if(log.isDebugEnabled())
                  {
                     log.debug("push callback.  Calling client now.");
                  }
                  if(callback != null)
                  {
                     Map returnPayload = callback.getReturnPayload();
                     if(returnPayload == null)
                     {
                        returnPayload = new HashMap();
                     }
                     returnPayload.put(Callback.SERVER_LOCATOR_KEY, serverLocator);
                     callback.setReturnPayload(returnPayload);
                  }
                  // sending internal invocation so server invoker we are sending to
                  // will know how pass onto it's client callback handler
                  InternalInvocation internalInvocation = new InternalInvocation(InternalInvocation.HANDLECALLBACK,
                                                                                 new Object[]{callback});
                  callBackClient.setSessionId(sessionId);
                  callBackClient.invoke(internalInvocation,
                                        callback.getRequestPayload());
               }
               else
               {
                  log.error("Can not handle callback since can not connect to client invoker.");
                  throw new HandleCallbackException("Can not handle callback since can not connect to client invoker.");
               }
            }
            catch(Throwable ex)
            {
               log.debug("Error dispatching callback to handler.", ex);
               throw new HandleCallbackException("Error dispatching callback to handler.", ex);
            }
         }
      }
      catch(Throwable thr)
      {
         log.error("Error handling callback.", thr);
         throw new HandleCallbackException("Error handling callback.", thr);
      }
   }

   private void persistCallback(InvocationRequest callback) throws IOException
   {
      callbackStore.add(callback);
   }

   /**
    * Calculates the percentage amount of free memory compared to max memory.  The calculations for this
    * is not always acurate.  The reason is that total memory used is usually less than the max allowed.  Thus,
    * the amount of free memory is relative to the total amount allocated at that point in time.  It is not
    * until the total amount of memory allocated is equal to the max it will be allowed to allocate.  At this point,
    * the amount of free memory becomes relavent.  Therefore, if the memory percentage ceiling is high, it might
    * not trigger until after free memory percentage is well below the ceiling.
    *
    * @return
    */
   private boolean shouldPersist()
   {
      return isMemLow();
   }

   private boolean isMemLow()
   {
      Runtime runtime = Runtime.getRuntime();
      long max = runtime.maxMemory();
      long total = runtime.totalMemory();
      long free = runtime.freeMemory();
      float percentage = 100 * free / total;
      if(max == total && memPercentCeiling >= percentage)
      {
         return true;
      }
      else
      {
         return false;
      }
   }

   /**
    * Returns the id for this handler
    *
    * @return
    */
   public String toString()
   {
      return getClass().getName() + " - id: " + getId();
   }

   /**
    * This method is required to be called upon removing a callback listener
    * so can clean up resources used by the handler.  In particular, should
    * call disconnect on internal Client.
    */
   public void destroy()
   {
      if(callBackClient != null)
      {
         callBackClient.disconnect();
         callBackClient = null;
      }
      if(callbackStore != null)
      {
         callbackStore.purgeFiles();
      }
   }
}