endpointdemuxlistener.java

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

/*
 *  $Id: EndpointDemuxListener.java,v 1.23 2002/06/20 19:33:11 jice 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 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.
 */
package net.jxta.impl.endpoint;

import net.jxta.endpoint.EndpointListener;
import net.jxta.endpoint.Message;
import net.jxta.impl.util.ResourceDispatcher;
import net.jxta.impl.util.ResourceAccount;
import net.jxta.impl.util.Cache;
import net.jxta.impl.util.CacheEntry;
import net.jxta.impl.util.CacheEntryListener;

import java.util.LinkedList;
import java.util.Iterator;

import org.apache.log4j.Category;
import org.apache.log4j.Priority;
import net.jxta.endpoint.EndpointAddress;

// NOTE: [e-mail] - 20020526
// There would be great value in making such an object the explicit interface
// between the endpoint and its clients, rather than a bland listener interface
// The client would have the ability to specify the threads limit, possibly
// setting it zero, and then could have access to the buffer.
// To implement that with a simple listener would mean that the endpoint
// has to TRUST the client that the listener does no-more than queuing and
// signaling or else, force a full and possibly redundant hand-shake in all
// cases, as is the case now.
// To be improved.

public class EndpointDemuxListener implements EndpointListener {

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

    // All EndpointDemuxListeners share one global resource
    // manager for threads and message queing. Its budget is hardcoded
    // for now.
    
    // Parameters read as follows: 100*1 thread authorizations are
    // set aside for granting reserved threads at account creation.
    // 100 additional threads may be authorized dynamicaly.
    // Each account may ask between 1 and 3 reserved threads.
    // At least one will always be granted as long as there are
    // less than 100 accounts. The other reservations are pulled from
    // whatever of the 100 extras is available at the time.
    // Each account can obtain its initialy reserved threads plus whatever
    // of the 100 extra is available at the time of allocation.
    // Any single account cannot allocate more than 6 of the extra
    // threads on top of its reserved threads. True means that round-robin
    // is in effect, so if a thread cannot be obtained at some point, it
    // may be granted asynchronously by reassigning one that becomes
    // available.

    static private ResourceDispatcher threadDispatcher =
	new ResourceDispatcher(100, 1, 3, 100, 50, true);


    // All next hop peers that send us messages share one global resource
    // manager message queing. Its budget is hardcoded for now.

    // Parameters read as follows: 200 * 1 message authorization are set
    // aside for granting reserved messages at account creation.
    // 400 additional messages may be authorized dynamically.
    // Each account may ask between 1 and 5 reserved messages.
    // At least 1 will always be granted as long as there are less than
    // 200 accounts. The other reservations are pulled from whatever of the
    // 400 extra is available at the time.
    // Each account can obtain its initialy reserved messages plus
    // whatever is available of the 400 extra at the time of allocation.
    // Any single account cannot allocate more than 100 of the extra
    // messages on top of its reserved messages. False means that
    // round-robin is NOT in effect, so if a message cannot be obtained
    // at some point, that's it, it will not be granted asynchronously. The
    // request is forgotten. The reason is, the message buffer is already
    // allocated; what is granted is the authorization to queue it. So
    // waiting is not an option, by definition. Note that the message
    // is discounted as soon as dequeued. We only manage resources occupied
    // by idle messages.
    // We used to count messages; now we count bytes. The figures are about
    // the same, given that we assume that most messages are 10K long or less.
    // The important side effect is that, with these numbers, it is no-longer
    // possible to send a message bigger than about 1 MB, since that is about
    // the maximum a single account can hold at any given time.

    static private ResourceDispatcher messageDispatcher =
	new ResourceDispatcher(200, 1*10240, 5*10240, 400*10240,
			       100*10240, false);

    // A canonical mapping of all the message originators.
    // This is a cache, since peers can disappear silently.
    // The number 100 is the maximum number of idle accounts that
    // we keep around in case the peer comes back.
    // Note: a number of accounts might not be the best criterion though.
    // since just a few stale accounts could hog a lot of resources.
    // May be we should just make that cache sensitive to the inconvenience
    // endured by live accounts.

    static class MyCacheListener implements CacheEntryListener {
	// This may only be called when something gets added to the
	// cache or when an item is made purgeable. In our case that means
	// all the synchro we need is already there.
	public void purged(CacheEntry entry) {
	    ((ResourceAccount) entry.getValue()).close();
	}
    }

    // We put a large hard limit on the cache because we detect the
    // need for purging normaly before that limit is reached, and we purge
    // it explicitly.
    static private Cache allSources = new Cache(100, new MyCacheListener());

    private LinkedList messageQueue = new LinkedList();
    private boolean keepGoing = true;
    private EndpointListener listener = null;
    private int nbOfQueuedMessages = 0;
    private int nbOfDeQueuedMessages = 0;
    private String name = null;

    private ResourceAccount myAccount = null;


    /**
     *  
     */
    private void enQueue(MessageFromSource m) {
	messageQueue.addLast(m);
	++nbOfQueuedMessages; // For statistics. Not used currently.
    }

    /**
     * Must be synchronized to call this.
     */
    private MessageFromSource deQueue() {
	if ((!keepGoing) || (messageQueue.size() == 0)) return null;
	Object o = messageQueue.removeFirst();
	++nbOfDeQueuedMessages; // For statistics. Not used currently.
	return (MessageFromSource) o;
    }

    /**
     *  Gets the listener attribute of the EndpointDemuxListener object
     *
     * @return    The listener value
     */
    public EndpointListener getListener() {
	return listener;
    }
    
    /**
     * Constructor for the EndpointDemuxListener object
     *
     * @param  listener  Description of the Parameter
     */
    public EndpointDemuxListener(EndpointListener listener) {
	this("Unknown", listener);
    }

    /**
     *Constructor for the EndpointDemuxListener object
     *
     * @param  name      Description of the Parameter
     * @param  listener  Description of the Parameter
     */
    public EndpointDemuxListener(String name, EndpointListener listener)
    {