proxyservice.java

JXTA&#8482 is a set of open, generalized peer-to-peer (P2P) protocols that allow any networked devi

/*
 * Copyright (c) 2005-2007 Sun Microsystems, Inc.  All rights reserved.
 *  
 *  The Sun Project JXTA(TM) Software License
 *  
 *  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 Sun Microsystems, Inc. for JXTA(TM) technology." 
 *     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.
 *  
 *  JXTA is a registered trademark of Sun Microsystems, Inc. in the United 
 *  States and other countries.
 *  
 *  Please see the license information page at :
 *  <http://www.jxta.org/project/www/license.html> for instructions on use of 
 *  the license in source files.
 *  
 *  ====================================================================
 *  
 *  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.proxy;

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.CacheEntry;
import net.jxta.impl.util.CacheEntryListener;
import net.jxta.impl.util.LRUCache;
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.ModuleImplAdvertisement;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.protocol.PeerGroupAdvertisement;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.service.Service;
import java.util.logging.Level;
import net.jxta.logging.Logging;
import java.util.logging.Logger;

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.platform.Module;


// 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());

    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 ModuleImplAdvertisement implAdvertisement = null;

    private final LRUCache<Integer, Requestor> searchRequests = new LRUCache<Integer, Requestor>(25); // Currently unused
    private final Map<String, PipeListenerList> pipeListeners = new TreeMap<String, PipeListenerList>();

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

    private static Map<String, PeerGroup> proxiedGroups = new HashMap<String, PeerGroup>(16);
    private static Map<String, String> passwords = new HashMap<String, String>(16);

    /**
     * {@inheritDoc}
     */
    public void init(PeerGroup group, ID assignedID, Advertisement implAdv) throws PeerGroupException {
        this.group = group;
        this.assignedID = assignedID;
        this.serviceName = assignedID.toString();
        this.implAdvertisement = (ModuleImplAdvertisement) implAdv;

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

        // 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);

        if (Logging.SHOW_CONFIG && LOG.isLoggable(Level.CONFIG)) {
            StringBuilder configInfo = new StringBuilder("Configuring JXME Proxy Service : " + assignedID);

            configInfo.append("\n\tImplementation :");
            configInfo.append("\n\t\tModule Spec ID: ").append(implAdvertisement.getModuleSpecID());
            configInfo.append("\n\t\tImpl Description : ").append(implAdvertisement.getDescription());
            configInfo.append("\n\t\tImpl URI : ").append(implAdvertisement.getUri());
            configInfo.append("\n\t\tImpl Code : ").append(implAdvertisement.getCode());
            configInfo.append("\n\tGroup Params :");
            configInfo.append("\n\t\tGroup : ").append(group.getPeerGroupName());
            configInfo.append("\n\t\tGroup ID : ").append(group.getPeerGroupID());
            configInfo.append("\n\t\tPeer ID : ").append(group.getPeerID());
            LOG.config(configInfo.toString());
        }
    }

    /**
     * {@inheritDoc}
     */
    public int startApp(String[] args) {

        Service needed = group.getEndpointService();

        if (null == needed) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("Stalled until there is a endpoint service");
            }

            return START_AGAIN_STALLED;
        }

        needed = group.getDiscoveryService();
        if (null == needed) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("Stalled until there is a discovery service");
            }

            return START_AGAIN_STALLED;
        }

        needed = group.getPipeService();
        if (null == needed) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("Stalled until there is a pipe service");
            }

            return START_AGAIN_STALLED;
        }

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

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("addListener " + serviceName + serviceParameter);
        }

        endpoint.addIncomingMessageListener(this, serviceName, serviceParameter);
        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            LOG.info("JXME Proxy Service started.");
        }
        return Module.START_OK;
    }

    /**
     * {@inheritDoc}
     */
    public void stopApp() {
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("removeListener " + serviceName + serviceParameter);
        }

        endpoint.removeIncomingMessageListener(serviceName, serviceParameter);

        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            LOG.info("JXME Proxy Service stopped.");
        }
    }

    /**
     * {@inheritDoc}
     */
    public ModuleImplAdvertisement getImplAdvertisement() {
        return implAdvertisement;
    }

    /**
     * {@inheritDoc}
     */
    public ProxyService getInterface() {
        return this;
    }

    /**
     * {@inheritDoc}
     */
    public synchronized void processIncomingMessage(Message message, EndpointAddress srcAddr, EndpointAddress dstAddr) {

        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 (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "could not create requestor", e);
            }
        }

        String request = popString(REQUEST_TAG, message);

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("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 = proxiedGroups.get(grpId);

        if (g != null) {
            if (g == this.group) {
                requestor.notifyError("Same group");
            } else if (!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 proxyService = new ProxyService();
            proxyService.init(g, assignedID, implAdvertisement);
            proxyService.startApp(null);
        } catch (Exception e) {
            requestor.notifyError(e.getMessage());
            return;