communicatorserver.java

Mobile 应用程序使用 Java Micro Edition (Java ME) 平台

/*
 * @(#)file      CommunicatorServer.java
 * @(#)author    Sun Microsystems, Inc.
 * @(#)version   1.60
 * @(#)lastedit      05/11/17
 *
 * Copyright 2006 Sun Microsystems, Inc. All rights reserved.
 * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 *
 */


package com.sun.jmx.snmp.daemon;



// java import
//
import java.io.ObjectInputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.util.Vector;
import java.util.Enumeration;

// jmx import
//
import javax.management.MBeanServer;
import javax.management.MBeanRegistration;
import javax.management.ObjectName;
import javax.management.NotificationListener;
import javax.management.NotificationFilter;
import javax.management.NotificationBroadcaster;
import javax.management.NotificationBroadcasterSupport;
import javax.management.MBeanNotificationInfo;
import javax.management.AttributeChangeNotification;
import javax.management.ListenerNotFoundException;
import javax.management.loading.ClassLoaderRepository;
import javax.management.MBeanServerFactory;

// jmx RI import
//
import com.sun.jmx.trace.Trace;
import java.util.NoSuchElementException;

// JSR 160 import
//
// XXX Revisit:
//   used to import com.sun.jmx.snmp.MBeanServerForwarder
// Now using JSR 160 instead. => this is an additional
// dependency to JSR 160.
//
import javax.management.remote.MBeanServerForwarder;

/**
 * Defines generic behavior for the server part of a connector or an adaptor.
 * Most connectors or adaptors extend <CODE>CommunicatorServer</CODE>
 * and inherit this behavior. Connectors or adaptors that do not fit into 
 * this model do not extend <CODE>CommunicatorServer</CODE>.
 * <p>
 * A <CODE>CommunicatorServer</CODE> is an active object, it listens for 
 * client requests  and processes them in its own thread. When necessary, a 
 * <CODE>CommunicatorServer</CODE> creates other threads to process multiple 
 * requests concurrently.
 * <p>
 * A <CODE>CommunicatorServer</CODE> object can be stopped by calling the 
 * <CODE>stop</CODE> method. When it is stopped, the 
 * <CODE>CommunicatorServer</CODE> no longer listens to client requests and 
 * no longer holds any thread or communication resources.
 * It can be started again by calling the <CODE>start</CODE> method.
 * <p>
 * A <CODE>CommunicatorServer</CODE> has a <CODE>State</CODE> attribute 
 * which reflects its  activity.
 * <p>
 * <TABLE>
 * <TR><TH>CommunicatorServer</TH>      <TH>State</TH></TR>
 * <TR><TD><CODE>stopped</CODE></TD>    <TD><CODE>OFFLINE</CODE></TD></TR>
 * <TR><TD><CODE>starting</CODE></TD>    <TD><CODE>STARTING</CODE></TD></TR>
 * <TR><TD><CODE>running</CODE></TD>     <TD><CODE>ONLINE</CODE></TD></TR>
 * <TR><TD><CODE>stopping</CODE></TD>     <TD><CODE>STOPPING</CODE></TD></TR>
 * </TABLE>
 * <p>
 * The <CODE>STARTING</CODE> state marks the transition 
 * from <CODE>OFFLINE</CODE> to <CODE>ONLINE</CODE>.
 * <p>
 * The <CODE>STOPPING</CODE> state marks the transition from 
 * <CODE>ONLINE</CODE> to <CODE>OFFLINE</CODE>. This occurs when the 
 * <CODE>CommunicatorServer</CODE> is finishing or interrupting active 
 * requests.
 * <p>
 * When a <CODE>CommunicatorServer</CODE> is unregistered from the MBeanServer,
 * it is stopped automatically.
 * <p>
 * When the value of the <CODE>State</CODE> attribute changes the 
 * <CODE>CommunicatorServer</CODE> sends a
 * <tt>{@link javax.management.AttributeChangeNotification}</tt> to the 
 * registered listeners, if any.
 *
 * <p><b>This API is a Sun Microsystems internal API  and is subject 
 * to change without notice.</b></p>
 * @version     1.60     11/17/05
 * @author      Sun Microsystems, Inc
 */

