requestcorrelator.java

来自「JGRoups源码」· Java 代码 · 共 923 行 · 第 1/3 页

        if(log.isTraceEnabled()) {
            log.trace(new StringBuffer("calling (").append((request_handler != null? request_handler.getClass().getName() : "null")).
                      append(") with request ").append(hdr.id));
        }

        try {
            retval=request_handler.handle(req);
        }
        catch(Throwable t) {
            if(log.isErrorEnabled()) log.error("error invoking method", t);
            retval=t;
        }

        if(!hdr.rsp_expected) // asynchronous call, we don't need to send a response; terminate call here
            return;

        if(transport == null) {
            if(log.isErrorEnabled()) log.error("failure sending response; no transport available");
            return;
        }

        // changed (bela Feb 20 2004): catch exception and return exception
        try {  // retval could be an exception, or a real value
            rsp_buf=marshaller != null? marshaller.objectToByteBuffer(retval) : Util.objectToByteBuffer(retval);
        }
        catch(Throwable t) {
            try {  // this call should succeed (all exceptions are serializable)
                rsp_buf=marshaller != null? marshaller.objectToByteBuffer(t) : Util.objectToByteBuffer(t);
            }
            catch(Throwable tt) {
                if(log.isErrorEnabled()) log.error("failed sending rsp: return value (" + retval + ") is not serializable");
                return;
            }
        }

        rsp=req.makeReply();
        if(rsp_buf != null)
            rsp.setBuffer(rsp_buf);
        rsp_hdr=new Header(Header.RSP, hdr.id, false, name);
        rsp.putHeader(name, rsp_hdr);
        if(log.isTraceEnabled())
            log.trace(new StringBuffer("sending rsp for ").append(rsp_hdr.id).append(" to ").append(rsp.getDest()));

        try {
            if(transport instanceof Protocol)
                ((Protocol)transport).passDown(new Event(Event.MSG, rsp));
            else if(transport instanceof Transport)
                ((Transport)transport).send(rsp);
            else
                if(log.isErrorEnabled()) log.error("transport object has to be either a " +
                                                   "Transport or a Protocol, however it is a " + transport.getClass());
        }
        catch(Throwable e) {
            if(log.isErrorEnabled()) log.error("failed sending the response", e);
        }
    }


    // .......................................................................





    /**
     * Associates an ID with an <tt>RspCollector</tt>
     */
    private static class RequestEntry {
        public RspCollector coll;

        public RequestEntry(RspCollector coll) {
            this.coll = coll;
        }
    }



    /**
     * The header for <tt>RequestCorrelator</tt> messages
     */
    public static final class Header extends org.jgroups.Header implements Streamable {
        public static final byte REQ = 0;
        public static final byte RSP = 1;

        /** Type of header: request or reply */
        public byte type=REQ;
        /**
         * The id of this request to distinguish among other requests from
         * the same <tt>RequestCorrelator</tt> */
        public long id=0;

        /** msg is synchronous if true */
        public boolean rsp_expected=true;

        /** The unique name of the associated <tt>RequestCorrelator</tt> */
        public String corrName=null;

        /** Stack<Address>. Contains senders (e.g. P --> Q --> R) */
        public java.util.Stack callStack=null;

        /** Contains a list of members who should receive the request (others will drop). Ignored if null */
        public java.util.List dest_mbrs=null;


        /**
         * Used for externalization
         */
        public Header() {}

        /**
         * @param type type of header (<tt>REQ</tt>/<tt>RSP</tt>)
         * @param id id of this header relative to ids of other requests
         * originating from the same correlator
         * @param rsp_expected whether it's a sync or async request
         * @param name the name of the <tt>RequestCorrelator</tt> from which
         */
        public Header(byte type, long id, boolean rsp_expected, String name) {
            this.type         = type;
            this.id           = id;
            this.rsp_expected = rsp_expected;
            this.corrName     = name;
        }

        /**
         */
        public String toString() {
            StringBuffer ret=new StringBuffer();
            ret.append("[Header: name=" + corrName + ", type=");
            ret.append(type == REQ ? "REQ" : type == RSP ? "RSP" : "<unknown>");
            ret.append(", id=" + id);
            ret.append(", rsp_expected=" + rsp_expected + ']');
            if(callStack != null)
                ret.append(", call stack=" + callStack);
            if(dest_mbrs != null)
                ret.append(", dest_mbrs=").append(dest_mbrs);
            return ret.toString();
        }


        public void writeExternal(ObjectOutput out) throws IOException {
            out.writeByte(type);
            out.writeLong(id);
            out.writeBoolean(rsp_expected);
            if(corrName != null) {
                out.writeBoolean(true);
                out.writeUTF(corrName);
            }
            else {
                out.writeBoolean(false);
            }
            out.writeObject(callStack);
            out.writeObject(dest_mbrs);
        }


        public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
            type         = in.readByte();
            id           = in.readLong();
            rsp_expected = in.readBoolean();
            if(in.readBoolean())
                corrName         = in.readUTF();
            callStack   = (java.util.Stack)in.readObject();
            dest_mbrs=(java.util.List)in.readObject();
        }

        public void writeTo(DataOutputStream out) throws IOException {
            out.writeByte(type);
            out.writeLong(id);
            out.writeBoolean(rsp_expected);

            if(corrName != null) {
                out.writeBoolean(true);
                out.writeUTF(corrName);
            }
            else {
                out.writeBoolean(false);
            }

            if(callStack != null) {
                out.writeBoolean(true);
                out.writeShort(callStack.size());
                Address mbr;
                for(int i=0; i < callStack.size(); i++) {
                    mbr=(Address)callStack.elementAt(i);
                    Util.writeAddress(mbr, out);
                }
            }
            else {
                out.writeBoolean(false);
            }

            Util.writeAddresses(dest_mbrs, out);
        }

        public void readFrom(DataInputStream in) throws IOException, IllegalAccessException, InstantiationException {
            boolean present;
            type=in.readByte();
            id=in.readLong();
            rsp_expected=in.readBoolean();

            present=in.readBoolean();
            if(present)
                corrName=in.readUTF();

            present=in.readBoolean();
            if(present) {
                callStack=new Stack();
                short len=in.readShort();
                Address tmp;
                for(short i=0; i < len; i++) {
                    tmp=Util.readAddress(in);
                    callStack.add(tmp);
                }
            }

            dest_mbrs=(List)Util.readAddresses(in, java.util.LinkedList.class);
        }

        public long size() {
            long retval=Global.BYTE_SIZE // type
                    + Global.LONG_SIZE // id
                    + Global.BYTE_SIZE; // rsp_expected

            retval+=Global.BYTE_SIZE; // presence for corrName
            if(corrName != null)
                retval+=corrName.length() +2; // UTF

            retval+=Global.BYTE_SIZE; // presence
            if(callStack != null) {
                retval+=Global.SHORT_SIZE; // number of elements
                if(callStack.size() > 0) {
                    Address mbr=(Address)callStack.firstElement();
                    retval+=callStack.size() * (Util.size(mbr));
                }
            }

            retval+=Util.size(dest_mbrs);
            return retval;
        }

    }




    /**
     * Listens for scheduler events and sets the current call chain (stack)
     * whenever a thread is started, or a suspended thread resumed. Does
     * this only for synchronous requests (<code>Runnable</code> is actually
     * a <code>Request</code>).
     */
    private class CallStackSetter implements SchedulerListener {
        public void started(Runnable r)   { setCallStack(r); }
        public void stopped(Runnable r)   { setCallStack(null); }
        public void suspended(Runnable r) { setCallStack(null); }
        public void resumed(Runnable r)   { setCallStack(r); }

        void setCallStack(Runnable r) {
            java.util.Stack new_stack;
            Message req;
            Header  hdr;
            Object  obj;

            if(r == null) {
                call_stack=null;
                return;
            }

            req=((Request)r).req;
            if(req == null)
                return;

            obj=req.getHeader(name);
            if(obj == null || !(obj instanceof Header))
                return;

            hdr=(Header)obj;
            if(hdr.rsp_expected == false)
                return;

            new_stack=hdr.callStack;
            if(new_stack != null)
                call_stack=(java.util.Stack)new_stack.clone();
        }
    }


    /**
     * The runnable for an incoming request which is submitted to the
     * dispatcher
     */
    private class Request implements Runnable {
        public final Message req;

        public Request(Message req) { this.req=req; }
        public void run() { handleRequest(req); }

        public String toString() {
            StringBuffer sb=new StringBuffer();
            if(req != null)
                sb.append("req=" + req + ", headers=" + req.printObjectHeaders());
            return sb.toString();
        }
    }

}