arraynotificationbuffer.java

java1.6众多例子参考

/*
 * @(#)ArrayNotificationBuffer.java	1.31 07/05/29
 *
 * Copyright 2006 Sun Microsystems, Inc. All rights reserved.
 * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 */

package com.sun.jmx.remote.internal;

import java.security.AccessController;
import java.security.PrivilegedAction;
import java.security.PrivilegedActionException;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.HashMap;
import java.util.Map;

import javax.management.InstanceNotFoundException;
import javax.management.MBeanServer;
import javax.management.MBeanServerDelegate;
import javax.management.MBeanServerNotification;
import javax.management.Notification;
import javax.management.NotificationBroadcaster;
import javax.management.NotificationFilter;
import javax.management.NotificationFilterSupport;
import javax.management.NotificationListener;
import javax.management.ObjectName;
import javax.management.QueryEval;
import javax.management.QueryExp;

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

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

/** A circular buffer of notifications received from an MBean server. */
/*
  There is one instance of ArrayNotificationBuffer for every
  MBeanServer object that has an attached ConnectorServer.  Then, for
  every ConnectorServer attached to a given MBeanServer, there is an
  instance of the inner class ShareBuffer.  So for example with two
  ConnectorServers it looks like this:

  ConnectorServer1 -> ShareBuffer1 -\
                                     }-> ArrayNotificationBuffer
  ConnectorServer2 -> ShareBuffer2 -/              |
                                                   |
						   v
                                              MBeanServer

  The ArrayNotificationBuffer has a circular buffer of
  NamedNotification objects.  Each ConnectorServer defines a
  notification buffer size, and this size is recorded by the
  corresponding ShareBuffer.  The buffer size of the
  ArrayNotificationBuffer is the maximum of all of its ShareBuffers.
  When a ShareBuffer is added or removed, the ArrayNotificationBuffer
  size is adjusted accordingly.

  An ArrayNotificationBuffer also has a BufferListener (which is a
  NotificationListener) registered on every NotificationBroadcaster
  MBean in the MBeanServer to which it is attached.  The cost of this
  potentially large set of listeners is the principal motivation for
  sharing the ArrayNotificationBuffer between ConnectorServers, and
  also the reason that we are careful to discard the
  ArrayNotificationBuffer (and its BufferListeners) when there are no
  longer any ConnectorServers using it.

  The synchronization of this class is inherently complex.  In an attempt
  to limit the complexity, we use just two locks:

  - globalLock controls access to the mapping between an MBeanServer
    and its ArrayNotificationBuffer and to the set of ShareBuffers for
    each ArrayNotificationBuffer.
  
  - the instance lock of each ArrayNotificationBuffer controls access
    to the array of notifications, including its size, and to the
    dispose flag of the ArrayNotificationBuffer.  The wait/notify
    mechanism is used to indicate changes to the array.

  If both locks are held at the same time, the globalLock must be
  taken first.

  Since adding or removing a BufferListener to an MBean can involve
  calling user code, we are careful not to hold any locks while it is
  done.
 */
public class ArrayNotificationBuffer implements NotificationBuffer {
    private boolean disposed = false;

    // FACTORY STUFF, INCLUDING SHARING

    private static final Object globalLock = new Object();
    private static final
	HashMap<MBeanServer,ArrayNotificationBuffer> mbsToBuffer =
	new HashMap<MBeanServer,ArrayNotificationBuffer>(1);
    private final Collection<ShareBuffer> sharers = new HashSet<ShareBuffer>(1);

    public static NotificationBuffer getNotificationBuffer(
            MBeanServer mbs, Map env) {

        if (env == null)
            env = Collections.emptyMap();

	//Find out queue size
	int queueSize = EnvHelp.getNotifBufferSize(env);

        ArrayNotificationBuffer buf;
        boolean create;
        NotificationBuffer sharer;
        synchronized (globalLock) {
            buf = mbsToBuffer.get(mbs);
            create = (buf == null);
            if (create) {
                buf = new ArrayNotificationBuffer(mbs, queueSize);
                mbsToBuffer.put(mbs, buf);
            }
            sharer = buf.new ShareBuffer(queueSize);
        }
        /* We avoid holding any locks while calling createListeners.
         * This prevents possible deadlocks involving user code, but
         * does mean that a second ConnectorServer created and started
         * in this window will return before all the listeners are ready,
         * which could lead to surprising behaviour.  The alternative
         * would be to block the second ConnectorServer until the first
         * one has finished adding all the listeners, but that would then
         * be subject to deadlock.
         */
        if (create)
            buf.createListeners();
        return sharer;
    }