public abstract class CommunicatorServer 
    implements Runnable, MBeanRegistration, NotificationBroadcaster, 
	       CommunicatorServerMBean {

    //
    // States of a CommunicatorServer
    //

    /**
     * Represents an <CODE>ONLINE</CODE> state.
     */
    public static final int ONLINE = 0 ;

    /**
     * Represents an <CODE>OFFLINE</CODE> state.
     */
    public static final int OFFLINE = 1 ;

    /**
     * Represents a <CODE>STOPPING</CODE> state.
     */
    public static final int STOPPING = 2 ;

    /**
     * Represents a <CODE>STARTING</CODE> state.
     */
    public static final int STARTING = 3 ;

    //
    // Types of connectors.
    //

    /**
     * Indicates that it is an RMI connector type.
     */
    //public static final int RMI_TYPE = 1 ;

    /**
     * Indicates that it is an HTTP connector type.
     */
    //public static final int HTTP_TYPE = 2 ;

    /**
     * Indicates that it is an HTML connector type.
     */
    //public static final int HTML_TYPE = 3 ;

    /**
     * Indicates that it is an SNMP connector type.
     */
    public static final int SNMP_TYPE = 4 ;

    /**
     * Indicates that it is an HTTPS connector type.
     */
    //public static final int HTTPS_TYPE = 5 ;

    //
    // Package variables
    //

    /**
     * The state of the connector server.
     */
     transient volatile int state = OFFLINE ;

    /**
     * The object name of the connector server.
     * @serial
     */
    ObjectName objectName ;

    MBeanServer topMBS;
    MBeanServer bottomMBS;

    /**
     */
    transient String dbgTag = null ;

    /**
     * The maximum number of clients that the CommunicatorServer can 
     * process concurrently.
     * @serial 
     */
    int maxActiveClientCount = 1 ;

    /**
     */
    transient int servedClientCount = 0 ;

    /**
     * The host name used by this CommunicatorServer.
     * @serial
     */
    String host = null ;

    /**
     * The port number used by this CommunicatorServer.
     * @serial
     */
    int port = -1 ;


    //
    // Private fields
    //

    /* This object controls access to the "state" and "interrupted" variables.
       If held at the same time as the lock on "this", the "this" lock must
       be taken first.  */
    private transient Object stateLock = new Object();

    private transient Vector<ClientHandler> 
            clientHandlerVector = new Vector<ClientHandler>() ;

    private transient Thread fatherThread = Thread.currentThread() ;
    private transient Thread mainThread = null ;

    private volatile boolean stopRequested = false ;
    private boolean interrupted = false;
    private transient Exception startException = null;

    // Notifs count, broadcaster and info
    private transient long notifCount = 0;
    private transient NotificationBroadcasterSupport notifBroadcaster = 
	new NotificationBroadcasterSupport();
    private transient MBeanNotificationInfo[] notifInfos = null;
    

    /**
     * Instantiates a <CODE>CommunicatorServer</CODE>.
     *
     * @param connectorType Indicates the connector type. Possible values are:
     * SNMP_TYPE.
     *
     * @exception <CODE>java.lang.IllegalArgumentException</CODE> 
     *            This connector type is not correct.
     */  
    public CommunicatorServer(int connectorType) 
	throws IllegalArgumentException {
        switch (connectorType) {
        case SNMP_TYPE :
            infoType = Trace.INFO_ADAPTOR_SNMP ;
            break;
        default:
            throw new IllegalArgumentException("Invalid connector Type") ;
        }
        dbgTag = makeDebugTag() ;
    }

    protected Thread createMainThread() {
	return new Thread (this, makeThreadName());
    }

    /**
     * Starts this <CODE>CommunicatorServer</CODE>.
     * <p>
     * Has no effect if this <CODE>CommunicatorServer</CODE> is 
     * <CODE>ONLINE</CODE> or <CODE>STOPPING</CODE>.
     * @param timeout Time in ms to wait for the connector to start.
     *        If <code>timeout</code> is positive, wait for at most
     *        the specified time. An infinite timeout can be specified
     *        by passing a <code>timeout</code> value equals 
     *        <code>Long.MAX_VALUE</code>. In that case the method
     *        will wait until the connector starts or fails to start. 
     *        If timeout is negative or zero, returns as soon as possible 
     *        without waiting.
     * @exception CommunicationException if the connectors fails to start.
     * @exception InterruptedException if the thread is interrupted or the
     *            timeout expires.
     */
    public void start(long timeout) 
	throws CommunicationException, InterruptedException {
	boolean start;

	synchronized (stateLock) {
	    if (state == STOPPING) { 
		// Fix for bug 4352451: 
		//     "java.net.BindException: Address in use".
		waitState(OFFLINE, 60000);
	    }
	    start = (state == OFFLINE);
	    if (start) {
		changeState(STARTING);
		stopRequested = false;
		interrupted = false;
		startException = null;
	    }
	}

	if (!start) {
            if (isTraceOn())
                trace("start","Connector is not OFFLINE") ;
	    return;
	}

	if (isTraceOn())
	    trace("start","--> Start connector ") ;

	mainThread = createMainThread();

	mainThread.start() ;
	
	if (timeout > 0) waitForStart(timeout);
    }
    
    /**
     * Starts this <CODE>CommunicatorServer</CODE>.
     * <p>
     * Has no effect if this <CODE>CommunicatorServer</CODE> is 
     * <CODE>ONLINE</CODE> or <CODE>STOPPING</CODE>.
     */
    public void start() {
	try {
	    start(0);
	} catch (InterruptedException x) {
	    // can not happen because of `0'
	    trace("start","interrupted: " + x);
	}
    }
    
    /**
     * Stops this <CODE>CommunicatorServer</CODE>.
     * <p> 
     * Has no effect if this <CODE>CommunicatorServer</CODE> is 
     * <CODE>OFFLINE</CODE> or  <CODE>STOPPING</CODE>.
     */
    public void stop() {
	synchronized (stateLock) {
	    if (state == OFFLINE || state == STOPPING) {
		if (isTraceOn())
		    trace("stop","Connector is not ONLINE") ;
		return;
	    }
	    changeState(STOPPING);
	    //
	    // Stop the connector thread
	    //
	    if (isTraceOn())
		trace("stop","Interrupt main thread") ;
	    stopRequested = true ;
	    if (!interrupted) {		    
		interrupted = true;
		mainThread.interrupt();
	    }
	}

	//
	// Call terminate on each active client handler
	//
	if (isTraceOn()) {
	    trace("stop","terminateAllClient") ;
	}
	terminateAllClient() ;
	
	// ----------------------
	// changeState
	// ----------------------
	synchronized (stateLock) {
	    if (state == STARTING)
		changeState(OFFLINE);
	}
    }

    /**
     * Tests whether the <CODE>CommunicatorServer</CODE> is active.
     *
     * @return True if connector is <CODE>ONLINE</CODE>; false otherwise.
     */
    public boolean isActive() {
	synchronized (stateLock) {
	    return (state == ONLINE);
	}
    }

    /**
     * <p>Waits until either the State attribute of this MBean equals the 
     * specified <VAR>wantedState</VAR> parameter, 
     * or the specified  <VAR>timeOut</VAR> has elapsed. 
     * The method <CODE>waitState</CODE> returns with a boolean value 
     * indicating whether the specified <VAR>wantedState</VAR> parameter 
     * equals the value of this MBean's State attribute at the time the method 
     * terminates.</p>
     *
     * <p>Two special cases for the <VAR>timeOut</VAR> parameter value are:</p>
     * <UL><LI> if <VAR>timeOut</VAR> is negative then <CODE>waitState</CODE> 
     *     returns immediately (i.e. does not wait at all),</LI>
     * <LI> if <VAR>timeOut</VAR> equals zero then <CODE>waitState</CODE> 
     *     waits untill the value of this MBean's State attribute 
     *     is the same as the <VAR>wantedState</VAR> parameter (i.e. will wait 
     *     indefinitely if this condition is never met).</LI></UL>
     * 
     * @param wantedState The value of this MBean's State attribute to wait 
     *        for. <VAR>wantedState</VAR> can be one of:
     * <ul>
     * <li><CODE>CommunicatorServer.OFFLINE</CODE>,</li> 
     * <li><CODE>CommunicatorServer.ONLINE</CODE>,</li>
     * <li><CODE>CommunicatorServer.STARTING</CODE>,</li> 
     * <li><CODE>CommunicatorServer.STOPPING</CODE>.</li>
     * </ul>
     * @param timeOut The maximum time to wait for, in milliseconds, 
     *        if positive. 
     * Infinite time out if 0, or no waiting at all if negative.
     *
     * @return true if the value of this MBean's State attribute is the
     *      same as the <VAR>wantedState</VAR> parameter; false otherwise.
     */
    public boolean waitState(int wantedState, long timeOut) {
        if (isTraceOn())
            trace("waitState", wantedState + "(0on,1off,2st) TO=" + timeOut +
		  " ; current state = " + getStateString());

	long endTime = 0;
	if (timeOut > 0)
	    endTime = System.currentTimeMillis() + timeOut;

	synchronized (stateLock) {
	    while (state != wantedState) {
		if (timeOut < 0) {
		    if (isTraceOn())
			trace("waitState", "timeOut < 0, return without wait");
		    return false;
		} else {
		    try {
			if (timeOut > 0) {
			    long toWait = endTime - System.currentTimeMillis();
			    if (toWait <= 0) {
				if (isTraceOn())
				    trace("waitState", "timed out");
				return false;
			    }
			    stateLock.wait(toWait);
			} else {  // timeOut == 0
			    stateLock.wait();
			}
		    } catch (InterruptedException e) {
			if (isTraceOn())
			    trace("waitState", "wait interrupted");
			return (state == wantedState);
		    }
		}
	    }
	    if (isTraceOn())
		trace("waitState", "returning in desired state");
	    return true;
	}
    }

    /**
     * <p>Waits until the communicator is started or timeout expires.
     *    
     * @param timeout Time in ms to wait for the connector to start.
     *        If <code>timeout</code> is positive, wait for at most
     *        the specified time. An infinite timeout can be specified
     *        by passing a <code>timeout</code> value equals