mcasttransport.java

JXTA&#8482 is a set of open, generalized peer-to-peer (P2P) protocols that allow any networked devi

/*
 * Copyright (c) 2001-2007 Sun Microsystems, Inc.  All rights reserved.
 *
 *  The Sun Project JXTA(TM) Software License
 *
 *  Redistribution and use in source and binary forms, with or without
 *  modification, are permitted provided that the following conditions are met:
 *
 *  1. Redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *
 *  2. Redistributions in binary form must reproduce the above copyright notice,
 *     this list of conditions and the following disclaimer in the documentation
 *     and/or other materials provided with the distribution.
 *
 *  3. The end-user documentation included with the redistribution, if any, must
 *     include the following acknowledgment: "This product includes software
 *     developed by Sun Microsystems, Inc. for JXTA(TM) technology."
 *     Alternately, this acknowledgment may appear in the software itself, if
 *     and wherever such third-party acknowledgments normally appear.
 *
 *  4. The names "Sun", "Sun Microsystems, Inc.", "JXTA" and "Project JXTA" must
 *     not be used to endorse or promote products derived from this software
 *     without prior written permission. For written permission, please contact
 *     Project JXTA at http://www.jxta.org.
 *
 *  5. Products derived from this software may not be called "JXTA", nor may
 *     "JXTA" appear in their name, without prior written permission of Sun.
 *
 *  THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES,
 *  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
 *  FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL SUN
 *  MICROSYSTEMS OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
 *  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 *  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA,
 *  OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 *  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 *  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
 *  EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 *  JXTA is a registered trademark of Sun Microsystems, Inc. in the United
 *  States and other countries.
 *
 *  Please see the license information page at :
 *  <http://www.jxta.org/project/www/license.html> for instructions on use of
 *  the license in source files.
 *
 *  ====================================================================
 *
 *  This software consists of voluntary contributions made by many individuals
 *  on behalf of Project JXTA. For more information on Project JXTA, please see
 *  http://www.jxta.org.
 *
 *  This license is based on the BSD license adopted by the Apache Foundation.
 */
package net.jxta.impl.endpoint.mcast;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.net.SocketException;
import java.net.URI;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.logging.Level;
import java.util.logging.Logger;

import net.jxta.document.Advertisement;
import net.jxta.document.AdvertisementFactory;
import net.jxta.document.Attribute;
import net.jxta.document.MimeMediaType;
import net.jxta.document.XMLElement;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.EndpointService;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.MessagePropagater;
import net.jxta.endpoint.MessengerEventListener;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.endpoint.WireFormatMessage;
import net.jxta.endpoint.WireFormatMessageFactory;
import net.jxta.exception.PeerGroupException;
import net.jxta.id.ID;
import net.jxta.logging.Logging;
import net.jxta.meter.MonitorResources;
import net.jxta.peergroup.PeerGroup;
import net.jxta.platform.Module;
import net.jxta.platform.ModuleClassID;
import net.jxta.platform.ModuleSpecID;
import net.jxta.protocol.ConfigParams;
import net.jxta.protocol.ModuleImplAdvertisement;
import net.jxta.protocol.TransportAdvertisement;

import net.jxta.impl.endpoint.EndpointServiceImpl;
import net.jxta.impl.endpoint.IPUtils;
import net.jxta.impl.endpoint.msgframing.MessagePackageHeader;
import net.jxta.impl.endpoint.transportMeter.TransportBindingMeter;
import net.jxta.impl.endpoint.transportMeter.TransportMeter;
import net.jxta.impl.endpoint.transportMeter.TransportMeterBuildSettings;
import net.jxta.impl.endpoint.transportMeter.TransportServiceMonitor;
import net.jxta.impl.meter.MonitorManager;
import net.jxta.impl.peergroup.StdPeerGroup;
import net.jxta.impl.protocol.TCPAdv;

/**
 * This class implements the IP Multicast Message Transport.
 * <p/>
 * <b>Important Note:</b> This implementation was formerly a portion of the TCP
 * Message Transport and currently uses the TCP Transport's configuration 
 * advertisement.
 *
 * @see net.jxta.endpoint.MessageTransport
 * @see net.jxta.endpoint.MessagePropagater
 * @see net.jxta.endpoint.EndpointService
 * @see <a href="http://spec.jxta.org/v1.0/docbook/JXTAProtocols.html#trans-tcpipt">JXTA Protocols Specification : Standard JXTA Transport Bindings</a>
 */
public class McastTransport implements Runnable, Module, MessagePropagater {

