proxyservice.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: ProxyService.java,v 1.13 2002/06/20 01:17:29 jice Exp $
 *
 */

package net.jxta.impl.proxy;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.net.URL;
import java.net.MalformedURLException;
import java.net.UnknownServiceException;
import java.util.*;

import net.jxta.platform.ModuleClassID;
import net.jxta.id.ID;
import net.jxta.id.IDFactory;
import net.jxta.peer.PeerID;
import net.jxta.peergroup.PeerGroup;
import net.jxta.peergroup.PeerGroupID;
import net.jxta.peergroup.PeerGroupFactory;
import net.jxta.exception.PeerGroupException;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.EndpointListener;
import net.jxta.endpoint.EndpointMessenger;
import net.jxta.endpoint.EndpointService;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.MessageElementEnumeration;
import net.jxta.document.MimeMediaType;
import net.jxta.document.Advertisement;
import net.jxta.document.AdvertisementFactory;
import net.jxta.discovery.DiscoveryListener;
import net.jxta.discovery.DiscoveryEvent;
import net.jxta.discovery.DiscoveryService;
import net.jxta.pipe.PipeService;
import net.jxta.pipe.InputPipe;
import net.jxta.pipe.OutputPipe;
import net.jxta.pipe.PipeMsgListener;
import net.jxta.pipe.OutputPipeListener;
import net.jxta.pipe.PipeMsgEvent;
import net.jxta.pipe.OutputPipeEvent;
import net.jxta.pipe.PipeID;
import net.jxta.protocol.DiscoveryResponseMsg;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.protocol.PeerGroupAdvertisement;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.service.Service;
import net.jxta.util.StringEnumeration;

import net.jxta.impl.endpoint.*;
import net.jxta.impl.util.Cache;
import net.jxta.impl.util.CacheEntry;
import net.jxta.impl.util.CacheEntryListener;

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

// FIXME: [e-mail] - 20020515
// All public methods are synchronized.
// None of them does anything blocking so that should be about OK, however
// first it is not 100% sure, second eventhough non-blocking, some of these
// operations could take a significant amount of time, which may be unfair
// to other threads that wish to enter for a quick operation.
// Making the locking finer-grain would require significant code rework, so
// it will have to do for now.

