piperesolver.java

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

/*
 *  $Id: PipeResolver.java,v 1.62 2002/02/02 01:18:14 hamada Exp $
 *
 *  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 THE APACHE SOFTWARE FOUNDATION 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.
 *
 */
package net.jxta.impl.pipe;

import java.io.ByteArrayInputStream;
import java.io.StringWriter;
import java.io.IOException;
import java.util.Vector;
import java.util.Enumeration;
import java.util.Hashtable;
import java.lang.reflect.InvocationTargetException;

import net.jxta.pipe.InputPipe;
import net.jxta.pipe.PipeService;
import net.jxta.exception.NoResponseException;
import net.jxta.exception.DiscardQueryException;
import net.jxta.exception.ResendQueryException;
import net.jxta.discovery.DiscoveryService;
import net.jxta.endpoint.EndpointService;
import net.jxta.document.Element;
import net.jxta.document.Advertisement;
import net.jxta.document.AdvertisementFactory;
import net.jxta.document.MimeMediaType;
import net.jxta.document.StructuredDocument;
import net.jxta.document.TextElement;
import net.jxta.document.StructuredDocumentFactory;
import net.jxta.document.StructuredTextDocument;
import net.jxta.protocol.ResolverResponseMsg;
import net.jxta.protocol.ResolverQueryMsg;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.resolver.QueryHandler;
import net.jxta.resolver.ResolverService;
import net.jxta.peergroup.PeerGroup;
import net.jxta.impl.protocol.ResolverQuery;
import net.jxta.impl.protocol.ResolverResponse;
import net.jxta.impl.peergroup.RefPeerGroup;
import net.jxta.impl.pipe.PipeResolver.PipeEntry;
import net.jxta.impl.util.JxtaTimerHandler;
import net.jxta.impl.util.JxtaTimer;

import org.apache.log4j.Category; import org.apache.log4j.Priority;
/**
 * This class implements the PipeServiceImpl ResolverServiceImpl.
 *
 */

public class PipeResolver implements JxtaTimerHandler, QueryHandler {
    private final static Category LOG = Category.getInstance(PipeResolver.class.getName());
    
    private final static String PipeResolverName = "JxtaPipeResolver";
    private final static String MsgTypeTag       = "MsgType";
    private final static String PipeIdTag        = "PipeId";
    private final static String TypeTag        = "Type";
    private final static String PeerIdTag        = "Peer";
    private final static String CachedTag        = "Cached";
    private final static String PeerAdvTag       = "PeerAdv";
    private final static String FoundTag         = "Found";
    private final static String QueryMsgType     = "Query";
    private final static String AnswerMsgType    = "Answer";
    
    private static long DefaultPipeLease     = 60 * 60 * 1000;  // 1 Hour
    private static long GcDelay              = 10 * 60 * 1000; // 10 minutes
    private static int qid = 0;
    
