udp_nio.java.txt

来自「JGRoups源码」· 文本 代码 · 共 1,513 行 · 第 1/4 页

        }

        str=props.getProperty("bind_addrs_exclude");
        if(str != null) {
            str=str.trim();
            exclude_list=Util.parseCommaDelimitedStrings(str);
            props.remove("bind_addrs_exclude");
        }

        str=props.getProperty("bind_port");
        if(str != null) {
            local_bind_port=Integer.parseInt(str);
            props.remove("bind_port");
        }

        str=props.getProperty("start_port");
        if(str != null) {
            local_bind_port=Integer.parseInt(str);
            props.remove("start_port");
        }

        str=props.getProperty("port_range");
        if(str != null) {
            port_range=Integer.parseInt(str);
            props.remove("port_range");
        }

        str=props.getProperty("mcast_addr");
        if(str != null) {
            mcast_addr_name=str;
            props.remove("mcast_addr");
        }

        str=props.getProperty("mcast_port");
        if(str != null) {
            mcast_port=Integer.parseInt(str);
            props.remove("mcast_port");
        }

        str=props.getProperty("ip_mcast");
        if(str != null) {
            ip_mcast=Boolean.valueOf(str).booleanValue();
            props.remove("ip_mcast");
        }

        str=props.getProperty("ip_ttl");
        if(str != null) {
            ip_ttl=Integer.parseInt(str);
            props.remove("ip_ttl");
        }

        str=props.getProperty("mcast_send_buf_size");
        if(str != null) {
            mcast_send_buf_size=Integer.parseInt(str);
            props.remove("mcast_send_buf_size");
        }

        str=props.getProperty("mcast_recv_buf_size");
        if(str != null) {
            mcast_recv_buf_size=Integer.parseInt(str);
            props.remove("mcast_recv_buf_size");
        }

        str=props.getProperty("ucast_send_buf_size");
        if(str != null) {
            ucast_send_buf_size=Integer.parseInt(str);
            props.remove("ucast_send_buf_size");
        }

        str=props.getProperty("ucast_recv_buf_size");
        if(str != null) {
            ucast_recv_buf_size=Integer.parseInt(str);
            props.remove("ucast_recv_buf_size");
        }

        str=props.getProperty("use_packet_handler");
        if(str != null) {
            use_packet_handler=Boolean.valueOf(str).booleanValue();
            props.remove("use_packet_handler");
        }


        // determine mcast_addr
        mcast_addr=new InetSocketAddress(mcast_addr_name, mcast_port);

        // handling of bind_addrs
        if(bind_addrs == null)
            bind_addrs=new ArrayList();
        if(bind_addrs.size() == 0) {
            try {
                String default_bind_addr=determineDefaultBindInterface();
                bind_addrs.add(default_bind_addr);
            }
            catch(SocketException ex) {
                if(log.isErrorEnabled()) log.error("failed determining the default bind interface: " + ex);
            }
        }
        if(exclude_list != null) {
            bind_addrs.removeAll(exclude_list);
        }
        if(bind_addrs.size() == 0) {
            if(log.isErrorEnabled()) log.error("no valid bind interface found, unable to listen for network traffic");
            return false;
        }
        else {

                if(log.isInfoEnabled()) log.info("bind interfaces are " + bind_addrs);
        }

        if(props.size() > 0) {
            log.error("UDP_NIO.setProperties(): the following properties are not recognized: " + props);

            return false;
        }
        return true;
    }


    /**
     * This prevents the up-handler thread to be created, which essentially is superfluous:
     * messages are received from the network rather than from a layer below.
     * DON'T REMOVE ! 
     */
    public void startUpHandler() {
    }

    /**
     * handle the UP event.
     * 
     * @param evt - the event being send from the stack
     */
    public void up(Event evt) {
        passUp(evt);

        switch(evt.getType()) {

            case Event.CONFIG:
                passUp(evt);
                 if(log.isInfoEnabled()) log.info("received CONFIG event: " + evt.getArg());
                handleConfigEvent((HashMap)evt.getArg());
                return;
        }

        passUp(evt);
    }

    /**
     * Caller by the layer above this layer. Usually we just put this Message
     * into the send queue and let one or more worker threads handle it. A worker thread
     * then removes the Message from the send queue, performs a conversion and adds the
     * modified Message to the send queue of the layer below it, by calling Down).
     */
    public void down(Event evt) {
        Message msg;
        Object dest_addr;

        if(evt.getType() != Event.MSG) {  // unless it is a message handle it and respond
            handleDownEvent(evt);
            return;
        }

        msg=(Message)evt.getArg();

        if(udp_hdr != null && udp_hdr.channel_name != null) {
            // added patch by Roland Kurmann (March 20 2003)
            msg.putHeader(name, udp_hdr);
        }

        dest_addr=msg.getDest();

        // Because we don't call Protocol.passDown(), we notify the observer directly (e.g. PerfObserver).
        // This way, we still have performance numbers for UDP
        if(observer != null)
            observer.passDown(evt);

        if(dest_addr == null) { // 'null' means send to all group members
            if(ip_mcast == false) {
                //sends a separate UDP message to each address
                sendMultipleUdpMessages(msg, members);
                return;
            }
        }

        try {
            sendUdpMessage(msg); // either unicast (dest != null) or multicast (dest == null)
        }
        catch(Exception e) {
            if(log.isErrorEnabled()) log.error("exception=" + e + ", msg=" + msg + ", mcast_addr=" + mcast_addr);
        }
    }






    /*--------------------------- End of Protocol interface -------------------------- */


    /* ------------------------------ Private Methods -------------------------------- */


    void handleMessage(Message msg) {

    }


    /**
     * Processes a packet read from either the multicast or unicast socket. Needs to be synchronized because
     * mcast or unicast socket reads can be concurrent
     */
    void handleIncomingUdpPacket(byte[] data, SocketAddress sender) {
        ByteArrayInputStream inp_stream;
        ObjectInputStream    inp;
        Message              msg=null;
        UdpHeader            hdr=null;
        Event                evt;
        Address              dst, src;
        short                version;

        try {
            // skip the first n bytes (default: 4), this is the version info
            inp_stream=new ByteArrayInputStream(data);
            inp=new ObjectInputStream(inp_stream);
            version=inp.readShort();

            if(Version.compareTo(version) == false) {
                if(warn)
                    log.warn("packet from " + sender + " has different version (" + version +
                               ") from ours (" + Version.version + "). This may cause problems");
            }

            msg=new Message();
            msg.readExternal(inp);
            dst=msg.getDest();
            src=msg.getSrc();
            if(src == null) {
                if(log.isErrorEnabled()) log.error("sender's address is null");
            }
            else {
                ((LogicalAddress)src).setPrimaryPhysicalAddress(sender);
            }

            // discard my own multicast loopback copy
            if((dst == null || dst.isMulticastAddress()) && src != null && local_addr.equals(src)) {
                if(trace)
                    log.trace("discarded own loopback multicast packet");

                // System.out.println("-- discarded " + msg.getObject());

                return;
            }

            evt=new Event(Event.MSG, msg);
            if(trace)
                log.trace("Message is " + msg + ", headers are " + msg.getHeaders());

            /* Because Protocol.up() is never called by this bottommost layer, we call up() directly in the observer.
             * This allows e.g. PerfObserver to get the time of reception of a message */
            if(observer != null)
                observer.up(evt, up_queue.size());

            hdr=(UdpHeader)msg.removeHeader(name);
        } catch(Throwable e) {
            if(log.isErrorEnabled()) log.error("exception=" + Util.getStackTrace(e));
            return;
        }

        if(hdr != null) {

            /* Discard all messages destined for a channel with a different name */
            String ch_name=null;

            if(hdr.channel_name != null)
                ch_name=hdr.channel_name;

            // Discard if message's group name is not the same as our group name unless the
            // message is a diagnosis message (special group name DIAG_GROUP)
            if(ch_name != null && group_name != null && !group_name.equals(ch_name) &&
                    !ch_name.equals(Util.DIAG_GROUP)) {

                    if(warn) log.warn("discarded message from different group (" +
                            ch_name + "). Sender was " + msg.getSrc());
                return;
            }
        }

        passUp(evt);
    }


    /**
     * Send a message to the address specified in dest
     */
    void sendUdpMessage(Message msg) throws Exception {
        Address            dest, src;
        ObjectOutputStream out;
        byte               buf[];
        DatagramPacket     packet;
        Message            copy;
        Event              evt; // for loopback messages

        dest=msg.getDest();  // if non-null: unicast, else multicast
        src=msg.getSrc();
        if(src == null) {
            src=local_addr_canonical; // no physical addresses present
            msg.setSrc(src);
        }

        if(trace)
            log.trace("sending message to " + msg.getDest() +
                    " (src=" + msg.getSrc() + "), headers are " + msg.getHeaders());

        // Don't send if destination is local address. Instead, switch dst and src and put in up_queue.
        // If multicast message, loopback a copy directly to us (but still multicast). Once we receive this,
        // we will discard our own multicast message
        if(dest == null || dest.isMulticastAddress() || dest.equals(local_addr)) {
            copy=msg.copy();
            copy.removeHeader(name);
            evt=new Event(Event.MSG, copy);

            /* Because Protocol.up() is never called by this bottommost layer, we call up() directly in the observer.
               This allows e.g. PerfObserver to get the time of reception of a message */
            if(observer != null)
                observer.up(evt, up_queue.size());
            if(trace) log.trace("looped back local message " + copy);

            // System.out.println("\n-- passing up packet id=" + copy.getObject());
            passUp(evt);
            // System.out.println("-- passed up packet id=" + copy.getObject());

            if(dest != null && !dest.isMulticastAddress())
                return; // it is a unicast message to myself, no need to put on the network
        }

        out_stream.reset();
        out=new ObjectOutputStream(out_stream);
        out.writeShort(Version.version);
        msg.writeExternal(out);
        out.flush(); // needed if out buffers its output to out_stream
        buf=out_stream.toByteArray();
        packet=new DatagramPacket(buf, buf.length, mcast_addr);

        //System.out.println("-- sleeping 4 secs");
        // Thread.sleep(4000);


        // System.out.println("\n-- sending packet " + msg.getObject());
        ct.send(packet);
        // System.out.println("-- sent " + msg.getObject());
    }


    void sendMultipleUdpMessages(Message msg, Vector dests) {
        Address dest;

        for(int i=0; i < dests.size(); i++) {
            dest=(Address)dests.elementAt(i);
            msg.setDest(dest);

            try {
                sendUdpMessage(msg);
            }
            catch(Exception e) {
                if(log.isDebugEnabled()) log.debug("exception=" + e);
            }
        }
    }





//
//    /**
//     * Workaround for the problem encountered in certains JDKs that a thread listening on a socket
//     * cannot be interrupted. Therefore we just send a dummy datagram packet so that the thread 'wakes up'
//     * and realizes it has to terminate. Should be removed when all JDKs support Thread.interrupt() on
//     * reads. Uses send_sock t send dummy packet, so this socket has to be still alive.