socketdispatcher.java

一个工作流设计及定义的系统,可以直接与数据库结合进行系统工作流程的定义及应用.

/*
 * Copyright (c) 2005, John Mettraux, OpenWFE.org
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without 
 * modification, are permitted provided that the following conditions are met:
 * 
 * . Redistributions of source code must retain the above copyright notice, this
 *   list of conditions and the following disclaimer.  
 * 
 * . Redistributions in binary form must reproduce the above copyright notice, 
 *   this list of conditions and the following disclaimer in the documentation 
 *   and/or other materials provided with the distribution.
 * 
 * . Neither the name of the "OpenWFE" nor the names of its contributors may be
 *   used to endorse or promote products derived from this software without
 *   specific prior written permission.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
 * POSSIBILITY OF SUCH DAMAGE.
 *
 * $Id: SocketDispatcher.java,v 1.9 2005/07/28 16:33:09 jmettraux Exp $
 */

//
// SocketDispatcher.java
//
// [e-mail]
//
// generated with 
// jtmpl 1.0.04 20.11.2001 John Mettraux ([e-mail])
//

package openwfe.org.engine.impl.dispatch;

import openwfe.org.Utils;
import openwfe.org.MapUtils;
import openwfe.org.ReflectionUtils;
import openwfe.org.ServiceException;
import openwfe.org.ApplicationContext;
import openwfe.org.time.Time;
import openwfe.org.engine.listen.reply.WarningReply;
import openwfe.org.engine.listen.reply.ListenerReply;
import openwfe.org.engine.workitem.WorkItem;
import openwfe.org.engine.workitem.InFlowWorkItem;
import openwfe.org.engine.workitem.WorkItemCoder;
import openwfe.org.engine.workitem.CodingException;
import openwfe.org.engine.dispatch.DispatchingException;
import openwfe.org.engine.dispatch.FatalDispatchingException;
import openwfe.org.engine.dispatch.AbstractWorkItemDispatcher;


/**
 * It dispatches workitems through sockets
 *
 * <p><font size=2>CVS Info :
 * <br>$Author: jmettraux $
 * <br>$Date: 2005/07/28 16:33:09 $
 * <br>$Id: SocketDispatcher.java,v 1.9 2005/07/28 16:33:09 jmettraux Exp $ </font>
 *
 * @author [e-mail]
 */
public class SocketDispatcher

    extends AbstractWorkItemDispatcher

