udp_nio.java.txt

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

        /** Vector<Connector>. A list of Connectors, one for each interface we listen on */
        Vector connectors=new Vector();

        boolean running=false;

        static final Log mylog=LogFactory.getLog(ConnectorTable.class);
        static final boolean mywarn=mylog.isWarnEnabled();




        public ConnectorTable(InetSocketAddress mcast_addr,
                              int receive_buffer_size, int receive_sock_buf_size,
                              boolean ip_mcast, Receiver receiver) throws IOException {
            this.receiver=receiver;
            this.mcastAddr=mcast_addr;
            this.receive_buffer=new byte[receive_buffer_size];

            if(ip_mcast) {
                mcast_sock=new MulticastSocket(mcast_addr.getPort());
                // changed Bela Dec 31 2003: if loopback is disabled other members on the same machine won't be able
                // to receive our multicasts
                // mcast_sock.setLoopbackMode(true); // do not get own multicasts
                mcast_sock.setReceiveBufferSize(receive_sock_buf_size);
            }
        }


        public Receiver getReceiver() {
            return receiver;
        }

        public void setReceiver(Receiver receiver) {
            this.receiver=receiver;
        }


        /** Get all interfaces, create one Connector per interface and call start() on it */
        public void start() throws Exception {
            Connector tmp;
            if(running)
                return;

            if(mcast_sock != null) {
                // Start the thread servicing the incoming multicasts
                t=new Thread(this, "ConnectorTable thread");
                t.setDaemon(true);
                t.start();
            }


            // Start all Connectors
            for(Iterator it=connectors.iterator(); it.hasNext();) {
                tmp=(Connector)it.next();
                tmp.start();
            }

            running=true;
        }


        public void stop() {
            Connector tmp;
            for(Iterator it=connectors.iterator(); it.hasNext();) {
                tmp=(Connector)it.next();
                tmp.stop();
            }
            connectors.clear();
            t=null;
            if(mcast_sock != null) {
                mcast_sock.close();
                mcast_sock=null;
            }
            running=false;
        }


        public void run() {
            // receive mcast packets on any interface of the list of interfaces we're listening on
            DatagramPacket p=new DatagramPacket(receive_buffer, receive_buffer.length);
            while(t != null && mcast_sock != null && !mcast_sock.isClosed()) {
                p.setData(receive_buffer, 0, receive_buffer.length);
                try {
                    receivePacket(p, mcast_sock, this);
                }
                catch(Throwable th) {
                    if(th == null || mcast_sock == null || mcast_sock.isClosed())
                        break;
                    if(mylog.isErrorEnabled()) mylog.error("exception=" + th);
                    Util.sleep(300); // so we don't get into 100% cpu spinning (should NEVER happen !)
                }
            }
            t=null;
        }


        /**
         * Returns a list of local addresses (one for each Connector)
         * @return List<SocketAddress>
         */
        public List getConnectorAddresses() {
            Connector c;
            ArrayList ret=new ArrayList();
            for(Iterator it=connectors.iterator(); it.hasNext();) {
                c=(Connector)it.next();
                ret.add(c.getLocalAddress());
            }
            return ret;
        }

        /** Sends a packet. If the destination is a multicast address, call send() on all connectors.
         * If destination is not null, send the message using <em>any</em> Connector: if we send a unicast
         * message, it doesn't matter to which interface we are bound; the kernel will choose the correct
         * interface based on the destination and the routing table. Note that the receiver will have the
         * interface which was chosen by the kernel to send the message as the receiver's address, so the
         * correct Connector will receive a possible response.
         * @param msg
         * @throws Exception
         */
        public void send(DatagramPacket msg) throws Exception {
            InetAddress dest;

            if(msg == null)
                return;
            dest=msg.getAddress();
            if(dest == null)
                throw new IOException("UDP_NIO.ConnectorTable.send(): destination address is null");

            if(dest.isMulticastAddress()) {
                // send to all Connectors
                for(int i=0; i < connectors.size(); i++) {
                    ((Connector)connectors.get(i)).send(msg);
                }
            }
            else {
                // send to a random connector
                Connector c=pickRandomConnector(connectors);
                c.send(msg);
            }
        }

        private Connector pickRandomConnector(Vector conns) {
            int size=conns.size();
            int index=((int)(Util.random(size))) -1;
            return (Connector)conns.get(index);
        }

        /**
         * Adds the given interface address to the list of interfaces on which the receiver mcast
         * socket has to listen.
         * Also creates a new Connector. Calling this method twice on the same interface will throw an exception
         * @param bind_interface
         * @param local_port
         * @param port_range
         * @param receive_buffer_size
         * @throws IOException
         */
        public void listenOn(String bind_interface, int local_port, int port_range, 
                             int receive_buffer_size, int receiver_sock_buf_size, int send_sock_buf_size,
                             int ip_ttl, Receiver receiver) throws IOException {
            if(bind_interface == null)
                return;

            NetworkInterface ni=NetworkInterface.getByInetAddress(InetAddress.getByName(bind_interface));
            if(ni == null)
                throw new IOException("UDP_NIO.ConnectorTable.listenOn(): bind interface for " +
                        bind_interface + " not found");

            Connector tmp=findConnector(ni);
            if(tmp != null) {
                if(mywarn) mylog.warn("connector for interface " + bind_interface +
                        " is already present (will be skipped): " + tmp);
                return;
            }

            // 1. join the group on this interface
            if(mcast_sock != null) {
                mcast_sock.joinGroup(mcastAddr, ni);

                    if(mylog.isInfoEnabled()) mylog.info("joining " + mcastAddr + " on interface " + ni);
            }

            // 2. create a new Connector
            tmp=new Connector(ni, local_port, port_range, receive_buffer_size, receiver_sock_buf_size,
                    send_sock_buf_size, ip_ttl, receiver);
            connectors.add(tmp);
        }

        private Connector findConnector(NetworkInterface ni) {
            for(int i=0; i < connectors.size(); i++) {
                Connector c=(Connector)connectors.elementAt(i);
                if(c.getBindInterface().equals(ni))
                    return c;
            }
            return null;
        }


        public void receive(DatagramPacket packet) {
            if(receiver != null) {
                receiver.receive(packet);
            }
        }


        public String toString() {
            StringBuffer sb=new StringBuffer();
            sb.append("*** todo: implement ***");
            return sb.toString();
        }


        public static void receivePacket(DatagramPacket packet, DatagramSocket sock, Receiver receiver) throws IOException {
            int len;

            sock.receive(packet);
            len=packet.getLength();
            if(len == 1 && packet.getData()[0] == 0) {
                if(mylog.isTraceEnabled()) mylog.trace("received dummy packet");
                return;
            }
            if(receiver != null)
                receiver.receive(packet);
        }
    }






    public static class MyReceiver implements Receiver {
        ConnectorTable t=null;

        public MyReceiver() {

        }

        public void setConnectorTable(ConnectorTable t) {
            this.t=t;
        }

        public void receive(DatagramPacket packet) {
            System.out.println("-- received " + packet.getLength() + " bytes from " + packet.getSocketAddress());
            InetAddress sender=packet.getAddress();
            byte[] buf=packet.getData();
            int len=packet.getLength();
            String tmp=new String(buf, 0, len);
            if(len > 4) {
                if(tmp.startsWith("rsp:")) {
                    System.out.println("-- received respose: \"" + tmp + '\"');
                    return;
                }
            }

            byte[] rsp_buf=("rsp: this is a response to " + tmp).getBytes();
            DatagramPacket response=new DatagramPacket(rsp_buf, rsp_buf.length, sender, packet.getPort());

            try {
                t.send(response);
            }
            catch(Exception e) {
                e.printStackTrace();
                System.err.println("MyReceiver: problem sending response to " + sender);
            }
        }
    }



    public static class MulticastReceiver implements Runnable {
        Unmarshaller m=null;
        DatagramSocket sock=null; // may be DatagramSocket or MulticastSocket

        public void run() {
            // receives packet from socket
            // calls Unmarshaller.receive()
        }

    }

    public static class Unmarshaller {
        Queue q=null;

        void receive(byte[] data, SocketAddress sender) {
            // if (q) --> q.add()
            // unserialize and call handleMessage()
        }
    }



    static void help() {
        System.out.println("UDP_NIO [-help] [-bind_addrs <list of interfaces>]");
    }



    public static void main(String[] args) {
        MyReceiver        r=new MyReceiver();
        ConnectorTable    ct;
        String            line;
        InetSocketAddress mcast_addr;
        BufferedReader    in=null;
        DatagramPacket    packet;
        byte[]            send_buf;
        int               receive_buffer_size=65000;
        boolean           ip_mcast=true;

        try {
            mcast_addr=new InetSocketAddress("230.1.2.3", 7500);
            ct=new ConnectorTable(mcast_addr, receive_buffer_size, 120000, ip_mcast, r);
            r.setConnectorTable(ct);
        }
        catch(Throwable t) {
            t.printStackTrace();
            return;
        }

        for(int i=0; i < args.length; i++) {
            if("-help".equals(args[i])) {
                help();
                continue;
            }
            if("-bind_addrs".equals(args[i])) {
                while(++i < args.length && !args[i].trim().startsWith("-")) {
                    try {
                        ct.listenOn(args[i], 0, 1, receive_buffer_size, 120000, 12000, 32, r);
                    }
                    catch(IOException e) {
                        e.printStackTrace();
                        return;
                    }
                }
            }
        }


        try {
            ct.start(); // starts all Connectors in turn
            in=new BufferedReader(new InputStreamReader(System.in));
            while(true) {
                System.out.print("> "); System.out.flush();
                line=in.readLine();
                if(line.startsWith("quit") || line.startsWith("exit"))
                    break;
                send_buf=line.getBytes();
                packet=new DatagramPacket(send_buf, send_buf.length, mcast_addr);
                ct.send(packet);
            }
        }
        catch(Exception e) {
            e.printStackTrace();
        }
        finally {
            if(ct != null)
                ct.stop();
        }
    }




}


interface Receiver {

    /** Called when data has been received on a socket. When the callback returns, the buffer will be
     * reused: therefore, if <code>buf</code> must be processed on a separate thread, it needs to be copied.
     * This method might be called concurrently by multiple threads, so it has to be reentrant
     * @param packet
     */
    void receive(DatagramPacket packet);
}