proxyservice.java

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

/*
 * Copyright (c) 2005 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.44 2006/08/31 17:04:47 hamada Exp $
 *
 */

package net.jxta.impl.proxy;

import java.io.IOException;
import java.io.StringReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.TreeMap;

import net.jxta.discovery.DiscoveryEvent;
import net.jxta.discovery.DiscoveryService;
import net.jxta.document.Advertisement;
import net.jxta.document.AdvertisementFactory;
import net.jxta.document.MimeMediaType;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.EndpointListener;
import net.jxta.endpoint.EndpointService;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.exception.PeerGroupException;
import net.jxta.id.ID;
import net.jxta.id.IDFactory;
import net.jxta.impl.util.Cache;
import net.jxta.impl.util.LRUCache;
import net.jxta.impl.util.CacheEntry;
import net.jxta.impl.util.CacheEntryListener;
import net.jxta.peer.PeerID;
import net.jxta.peergroup.PeerGroup;
import net.jxta.peergroup.PeerGroupID;
import net.jxta.pipe.InputPipe;
import net.jxta.pipe.OutputPipe;
import net.jxta.pipe.OutputPipeEvent;
import net.jxta.pipe.OutputPipeListener;
import net.jxta.pipe.PipeMsgEvent;
import net.jxta.pipe.PipeMsgListener;
import net.jxta.pipe.PipeService;
import net.jxta.protocol.DiscoveryResponseMsg;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.protocol.PeerGroupAdvertisement;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.protocol.ModuleImplAdvertisement;
import net.jxta.service.Service;

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

// 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 Logger LOG = Logger.getLogger(ProxyService.class.getName());

    protected final static boolean LOG_MESSAGES = true;

    public final static int DEFAULT_THRESHOLD = 2;
    public final static int DEFAULT_LIFETIME = 1000 * 60 * 30; // 30 minutes
    /************************************************************************
     *  Define the proxy message tags
     **********************************************************************/
    public static final String REQUEST_TAG = "request";
    public static final String RESPONSE_TAG = "response";

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

    /************************************************************************
     *  Define the proxy request types
     **********************************************************************/
    public static final String REQUEST_JOIN = "join";
    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";

    private PeerGroup group = null;
    private ID assignedID = 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 LRUCache searchRequests; // Currently unused
    private Map pipeListeners;
    private Cache pendingPipes;
    private Cache resolvedPipes;

    private static Map proxiedGroups = new HashMap(16);
    private static Map passwords = new HashMap(16);

    /************************************************************************
     *  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.assignedID = assignedID;
        this.serviceName = assignedID.toString();
        this.implAdv = implAdv;

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

        searchRequests = new LRUCache(25);
        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) {

        Service needed = group.getEndpointService();

        if(null == needed) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Stalled until there is a endpoint service");
            }

            return START_AGAIN_STALLED;
        }

        needed = group.getDiscoveryService();

        if(null == needed) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Stalled until there is a discovery service");
            }

            return START_AGAIN_STALLED;
        }

        needed = group.getPipeService();

        if(null == needed) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Stalled until there is a pipe service");
            }

            return START_AGAIN_STALLED;
        }

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

        if (LOG.isEnabledFor(Level.INFO)) {
            LOG.info("addListener "+serviceName+serviceParameter);
        }

        endpoint.addIncomingMessageListener(this, serviceName, serviceParameter);

        return 0;
    }

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

        endpoint.removeIncomingMessageListener(serviceName, serviceParameter);
    }


    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, srcAddr);
            // Commented out the above line and added the following three lines.
            // The change allows to reduce the traffice going to a JXME peer
            // by able to remove ERM completly. As a side effect (severe one)
            // JXTA Proxy and JXTA relay need to be running on the same peer.
            // This changes should be pulled out as soon as ERM is implemented
            // in a more inteligent and effective way so that it doesn't
            // have any impact on JXME peers.
            EndpointAddress relayAddr = new EndpointAddress( "relay",
                    srcAddr.getProtocolAddress(),
                    srcAddr.getServiceName(),
                    srcAddr.getServiceParameter() );

            requestor = Requestor.createRequestor(group, message, relayAddr, 0);
        } catch (IOException e) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("could not create requestor", e);
            }
        }

        String request = popString(REQUEST_TAG, message);
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("request = "+request + " requestor " + requestor);
        }

        if (request != null && requestor != null) {
            if (REQUEST_JOIN.equals(request)) {
                handleJoinRequest(requestor,
                                  popString(ID_TAG, message),
                                  popString(ARG_TAG, message));
            } else 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),
                                    popString(THRESHOLD_TAG, message));
            } else if ("listen".equals(request)) {
                handleListenRequest(requestor,
                                    popString(ID_TAG, message));
            } else if ("close".equals(request)) {
                handleCloseRequest(requestor,
                                   popString(ID_TAG, message));
            } else if ("send".equals(request)) {
                handleSendRequest(requestor,
                                  popString(ID_TAG, message),
                                  message);
            }
        }
    }

    // Right now there's a security hole: passwd come in clear.
    // And not much is done for stopping clients to use the new group
    // without being authenticated. We also never get rid of these
    // additional groups.
    private synchronized void handleJoinRequest(Requestor requestor,
            String grpId,
            String passwd) {

        PeerGroup g = (PeerGroup) proxiedGroups.get(grpId);

        if (g != null) {
            if (g == this.group) {
                requestor.notifyError("Same group");
            } else if (! ((String) passwords.get(grpId)).equals(passwd)) {
                requestor.notifyError("Incorrect password");
            } else {
                requestor.notifySuccess();
            }
            return;
        }

        try {
            g = group.newGroup((PeerGroupID) IDFactory.fromURI(new URI(grpId)));
            g.getRendezVousService().startRendezVous();
        } catch (Exception ge) {
            requestor.notifyError(ge.getMessage());
            return;
        }

        // XXX check membership here. (would work only for single passwd grps)
        // For now, assume join is always welcome.

        // So far so good. Try to start a proxy in that grp.
        try {
            // Fork this proxy into the new grp.
            ProxyService p = new ProxyService();
            p.init(g, assignedID, implAdv);
            p.startApp(null);
        } catch(Exception e) {
            requestor.notifyError(e.getMessage());
            return;
        }
        // set non-deft passwd
        passwords.put(grpId, passwd);
        proxiedGroups.put(grpId, g);
        requestor.notifySuccess();
    }

    /**
     */
    private void handleCreateRequest(Requestor requestor,
                                     String type,
                                     String name,
                                     String id,
                                     String arg) {

        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("handleCreateRequest type=" + type +
                      " name=" + name +
                      " id=" + id +
                      " arg=" + arg);
        }

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