remotedispatcher.java

提供ESB 应用mule源代码 提供ESB 应用mule源代码

/*
 * $Id: RemoteDispatcher.java 11249 2008-03-07 14:59:58Z tcarlson $
 * --------------------------------------------------------------------------------------
 * Copyright (c) MuleSource, Inc.  All rights reserved.  http://www.mulesource.com
 *
 * The software in this package is published under the terms of the CPAL v1.0
 * license, a copy of which has been included with this distribution in the
 * LICENSE.txt file.
 */

package org.mule.module.client;

import org.mule.DefaultMuleEvent;
import org.mule.DefaultMuleMessage;
import org.mule.DefaultMuleSession;
import org.mule.MuleServer;
import org.mule.MuleSessionHandler;
import org.mule.NullSessionHandler;
import org.mule.RegistryContext;
import org.mule.RequestContext;
import org.mule.api.FutureMessageResult;
import org.mule.api.MuleEvent;
import org.mule.api.MuleException;
import org.mule.api.MuleMessage;
import org.mule.api.MuleSession;
import org.mule.api.config.MuleProperties;
import org.mule.api.endpoint.EndpointBuilder;
import org.mule.api.endpoint.EndpointFactory;
import org.mule.api.endpoint.ImmutableEndpoint;
import org.mule.api.endpoint.OutboundEndpoint;
import org.mule.api.lifecycle.Disposable;
import org.mule.api.security.Credentials;
import org.mule.api.transformer.wire.WireFormat;
import org.mule.api.transport.DispatchException;
import org.mule.module.client.i18n.ClientMessages;
import org.mule.module.client.remoting.RemoteDispatcherException;
import org.mule.module.client.remoting.ServerHandshake;
import org.mule.module.client.remoting.UnsupportedWireFormatException;
import org.mule.module.client.remoting.notification.RemoteDispatcherNotification;
import org.mule.security.MuleCredentials;
import org.mule.transformer.TransformerUtils;
import org.mule.transport.AbstractConnector;
import org.mule.util.ClassUtils;
import org.mule.util.IOUtils;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.util.Map;

import edu.emory.mathcs.backport.java.util.concurrent.Callable;
import edu.emory.mathcs.backport.java.util.concurrent.Executor;

import org.apache.commons.lang.SerializationUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * <code>RemoteDispatcher</code> is used to make and receive requests to a remote
 * Mule instance. It is used to proxy requests to Mule using the Server URL as the
 * transport channel.
 */

public class RemoteDispatcher implements Disposable
{

    /**
     * logger used by this class
     */
    protected static final Log logger = LogFactory.getLog(RemoteDispatcher.class);

    /**
     * dispatch destination
     */
    private OutboundEndpoint asyncServerEndpoint;
    private OutboundEndpoint syncServerEndpoint;
    private Credentials credentials = null;

    /**
     * an ExecutorService for async messages (optional)
     */
    private Executor asyncExecutor;

    /**
     * calls made to a remote server are serialised using a wireformat
     */
    private WireFormat wireFormat;

    protected RemoteDispatcher(String endpoint, Credentials credentials) throws MuleException
    {
        this(endpoint);
        this.credentials = credentials;
    }

    protected RemoteDispatcher(String endpoint) throws MuleException
    {
        EndpointFactory endpointFactory = RegistryContext.getRegistry().lookupEndpointFactory();
        asyncServerEndpoint = endpointFactory.getOutboundEndpoint(endpoint);

        EndpointBuilder endpointBuilder = endpointFactory.getEndpointBuilder(endpoint);
        endpointBuilder.setRemoteSync(true);
        syncServerEndpoint = RegistryContext.getRegistry().lookupEndpointFactory().getOutboundEndpoint(
            endpointBuilder);

        wireFormat = requestWireFormat();
    }