{

    private final static org.apache.log4j.Logger log = org.apache.log4j.Logger
        .getLogger(SocketDispatcher.class.getName());

    //
    // CONSTANTS (definitions)

    public final static String HOST
        = "host";

    public final static String PORT
        = "port";

    private final static int DEFAULT_PORT = 7007;

    public final static String SOCKET_TIMEOUT
        = "socketTimeout";

    public final static String SOCKET_FACTORY
        = "socketFactory";

    public final static String RETRY_COUNT
        = "retryCount";

    /*
    public final static String RETRY_FREQUENCY
        = "retryFrequency";
    */

    //
    // FIELDS

    private String targetHost = null;
    private int targetPort = -1;
    private int retryCount = 10;
    //protected long retryFrequency = 300000; // 5 minutes
    private int socketTimeout = 120000;
    private javax.net.SocketFactory socketFactory = null;

    //
    // CONSTRUCTORS

    public void init 
        (final String serviceName, 
         final ApplicationContext context, 
         final java.util.Map serviceParams)
    throws 
        ServiceException
    {
        super.init(serviceName, context, serviceParams);

        //
        // host
        
        this.targetHost = (String)serviceParams.get(HOST);

        if (this.targetHost == null) this.targetHost = "127.0.0.1";

        //
        // port
        
        this.targetPort = MapUtils.getAsInt(serviceParams, PORT, DEFAULT_PORT);
        
        //
        // retryCount
        
        this.retryCount = MapUtils.getAsInt(serviceParams, RETRY_COUNT, 10);
        
        /*
        //
        // retryFrequency
        
        try
        {
            this.retryFrequency =
                Long.parseLong((String)serviceParams.get(RETRY_FREQUENCY));
        }
        catch (Exception e)
        {
        }
        */

        //
        // socket timeout
        
        this.socketTimeout = 
            MapUtils.getAsInt(serviceParams, SOCKET_TIMEOUT, 120000);

        //
        // socketFactory
        
        String sFactoryClass = (String)serviceParams.get(SOCKET_FACTORY);
        try
        {
            /*
            Class factoryClass = Class.forName(sFactoryClass);

            java.lang.reflect.Method getDefault = factoryClass
                .getDeclaredMethod("getDefault", new Class[] {});

            this.socketFactory = (javax.net.SocketFactory)getDefault
                .invoke(null, new Object[] {});
            */

            this.socketFactory = 
                (javax.net.SocketFactory)ReflectionUtils.invokeStatic
                    (Class.forName(sFactoryClass),
                     "getDefault",
                     new Class[] {},
                     new Object[] {});
        }
        catch (final Exception e)
        {
            if (sFactoryClass != null)
            {
                log.warn
                    ("Failed to use custom SocketFactory '"+sFactoryClass+"'", 
                     e);
            }

            this.socketFactory = javax.net.SocketFactory.getDefault();

            log.debug
                ("Using default socket factory '"+
                 this.socketFactory.getClass().getName()+"'");
        }
    }

    //
    // METHODS

    /**
     * The bulk of the encoding work taken out of the dispatch method
     */
    protected byte[] encodeWorkitem 
        (final WorkItemCoder coder, final WorkItem wi)
    throws 
        DispatchingException
    {
        final long start = System.currentTimeMillis();

        try
        {
            final java.io.ByteArrayOutputStream baos = 
                new java.io.ByteArrayOutputStream();

            //
            // advertise encoder used

            final String ad = getWorkItemCoderName() + "\n";
            baos.write(ad.getBytes(Utils.getEncoding()));

            //log.debug("advertise() coderName >"+getWorkItemCoderName()+"<");

            //
            // set dispatch time

            if (wi instanceof InFlowWorkItem)
            {
                ((InFlowWorkItem)wi)
                    .setDispatchTime(Time.toIsoDate());
            }

            //
            // encode workitem

            byte[] encodedWorkItem = (byte[])coder
                .encode(wi, getContext(), getParams());

            baos.write(encodedWorkItem);
            baos.flush();

            log.debug
                ("encodeWorkitem() took "+(System.currentTimeMillis() - start));

            return baos.toByteArray();
        }
        catch (final Exception e)
        {
            throw new DispatchingException
                ("Failed to encode workitem", e);
        }
    }

    /**
     * Dispatches the workitem over a socket connection.
     */
    public Object dispatch (final WorkItem wi)
        throws DispatchingException
    {
        log.debug
            ("dispatch() to "+this.targetHost+":"+this.targetPort);

        final long start = System.currentTimeMillis();

        ListenerReply reply = null;

        final WorkItemCoder coder = instantiateEncoder();

        byte[] encodedWorkitem = encodeWorkitem(coder, wi);

        for (int i=0; i<this.retryCount; i++)
        {
            log.debug("dispatch() attempt #"+i);

            java.net.Socket socket = null;

            java.io.OutputStream os = null;
            java.io.InputStream is = null;

            try
            {
                // 
                // establish connection

                socket = this.socketFactory
                    .createSocket(this.targetHost, this.targetPort);

                socket.setSoTimeout(this.socketTimeout);
                socket.setTcpNoDelay(true);

                os = socket.getOutputStream();
                is = socket.getInputStream();

                os.write(encodedWorkitem);
                os.flush();

                reply = coder.getReplyCoder().decode(is);

                if (reply instanceof WarningReply)
                {
                    final WarningReply wr = (WarningReply)reply;

                    if (i == 0)
                        //
                        // showing these warnings only at first failure...
                    {
                        log.warn("dispatch() problem \""+wr.getMessage()+"\"");

                        if (wr.getCause() != null)
                        {
                            log.warn("dispatch() problem", wr.getCause());
                        }
                    }

                    throw new DispatchingException
                        (""+wr.getClass().getName()+"\n"+wr.getMessage());
                }

                break; // dispatching is successful
            }
            catch (final Exception e)
            {
                log.info
                    ("dispatch() try #"+i, e);

                if (i >= this.retryCount-1) 
                {
                    // flow will get frozen

                    throw new DispatchingException
                        ("last dispatch attempt failed", e);
                }

                Thread.currentThread().yield();
                    // yield priority to other threads...
            }
            finally
            {
                try
                {
                    os.flush();
                    socket.close();
                        // close connection
                }
                catch (final Throwable t)
                {
                    // ignore
                }
            }
        }

        log.debug("dispatch() took "+(System.currentTimeMillis() - start));

        //
        // handle reply

        return handleReply(reply);
            // this method from parent class will throw an exception
            // if the reply is fatal
    }

}