    private ResolverService     resolver = null;
    private Hashtable       pipes = null;
    private Hashtable cachedPipes = null;
    private Vector     localAddrs = null;
    private RefPeerGroup  myGroup = null;
    private EndpointService     endpoint = null;
    private DiscoveryService   discovery = null;
    private String    localPeerId = null;
    private JxtaTimer     gcTimer = null;
    private Hashtable   listeners = new Hashtable();
    
    
    /**
     * Constructor for the PipeResolver object
     *
     * @param  g  group for which this PipeResolver operates in
     * @since 1.0
     */
    public PipeResolver(PeerGroup g) {
        
        myGroup        = (RefPeerGroup) g;
        localPeerId    = myGroup.getPeerID().toString();
        endpoint       = myGroup.getEndpointService();
        pipes          = new Hashtable();
        cachedPipes    = new Hashtable();
        resolver       = g.getResolverService();
        discovery      = myGroup.getDiscoveryService();
        
        // Register to the Generic ResolverServiceImpl
        resolver.registerHandler(PipeResolverName, this);
        
        // Create our garbage collection timer
        gcTimer = new JxtaTimer(this, GcDelay, true);
    }
    
    
    protected void stop() {
        
        resolver.unregisterHandler(PipeResolverName);
        
    }
    /**
     *Description of the Method
     *
     * @param  ipId  Description of Parameter
     * @param  peer  Description of Parameter
     * @since 1.0
     */
    public void registerCached(String ipId, String peer) {
        
        if ( LOG.isDebugEnabled() ) {
            if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("registerCached:");
            if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("      Id = " + ipId);
            if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("    peer = " + peer);
        }
        
        PipeEntry entry = null;
        if (cachedPipes.containsKey(ipId)) {
            // There already is an entry for this pipe.
            // Reset the lease on the entry
            entry = (PipeEntry) cachedPipes.get(ipId);
            if (entry != null) {
                // That should be the case;
                // Update the peer information and the lease.
                entry.peer = peer;
                resetLease(entry);
            } else {
                if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("registerCached. unexpected null pointer");
            }
            return;
        }
        
        entry = new PipeEntry(peer);
        resetLease(entry);
        try {
            Object obj = cachedPipes.put(ipId, entry);
        } catch (Exception e) {
            // XXX: is there something we should do here ?
            if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("registerCached failed " + e);
            return;
        }
        notifyWaitingThreads();
    }
    
    
    /**
     *Description of the Method
     *
     * @param  ipId  Description of Parameter
     * @param  ip    Description of Parameter
     * @since
     */
    public void register(String ipId, InputPipe ip) {
        
        if ( LOG.isDebugEnabled() ) {
            if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("register " + ipId);
        }
        
        try {
            Object obj = pipes.put(ipId, ip);
        } catch (Exception e) {
            // XXX: is there something we should do here ?
            return;
        }
        notifyWaitingThreads();
    }
    
    
    /**
     *Description of the Method
     *
     * @param  ipId  Description of Parameter
     * @since 1.0
     */
    public void forget(String ipId) {
        if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("forget " + ipId);
        Object obj = pipes.remove(ipId);
    }
    
    
    /**
     *Description of the Method
     *
     * @param  ipId  Description of Parameter
     * @since 1.0
     */
    public void forgetCached(String ipId) {
        if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("forgetCached " + ipId);
        Object obj = cachedPipes.remove(ipId);
    }
    
    
    /**
     *Description of the Method
     *
     * @param  ipId  Description of Parameter
     * @return       Description of the Returned Value
     * @since 1.0
     */
    public String findLocal(String ipId) {
        
        if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("findLocal " + ipId);
        String peer = null;
        
        // First look if the pipe is a local InputPipe
        Object obj = pipes.get(ipId);
        
        if (obj != null) {
            // Found it. Return the local PeerId
            if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("findLocal: found local InputPipe for " + ipId);
            return localPeerId;
        }
        
        // Look into our cached set of addresses
        peer = findCached(ipId);
        return peer;
    }
    
    

    /**
     * nocache is set to true if we want answer only from the peer that
     * has the inputpipe. If nocache is set to false, peers that have
     * cached information about that pipe can answer.
     *
     * @param  ipId     Description of Parameter
     * @param  timeout  Description of Parameter
     * @param  nocache  Description of Parameter
     * @return          Description of the Returned Value
     * @since 1.0
     */
    public String find(PipeAdvertisement pAdv, long timeout, boolean nocache) {
        
        String ipId = pAdv.getPipeID().toString();
        
        if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("find " + ipId);
        // First look into the local various tables to see
        // if we know this pipe.
        
        String peer = findLocal(ipId);
        
        if (peer == null) {
            Enumeration enum = null;
            // We don't know about this pipe: send a query
            // Finding a pipe remotely is asynchronos: try again later
            
            peer = findRemote(pAdv, timeout, nocache);
            return peer;
        } else {
            // call any listener for ipid
            // [e-mail]
            // Right now, we force the type to be UnicastType. I am not sure, however,
            // that the type of the pipe is really relevant to the event listener, since
            // the type is now embedded within the pipe advertisement anyway.
            // The query id is also random, since no query has been sent.
            // This probably will have to be revisited.
            
            if ( listeners.containsKey(ipId) ) {
                Event newevent = new Event(this,
                peer,
                ipId,
                PipeService.UnicastType,
                0 );
                Listener pl = (Listener) listeners.get(ipId);
                pl.pipeResolverEvent(newevent);
            }
            return peer;
        }
    }
    
    /**
     * Send a unicast query to verify that the InputPipe is still
     * where it's supposed to be.
     * @param  peer  destination peer
     * @param  ipId  pipe id
     * @param  type  type of the pipe
     * @since 1.0
     */
    public void sendVerify(String peer,
    String ipId,
    String type) {
        
        if ( LOG.isDebugEnabled() ) {
            if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("sendVerify:");
            if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("    pipe= " + ipId);
            if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("    peer= " + peer);
        }
        
        try {
            // First create the request message.
            StructuredTextDocument doc = (StructuredTextDocument)
            StructuredDocumentFactory.newStructuredDocument(
            new MimeMediaType("text/xml"),
            "jxta:PipeResolver");
            
            Element e = null;
            e = doc.createElement(MsgTypeTag, QueryMsgType);
            doc.appendChild(e);
            e = doc.createElement(PipeIdTag, ipId);
            doc.appendChild(e);
            e = doc.createElement(TypeTag, type);