    protected WireFormat requestWireFormat() throws MuleException
    {
        MuleMessage msg = new DefaultMuleMessage(ServerHandshake.SERVER_HANDSHAKE_PROPERTY);
        MuleMessage result = syncServerEndpoint.send(new DefaultMuleEvent(msg, syncServerEndpoint,
                new DefaultMuleSession(msg, new NullSessionHandler(), MuleServer.getMuleContext()), true));

        if(result==null)
        {
            throw new RemoteDispatcherException(ClientMessages.failedToDispatchActionNoResponseFromServer("request wire format", 5000));
        }

        ServerHandshake handshake;
        try
        {
            ByteArrayInputStream in = new ByteArrayInputStream(result.getPayloadAsBytes());
            handshake = (ServerHandshake) SerializationUtils.deserialize(in);
        }
        catch (Exception e)
        {
            throw new RemoteDispatcherException(ClientMessages.failedToDeserializeHandshakeFromServer(), e);
        }

        try
        {
            WireFormat wf = (WireFormat)ClassUtils.instanciateClass(handshake.getWireFormatClass(),
                    ClassUtils.NO_ARGS, getClass());

            return wf;
        }
        catch (Exception e)
        {
            throw new UnsupportedWireFormatException(handshake.getWireFormatClass(), e);
        }
    }

    protected void setExecutor(Executor e)
    {
        this.asyncExecutor = e;
    }

    /**
     * Dispatcher an event asynchronously to a components on a remote Mule instance.
     * Users can endpoint a url to a remote Mule server in the constructor of a Mule
     * client, by default the default Mule server url tcp://localhost:60504 is used.
     *
     * @param component the name of the Mule components to dispatch to
     * @param payload the object that is the payload of the event
     * @param messageProperties any properties to be associated with the payload. as
     *            null
     * @throws org.mule.api.MuleException if the dispatch fails or the components or
     *             transfromers cannot be found
     */
    public void dispatchToRemoteComponent(String component, Object payload, Map messageProperties)
        throws MuleException
    {
        doToRemoteComponent(component, payload, messageProperties, false);
    }

    /**
     * sends an event synchronously to a components on a remote Mule instance. Users
     * can endpoint a url to a remote Mule server in the constructor of a Mule
     * client, by default the default Mule server url tcp://localhost:60504 is used.
     *
     * @param component the name of the Mule components to send to
     * @param payload the object that is the payload of the event
     * @param messageProperties any properties to be associated with the payload. as
     *            null
     * @return the result message if any of the invocation
     * @throws org.mule.api.MuleException if the dispatch fails or the components or
     *             transfromers cannot be found
     */
    public MuleMessage sendToRemoteComponent(String component, Object payload, Map messageProperties)
        throws MuleException
    {
        return doToRemoteComponent(component, payload, messageProperties, true);
    }

    /**
     * sends an event to a components on a remote Mule instance, while making the
     * result of the event trigger available as a Future result that can be accessed
     * later by client code. Users can endpoint a url to a remote Mule server in the
     * constructor of a Mule client, by default the default Mule server url
     * tcp://localhost:60504 is used.
     *
     * @param component the name of the Mule components to send to
     * @param transformers a comma separated list of transformers to apply to the
     *            result message
     * @param payload the object that is the payload of the event
     * @param messageProperties any properties to be associated with the payload. as
     *            null
     * @return the result message if any of the invocation
     * @throws org.mule.api.MuleException if the dispatch fails or the components or
     *             transfromers cannot be found
     */
    public FutureMessageResult sendAsyncToRemoteComponent(final String component,
                                                          String transformers,
                                                          final Object payload,
                                                          final Map messageProperties) throws MuleException
    {
        Callable callable = new Callable()
        {
            public Object call() throws Exception
            {
                return doToRemoteComponent(component, payload, messageProperties, true);
            }
        };

        FutureMessageResult result = new FutureMessageResult(callable);

        if (asyncExecutor != null)
        {
            result.setExecutor(asyncExecutor);
        }

        if (transformers != null)
        {
            result.setTransformers(TransformerUtils.getTransformers(transformers));
        }

        result.execute();
        return result;
    }