srdi.java

jxta_src_2.41b jxta 2.41b 最新版源码 from www.jxta.org

/*
 *  Copyright (c) 2001 Sun Microsystems, Inc.  All rights
 *  reserved.
 *
 *  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 the
 *  Sun Microsystems, Inc. for Project JXTA."
 *  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.
 *  ====================================================================
 *
 *  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.
 *
 *  $Id: Srdi.java,v 1.47 2006/04/06 16:36:27 hamada Exp $
 */
package net.jxta.impl.cm;

import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
import java.math.BigInteger;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Enumeration;
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 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.rendezvous.rpv.PeerViewEvent;
import net.jxta.impl.util.JxtaHash;
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.RendezvousEvent;
import net.jxta.rendezvous.RendezvousListener;
import net.jxta.resolver.ResolverService;

import org.apache.log4j.Level;
import org.apache.log4j.Logger;

/**
 * Srdi is a service which provides SRDI functionalities such as :
 *
 * <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="http://spec.jxta.org/nonav/v1.0/docbook/JXTAProtocols.html#proto-prp" target="_blank">JXTA Protocols Specification : Peer Resolver Protocol</a>
 */
public class Srdi implements Runnable, RendezvousListener {

    /**
     *  Log4J 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 volatile boolean republish = true;

    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 (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("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 (LOG.isEnabledFor(Level.WARN)) {
                                LOG.warn("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 (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("could not get credential", all);
                }
            }
        }
    }


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

        if (stop) {
            return;
        }

        stop = true;

        group.getRendezVousService().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 rpv = getGlobalPeerView();

        if (srdiMsg.getTTL() < 1 || !group.isRendezvous() || rpv.size() < RPV_REPLICATION_THRESHOLD) {
            return;
        }

        Iterator allEntries = srdiMsg.getEntries().iterator();
        Map bins = new HashMap(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 = (SrdiMessageImpl) bins.get(destPeer);

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

        Iterator peers = bins.keySet().iterator();

        while (peers.hasNext()) {
            PeerID destPeer = (PeerID) peers.next();
            SrdiMessageImpl msg = (SrdiMessageImpl) bins.get(destPeer);

            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("[" + 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
     *
     * @param  peer  peer to push message to, if peer is null it is
     *               the message is propagated
     * @param  srdi  SRDI message to send
     */
    public void pushSrdi(ID peer, SrdiMessage srdi) {

        try {
            ResolverSrdiMsg resSrdi = new ResolverSrdiMsgImpl(handlername, credential, srdi.toString());

            if (null == peer) {
                resolver.sendSrdi(null, resSrdi);
            } else {
                resolver.sendSrdi(peer.toString(), resSrdi);
            }
        } catch (Exception e) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Failed to send srdi message", e);
            }
        }
    }