DiscoveryComm.java

package org.objectweb.jonas.discovery;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.net.SocketException;
import java.net.UnknownHostException;

import javax.management.MBeanServer;
import javax.management.remote.JMXServiceURL;

import org.objectweb.jonas.common.Log;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

public class DiscoveryComm implements Runnable {
    private int              RECEIVE_BUFFER_SIZE = 1024;
    private MulticastSocket  multicastSocket;
    private DatagramSocket unicastSocket;
    private DiscoveryManager dm;
    private int port;
    private InetAddress destAddress;
    private boolean notStopped = true;
    private int ttl = 1; // why 1 ??

    private String jonasName = null;
    private String domainName = null;
    private String[] urls = null;

    private static Logger logger = Log.getLogger(Log.JONAS_DISCOVERY_PREFIX);
    public DiscoveryComm(DiscoveryManager dm) {
        this.port = dm.getListeningPort();
        try {
            this.destAddress = InetAddress.getByName(dm.getListeningIp());
            this.ttl = dm.getTimeToLive();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        this.dm = dm;
    }
    
    private void join() {
        try {
            multicastSocket = new MulticastSocket(port);
            multicastSocket.setTimeToLive(ttl);
            multicastSocket.joinGroup(destAddress);
            logger.log(BasicLevel.DEBUG, "multicast ip address is "
                    + destAddress);

            logger.log(BasicLevel.DEBUG, "multicast port is " + port);

        } catch (IOException e) {
            logger.log(BasicLevel.ERROR, "io problem");
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    public void sendNotif(DiscEvent msg) {
        try {
            //send it on the multicast address
            //after transforming the object to a datagram
            logger.log(BasicLevel.DEBUG, msg);
            byte[] messageBytes = DiscEvent.objectToBytes(msg);
            multicastSocket.send(new DatagramPacket(messageBytes,
                    messageBytes.length, destAddress, port));
        } catch (IOException e1) {
            logger.log(BasicLevel.ERROR, "DiscoveryComm: Error to send notification");
            e1.printStackTrace();
        }

    }
    private void sendResponse(DiscEvent msg, InetAddress destAddress, int port) {
        logger.log(BasicLevel.DEBUG, "DiscoveryComm : The message to send is "
                + msg);
        byte[] messageBytes = DiscEvent.objectToBytes(msg);
        if (messageBytes != null) {
            try {
                // send the unicast response to the discovery client
                unicastSocket.send(new DatagramPacket(messageBytes,
                        messageBytes.length, destAddress, port));
            } catch (IOException e) {
                logger.log(BasicLevel.ERROR, "DiscoveryComm: Error to send response to discovery message");
                e.printStackTrace();
            }
        }
    }

    public DiscEvent createNotifMessage(String state) throws Exception {
        String theHostName;
        MBeanServer mbeanServer;
        try {
            theHostName = InetAddress.getLocalHost().getHostName();
        } catch (UnknownHostException e) {
            e.printStackTrace();
            return null;
        }

        if (!state.equals(DiscEvent.RUNNING)) {
            urls = null;
        }
        // In the case of a notification, the field port is not important since the
        // notifier is not waiting for an acknowledfement.
        DiscEvent resp = new DiscEvent(theHostName, port, jonasName, domainName, urls);
        resp.setState(state);
        return resp;
    }

    public void run()  {
        // Create a DatagramPacket to receive messages
        DatagramPacket datagram = new DatagramPacket(new byte[RECEIVE_BUFFER_SIZE],
                RECEIVE_BUFFER_SIZE);
        // Object reveived in a message
        Object objReceived = null;
        ObjectInputStream in = null;
        
        // Join the group in order to receive multicast messages
        join();
        // Create notification message containing a discovery event with state RUNNING
        DiscEvent msg = null;
        try {
            msg = createNotifMessage(DiscEvent.RUNNING);
        } catch (Exception e) {
            logger.log(BasicLevel.ERROR,
                "DiscoveryComm:  Unable to create a notification message");
            e.printStackTrace();
        }
        if (msg != null) {
            // Multicast the message
            sendNotif(msg);
        }
        // Wait for notification messages
        // Create the socket to be used for responding
        try {
            unicastSocket = new DatagramSocket();
        } catch (SocketException e3) {
            e3.printStackTrace();
            return;
        }
        try {
            while (notStopped) {
                multicastSocket.receive(datagram);
                in = new ObjectInputStream(new ByteArrayInputStream(datagram.getData()));
                objReceived = in.readObject();

                if (objReceived != null) {
                    // The DiscEvents are ignored
                    if (objReceived instanceof DiscMessage) {
                        if (objReceived instanceof DiscEvent) {
                            logger.log(BasicLevel.DEBUG,
                                    "This discovery event is ignored " + objReceived);
                        } else {
                            DiscMessage request = (DiscMessage) objReceived;
                            logger.log(BasicLevel.DEBUG,
                                    "A dicovery message is received "
                                    + objReceived);

                            /* do not create  a new discovery event object with state RUNNING but 
                             * reuse the one created above
                             */
                            /*
                            DiscEvent msg = null;
                            try {
                                msg = createNotifMessage(DiscEvent.RUNNING);
                            } catch (Exception e) {
                                logger.log(BasicLevel.ERROR,
                                    "DiscoveryComm:  Unable to create a notification message");
                                e.printStackTrace();
                            }
                            */
                            if (msg != null) {
                                InetAddress destAddress = InetAddress.getByName(request.getSourceAddress());
                                int destPort = request.getSourcePort();
                                sendResponse(msg, destAddress, destPort);
                            }
                        }
                    }
                }
            }
        } catch (SocketException e) {
            logger.log(BasicLevel.ERROR, "Socket closed: " + e);
            notStopped = false;
        } catch (IOException e1) {
            e1.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    public void stop() {
        // send a notification message of type STOPPING
        DiscEvent msg = null;
        try {
            msg = createNotifMessage(DiscEvent.STOPPING);
        } catch (Exception e) {
            e.printStackTrace();
        }
        if (msg != null) {
            sendNotif(msg);
        }
        Thread.interrupted();
    }

    protected void setJonasName(String jonasName) {
        this.jonasName = jonasName;
    }
    protected void setDomainName(String domainName) {
        this.domainName = domainName;
    }
    protected void setUrls(String[] urls) {
        this.urls = urls;
    }
}

---