tp.java

来自「JGRoups源码」· Java 代码 · 共 1,778 行 · 第 1/5 页

                interfaces.add(intf);
            }
        }
        return interfaces;
    }

    private static String print(java.util.List interfaces) {
        StringBuffer sb=new StringBuffer();
        boolean first=true;
        NetworkInterface intf;
        for(Iterator it=interfaces.iterator(); it.hasNext();) {
            intf=(NetworkInterface)it.next();
            if(first) {
                first=false;
            }
            else {
                sb.append(", ");
            }
            sb.append(intf.getName());
        }
        return sb.toString();
    }


    protected void handleDownEvent(Event evt) {
        switch(evt.getType()) {

        case Event.TMP_VIEW:
        case Event.VIEW_CHANGE:
            synchronized(members) {
                view=(View)evt.getArg();
                members.clear();
                Vector tmpvec=view.getMembers();
                members.addAll(tmpvec);
            }
            break;

        case Event.GET_LOCAL_ADDRESS:   // return local address -> Event(SET_LOCAL_ADDRESS, local)
            passUp(new Event(Event.SET_LOCAL_ADDRESS, local_addr));
            break;

        case Event.CONNECT:
            channel_name=(String)evt.getArg();
            header=new TpHeader(channel_name);
            setThreadNames();

            // removed March 18 2003 (bela), not needed (handled by GMS)
            // changed July 2 2003 (bela): we discard CONNECT_OK at the GMS level anyway, this might
            // be needed if we run without GMS though
            passUp(new Event(Event.CONNECT_OK));
            break;

        case Event.DISCONNECT:
            unsetThreadNames();
            passUp(new Event(Event.DISCONNECT_OK));
            break;

        case Event.CONFIG:
            if(log.isDebugEnabled()) log.debug("received CONFIG event: " + evt.getArg());
            handleConfigEvent((HashMap)evt.getArg());
            break;
        }
    }




    protected void setThreadNames() {
        if(channel_name != null) {
            String tmp, prefix=Global.THREAD_PREFIX;
            if(incoming_packet_handler != null) {
                tmp=incoming_packet_handler.getName();
                if(tmp != null && tmp.indexOf(prefix) == -1) {
                    tmp+=prefix + channel_name + ")";
                    incoming_packet_handler.setName(tmp);
                }
            }
            if(incoming_msg_handler != null) {
                tmp=incoming_msg_handler.getName();
                if(tmp != null && tmp.indexOf(prefix) == -1) {
                    tmp+=prefix + channel_name + ")";
                    incoming_msg_handler.setName(tmp);
                }
            }
            if(outgoing_packet_handler != null) {
                tmp=outgoing_packet_handler.getName();
                if(tmp != null && tmp.indexOf(prefix) == -1) {
                    tmp+=prefix + channel_name + ")";
                    outgoing_packet_handler.setName(tmp);
                }
            }
            if(diag_handler != null) {
                tmp=diag_handler.getName();
                if(tmp != null && tmp.indexOf(prefix) == -1) {
                    tmp+=prefix + channel_name + ")";
                    diag_handler.setName(tmp);
                }
            }
        }
    }


    protected void unsetThreadNames() {
        if(channel_name != null) {
            String tmp, prefix=Global.THREAD_PREFIX;
            int index;

            tmp=incoming_packet_handler != null? incoming_packet_handler.getName() : null;
            if(tmp != null) {
                index=tmp.indexOf(prefix);
                if(index > -1) {
                    tmp=tmp.substring(0, index);
                    incoming_packet_handler.setName(tmp);
                }
            }

            tmp=incoming_msg_handler != null? incoming_msg_handler.getName() : null;
            if(tmp != null) {
                index=tmp.indexOf(prefix);
                if(index > -1) {
                    tmp=tmp.substring(0, index);
                    incoming_msg_handler.setName(tmp);
                }
            }

            tmp=outgoing_packet_handler != null? outgoing_packet_handler.getName() : null;
            if(tmp != null) {
                index=tmp.indexOf(prefix);
                if(index > -1) {
                    tmp=tmp.substring(0, index);
                    outgoing_packet_handler.setName(tmp);
                }
            }
            tmp=diag_handler != null? diag_handler.getName() : null;
            if(tmp != null) {
                index=tmp.indexOf(prefix);
                if(index > -1) {
                    tmp=tmp.substring(0, index);
                    diag_handler.setName(tmp);
                }
            }
        }
    }



    protected void handleConfigEvent(HashMap map) {
        if(map == null) return;
        if(map.containsKey("additional_data")) {
            additional_data=(byte[])map.get("additional_data");
            if(local_addr instanceof IpAddress)
                ((IpAddress)local_addr).setAdditionalData(additional_data);
        }
    }



    /* ----------------------------- End of Private Methods ---------------------------------------- */



    /* ----------------------------- Inner Classes ---------------------------------------- */

    static class IncomingQueueEntry {
        Address   dest=null;
        Address   sender=null;
        byte[]    buf;
        int       offset, length;

        IncomingQueueEntry(Address dest, Address sender, byte[] buf, int offset, int length) {
            this.dest=dest;
            this.sender=sender;
            this.buf=buf;
            this.offset=offset;
            this.length=length;
        }
    }




    /**
     * This thread fetches byte buffers from the packet_queue, converts them into messages and passes them up
     * to the higher layer (done in handleIncomingUdpPacket()).
     */
    class IncomingPacketHandler implements Runnable {
        Thread t=null;

        String getName() {
            return t != null? t.getName() : null;
        }

        void setName(String thread_name) {
            if(t != null)
                t.setName(thread_name);
        }

        void start() {
            if(t == null || !t.isAlive()) {
                t=new Thread(Util.getGlobalThreadGroup(), this, "IncomingPacketHandler");
                t.setDaemon(true);
                t.start();
            }
        }

        void stop() {
            Thread tmp=t;
            t=null;
            incoming_packet_queue.close(true); // should terminate the packet_handler thread too
            if(tmp != null) {
                try {
                    tmp.join(10000);
                }
                catch(InterruptedException e) {
                }
                if(tmp.isAlive()) {
                    if(log.isWarnEnabled())
                        log.warn("IncomingPacketHandler thread was interrupted, but is still alive");
                }
            }
        }

        public void run() {
            IncomingQueueEntry entry;
            while(!incoming_packet_queue.closed() && Thread.currentThread().equals(t)) {
                try {
                    entry=(IncomingQueueEntry)incoming_packet_queue.remove();
                    handleIncomingPacket(entry.dest, entry.sender, entry.buf, entry.offset, entry.length);
                }
                catch(QueueClosedException closed_ex) {
                    break;
                }
                catch(Throwable ex) {
                    if(log.isErrorEnabled())
                        log.error("error processing incoming packet", ex);
                }
            }
            if(trace) log.trace("incoming packet handler terminating");
        }
    }


    class IncomingMessageHandler implements Runnable {
        Thread t;
        int i=0;


        String getName() {
            return t != null? t.getName() : null;
        }

        void setName(String thread_name) {
            if(t != null)
                t.setName(thread_name);
        }

        public void start() {
            if(t == null || !t.isAlive()) {
                t=new Thread(Util.getGlobalThreadGroup(), this, "IncomingMessageHandler");
                t.setDaemon(true);
                t.start();
            }
        }


        public void stop() {
            incoming_msg_queue.close(true);
            t=null;
        }

        public void run() {
            Message msg;
            while(!incoming_msg_queue.closed() && Thread.currentThread().equals(t)) {
                try {
                    msg=(Message)incoming_msg_queue.remove();
                    handleIncomingMessage(msg);
                }
                catch(QueueClosedException closed_ex) {
                    break;
                }
                catch(Throwable ex) {
                    if(log.isErrorEnabled())
                        log.error("error processing incoming message", ex);
                }
            }
            if(trace) log.trace("incoming message handler terminating");
        }
    }


    /**
     * This thread fetches byte buffers from the outgoing_packet_queue, converts them into messages and sends them
     * using the unicast or multicast socket
     */
    class OutgoingPacketHandler implements Runnable {
        Thread             t=null;
        byte[]             buf;
        DatagramPacket     packet;


        String getName() {
            return t != null? t.getName() : null;
        }

        void setName(String thread_name) {
            if(t != null)
                t.setName(thread_name);
        }

        void start() {
            if(t == null || !t.isAlive()) {
                t=new Thread(Util.getGlobalThreadGroup(), this, "OutgoingPacketHandler");
                t.setDaemon(true);
                t.start();
            }
        }

        void stop() {
            Thread tmp=t;
            t=null;
            if(tmp != null) {
                tmp.interrupt();
            }
        }

        public void run() {
            Message msg;

            while(t != null && Thread.currentThread().equals(t)) {
                try {
                    msg=(Message)outgoing_queue.take();
                    handleMessage(msg);
                }
                catch(QueueClosedException closed_ex) {
                    break;
                }
                catch(InterruptedException interruptedEx) {
                }
                catch(Throwable th) {
                    if(log.isErrorEnabled()) log.error("exception sending packet", th);
                }
                msg=null; // let's give the garbage collector a hand... this is probably useless though
            }
            if(trace) log.trace("outgoing message handler terminating");
        }

        protected void handleMessage(Message msg) throws Throwable {
            Address dest=msg.getDest();
            send(msg, dest, dest == null || dest.isMulticastAddress());
        }


    }