DiscoveryClientListener.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.Notification;

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

public class DiscoveryClientListener implements Runnable {
    public static final String DISCOVERY_TYPE = "jonas.management.discovery";

    private static int RECEIVE_BUFFER_SIZE = 1024;
    private DiscoveryClient discoveryClient;
    private MulticastSocket multicastSocket;
    private int port;
    private InetAddress destAddress;
    private int             ttl;
    private DatagramSocket  unicastSocket;
    private String          sourceIp;
    private int             sourcePort;
    private boolean         notStopped          = true;
    private long            timeout             = 1000;
    private long            sequenceNumber      = 0;
    private static Logger   logger              = Log.getLogger(Log.JONAS_DISCOVERY_PREFIX);
    public DiscoveryClientListener(DiscoveryClient discoveryClient) {
        this.port = discoveryClient.getListeningPort();
        try {
            this.destAddress = InetAddress.getByName(discoveryClient.getListeningIp());
            this.ttl = discoveryClient.getTimeToLive();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        this.timeout = discoveryClient.getTimeout();
        this.sourcePort = discoveryClient.getSourcePort();
        this.sourceIp = discoveryClient.getSourceIp();
        this.discoveryClient = discoveryClient;
    }

    public void sendDiscoveryMessage(DiscMessage msg) {
        logger.log(BasicLevel.DEBUG, "DiscoveryClient : The message to send is "
                + msg);
        //send the message on the multicast socket
        //after packing it into a datagram
        byte[] messageBytes = DiscMessage.objectToBytes(msg);
        if (messageBytes != null) {
                try {
                        multicastSocket.send(new DatagramPacket(messageBytes,
                    messageBytes.length, destAddress, port));
                } catch (IOException e1) {
                        logger.log(BasicLevel.ERROR,
                        "DiscoveryClient :  Error to send discovery message");
                        e1.printStackTrace();
                }
        }
    }

    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;

        // Create a multicast socket
        try {
            multicastSocket = new MulticastSocket(port);
            multicastSocket.setTimeToLive(ttl);
            /*
             * Not necessary as the DiscoveryClient does not waits for multicast events
             */
            //multicastSocket.joinGroup(destAddress);
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

        // Create a unicast socket to receive responses
        try {
            unicastSocket = new DatagramSocket(sourcePort);
        } catch (SocketException e2) {
            logger.log(BasicLevel.ERROR,
            "DiscoveryClient : Unable to create a Datagram socket");
            e2.printStackTrace();
            return;
        }

        // Prepare a discovery message
        DiscMessage msg = new DiscMessage(sourceIp, sourcePort);
        //  First send a discovery message
        sendDiscoveryMessage(msg);
        logger.log(BasicLevel.DEBUG, " DiscoveryClient: Sent Message is" + msg);

        //  wait for responses during a Timeout period on the unicast socket.
        long lastTime = timeout + System.currentTimeMillis();
        DiscEvent event = null;
        try {
            while ((notStopped) && System.currentTimeMillis() <= lastTime) {
                unicastSocket.receive(datagram);
                in = new ObjectInputStream(new ByteArrayInputStream(datagram.getData()));
                objReceived = in.readObject();
                if (objReceived != null) {
                    if (objReceived instanceof DiscEvent) {
                        event = (DiscEvent) objReceived;
                        handleReceivedMessage(event);
                    }
                }
            }
        } catch (SocketException e) {
            logger.log(BasicLevel.ERROR, "DiscoveryClient : Socket closed" + e);
            notStopped = false;
        } catch (IOException e1) {
            e1.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    private void handleReceivedMessage(DiscEvent msg) {
        logger.log(BasicLevel.DEBUG, "discovery event received: " + msg);
        
        // create a JMX notification for all listeners of a discovery event
        Notification notif = new Notification(DISCOVERY_TYPE, discoveryClient,
                sequenceNumber++, System.currentTimeMillis(), msg.getState());
        notif.setUserData(msg);
        discoveryClient.sendNotification(notif);
    }

    public void stop() {
        notStopped = false;
        Thread.interrupted();
    }
}

---