clientnotifforwarder.java

JAVA的一些源码 JAVA2 STANDARD EDITION DEVELOPMENT KIT 5.0

/*
 * @(#)ClientNotifForwarder.java	1.39 05/01/04
 * 
 * Copyright 2005 Sun Microsystems, Inc. All rights reserved.
 * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 */

package com.sun.jmx.remote.internal;

import java.io.IOException;
import java.io.NotSerializableException;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executor;

import java.security.AccessController;
import java.security.PrivilegedAction;
import javax.security.auth.Subject;

import javax.management.Notification;
import javax.management.NotificationListener;
import javax.management.NotificationFilter;
import javax.management.ObjectName;
import javax.management.MBeanServerNotification;
import javax.management.InstanceNotFoundException;
import javax.management.ListenerNotFoundException;

import javax.management.remote.NotificationResult;
import javax.management.remote.TargetedNotification;

import com.sun.jmx.remote.util.ClassLogger;
import com.sun.jmx.remote.util.EnvHelp;


public abstract class ClientNotifForwarder {
    public ClientNotifForwarder(Map env) {
	this(null, env);
    }

    private static int threadId;

    /* An Executor that allows at most one executing and one pending
       Runnable.  It uses at most one thread -- as soon as there is
       no pending Runnable the thread can exit.  Another thread is
       created as soon as there is a new pending Runnable.  This
       Executor is adapted for use in a situation where each Runnable
       usually schedules up another Runnable.  On return from the
       first one, the second one is immediately executed.  So this
       just becomes a complicated way to write a while loop, but with
       the advantage that you can replace it with another Executor,
       for instance one that you are using to execute a bunch of other
       unrelated work.

       You might expect that a java.util.concurrent.ThreadPoolExecutor 
       with corePoolSize=0 and maximumPoolSize=1 would have the same
       behaviour, but it does not.  A ThreadPoolExecutor only creates
       a new thread when a new task is submitted and the number of
       existing threads is < corePoolSize.  This can never happen when
       corePoolSize=0, so new threads are never created.  Surprising,
       but there you are.
    */
    private static class LinearExecutor implements Executor {
	public synchronized void execute(Runnable command) {
	    if (this.command != null)
		throw new IllegalArgumentException("More than one command");
	    this.command = command;
	    if (thread == null) {
		thread = new Thread() {
		    public void run() {
			while (true) {
			    Runnable r;
			    synchronized (LinearExecutor.this) {
				if (LinearExecutor.this.command == null) {
				    thread = null;
				    return;
				} else {
				    r = LinearExecutor.this.command;
				    LinearExecutor.this.command = null;
				}
			    }
			    r.run();
			}
		    }
		};
		thread.setDaemon(true);
		thread.setName("ClientNotifForwarder-" + ++threadId);
		thread.start();
	    }
	}

	private Runnable command;
	private Thread thread;
    }

    public ClientNotifForwarder(ClassLoader defaultClassLoader, Map env) {
	maxNotifications = EnvHelp.getMaxFetchNotifNumber(env);
	timeout = EnvHelp.getFetchTimeout(env);

        /* You can supply an Executor in which the remote call to
           fetchNotifications will be made.  The Executor's execute
           method reschedules another task, so you must not use
           an Executor that executes tasks in the caller's thread.  */
	Executor ex = (Executor)
	    env.get("jmx.remote.x.fetch.notifications.executor");
	if (ex == null)
            ex = new LinearExecutor();
        else if (logger.traceOn())
            logger.trace("ClientNotifForwarder", "executor is " + ex);

	this.defaultClassLoader = defaultClassLoader;
	this.executor = ex;
    }

    /**
     * Called to to fetch notifications from a server.
     */
    abstract protected NotificationResult fetchNotifs(long clientSequenceNumber,
						      int maxNotifications,
						      long timeout)
	    throws IOException, ClassNotFoundException;