public class ProxyService implements Service,
				     EndpointListener,
				     PipeMsgListener,
				     OutputPipeListener,
				     CacheEntryListener
{

    private final static Category LOG = Category.getInstance(ProxyService.class.getName());

    protected final static boolean LOG_MESSAGES = false;

    public final static int DEFAULT_THRESHOLD = 2;
    public final static int DEFAULT_LIFETIME = 1000 * 60 * 30; // 30 minutes

    private PeerGroup group = null;
    private String serviceName = null;
    private String serviceParameter = null;
    private EndpointService endpoint = null;
    private DiscoveryService discovery = null;
    private PipeService pipe = null;
    private Advertisement implAdv = null;

    private Map searchRequests; // Currently unused
    private Map pipeListeners;
    private Cache pendingPipes;
    private Cache resolvedPipes;

    /************************************************************************
     *  Methods that are part of the Service Interface
     **********************************************************************/
    public Service getInterface() {
        return this;
    }

    public Advertisement getImplAdvertisement() {
        return implAdv;
    }

    public void init(PeerGroup group,
                     ID assignedID,
                     Advertisement implAdv) throws PeerGroupException {
        this.group = group;
        this.serviceName = assignedID.toString();
        this.implAdv = implAdv;

        serviceParameter = group.getPeerGroupID().toString();

        searchRequests = new TreeMap();
        pipeListeners = new TreeMap();

	// Pending pipes cost only memory, so it is not a problrm to
	// wait for the GC to cleanup things. No CacheEntryListener.
        pendingPipes = new Cache(200, null);

	// Resolved pipes cost non-memory resources, so we need to close
	// them as early as we forget them. Need a CacheEntryListener (this).
        resolvedPipes = new Cache(200, this);
    }

    public int startApp(String[] args) {
        if (LOG.isEnabledFor(Priority.INFO)) {
            LOG.info("addListener "+serviceName+serviceParameter);
        }

        endpoint = group.getEndpointService();
        discovery = group.getDiscoveryService();
        pipe = group.getPipeService();

        endpoint.addListener(serviceName+serviceParameter, this);

        return 0;
    }

    public void stopApp() {
        if (LOG.isEnabledFor(Priority.INFO)) {
            LOG.info("removeListener "+serviceName+serviceParameter);
        }

        endpoint.removeListener(serviceName+serviceParameter, this);
    }

    /************************************************************************
     *  Define the proxy message tags
     **********************************************************************/
    public static final String REQUEST_TAG = "proxy:request";
    public static final String RESPONSE_TAG = "proxy:response";

    public static final String REQUESTID_TAG = "proxy:requestId";
    public static final String TYPE_TAG = "proxy:type";
    public static final String NAME_TAG = "proxy:name";
    public static final String ID_TAG = "proxy:id";
    public static final String ARG_TAG = "proxy:arg";
    public static final String ATTRIBUTE_TAG = "proxy:attr";
    public static final String VALUE_TAG = "proxy:value";
    public static final String ERROR_MESSAGE_TAG = "proxy:error";

    /************************************************************************
     *  Define the proxy request types
     **********************************************************************/
    public static final String REQUEST_CREATE = "create";
    public static final String REQUEST_SEARCH = "search";
    public static final String REQUEST_LISTEN = "listen";
    public static final String REQUEST_CLOSE = "close";
    public static final String REQUEST_SEND = "send";

    /************************************************************************
     *  Define the proxy response types
     **********************************************************************/
    public static final String RESPONSE_SUCCESS = "success";
    public static final String RESPONSE_ERROR = "error";
    public static final String RESPONSE_INFO = "info";
    public static final String RESPONSE_RESULT = "result";
    public static final String RESPONSE_MESSAGE = "data";

    /************************************************************************
     *  Define the proxy type tags
     **********************************************************************/
    public static final String TYPE_PEER = "PEER";
    public static final String TYPE_GROUP = "GROUP";
    public static final String TYPE_PIPE = "PIPE";

    public synchronized void processIncomingMessage(Message message,         
				       EndpointAddress srcAddr, 
				       EndpointAddress dstAddr) {
        if (LOG_MESSAGES) {
            logMessage(message, LOG);
        }

        Requestor requestor = null;
        try {
            requestor = Requestor.createRequestor(group, message);
        } catch (IOException e) {
            if (LOG.isEnabledFor(Priority.WARN)) {
                LOG.warn("could not create requestor", e);
            }
        }
        
        String request = popString(REQUEST_TAG, message);
        if (LOG.isEnabledFor(Priority.DEBUG)) {
            LOG.debug("request = "+request + " object " + this);
        }
        if (request != null && requestor != null) {
            if (REQUEST_CREATE.equals(request)) {
                handleCreateRequest(requestor, 
                                    popString(TYPE_TAG, message), 
                                    popString(NAME_TAG, message), 
                                    popString(ID_TAG, message), 
                                    popString(ARG_TAG, message));
            } else if (REQUEST_SEARCH.equals(request)) {
                handleSearchRequest(requestor, 
                                    popString(TYPE_TAG, message), 
                                    popString(ATTRIBUTE_TAG, message), 
                                    popString(VALUE_TAG, message));
            } else if ("listen".equals(request)) {
                handleListenRequest(requestor, 
                                    popString(NAME_TAG, message), 
                                    popString(ID_TAG, message), 
                                    popString(ARG_TAG, message));
            } else if ("close".equals(request)) {
                handleCloseRequest(requestor, 
                                   popString(ID_TAG, message));
            } else if ("send".equals(request)) {
                handleSendRequest(requestor, 
                                  popString(NAME_TAG, message), 
                                  popString(ID_TAG, message), 
                                  popString(ARG_TAG, message), 
                                  message);
            }
        }
    }

    /**
     */
    private void handleCreateRequest(Requestor requestor, 
                                    String type, 
                                    String name, 
                                    String id, 
                                    String arg) {
        if (LOG.isEnabledFor(Priority.DEBUG)) {
            LOG.debug("handleCreateRequest type=" + type + 
                       " name=" + name +
                       " id=" + id +
                       " arg=" + arg);
        }

        if (name == null) {
            name = ""; // default name
        }

        if (TYPE_PEER.equals(type)) {
            PeerAdvertisement adv = createPeerAdvertisement(name, id);
            if (adv != null) {
                try {
                    discovery.publish(adv, DiscoveryService.PEER);
                } catch (Exception e) {
                    if (LOG.isEnabledFor(Priority.WARN)) {
                        LOG.warn("Could not publish peer advertisement");
                    }
                }

                requestor.send(adv, RESPONSE_SUCCESS);
            } else {
                requestor.notifyError("could not create advertisement");
            }
        } else if (TYPE_GROUP.equals(type)) {
            PeerGroupAdvertisement adv = createGroupAdvertisement(name, id);
            if (adv != null) {
                try {
                    discovery.publish(adv, DiscoveryService.GROUP);
                } catch (Exception e) {
                    if (LOG.isEnabledFor(Priority.WARN)) {
                        LOG.warn("Could not publish group advertisement");
                    }
                }

                requestor.send(adv, RESPONSE_SUCCESS);
            } else {
                requestor.notifyError("could not create advertisement");
            }
        } else if (TYPE_PIPE.equals(type)) {
            if (arg == null) {
                arg = PipeService.UnicastType; // default pipe type
            }

            PipeAdvertisement adv = createPipeAdvertisement(name, id, arg);
            if (adv != null) {
                try {
                    discovery.publish(adv, DiscoveryService.ADV);
                } catch (Exception e) {
                    if (LOG.isEnabledFor(Priority.WARN)) {
                        LOG.warn("Could not publish pipe advertisement");
                    }
                }

                requestor.send(adv, RESPONSE_SUCCESS);
            } else {
                requestor.notifyError("could not create advertisement");
            }
        } else {
            requestor.notifyError("unsupported type");
        }
    }

    /**
     */
    private void handleSearchRequest(Requestor requestor, 
                                    String type, 
                                    String attribute, 
                                    String value) {
        if (LOG.isEnabledFor(Priority.DEBUG)) {
            LOG.debug("handleSearchRequest type=" + type + 
                      " attribute=" + attribute +
                      " value=" + value);
        }

        int discoveryType;

        if (TYPE_PEER.equals(type)) {
            discoveryType = DiscoveryService.PEER;
        } else if (TYPE_GROUP.equals(type)) {