    /**
     * Logger
     */
    private static final Logger LOG = Logger.getLogger(McastTransport.class.getName());

    /**
     * Well known service class identifier: mcast message transport
     */
    public final static ModuleClassID MCAST_TRANSPORT_CLASSID =
            ModuleClassID.create(URI.create("urn:jxta:uuid-0C801F65D38F421C9884D706B337B81105"));

    /**
     * Well known service spec identifier: mcast message transport
     */
    public final static ModuleSpecID MCAST_TRANSPORT_SPECID =
            ModuleSpecID.create(URI.create("urn:jxta:uuid-0C801F65D38F421C9884D706B337B8110106"));

    /**
     * The Protocol name we will use for our endpoint addresses.
     */
    private String protocolName = "mcast";

    /**
     * Our Source Addres.
     */
    private EndpointAddress ourSrcAddr = null;

    /**
     * The Source Address Element we attach to all of the messages we send.
     */
    private MessageElement msgSrcAddrElement = null;

    /**
     * The name of the  local interface that we bind to.
     */
    private String interfaceAddressStr;

    /**
     * The address of the local interface address that be bind to.
     */
    private InetAddress usingInterface;

    /**
     * If {@code true} then we are closed otherwise {@code false}
     */
    private boolean isClosed = false;

    /**
     * The name of multicast address we will send/receive upon.
     */
    private String multicastAddress = "224.0.1.85";

    /**
     * The multicast address we will send/receive upon.
     */
    private InetAddress multicastInetAddress;

    /**
     * The port number will send/receive upon.
     */
    private int multicastPort = 1234;

    /**
     * The "return address" we will advertise.
     */
    private EndpointAddress publicAddress = new EndpointAddress(protocolName, multicastAddress + ":" + Integer.toString(multicastPort), null, null);

    /**
     * The maximum size of multicast messages we will send and the size of the
     * {@code DatagramPacket}s we will allocate.
     */
    private int multicastPacketSize = 16384;

    /**
     * The socket we use to send and receive.
     */
    private MulticastSocket multicastSocket = null;

    /**
     * Daemon thread which services the multicast socket and receives datagrams.
     */
    private Thread multicastThread = null;

    /**
     * Thread pooling/queing multicast datagram processor.
     */
    private DatagramProcessor multicastProcessor;

    /**
     * The peer group we are working for.
     */
    private PeerGroup group = null;

    /**
     * The Module Class ID we were assigned in init().
     */
    private ID assignedID = null;

    /**
     * The impl advertisement we were provided in init().
     */
    private ModuleImplAdvertisement implAdvertisement = null;

    /**
     * The endpoint service we are working for.
     */
    private EndpointService endpoint = null;

    private TransportMeter multicastTransportMeter;
    private TransportBindingMeter multicastTransportBindingMeter;
    private transient boolean disabled = false;

    /**
     * Construct a new McastTransport instance
     */
    public McastTransport() {
    }

    /**
     * {@inheritDoc}
     */
    public boolean equals(Object target) {
        if (this == target) {
            return true;
        }

        if (target instanceof McastTransport) {
            McastTransport likeMe = (McastTransport) target;
            return getProtocolName().equals(likeMe.getProtocolName()) && getPublicAddress().equals(likeMe.getPublicAddress());
        }
        return false;
    }

    /**
     * {@inheritDoc}
     */
    public int hashCode() {
        return getPublicAddress().hashCode();
    }

    /**
     * {@inheritDoc}
     */
    public void init(PeerGroup group, ID assignedID, Advertisement impl) throws PeerGroupException {
        this.group = group;
        this.assignedID = assignedID;
        this.implAdvertisement = (ModuleImplAdvertisement) impl;

        ConfigParams configAdv = group.getConfigAdvertisement();

        // Get out invariable parameters from the implAdv
        XMLElement param = (XMLElement) implAdvertisement.getParam();

        if (param != null) {
            Enumeration<XMLElement> list = param.getChildren("Proto");
            if (list.hasMoreElements()) {
                XMLElement pname = list.nextElement();
                protocolName = pname.getTextValue();
            }
        }

        // Get our peer-defined parameters in the configAdv
        param = (XMLElement) configAdv.getServiceParam(PeerGroup.tcpProtoClassID);
        if (null == param) {
            throw new IllegalArgumentException(TransportAdvertisement.getAdvertisementType() + " could not be located.");
        }

        Enumeration<XMLElement> tcpChilds = param.getChildren(TransportAdvertisement.getAdvertisementType());