rendaddrcompactor.java

jxme的一些相关程序,主要是手机上程序开发以及手机和计算机通信的一些程序资料,程序编译需要Ant支持

/*
 * 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: RendAddrCompactor.java,v 1.2 2002/03/04 21:43:00 echtcherbina Exp $
 */

package net.jxta.impl.rendezvous;

import java.util.Enumeration;
import java.io.ByteArrayInputStream;
import java.util.Vector;
import java.util.EmptyStackException;
import java.util.Stack;
import java.io.IOException;
import java.net.*;

import org.apache.log4j.Category; import org.apache.log4j.Priority;

import net.jxta.service.Service;
import net.jxta.endpoint.EndpointService;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.rendezvous.RendezVousService;
import net.jxta.discovery.*;
import net.jxta.protocol.*;
import net.jxta.document.*;
import net.jxta.peergroup.PeerGroup;
import net.jxta.exception.ServiceNotFoundException;
import net.jxta.id.*;
import net.jxta.peer.*;
import net.jxta.impl.peergroup.RefPeerGroup;

public class RendAddrCompactor  implements Runnable {
    private static final Category LOG = Category.getInstance(RendAddrCompactor.class.getName());

    /**
     *  The task of this thread is to determine which of the
     *  rendezvous points is reachable, and which is not.
     *  the thread is started when a group is initialized.
     *  It starts with what's described in the peer/group
     *  advertisement and adds entries one by one to the list
     *  reachable addresses.  The reason behind using EndpointService
     *  and not a direct socket is to ensure we able to discover
     *  routes if the address is not directly reachable
     */


    private PeerGroup myGroup;
    private EndpointService endpoint;
    private DiscoveryService discovery;
    private RendezVousService rendezvous = null;
    private String localPeerId = null;
    private Vector initialParams = null;
    private Thread myThread = null;
    protected boolean stopping = false;
    private RdvListener rdvListener = null;

    private static final int MaxNbOfRdvs = 3;
    private static final long DiscoveryLongNap = 40 * 60 * 1000; // 40 minutes
    private static final long DiscoveryMediumNap = 20 * 60 * 1000; // 20 minutes
    private static final long DiscoveryShortNap = 10 * 60 * 1000; // 10 Minutes

    private static final int MaxNbOfConnectThreads = 5; // Number of parallele connection (threads)

    private EndpointAddress mkAddress(String destPeer) {
        try {
            PeerID asID = (PeerID) IDFactory.fromURL(new URL(destPeer));
            String asString = "jxta://" + asID.getUniqueValue().toString();

            EndpointAddress addr = endpoint.newEndpointAddress(asString);

            return addr;
        } catch (Exception e) {
            if (LOG.isEnabledFor(Priority.WARN)) LOG.warn("Invalid peerID string " + destPeer);
            return null;
        }
    }

    private EndpointAddress mkAddress(String destPeer,
                                      String serv, String parm) {

        EndpointAddress addr = mkAddress(destPeer);
        addr.setServiceName(serv);
        addr.setServiceParameter(parm);
        return addr;
    }

    public RendAddrCompactor(PeerGroup pg, Vector initialCompactorParams) {
        myGroup = pg;
        initialParams = initialCompactorParams;
        endpoint = myGroup.getEndpointService();
        localPeerId = myGroup.getPeerID().toString();
	rdvListener = new RdvListener (this);

        PeerGroup parent = ((RefPeerGroup) myGroup).getParentGroup();

        if (parent != null) {
            discovery = parent.getDiscoveryService();
        } else {
            discovery = myGroup.getDiscoveryService();
        }

        // Get the rdv service we're working for.
        rendezvous = myGroup.getRendezVousService();

        if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("Start compacting");
        myThread  = new Thread(this,"RendezvousService:RendezvousCompactor");
        myThread.start();
    }

    public void shutdown() {
        stopping = true;
        myThread.interrupt();
        // FIXME: [20011014 - [e-mail]]
        // for other thread we do not have a list so they'll have to terminate
        // when they notice the stopping flag.
    }

    public void run() {

        try {
            Vector params = (Vector) initialParams.clone();

            RdvMonitor monitor = new RdvMonitor(myGroup,
                                                (RendezVousServiceImpl) rendezvous);
            rendezvous.setMonitor(monitor);

            // If the group is either the Platform or direct descendant,
            // get the endpoints, because at this point, it is very likely that
            // this peer cannot find routes, since it is not connected to the
            // Network just yet.
            PeerGroup grandPa = null;
            PeerGroup parent = ((RefPeerGroup) myGroup).getParentGroup();
            if (parent != null) grandPa =
                                    ((RefPeerGroup) parent).getParentGroup();

            if (grandPa == null) {
                if (params != null) {
                    for(int i=0; i < params.size(); i++) {
                        String str = (String)params.elementAt(i);
                        EndpointAddress address =
                        endpoint.newEndpointAddress(str);
                        try {
                            rendezvous.connectToRendezVous(address);
                        } catch (Exception e) {
                            if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("Cannot connect to Rendezvous at " + str + ":" + e);
                        }
                    }
                }
                if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("Network peergroup Done compacting");
                return;
            }

            // This is an ordinary group. Search for a few advertisement of this
            // group and find rendezvous.
            if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("Regular group... Find rendezvous");
            String pgId = myGroup.getPeerGroupID().toString();

            // We will try to get as many Rendezvous advertisements as we can
            params = searchLocal(pgId);
            connect(params, MaxNbOfConnectThreads);

            // Now, until we have enough rendezvous, keep updating the group
            // advertisement, looking for new rendezvous, and try to connect to
            // them.
            Enumeration enum = null;
            Enumeration rdvsEnum = null;
            Vector rdvs = null;
            int count = 0;

            while (true) {
                if (stopping) return;

                rdvsEnum = rendezvous.getConnectedRendezVous();
                if ( !rdvsEnum.hasMoreElements() ) {
                    rdvs  = null;
                } else {
                    rdvs = new Vector();
                    while (rdvsEnum.hasMoreElements()) {
                        try {
                            rdvs.addElement(rdvsEnum.nextElement());
                        } catch (Exception e) {
                            break;
                        }
                    }
                }

                boolean connected = false;

                if ((rdvs != null) && (rdvs.size() > 0)) {
                    connected = true;
                }

		searchRemote(pgId);

                if (connected && (rdvs.size() >= MaxNbOfRdvs)) {
                    // We got enough.. Just sleep for now
                    try {
                        Thread.sleep(DiscoveryLongNap);
                    } catch (Exception e) {
                        if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("Exception " + e);
                        continue;
                    }
                    continue;
                } else {
		    // We are connected, but we do not enough rendezvous yet.
		    if (connected) {
			try {
			    Thread.sleep(DiscoveryMediumNap);
			} catch (Exception e) {
			    if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("Exception " + e);
			    continue;
			}
		    } else {
			// We are not connected. We need to find a rendezvous
			try {
			    Thread.sleep(DiscoveryShortNap);
			} catch (Exception e) {