srdi.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.cm;

import net.jxta.credential.Credential;
import net.jxta.document.MimeMediaType;
import net.jxta.document.StructuredDocument;
import net.jxta.id.ID;
import net.jxta.id.IDFactory;
import net.jxta.impl.protocol.ResolverSrdiMsgImpl;
import net.jxta.impl.protocol.SrdiMessageImpl;
import net.jxta.impl.util.JxtaHash;
import net.jxta.logging.Logging;
import net.jxta.membership.MembershipService;
import net.jxta.peer.PeerID;
import net.jxta.peergroup.PeerGroup;
import net.jxta.protocol.RdvAdvertisement;
import net.jxta.protocol.ResolverQueryMsg;
import net.jxta.protocol.ResolverSrdiMsg;
import net.jxta.protocol.SrdiMessage;
import net.jxta.rendezvous.RendezVousService;
import net.jxta.rendezvous.RendezVousStatus;
import net.jxta.rendezvous.RendezvousEvent;
import net.jxta.rendezvous.RendezvousListener;
import net.jxta.resolver.ResolverService;

import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
import java.math.BigInteger;
import java.net.URI;
import java.net.URISyntaxException;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.Vector;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Srdi is a service which provides SRDI functionalities such as :
 * <p/>
 * <ul>
 * <li>pushing of SRDI messages to a another peer/propagate</li>
 * <li>replication of an SRDI Message to other peers in a given peerview</li>
 * <li>given an expression SRDI provides a independently calculated starting point</li>
 * <li>Forwarding a ResolverQuery, and taking care of hopCount, random selection</li>
 * <li>registers with the RendezvousService to determine when to share SrdSRDIi Entries</li>
 * and whether to push deltas, or full a index</li>
 * <li>provides a SrdiInterface giving to provide a generic SRDI message definition</li>
 * </ul>
 * <p/>
 * If Srdi is started as a thread it performs periodic SRDI pushes of
 * indices and also has the ability to respond to rendezvous events.
 * <p/>
 * ResolverSrdiMessages define a ttl, to indicate to the receiving service
 * whether to replicate such message or not.
 * <p/>
 * In addition A ResolverQuery defines a hopCount to indicate how many
 * hops a query has been forwarded. This element could be used to detect/stop a
 * query forward loopback hopCount is checked to make ensure a query is not
 * forwarded more than twice.
 *
 * @see <a href="https://jxta-spec.dev.java.net/nonav/JXTAProtocols.html#proto-prp" target="_blank">JXTA Protocols Specification : Peer Resolver Protocol</a>
 */
public class Srdi implements Runnable, RendezvousListener {

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

    private PeerGroup group = null;
    private String handlername = null;
    private SrdiInterface srdiService = null;
    private SrdiIndex srdiIndex;
    private long connectPollInterval = 0;
    private long pushInterval = 0;

    private volatile boolean stop = false;
    private AtomicBoolean republishSignal = new AtomicBoolean(false);

    private ResolverService resolver;
    private MembershipService membership;
    private final JxtaHash jxtaHash = new JxtaHash();
    private CredentialListener membershipCredListener = null;
    private Credential credential = null;
    private StructuredDocument credentialDoc = null;
    private final String rdvEventLock;

    /**
     * Random number generator used for random result selection
     */
    private final static Random random = new Random();

    // This ought be to configurable/based on a function applied to the rpv size
    /**
     * Replication threshold (minimum number of rdv's in peer view before replication)
     */
    public final static int RPV_REPLICATION_THRESHOLD = 3;

    /**
     * Listener we use for membership property events.
     */
    private class CredentialListener implements PropertyChangeListener {