    abstract protected Integer addListenerForMBeanRemovedNotif() 
	throws IOException, InstanceNotFoundException;

    abstract protected void removeListenerForMBeanRemovedNotif(Integer id)
	throws IOException, InstanceNotFoundException, 
	       ListenerNotFoundException;

    /**
     * Used to send out a notification about lost notifs
     */
    abstract protected void lostNotifs(String message, long number);


    public synchronized void addNotificationListener(Integer listenerID,
                                        ObjectName name, 
                                        NotificationListener listener,
                                        NotificationFilter filter,
                                        Object handback,
					Subject delegationSubject)
	    throws IOException, InstanceNotFoundException {

	if (logger.traceOn()) {
	    logger.trace("addNotificationListener",
			 "Add the listener "+listener+" at "+name);
	}

	infoList.put(listenerID, 
		     new ClientListenerInfo(listenerID, 
					    name,
					    listener,
					    filter,
					    handback,
					    delegationSubject));
    

	init(false);
    }            

    public synchronized Integer[]
	removeNotificationListener(ObjectName name,
				   NotificationListener listener)
	throws ListenerNotFoundException, IOException {

        beforeRemove();
	    
	if (logger.traceOn()) {
	    logger.trace("removeNotificationListener",
			 "Remove the listener "+listener+" from "+name);
	}
	    
	ArrayList ids = new ArrayList();
	ArrayList values = new ArrayList(infoList.values());
	for (int i=values.size()-1; i>=0; i--) {
	    ClientListenerInfo li = (ClientListenerInfo)values.get(i);

	    if (li.sameAs(name, listener)) {
		ids.add(li.getListenerID());

		infoList.remove(li.getListenerID());
	    }
	}

	if (ids.isEmpty())
	    throw new ListenerNotFoundException("Listener not found");

	return (Integer[])ids.toArray(new Integer[0]);
    }

    public synchronized Integer
	removeNotificationListener(ObjectName name, 
				   NotificationListener listener,
				   NotificationFilter filter,
				   Object handback)
	    throws ListenerNotFoundException, IOException {

	if (logger.traceOn()) {
	    logger.trace("removeNotificationListener",
			 "Remove the listener "+listener+" from "+name);
	}

        beforeRemove();

	Integer id = null;

	ArrayList values = new ArrayList(infoList.values());
	for (int i=values.size()-1; i>=0; i--) {
	    ClientListenerInfo li = (ClientListenerInfo)values.get(i);
	    if (li.sameAs(name, listener, filter, handback)) {
		id=li.getListenerID();

		infoList.remove(id);

		break;		    
	    }
	}

	if (id == null)
	    throw new ListenerNotFoundException("Listener not found");

	return id;	
    }

    public synchronized Integer[] removeNotificationListener(ObjectName name) {
	if (logger.traceOn()) {
	    logger.trace("removeNotificationListener",
			 "Remove all listeners registered at "+name);
	}

	ArrayList ids = new ArrayList();

	ArrayList values = new ArrayList(infoList.values());
	for (int i=values.size()-1; i>=0; i--) {
	    ClientListenerInfo li = (ClientListenerInfo)values.get(i);
	    if (li.sameAs(name)) {
		ids.add(li.getListenerID());
		    
		infoList.remove(li.getListenerID());
	    }
	}

	return (Integer[]) ids.toArray(new Integer[0]);
    }

    public synchronized ListenerInfo[] getListenerInfo() {
	return (ListenerInfo[])infoList.values().toArray(new ListenerInfo[0]);
    }

