/*
* JBoss, Home of Professional Open Source
* Copyright 2005, JBoss Inc., and individual contributors as indicated
* by the @authors tag. See the copyright.txt in the distribution for a
* full listing of individual contributors.
*
* This is free software; you can redistribute it and/or modify it
* under the terms of the GNU Lesser General Public License as
* published by the Free Software Foundation; either version 2.1 of
* the License, or (at your option) any later version.
*
* This software is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this software; if not, write to the Free
* Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
* 02110-1301 USA, or see the FSF site: http://www.fsf.org.
*/
package org.jboss.remoting.callback;

import org.jboss.logging.Logger;
import org.jboss.remoting.Client;
import org.jboss.remoting.ConnectionListener;
import org.jboss.remoting.InvocationRequest;
import org.jboss.remoting.InvokerLocator;
import org.jboss.remoting.Remoting;
import org.jboss.remoting.SerializableStore;
import org.jboss.remoting.ServerInvoker;
import org.jboss.remoting.invocation.InternalInvocation;
import org.jboss.remoting.security.SSLServerSocketFactoryServiceMBean;
import org.jboss.remoting.security.SSLSocketBuilder;
import org.jboss.remoting.security.SSLSocketBuilderMBean;
import org.jboss.remoting.security.SSLSocketFactoryService;

import javax.management.MBeanServer;
import javax.management.MBeanServerInvocationHandler;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import javax.net.SocketFactory;
import javax.net.ssl.SSLServerSocketFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;


/**
 * Responsible for all callbacks in remoting at invoker level (on the server side).
 * This is created within the ServerInvoker and passed to the server handler as a
 * proxy for the client's callback handler.
 * <p/>
 * Will determin internally if is using pull or push mechanism for delivering callbacks.
 * If is push, will create a Client to call back on the callback server.
 *
 * @author <a href="mailto:telrod@e2technologies.net">Tom Elrod</a>
 */