    /* Ensure that this buffer is no longer the one that will be returned by
     * getNotificationBuffer.  This method is idempotent - calling it more
     * than once has no effect beyond that of calling it once.
     */
    static void removeNotificationBuffer(MBeanServer mbs) {
        synchronized (globalLock) {
            mbsToBuffer.remove(mbs);
        }
    }

    void addSharer(ShareBuffer sharer) {
        synchronized (globalLock) {
            synchronized (this) {
                if (sharer.getSize() > queueSize)
                    resize(sharer.getSize());
            }
            sharers.add(sharer);
        }
    }

    private void removeSharer(ShareBuffer sharer) {
        boolean empty;
        synchronized (globalLock) {
            sharers.remove(sharer);
            empty = sharers.isEmpty();
            if (empty)
                removeNotificationBuffer(mBeanServer);
            else {
                int max = 0;
                for (ShareBuffer buf : sharers) {
                    int bufsize = buf.getSize();
                    if (bufsize > max)
                        max = bufsize;
                }
                if (max < queueSize)
                    resize(max);
            }
        }
        if (empty) {
            synchronized (this) {
                disposed = true;
                // Notify potential waiting fetchNotification call
                notifyAll();
            }
            destroyListeners();
        }
    }

    private synchronized void resize(int newSize) {
	if (newSize == queueSize)
	    return;
	while (queue.size() > newSize)
	    dropNotification();
	queue.resize(newSize);
	queueSize = newSize;
    }

    private class ShareBuffer implements NotificationBuffer {
	ShareBuffer(int size) {
	    this.size = size;
	    addSharer(this);
	}

	public NotificationResult
	    fetchNotifications(NotificationBufferFilter filter,
			       long startSequenceNumber,
			       long timeout,
			       int maxNotifications)
		throws InterruptedException {
	    NotificationBuffer buf = ArrayNotificationBuffer.this;
	    return buf.fetchNotifications(filter, startSequenceNumber,
					  timeout, maxNotifications);
	}

	public void dispose() {
	    ArrayNotificationBuffer.this.removeSharer(this);
	}

	int getSize() {
	    return size;
	}

	private final int size;
    }


    // ARRAYNOTIFICATIONBUFFER IMPLEMENTATION

    private ArrayNotificationBuffer(MBeanServer mbs, int queueSize) {
        if (logger.traceOn())
            logger.trace("Constructor", "queueSize=" + queueSize);

        if (mbs == null || queueSize < 1)
            throw new IllegalArgumentException("Bad args");

        this.mBeanServer = mbs;
        this.queueSize = queueSize;
        this.queue = new ArrayQueue<NamedNotification>(queueSize);
        this.earliestSequenceNumber = System.currentTimeMillis();
        this.nextSequenceNumber = this.earliestSequenceNumber;

        logger.trace("Constructor", "ends");
    }

    private synchronized boolean isDisposed() {
	return disposed;
    }

    // We no longer support calling this method from outside.
    // The JDK doesn't contain any such calls and users are not
    // supposed to be accessing this class.
    public void dispose() {
        throw new UnsupportedOperationException();
    }

    /**
     * <p>Fetch notifications that match the given listeners.</p>
     *
     * <p>The operation only considers notifications with a sequence
     * number at least <code>startSequenceNumber</code>.  It will take
     * no longer than <code>timeout</code>, and will return no more
     * than <code>maxNotifications</code> different notifications.</p>
     *
     * <p>If there are no notifications matching the criteria, the
     * operation will block until one arrives, subject to the
     * timeout.</p>
     *
     * @param filter an object that will add notifications to a
     * {@code List<TargetedNotification>} if they match the current
     * listeners with their filters.
     * @param startSequenceNumber the first sequence number to
     * consider.
     * @param timeout the maximum time to wait.  May be 0 to indicate
     * not to wait if there are no notifications.
     * @param maxNotifications the maximum number of notifications to