    /*
     * Called when a connector is doing reconnection. Like <code>postReconnection</code>,
     * this method is intended to be called only by a client connetor:
     * <code>RMIConnector</code/> and <code/>ClientIntermediary</code>.
     * Call this method will set the flag beingReconnection to <code>true</code>,
     * and the thread used to fetch notifis will be stopped, a new thread can be
     * created only after the method <code>postReconnection</code> is called.
     *
     * It is caller's responsiblity to not re-call this method before calling
     * <code>postReconnection.
     */ 
    public synchronized ClientListenerInfo[] preReconnection() throws IOException {
	if (state == TERMINATED || beingReconnected) { // should never
	    throw new IOException("Illegal state.");
	}

	final ClientListenerInfo[] tmp = (ClientListenerInfo[]) 
            infoList.values().toArray(new ClientListenerInfo[0]);


	beingReconnected = true;

	infoList.clear();

	if (currentFetchThread == Thread.currentThread()) {
	    /* we do not need to stop the fetching thread, because this thread is
	       used to do restarting and it will not be used to do fetching during
	       the re-registering the listeners.*/
	    return tmp;
	}

	while (state == STARTING) {
	    try {
		wait();
	    } catch (InterruptedException ire) {
		IOException ioe = new IOException(ire.toString());
		EnvHelp.initCause(ioe, ire);

		throw ioe;
	    }
	}

	if (state == STARTED) {
	    setState(STOPPING);
	}

	return tmp;
    }

    /**
     * Called after reconnection is finished.
     * This method is intended to be called only by a client connetor:
     * <code>RMIConnector</code/> and <code/>ClientIntermediary</code>.
     */
    public synchronized void postReconnection(ClientListenerInfo[] listenerInfos)
	throws IOException {

	if (state == TERMINATED) {
	    return;
	}

	while (state == STOPPING) {
	    try {
		wait();
	    } catch (InterruptedException ire) {
		IOException ioe = new IOException(ire.toString());
		EnvHelp.initCause(ioe, ire);
		throw ioe;
	    }
	}

	final boolean trace = logger.traceOn();
	final int len   = listenerInfos.length;

	for (int i=0; i<len; i++) {
	    if (trace) {
		logger.trace("addNotificationListeners",
			     "Add a listener at "+
			     listenerInfos[i].getListenerID());
	    }

	    infoList.put(listenerInfos[i].getListenerID(), listenerInfos[i]);
	}

	beingReconnected = false;
	notifyAll();

	if (currentFetchThread == Thread.currentThread()) {
	    // no need to init, simply get the id
	    try {
		mbeanRemovedNotifID = addListenerForMBeanRemovedNotif();
	    } catch (Exception e) {
		final String msg =
		    "Failed to register a listener to the mbean " +
		    "server: the client will not do clean when an MBean " +
		    "is unregistered";
		if (logger.traceOn()) {
		    logger.trace("init", msg, e);
		}		 
	    } 
	} else if (listenerInfos.length > 0) { // old listeners re-registered
	    init(true);
	} else if (infoList.size() > 0) {
	    // but new listeners registered during reconnection
	    init(false);
	}
    }

    public synchronized void terminate() {
	if (state == TERMINATED) {
	    return;
	}

	if (logger.traceOn()) {
	    logger.trace("terminate", "Terminating...");
	}

	if (state == STARTED) {
	   infoList.clear();
	}

	setState(TERMINATED);
    }

// -------------------------------------------------
// private classes
// -------------------------------------------------
    //
    private class NotifFetcher implements Runnable {
	public void run() {
            synchronized (ClientNotifForwarder.this) {
		currentFetchThread = Thread.currentThread();

                if (state == STARTING)
                    setState(STARTED);
            }

	    if (defaultClassLoader != null) {
		AccessController.doPrivileged(new PrivilegedAction() {
			public Object run() {
			    Thread.currentThread().
				setContextClassLoader(defaultClassLoader);
			    return null;
			}
		    });
	    }

	    NotificationResult nr = null;
	    if (!shouldStop() && (nr = fetchNotifs()) != null) {
		// nr == null means got exception

		final TargetedNotification[] notifs =
		    nr.getTargetedNotifications();
		final int len = notifs.length;
		final HashMap listeners;
		final Integer myListenerID;

		long missed = 0;

		synchronized(ClientNotifForwarder.this) {
		    // check sequence number.
		    //