public class ServerInvokerCallbackHandler
implements AsynchInvokerCallbackHandler, ConnectionListener
{

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

   private static boolean trace = log.isTraceEnabled();

   private InvocationRequest invocation;
   private Client callBackClient;
   private ArrayList callbacks = new ArrayList();
   private String sessionId;
   private String listenerId;
   private String clientSessionId;
   private InvokerLocator serverLocator;
   private int blockingTimeout = ServerInvoker.DEFAULT_BLOCKING_TIMEOUT;
   

   private SerializableStore callbackStore = null;
   private CallbackErrorHandler callbackErrorHandler = 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 any callback error handler that
    * should be used.  This key should be used when setting up
    * config in the invoker.
    */
   public static final String CALLBACK_ERROR_HANDLER_KEY = "callbackErrorHandler";

   /**
    * 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 key used for storing a CallbackListener in the return metadata map of a Callback.
    */
   public static final String CALLBACK_LISTENER = "callbackListener";

   /*
    * The key used to indicate if callback acknowledgement should be handled by Remoting
    * or the application.  If it is present in the callback's return payload and
    * set to true, Remoting will handle the acknowledgement for push callbacks.
    */
   public static final String REMOTING_ACKNOWLEDGES_PUSH_CALLBACKS = "remotingAcknowledgesPushCallbacks";

   /**
    * This key used to identify a Callback to be acknowledged.
    */
   public static final String CALLBACK_ID = "callbackId";
   
   /** This key is used to identify the timeout used by a callback client */
   public static final String CALLBACK_TIMEOUT = "callbackTimeout";
   
   /** The key used to pass to the callback client a reference to ServerInvoker */
   public static final String SERVER_INVOKER = "serverInvoker";
   
   /** The key used to pass to the callback client a reference this ServerInvokerCallbackHandler */
   public static final String SERVER_INVOKER_CALLBACK_HANDLER = "serverInvokerCallbackHandler";

   /**
    * 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

   /**
    * Maps an ID to a CallbackListener for a Callback waiting to be acknowledged.
    */
   private Map idToListenerMap = Collections.synchronizedMap(new HashMap());


   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);
   }
   
   public void connect() throws Exception
   {
      if (callBackClient != null)
      {
         if (callBackClient.isConnected())
            return;
         callBackClient.connect();
      }
         
   }

   private void init(InvocationRequest invocation, ServerInvoker owner) throws Exception
   {
      clientSessionId = invocation.getSessionId();
      sessionId = invocation.getSessionId();
      
      Map metadata = null;
      if (owner.getConfiguration() == null)
      {
         metadata = new HashMap();
      }
      else
      {
         metadata = new HashMap(owner.getConfiguration());
      }
      if(invocation.getRequestPayload() != null)
      {
         metadata.putAll(invocation.getRequestPayload());
      }

      listenerId = (String) metadata.get(Client.LISTENER_ID_KEY);
      if(listenerId != null)
      {
         sessionId = sessionId + "+" + listenerId;
      }
      log.debug("Session id for callback handler is " + sessionId);

      if(invocation.getLocator() != null)
      {
         Object val = metadata.get(CALLBACK_TIMEOUT);
         if (val instanceof String)
         {
            try
            {
               Integer.parseInt((String) val);
               metadata.put(ServerInvoker.TIMEOUT, val);
               log.debug(this + " using callbackTimeout value " + val);
            }
            catch (NumberFormatException e)
            {
               log.warn("callbackTimeout value must have valid numeric format: " + val);
            }
         }
         else if (val != null)
         {
            log.warn("callbackTimeout value must be a String: " + val);
         }
         
         // need to configure callback client with ssl config if one exists for server
         configureSocketFactory(metadata, owner);
         
         metadata.put(SERVER_INVOKER, owner);
         metadata.put(SERVER_INVOKER_CALLBACK_HANDLER, this);

         callBackClient = new Client(invocation.getLocator(), invocation.getSubsystem(), metadata);
         createCallbackErrorHandler(owner, invocation.getSubsystem());
      }
      else
      {
         createCallbackStore(owner, sessionId);
      }
      
      Object val = metadata.get(ServerInvoker.BLOCKING_TIMEOUT);
      if (val != null)
      {
         if (val instanceof String)
         {
            try
            {
               blockingTimeout = Integer.parseInt((String) val);
            }
            catch (NumberFormatException e)
            {
               log.warn("Error converting " + ServerInvoker.BLOCKING_TIMEOUT + " to type long.  " + e.getMessage());
            }
         }
         else
         {
            log.warn("Value for " + ServerInvoker.BLOCKING_TIMEOUT + " configuration must be of type " + String.class.getName() +
                     " and is " + val.getClass().getName());
         }
      }
   }

   /**
    * Will check to see if the server invoker associated with this callback client
    * needs to have associated ssl config
    * @param clientConfig
    * @param serverInvoker
    */
   private void configureSocketFactory(Map clientConfig, ServerInvoker serverInvoker) throws Exception
   {
      // If a SocketFactory already exists, then all we have to do is tell the
      // client invoker to use it.
      if (serverInvoker.getSocketFactory() != null)
      {
         clientConfig.put(Remoting.CUSTOM_SOCKET_FACTORY, serverInvoker.getSocketFactory());
         return;
      }

      if (clientConfig == null)
         clientConfig = new HashMap();

      // If a constructed SocketFactory was passed in through config map, the client invoker
      // will use it.  Also, we store it in server invoker for future use.
      if (clientConfig.containsKey(Remoting.CUSTOM_SOCKET_FACTORY))
      {
         serverInvoker.setSocketFactory((SocketFactory) clientConfig.get(Remoting.CUSTOM_SOCKET_FACTORY));
         return;
      }

      // If a SocketFactory has not been created already, we need to make sure that the client
      // invoker creates a suitable SocketFactory.

      // First, see if we can convert server socket factory into socket factory.
      String serverSocketFactoryString = (String) clientConfig.get(ServerInvoker.SERVER_SOCKET_FACTORY);
      if (serverSocketFactoryString != null && serverSocketFactoryString.length() > 0)
      {
         MBeanServer server = serverInvoker.getMBeanServer();
         try
         {
            ObjectName serverSocketFactoryObjName = new ObjectName(serverSocketFactoryString);
            if (server != null)
            {
               boolean isCorrectType = server.isInstanceOf(serverSocketFactoryObjName, SSLServerSocketFactoryServiceMBean.class.getName());
               if (isCorrectType)
               {
                  SSLSocketBuilderMBean sslSocketBuilder = (SSLSocketBuilderMBean)server.
                  getAttribute(serverSocketFactoryObjName, "SSLSocketBuilder");
                  if (sslSocketBuilder != null)
                  {
                     SSLSocketBuilder clonedSSLSocketBuilder = (SSLSocketBuilder) sslSocketBuilder.clone();
                     boolean shouldUseDefault = sslSocketBuilder.getUseSSLServerSocketFactory();
                     clonedSSLSocketBuilder.setUseSSLSocketFactory(shouldUseDefault);
                     boolean useClientMode = sslSocketBuilder.isServerSocketUseClientMode();
                     clonedSSLSocketBuilder.setSocketUseClientMode(useClientMode);
                     SSLSocketFactoryService sslSocketFactoryService = new SSLSocketFactoryService();
                     sslSocketFactoryService.setSSLSocketBuilder(clonedSSLSocketBuilder);
                     sslSocketFactoryService.start();
                     clientConfig.put(Remoting.CUSTOM_SOCKET_FACTORY, sslSocketFactoryService);
                     clientConfig.put(SSLSocketBuilder.REMOTING_SOCKET_USE_CLIENT_MODE, "false");
                     // JBREM-536
                     clientConfig.put("hostnameVerifier", "org.jboss.test.remoting.transport.http.ssl.config.SelfIdentifyingHostnameVerifier");
                     serverInvoker.setSocketFactory(sslSocketFactoryService);
                     return;
                  }
               }
            }
         }
         catch (MalformedObjectNameException ignored)
         {
         }
      }

      // Otherwise, if we need an SSLSocketFactory, the client invoker will create it from
      // configuration parameters.  Tell it to use client mode (unless explictly configured
      // otherwise).
      if (serverInvoker.getServerSocketFactory() instanceof SSLServerSocketFactory)
      {
         if (!clientConfig.containsKey(SSLSocketBuilder.REMOTING_SOCKET_USE_CLIENT_MODE))
            clientConfig.put(SSLSocketBuilder.REMOTING_SOCKET_USE_CLIENT_MODE, "false");
      }
   }

   public String getCallbackSessionId()
   {
      return sessionId;
   }

   public String getClientSessionId()
   {
      return clientSessionId;
   }

   public String getSubsystem()
   {
	   return invocation.getSubsystem();
   }

   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(trace) { 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(trace) { 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 createCallbackErrorHandler(ServerInvoker owner, String subsystem) throws Exception
   {
      Map config = owner.getConfiguration();
      if(config != null)
      {
         // should either be a fully qualified class name or a mbean object name
         String errorHandlerName = (String) config.get(CALLBACK_ERROR_HANDLER_KEY);
         if(errorHandlerName != null)
         {
            // will first try as a MBean
            try
            {
               MBeanServer server = owner.getMBeanServer();
               ObjectName errorHandlerObjectName = new ObjectName(errorHandlerName);
               if(server != null)
               {
                  callbackErrorHandler = (CallbackErrorHandler)
                        MBeanServerInvocationHandler.newProxyInstance(server,
                                                                      errorHandlerObjectName,
                                                                      CallbackErrorHandler.class,
                                                                      false);
               }
            }
            catch(Exception ex)
            {
               log.debug("Could not create callback error handler from the configration value " +
                         "given (" + errorHandlerName + ") as an MBean.");
               if(trace) { log.trace("Error is: " + ex.getMessage(), ex); }
               callbackErrorHandler = null;
            }

            // now try by class name
            if(callbackErrorHandler == null)
            {
               try
               {
                  Class errorHandlerClass = Class.forName(errorHandlerName);
                  callbackErrorHandler = (CallbackErrorHandler) errorHandlerClass.newInstance();
               }
               catch(Exception e)
               {
                  log.debug("Could not create callback error handler from the configuration value " +
                            "given (" + errorHandlerName + ") as a fully qualified class name.");
                  if(trace) { log.trace("Error is: " + e.getMessage(), e); }
               }
            }
         }
      }

      // if still null, then just use default
      if(callbackErrorHandler == null)
      {
         callbackErrorHandler = new DefaultCallbackErrorHandler();
      }

      // set configuration for the error handler.
      Map errorHandlerConfig = new HashMap();
      errorHandlerConfig.putAll(owner.getConfiguration());
      errorHandlerConfig.put(CallbackErrorHandler.HANDLER_SUBSYSTEM, subsystem);
      callbackErrorHandler.setConfig(errorHandlerConfig);
      callbackErrorHandler.setServerInvoker(owner);
      callbackErrorHandler.setCallbackHandler(this);

   }

   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.
    */
   public static String getId(InvocationRequest invocation)
   {
      String sessionId = invocation.getSessionId();
      Map metadata = invocation.getRequestPayload();
      if(metadata != null)
      {
         String listenerId = (String) metadata.get(Client.LISTENER_ID_KEY);
         if(listenerId != null)
         {
            sessionId = sessionId + "+" + listenerId;
         }
      }
      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.
    */
   public String getId()
   {
      return sessionId;
   }

   public List getCallbacks(Map metadata)
   {
      log.trace("entering getCallbacks()");
      
      boolean blocking = false;
      int currentBlockingTimeout = blockingTimeout;
      
      if (metadata != null)
      {
         Object val = metadata.get(ServerInvoker.BLOCKING_MODE);
         if (ServerInvoker.BLOCKING.equals(val))
            blocking = true;
         
         val = metadata.get(ServerInvoker.BLOCKING_TIMEOUT);
         if (val != null)
         {
            if (val instanceof String)
            {
               try
               {
                  currentBlockingTimeout = Integer.parseInt((String) val);
               }
               catch (NumberFormatException e)
               {
                  log.warn("Error converting " + ServerInvoker.BLOCKING_TIMEOUT + " to type long.  " + e.getMessage());
               }
            }
            else
            {
               log.warn("Value for " + ServerInvoker.BLOCKING_TIMEOUT + " configuration must be of type " + String.class.getName() +
                        " and is " + val.getClass().getName());
            }
         }
      }

      if (trace)
      {
         log.trace("block: " + blocking);
         log.trace("blocking timeout: " + currentBlockingTimeout);
      }

      synchronized (callbacks)
      {
         List callbackList = constructCallbackList();
         if (blocking && callbackList.isEmpty())
         {
            try
            {
               callbacks.wait(currentBlockingTimeout);
               callbackList = constructCallbackList();
            }
            catch (InterruptedException e)
            {
               log.debug("unexpected interrupt");
            }
         }
         
         if (trace) log.trace("callbackList.size(): " + callbackList.size());
         return callbackList;
      }
   }
   
   private List constructCallbackList()
   {
      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(Callback callback) throws HandleCallbackException
   {
      handleCallback(callback, false, false);
   }

   /**
    * For push callbacks, will send the callback to the server invoker on the
    * client side, hand off processing to a separate thread, and return.<p>
    *
    * For pull callbacks, behaves the same as handleCallback(Callback callback).<p>
    *
    * @param callback
    * @throws HandleCallbackException
    */
   public void handleCallbackOneway(Callback callback) throws HandleCallbackException
   {
      handleCallback(callback, true, false);
   }

   /**
    * For push callbacks:<p>
    *   if serverSide == false, will send the callback to the server invoker on
    *   the client side, hand off processing to a separate thread, and return.<p>
    *
    *   if serverside == true, will hand off to a separate thread the sending
    *   of the callback and will then return.<p>
    *
    * For pull callbacks, behaves the same as handleCallback(Callback callback).<p>
    *
    * @param callback
    * @param serverSide
    * @throws HandleCallbackException
    */
   public void handleCallbackOneway(Callback callback, boolean serverSide)
      throws HandleCallbackException
   {
      handleCallback(callback, true, serverSide);
   }

   /**
    * For push callbacks:<p>
    *
    *   if asynch == false, behaves the same as handleCallback(Callback callback).<p>
    *
    *   if asynch == true:<p>
    *
    *     if serverSide == false, will send the callback to the server invoker on the client side,
    *     hand off processing to a separate thread, and return.<p>
    *
    *     if serverside == true, will hand off to a separate thread the sending of the callback and
    *     will then return.<p>
    *
    * For pull callbacks, behaves the same as handleCallback(Callback callback).<p>
    */
   public void handleCallback(Callback callback, boolean asynch, boolean serverSide)
      throws HandleCallbackException
   {
      try
      {
         Object callbackId = checkForCallbackListener(callback);

         if(callBackClient == null)
         {
            // need to check if should persist callback instead of keeping in memory
            if(shouldPersist())
            {
               try
               {
                  synchronized (callbacks)
                  {
                     persistCallback(callback);
                     callbacks.notify();
                  }
                  
                  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(trace){ log.debug(this + " got PULL callback. Adding to callback list ..."); }
                  callbacks.add(callback);
                  callbacks.notify();
               }
            }
         }
         else
         {
            // non null callback client

            try
            {
               if(trace){ log.debug(this + " got PUSH callback " + callback); }

               boolean handleAcknowledgement = false;

               if(callback != null)
               {
                  Map returnPayload = callback.getReturnPayload();

                  if(returnPayload == null)
                  {
                     returnPayload = new HashMap();
                  }
                  else
                  {
                     Object o = returnPayload.remove(REMOTING_ACKNOWLEDGES_PUSH_CALLBACKS);
                     if (!asynch)
                     {
                        if (o instanceof String  && Boolean.valueOf((String)o).booleanValue() ||
                            o instanceof Boolean && ((Boolean)o).booleanValue())
                           handleAcknowledgement = true;
                     }
                  }

                  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);

               if (asynch)
               {
                  if(trace){ log.debug(this + " sending ASYNCHRONOUSLY the callback to the client"); }
                  callBackClient.
                     invokeOneway(internalInvocation, callback.getRequestPayload(), serverSide);
               }
               else
               {
                  if(trace){ log.debug(this + " sending SYNCHRONOUSLY the callback to the client"); }
                  callBackClient.invoke(internalInvocation, callback.getRequestPayload());
               }

               handlePushCallbackAcknowledgement(callbackId, handleAcknowledgement);
            }
            catch(Throwable ex)
            {
               if(callbackErrorHandler == null)
               {
                  // no callback handler, rethrowing the exception
                  throw ex;
               }

               if (trace) { log.trace(this + " handing the error over to " + callbackErrorHandler); }

               // a well behaved callback error handler will perform clean up and then rethrow the
               // exception so the client application has a chance to find out about the error
               // condition
               callbackErrorHandler.handleError(ex);
            }
         }
      }
      catch(Throwable t)
      {
         log.error("Error handling callback", t);
         throw new HandleCallbackException("Error handling callback", t);
      }
   }

   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.
    */
   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;
      }
   }

   Object checkForCallbackListener(Callback callback)
   {
      Map returnPayload = callback.getReturnPayload();
      if (returnPayload == null)
         return null;

      Object listenerObject = returnPayload.remove(CALLBACK_LISTENER);
      if (listenerObject == null)
         return null;

      Object callbackId = returnPayload.get(CALLBACK_ID);
      if (callbackId == null)
      {
         log.error("CALLBACK_ID is null: unable to acknowledge callback");
         return null;
      }

      if (listenerObject instanceof CallbackListener)
      {
         if(listenerId != null)
         {
            returnPayload.put(Client.LISTENER_ID_KEY, listenerId);
            idToListenerMap.put(callbackId, listenerObject);
            return callbackId;
         }
         else
         {
            log.error("LISTENER_ID_KEY is null: unable to acknowledge callback");
            return null;
         }
      }
      else
      {
         log.error("callback preprocess listener has wrong type: " + listenerObject);
         return null;
      }
   }

   private void handlePushCallbackAcknowledgement(Object callbackId, boolean handleAck)
   {
      if (!handleAck)
         return;

      if (callbackId == null)
      {
         log.error("Unable to acknowledge push callback: callback id is null");
         return;
      }

      CallbackListener listener = (CallbackListener) idToListenerMap.get(callbackId);
      if (listener == null)
      {
         log.error("Unable to acknowledge push callback: listener is null");
         return;
      }

      listener.acknowledgeCallback(this, callbackId, null);
   }

   /**
    * Calls listeners to acknowledge callbacks
    * @param invocation carries identities of Callbacks to acknowledge and,
    *                   optionally, responses
    */
   public void acknowledgeCallbacks(InternalInvocation invocation) throws Exception
   {
      Object[] params = invocation.getParameters();
      if (params == null)
         return;

      List callbackIds = (List) params[0];
      List responses = (List) params[1];
      if (callbackIds == null || callbackIds.size() == 0)
         return;

      Iterator idsIterator = callbackIds.iterator();
      Iterator responseIterator = null;
      if (responses != null)
         responseIterator = responses.iterator();

      Object callbackId = null;
      Object response = null;
      while(idsIterator.hasNext())
      {
         callbackId = idsIterator.next();
         if (responseIterator != null)
            response = responseIterator.next();

         CallbackListener listener = (CallbackListener) idToListenerMap.remove(callbackId);

         if (listener == null)
         {
            log.warn("Cannot acknowledge callback: unrecognized id: " + callbackId);
            continue;
         }

         listener.acknowledgeCallback(this, callbackId, response);
      }
   }

   public String toString()
   {
      return "ServerInvokerCallbackHandler[" + 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 synchronized void destroy()
   {
      if(callBackClient != null)
      {
         callBackClient.disconnect();
         callBackClient = null;
      }

      if(callbackStore != null)
      {
         callbackStore.purgeFiles();
         callbackStore = null;
      }
   }

   public void handleConnectionException(Throwable throwable, Client client)
   {
      if (clientSessionId.equals(client.getSessionId()))
      {
         destroy();
         log.debug(this + " shut down");
      }
   }
}
