rendezvousserviceprovider.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.rendezvous;

import net.jxta.document.MimeMediaType;
import net.jxta.document.StructuredDocument;
import net.jxta.document.StructuredDocumentFactory;
import net.jxta.document.XMLDocument;
import net.jxta.document.XMLElement;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.EndpointListener;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.TextDocumentMessageElement;
import net.jxta.id.ID;
import net.jxta.id.IDFactory;
import net.jxta.impl.id.UUID.UUID;
import net.jxta.impl.rendezvous.rendezvousMeter.RendezvousMeter;
import net.jxta.impl.rendezvous.rendezvousMeter.RendezvousMeterBuildSettings;
import net.jxta.impl.rendezvous.rendezvousMeter.RendezvousServiceMonitor;
import net.jxta.logging.Logging;
import net.jxta.peergroup.PeerGroup;
import net.jxta.platform.Module;
import net.jxta.protocol.PeerAdvertisement;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Enumeration;
import java.util.Vector;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * This abstract class must be extended for all RendezVous Service providers
 * that are managed by RendezVousServiceImpl.
 * <p/>
 * Implementors of providers are responsible for using appropriate
 * synchronization. The RendezvousServiceImpl provides synchronization control
 * only only those methods which involve changing the active provider.
 */
public abstract class RendezVousServiceProvider implements EndpointListener {

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

    protected static final String PropSName = "JxtaPropagate";

    protected static final String RDV_MSG_NAMESPACE_NAME = "jxta";

    protected final String PropPName;
    protected final String PROP_HDR_ELEMENT_NAME;

    /**
     * Maximum TTL we will allow for propagation and repropagation issued by
     * this peer.
     */
    protected int MAX_TTL;

    protected final PeerGroup group;
    protected final RendezVousServiceImpl rdvService;
    protected boolean closed = false;

    private PeerAdvertisement cachedPeerAdv = null;
    private int cachedPeerAdvModCount = -1;
    private XMLDocument cachedPeerAdvDoc = null;

    protected RendezvousServiceMonitor rendezvousServiceMonitor = null;
    protected RendezvousMeter rendezvousMeter = null;

    /**
     *
     * @param group the peergroup
     * @param rdvService the implementation
     */
    protected RendezVousServiceProvider(PeerGroup group, RendezVousServiceImpl rdvService) {

        this.group = group;
        this.rdvService = rdvService;

        PropPName = this.group.getPeerGroupID().getUniqueValue().toString();
        PROP_HDR_ELEMENT_NAME = RendezVousPropagateMessage.Name + PropPName;
    }

    /**
     * {@inheritDoc}
     * <p/>
     * EndpointListener for the JxtaPropagate/<peergroup-unique value>
     */
    public void processIncomingMessage(Message msg, EndpointAddress srcAddr, EndpointAddress dstAddr) {

        RendezVousPropagateMessage propHdr = checkPropHeader(msg);

        if (null != propHdr) {
            // Get the destination real destination of the message
            String sName = propHdr.getDestSName();
            String sParam = propHdr.getDestSParam();

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Processing " + msg + "(" + propHdr.getMsgId() + ") for " + sName + "/" + sParam + " from " + srcAddr);
            }

            // Check if we have a local listener for this message
            processReceivedMessage(msg, propHdr, srcAddr, new EndpointAddress(dstAddr, sName, sParam));
        }
    }

    protected XMLDocument getPeerAdvertisementDoc() {
        PeerAdvertisement newPadv;

        synchronized (this) {
            newPadv = group.getPeerAdvertisement();
            int newModCount = newPadv.getModCount();

            if ((cachedPeerAdv != newPadv) || (cachedPeerAdvModCount != newModCount)) {
                cachedPeerAdv = newPadv;
                cachedPeerAdvModCount = newModCount;
            } else {
                newPadv = null;
            }

            if (null != newPadv) {
                cachedPeerAdvDoc = (XMLDocument) cachedPeerAdv.getDocument(MimeMediaType.XMLUTF8);
            }
        }

        return cachedPeerAdvDoc;
    }

    /**
     * Supply arguments and starts this service if it hadn't started by itself.
     * <p/>
     * Currently this service starts by itself and does not expect arguments.
     * @return 0 if successful
     * @param arg argument params
     */
    protected int startApp(String[] arg) {

        // All propagated messages originated by RendezvousService.propagate are handled by the
        // rendezvous service before being delivered to their local recipient.
        // This includes:
        // messages delivered here via netWorkPropagation. Therefore the rdv service has a special
        // endpointService listener for that purpose.
        // messages delivered here by rdv-to-rdv walk.
        try {
            if (!rdvService.endpoint.addIncomingMessageListener(this, PropSName, PropPName)) {
                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.severe("Cannot register the propagation listener (already registered)");
                }

                return -1;
            }
        } catch (Exception ez1) {
            // Not much we can do here.
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "Failed registering the endpoint listener", ez1);
            }

            return -1;
        }

        try {
            // Update the peeradv with our status
            if (rdvService.isRendezVous()) {
                XMLDocument params = (XMLDocument) StructuredDocumentFactory.newStructuredDocument(MimeMediaType.XMLUTF8, "Parm");
                XMLElement e = params.createElement("Rdv", Boolean.TRUE.toString());

                params.appendChild(e);
                group.getPeerAdvertisement().putServiceParam(rdvService.getAssignedID(), params);
            } else {
                group.getPeerAdvertisement().removeServiceParam(rdvService.getAssignedID());
            }
        } catch (Exception ignored) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Could not update Rdv Params in Peer Advertisement", ignored);
            }
        }

        return Module.START_OK;
    }

    /**
     * Ask this service to stop.
     */
    protected void stopApp() {
        EndpointListener shouldbeMe = rdvService.endpoint.removeIncomingMessageListener(PropSName, PropPName);

        if ((null != shouldbeMe) && (this != shouldbeMe)) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("Unregistered listener was not as expected." + this + " != " + shouldbeMe);
            }
        }

        // Update the peeradv. We are not a rdv.
        group.getPeerAdvertisement().removeServiceParam(rdvService.getAssignedID());
    }

    /**
     * Set the RendezvousServiceMonitor. The RendezvousServiceMonitor is used to
     * meter the activity of the RendezvousService.
     *
     * @param rendezvousServiceMonitor The monitor.
     * @see net.jxta.impl.meter.MonitorManager
     */
    public void setRendezvousServiceMonitor(RendezvousServiceMonitor rendezvousServiceMonitor) {
        if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING) {
            this.rendezvousServiceMonitor = rendezvousServiceMonitor;