communicatorserver.java

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

     *        <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.
     * 
     */
    private void waitForStart(long timeout) 
	throws CommunicationException, InterruptedException {
        if (isTraceOn())
            trace("waitForStart", "Timeout=" + timeout +
		  " ; current state = " + getStateString());
	
	final long startTime = System.currentTimeMillis();
 
	synchronized (stateLock) {
	    while (state == STARTING) {
		// Time elapsed since startTime...
		//
		final long elapsed = System.currentTimeMillis() - startTime;

		// wait for timeout - elapsed.
		// A timeout of Long.MAX_VALUE is equivalent to something
		// like 292271023 years - which is pretty close to 
		// forever as far as we are concerned ;-)
		//
		final long remainingTime = timeout-elapsed;

		// If remainingTime is negative, the timeout has elapsed.
		// 
		if (remainingTime < 0) {
		    if (isTraceOn())
			trace("waitForStart", 
			      "timeout < 0, return without wait");
		    throw new InterruptedException("Timeout expired");
		}

		// We're going to wait until someone notifies on the
		// the stateLock object, or until the timeout expires,
		// or until the thread is interrupted.
		//
		try {
		    stateLock.wait(remainingTime);
		} catch (InterruptedException e) {
		    if (isTraceOn())
			trace("waitForStart", "wait interrupted");

		    // If we are now ONLINE, then no need to rethrow the
		    // exception... we're simply going to exit the while
		    // loop. Otherwise, throw the InterruptedException.
		    //
		    if (state != ONLINE) throw e;
		}
	    }

	    // We're no longer in STARTING state
	    //
	    if (state == ONLINE) {
		// OK, we're started, everything went fine, just return
		//
		if (isTraceOn()) trace("waitForStart", "started");
		return;
	    } else if (startException instanceof CommunicationException) {
		// There was some exception during the starting phase.
		// Cast and throw...
		//
		throw (CommunicationException)startException;
	    } else if (startException instanceof InterruptedException) {
		// There was some exception during the starting phase.
		// Cast and throw...
		//
		throw (InterruptedException)startException;
	    } else if (startException != null) {
		// There was some exception during the starting phase.
		// Wrap and throw...
		//
		throw new CommunicationException(startException, 
						 "Failed to start: "+
						 startException);
	    } else { 
		// We're not ONLINE, and there's no exception...
		// Something went wrong but we don't know what...
		//
		throw new CommunicationException("Failed to start: state is "+
						 getStringForState(state));
	    }
	}
    }

    /**
     * Gets the state of this <CODE>CommunicatorServer</CODE> as an integer.
     *
     * @return <CODE>ONLINE</CODE>, <CODE>OFFLINE</CODE>, 
     *         <CODE>STARTING</CODE> or <CODE>STOPPING</CODE>.
     */
    public int getState() {
	synchronized (stateLock) {
	    return state ;
	}
    }

    /**
     * Gets the state of this <CODE>CommunicatorServer</CODE> as a string.
     *
     * @return One of the strings "ONLINE", "OFFLINE", "STARTING" or 
     *         "STOPPING".
     */
    public String getStateString() {
        return getStringForState(state) ;
    }

    /**
     * Gets the host name used by this <CODE>CommunicatorServer</CODE>.
     *
     * @return The host name used by this <CODE>CommunicatorServer</CODE>.
     */
    public String getHost() {
        try {
            host = InetAddress.getLocalHost().getHostName();
        } catch (Exception e) {
            host = "Unknown host";
        }
        return host ;
    }

    /**
     * Gets the port number used by this <CODE>CommunicatorServer</CODE>.
     *
     * @return The port number used by this <CODE>CommunicatorServer</CODE>.
     */
    public int getPort() {
	synchronized (stateLock) {
	    return port ;
	}
    }

    /**
     * Sets the port number used by this <CODE>CommunicatorServer</CODE>.
     *
     * @param port The port number used by this 
     *             <CODE>CommunicatorServer</CODE>.
     *
     * @exception java.lang.IllegalStateException This method has been invoked
     * while the communicator was ONLINE or STARTING.
     */
    public void setPort(int port) throws java.lang.IllegalStateException {
	synchronized (stateLock) {
	    if ((state == ONLINE) || (state == STARTING))
		throw new IllegalStateException("Stop server before " +
						"carrying out this operation");
	    this.port = port;
	    dbgTag = makeDebugTag();
	}
    }

    /**
     * Gets the protocol being used by this <CODE>CommunicatorServer</CODE>.
     * @return The protocol as a string.
     */
    public abstract String getProtocol() ;

    /**
     * Gets the number of clients that have been processed by this 
     * <CODE>CommunicatorServer</CODE>  since its creation.
     *
     * @return The number of clients handled by this 
     *         <CODE>CommunicatorServer</CODE>
     *         since its creation. This counter is not reset by the 
     *         <CODE>stop</CODE> method.
     */
    int getServedClientCount() {
        return servedClientCount ;
    }
  
    /**
     * Gets the number of clients currently being processed by this 
     * <CODE>CommunicatorServer</CODE>.
     *
     * @return The number of clients currently being processed by this 
     *         <CODE>CommunicatorServer</CODE>.
     */
    int getActiveClientCount() {
        int result = clientHandlerVector.size() ;
        return result ;
    }

    /**
     * Gets the maximum number of clients that this
     * <CODE>CommunicatorServer</CODE> can  process concurrently.
     *
     * @return The maximum number of clients that this 
     *         <CODE>CommunicatorServer</CODE> can 
     *         process concurrently.
     */
    int getMaxActiveClientCount() {
        return maxActiveClientCount ;
    }

    /**
     * Sets the maximum number of clients this 
     * <CODE>CommunicatorServer</CODE> can process concurrently.
     *
     * @param c The number of clients.
     *
     * @exception java.lang.IllegalStateException This method has been invoked
     * while the communicator was ONLINE or STARTING.
     */
    void setMaxActiveClientCount(int c) 
	throws java.lang.IllegalStateException {
	synchronized (stateLock) {
	    if ((state == ONLINE) || (state == STARTING)) {
		throw new IllegalStateException(
			  "Stop server before carrying out this operation");
	    }	    
	    maxActiveClientCount = c ;
	}
    }

    /**
     * For SNMP Runtime internal use only.
     */
    void notifyClientHandlerCreated(ClientHandler h) {
        clientHandlerVector.addElement(h) ;
    }

    /**
     * For SNMP Runtime internal use only.
     */
    synchronized void notifyClientHandlerDeleted(ClientHandler h) {
        clientHandlerVector.removeElement(h);
	notifyAll();
    }

    /**
     * The number of times the communicator server will attempt
     * to bind before giving up.
     **/
    protected int getBindTries() {
	return 50;
    }

    /**
     * The delay, in ms, during which the communicator server will sleep before
     * attempting to bind again.
     **/
    protected long getBindSleepTime() {
	return 100;
    }

    /**
     * For SNMP Runtime internal use only.
     * <p>
     * The <CODE>run</CODE> method executed by this connector's main thread.
     */
    public void run() {
        
        // Fix jaw.00667.B
        // It seems that the init of "i" and "success"
        // need to be done outside the "try" clause...
        // A bug in Java 2 production release ?
        //
        int i = 0;
        boolean success = false;
        
        // ----------------------
        // Bind 
        // ----------------------
        try {
            // Fix for bug 4352451: "java.net.BindException: Address in use".
            //
	    final int  bindRetries = getBindTries();
	    final long sleepTime   = getBindSleepTime();
            while (i < bindRetries && !success) {
                try {
                    // Try socket connection.
                    //
                    doBind();
                    success = true;
                } catch (CommunicationException ce) {
                    i++;
                    try {
                        Thread.sleep(sleepTime);
                    } catch (InterruptedException ie) {
			throw ie;
                    }
                }
            }
            // Retry last time to get correct exception.
            //
            if (!success) {
                // Try socket connection.
                //
                doBind();
            }

        } catch(Exception x) {
            if (isDebugOn()) {
                debug("run","Unexpected exception = "+x) ;
            }
	    synchronized(stateLock) {
		startException = x;
		changeState(OFFLINE);
	    }
            if (isTraceOn()) {
                trace("run","State is OFFLINE") ;
            }
	    doError(x);
	    return;
        }

        try {
            // ----------------------
            // State change
            // ----------------------
            changeState(ONLINE) ;
            if (isTraceOn()) {
                trace("run","State is ONLINE") ;
            }

            // ----------------------
            // Main loop
            // ----------------------
            while (!stopRequested) {
                servedClientCount++;	
                doReceive() ;
                waitIfTooManyClients() ;
                doProcess() ;
            }
            if (isTraceOn()) {
                trace("run","Stop has been requested") ;
            }

        } catch(InterruptedException x) {
            if (isTraceOn()) {
                trace("run","Interrupt caught") ;
            }
            changeState(STOPPING);
        } catch(Exception x) {
            if (isDebugOn()) {
                debug("run","Unexpected exception = "+x) ;
            }
            changeState(STOPPING);
	} finally {
	    synchronized (stateLock) {
		interrupted = true;
		Thread.currentThread().interrupted();
	    }

	    // ----------------------
	    // unBind
	    // ----------------------
	    try {
		doUnbind() ;
		waitClientTermination() ;
		changeState(OFFLINE);
		if (isTraceOn()) {
		    trace("run","State is OFFLINE") ;
		}
	    } catch(Exception x) {
		if (isDebugOn()) {
		    debug("run","Unexpected exception = "+x) ;
		}
		changeState(OFFLINE);
	    }
	    
	}
    }

    /**
     */
    protected abstract void doError(Exception e) throws CommunicationException;

    //
    // To be defined by the subclass.
    //
    // Each method below is called by run() and must be subclassed.
    // If the method sends an exception (Communication or Interrupt), this
    // will end up the run() method and switch the connector offline.
    //
    // If it is a CommunicationException, run() will call
    //       Debug.printException().
    //
    // All these methods should propagate the InterruptedException to inform
    // run() that the connector must be switch OFFLINE.
    //
    //
    //
    // doBind() should do all what is needed before calling doReceive().
    // If doBind() throws an exception, doUnbind() is not to be called 
    // and run() ends up.
    //

    /**
     */
    protected abstract void doBind() 
	throws CommunicationException, InterruptedException ;

    /**
     * <CODE>doReceive()</CODE> should block until a client is available.
     * If this method throws an exception, <CODE>doProcess()</CODE> is not 
     * called but <CODE>doUnbind()</CODE> is called then <CODE>run()</CODE> 
     * stops.
     */
    protected abstract void doReceive() 
	throws CommunicationException, InterruptedException ;

    /**
     * <CODE>doProcess()</CODE> is called after <CODE>doReceive()</CODE>: 
     * it should process the requests of the incoming client.
     * If it throws an exception, <CODE>doUnbind()</CODE> is called and 
     * <CODE>run()</CODE> stops.
     */
    protected abstract void doProcess() 
	throws CommunicationException, InterruptedException ;

    /**
     * <CODE>doUnbind()</CODE> is called whenever the connector goes 
     * <CODE>OFFLINE</CODE>, except if <CODE>doBind()</CODE> has thrown an 
     * exception.
     */
    protected abstract void doUnbind() 
	throws CommunicationException, InterruptedException ;

    /**
     * Get the <code>MBeanServer</code> object to which incoming requests are
     * sent.  This is either the MBean server in which this connector is
     * registered, or an <code>MBeanServerForwarder</code> leading to that
     * server.
     */
    public synchronized MBeanServer getMBeanServer() {
        return topMBS;
    }

    /**
     * Set the <code>MBeanServer</code> object to which incoming
     * requests are sent.  This must be either the MBean server in
     * which this connector is registered, or an
     * <code>MBeanServerForwarder</code> leading to that server.  An
     * <code>MBeanServerForwarder</code> <code>mbsf</code> leads to an
     * MBean server <code>mbs</code> if
     * <code>mbsf.getMBeanServer()</code> is either <code>mbs</code>
     * or an <code>MBeanServerForwarder</code> leading to
     * <code>mbs</code>.
     *
     * @exception IllegalArgumentException if <code>newMBS</code> is neither
     * the MBean server in which this connector is registered nor an
     * <code>MBeanServerForwarder</code> leading to that server.
     *
     * @exception IllegalStateException This method has been invoked
     * while the communicator was ONLINE or STARTING.
     */
    public synchronized void setMBeanServer(MBeanServer newMBS)
	    throws IllegalArgumentException, IllegalStateException {
	synchronized (stateLock) {
	    if (state == ONLINE || state == STARTING)