DiscoveryListener.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.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 DiscoveryListener implements Runnable {
    public static final String DISCOVERY_TYPE = "jonas.management.discovery";

    private static int RECEIVE_BUFFER_SIZE = 1024;
    private int             port;
    private InetAddress     destAddress;
    private Enroller        enroller;
    private int             ttl = 1; // why 1 ??
    private MulticastSocket multicastSocket;
    private boolean         notStopped          = true;
    private long            sequenceNumber      = 0;
    private static Logger logger = Log.getLogger(Log.JONAS_DISCOVERY_PREFIX);

    public DiscoveryListener(Enroller enroller) {
        this.port = enroller.getListeningPort();
        try {
            this.destAddress = InetAddress.getByName(enroller.getListeningIp());
            this.ttl = enroller.getTimeToLive();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        this.enroller = enroller;
    }
    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();
        }
    }

    private void handleDiscEvent(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, enroller,
                sequenceNumber++, System.currentTimeMillis(), msg.getState());
        notif.setUserData(msg);
        enroller.sendNotification(notif);
    }

    public void run() {
        // Create a DatagramPacket to receive messages (DiscoverEvents)
        DatagramPacket datagram = new DatagramPacket(new byte[RECEIVE_BUFFER_SIZE],
                RECEIVE_BUFFER_SIZE);
        // Object contained in a message
        Object objReceived = null;
        ObjectInputStream in = null;

        // Join the group in order to receive multicast messages
        join();

        try {
            while (notStopped) {
                multicastSocket.receive(datagram);
                in = new ObjectInputStream(new ByteArrayInputStream(datagram.getData()));
                objReceived = in.readObject();

                if (objReceived != null) {
                    if (objReceived instanceof DiscEvent) {
                        // Treat DiscEvents
                        DiscEvent msg = (DiscEvent) objReceived;
                        handleDiscEvent(msg);
                    }
                }
            }
        }
        catch (SocketException e) {
            logger.log(BasicLevel.ERROR, "Enroller: Socket closed" + e);
            notStopped = false;
        } catch (IOException e1) {
            e1.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    public void stopListener() {
        notStopped = false;
    }

}

---