/*
 * ################################################################
 *
 * ProActive Parallel Suite(TM): The Java(TM) library for
 *    Parallel, Distributed, Multi-Core Computing for
 *    Enterprise Grids & Clouds
 *
 * Copyright (C) 1997-2011 INRIA/University of
 *                 Nice-Sophia Antipolis/ActiveEon
 * Contact: proactive@ow2.org or contact@activeeon.com
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Affero General Public License
 * as published by the Free Software Foundation; version 3 of
 * the License.
 *
 * This library 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
 * Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA
 *
 * If needed, contact us to obtain a release under GPL Version 2 or 3
 * or a different license than the AGPL.
 *
 *  Initial developer(s):               The ActiveEon Team
 *                        http://www.activeeon.com/
 *  Contributor(s):
 *
 * ################################################################
 * $$ACTIVEEON_INITIAL_DEV$$
 */
package org.objectweb.proactive.extensions.pamr.client;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.apache.log4j.Logger;
import org.objectweb.proactive.core.body.future.MethodCallResult;
import org.objectweb.proactive.core.body.request.Request;
import org.objectweb.proactive.core.remoteobject.SynchronousReplyImpl;
import org.objectweb.proactive.core.runtime.ProActiveRuntimeImpl;
import org.objectweb.proactive.core.util.converter.remote.ProActiveMarshaller;
import org.objectweb.proactive.core.util.log.ProActiveLogger;
import org.objectweb.proactive.extensions.pamr.PAMRConfig;
import org.objectweb.proactive.extensions.pamr.exceptions.PAMRException;
import org.objectweb.proactive.extensions.pamr.protocol.message.DataRequestMessage;
import org.objectweb.proactive.extensions.pamr.remoteobject.message.PAMRMessage;
import org.objectweb.proactive.utils.NamedThreadFactory;


/** Executes a ProActive {@link Request} received and send the response.
 *
 * @since ProActive 4.1.0
 */
public class ProActiveMessageHandler implements MessageHandler {

    public static final Logger logger = ProActiveLogger.getLogger(PAMRConfig.Loggers.PAMR_CLIENT);

    /** {@link Request} are handled by a threadpool */
    final private ExecutorService tpe;

    /** Local agent */
    private Agent agent;

    public ProActiveMessageHandler(Agent agent) {
        this.agent = agent;

        /* DO NOT USE A FIXED THREAD POOL
         * 
         * Each time a message arrives, it is handled by a task submitted to 
         * this executor service. Each task can a perform remote calls. If 
         * the number of workers is fixed it can lead to deadlock.
         * 
         * Reentrant calls is the most obvious case of deadlock. But the same 
         * issue can occur with remote calls. 
         */
        ThreadFactory tf = new NamedThreadFactory("ProActive PAMR message handler");
        tpe = Executors.newCachedThreadPool(tf);

    }

    public void pushMessage(DataRequestMessage message) {
        if (logger.isTraceEnabled()) {
            logger.trace("pushing message " + message + " into the executor queue");
        }

        ProActiveMessageProcessor pmp = new ProActiveMessageProcessor(message, agent);
        tpe.submit(pmp);
    }

    /** Process one ProActive {@link Request} */
    private class ProActiveMessageProcessor implements Runnable {
        /** the request*/
        private final DataRequestMessage _toProcess;
        /** the local agent*/
        private final Agent agent;
        /** serialization*/
        private final ProActiveMarshaller marshaller;

        public ProActiveMessageProcessor(DataRequestMessage msg, Agent agent) {
            this._toProcess = msg;
            this.agent = agent;
            // get the runtime URL
            // if the local Agent has received a DataRequestMessage,
            // means that a ProActiveRuntime exists on this machine
            String runtimeUrl = ProActiveRuntimeImpl.getProActiveRuntime().getURL();
            this.marshaller = new ProActiveMarshaller(runtimeUrl);
        }

        public void run() {
            ClassLoader savedClassLoader = Thread.currentThread().getContextClassLoader();
            try {

                // Handle the message
                Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());

                PAMRMessage message;
                try {
                    message = (PAMRMessage) this.marshaller.unmarshallObject(_toProcess.getData());
                } catch (Throwable t) {
                    new PAMRException("Failed to unmarshall incoming message", t);
                    SynchronousReplyImpl sr = new SynchronousReplyImpl(new MethodCallResult(null, t));
                    agent.sendReply(_toProcess, this.marshaller.marshallObject(sr));
                    return;
                }

                if (logger.isTraceEnabled()) {
                    logger.trace("Processing message: " + message);
                }
                Object result = message.processMessage(); // Cannot throw an exception

                byte[] resultBytes;
                try {
                    resultBytes = this.marshaller.marshallObject(result);
                } catch (Throwable t) {
                    new PAMRException("Failed to marshall the result bytes", t);
                    SynchronousReplyImpl sr = new SynchronousReplyImpl(new MethodCallResult(null, t));
                    agent.sendReply(_toProcess, this.marshaller.marshallObject(sr));
                    return;
                }

                try {
                    agent.sendReply(_toProcess, resultBytes);
                } catch (Throwable t) {
                    logger.info("Failed to send the PAMR reply to " + this._toProcess +
                        ". The router should discover the disconnection and unlock the caller", t);
                    return;
                }
            } catch (PAMRException e) {
                logger.info("Failed to send the PAMR error reply to " + this._toProcess +
                    ". The router should discover the disconnection and unlock the caller", e);
            } catch (IOException e) {
                logger.info("Failed to send the PAMR error reply to " + this._toProcess +
                    ". The router should discover the disconnection and unlock the caller", e);
            } finally {
                Thread.currentThread().setContextClassLoader(savedClassLoader);
            }
        }
    }
}

@@ -5096,32 +5096,50 @@
                 
+ PAMRException e =
  new PAMRExcepti
@@ -5179,17 +5179,98 @@
  message
-%22
+ on %22 +%0A                        this.agent.getAgentID() + %22for %22 + this._toProcess
 , t);%0A  
@@ -5357,33 +5357,33 @@
 allResult(null, 
-t
+e
 ));%0A            
@@ -5914,16 +5914,34 @@
         
+ PAMRException e =
  new PAM
@@ -5987,17 +5987,94 @@
 lt bytes
-%22
+ on %22 +%0A                        this.agent.getAgentID() + %22 for %22 + _toProcess
 , t);%0A  
@@ -6169,17 +6169,17 @@
 t(null, 
-t
+e
 ));%0A    

Fix typo in exception handling