        /**
         * {@inheritDoc}
         */
        public void propertyChange(PropertyChangeEvent evt) {
            if ("defaultCredential".equals(evt.getPropertyName())) {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("New default credential event");
                }

                synchronized (Srdi.this) {
                    credential = (Credential) evt.getNewValue();
                    credentialDoc = null;
                    if (null != credential) {
                        try {
                            credentialDoc = credential.getDocument(MimeMediaType.XMLUTF8);
                        } catch (Exception all) {
                            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                                LOG.log(Level.WARNING, "Could not generate credential document", all);
                            }
                        }
                    }
                }
            }
        }
    }


    /**
     * Interface for pushing entries.
     */
    public interface SrdiInterface {

        /**
         * Pushe SRDI entries.
         *
         * @param all if true then push all entries otherwise just push
         *            those which have changed since the last push.
         */
        void pushEntries(boolean all);
    }

    /**
     * Starts the Srdi Service. wait for connectPollInterval prior to
     * pushing the index if connected to a rdv, otherwise index is
     * as soon as the Rendezvous connect occurs
     *
     * @param group               group context to operate in
     * @param handlername         the SRDI handlername
     * @param srdiService         the service utilizing this Srdi, for purposes of
     *                            callback push entries on events such as rdv connect/disconnect, etc.
     * @param srdiIndex           The index instance associated with this service
     * @param connectPollInterval initial timeout before the very first push of entries in milliseconds
     * @param pushInterval        the Interval at which the deltas are pushed in milliseconds
     */
    public Srdi(PeerGroup group, String handlername, SrdiInterface srdiService, SrdiIndex srdiIndex, long connectPollInterval, long pushInterval) {

        this.group = group;
        this.handlername = handlername;
        this.srdiService = srdiService;
        this.srdiIndex = srdiIndex;
        this.connectPollInterval = connectPollInterval;
        this.pushInterval = pushInterval;
        this.rdvEventLock = new String(handlername);
        membership = group.getMembershipService();

        resolver = group.getResolverService();

        group.getRendezVousService().addListener(this);

        synchronized (this) {
            membershipCredListener = new CredentialListener();
            membership.addPropertyChangeListener("defaultCredential", membershipCredListener);

            try {
                credential = membership.getDefaultCredential();

                if (null != credential) {
                    credentialDoc = credential.getDocument(MimeMediaType.XMLUTF8);
                } else {
                    credentialDoc = null;
                }
            } catch (Exception all) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "could not get credential", all);
                }
            }
        }
    }

    /**
     * stop the current running thread
     */
    public synchronized void stop() {

        if (stop) {
            return;
        }

        stop = true;

        RendezVousService rendezvous = group.getRendezVousService();

        if (null != rendezvous) {
            rendezvous.removeListener(this);
        }

        membership.removePropertyChangeListener("defaultCredential", membershipCredListener);
        membershipCredListener = null;

        // wakeup and die
        synchronized (rdvEventLock) {
            rdvEventLock.notify();
        }
    }

    /**
     * Replicates a SRDI message to other rendezvous'
     * entries are replicated by breaking out entries out of the message
     * and sorted out into rdv distribution bins. after which smaller messages
     * are sent to other rdv's
     *
     * @param srdiMsg srdi message to replicate
     */

    public void replicateEntries(SrdiMessage srdiMsg) {

        List<PeerID> rpv = getGlobalPeerView();

        if (srdiMsg.getScope() < SrdiMessage.REPLICATE || !group.isRendezvous() || rpv.size() < RPV_REPLICATION_THRESHOLD) {
            return;
        }

        Iterator allEntries = srdiMsg.getEntries().iterator();
        Map<PeerID, SrdiMessageImpl> bins = new HashMap<PeerID, SrdiMessageImpl>(rpv.size());

        while (allEntries.hasNext()) {
            SrdiMessage.Entry entry = (SrdiMessage.Entry) allEntries.next();
            PeerID destPeer = getReplicaPeer(srdiMsg.getPrimaryKey() + entry.key + entry.value);

            if (destPeer == null || destPeer.equals(group.getPeerID())) {
                // don't replicate message back to ourselves
                continue;
            }
            SrdiMessageImpl sm = bins.get(destPeer);

            if (sm == null) {
                sm = new SrdiMessageImpl();
                sm.setPrimaryKey(srdiMsg.getPrimaryKey());
                sm.setPeerID(srdiMsg.getPeerID());
                bins.put(destPeer, sm);
            }
            sm.addEntry(entry);
        }

        for (PeerID destPeer : bins.keySet()) {
            SrdiMessageImpl msg = bins.get(destPeer);

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("[" + group.getPeerGroupName() + " / " + handlername + "] Forwarding replica Srdi to " + destPeer);
            }
            pushSrdi(destPeer, msg);
        }
    }

    /**
     * Push an SRDI message to a peer
     * ttl is 1, and therefore services receiving this message could
     * choose to replicate this message
     *