/* * Copyright (c) 2003, 2008, Oracle and/or its affiliates. All rights reserved. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * * This code is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License version 2 only, as * published by the Free Software Foundation. Oracle designates this * particular file as subject to the "Classpath" exception as provided * by Oracle in the LICENSE file that accompanied this code. * * This code is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License * version 2 for more details (a copy is included in the LICENSE file that * accompanied this code). * * You should have received a copy of the GNU General Public License version * 2 along with this work; if not, write to the Free Software Foundation, * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. * * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA * or visit www.oracle.com if you need additional information or have any * questions. */ 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<String, ?> 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 * return. May be 0 to indicate a wait for eligible notifications * that will return a usable <code>nextSequenceNumber</code>. The * {@link TargetedNotification} array in the returned {@link * NotificationResult} may contain more than this number of * elements but will not contain more than this number of * different notifications. */ public NotificationResult fetchNotifications(NotificationBufferFilter filter, long startSequenceNumber, long timeout, int maxNotifications) throws InterruptedException { logger.trace("fetchNotifications", "starts"); if (startSequenceNumber < 0 || isDisposed()) { synchronized(this) { return new NotificationResult(earliestSequenceNumber(), nextSequenceNumber(), new TargetedNotification[0]); } } // Check arg validity if (filter == null || startSequenceNumber < 0 || timeout < 0 || maxNotifications < 0) { logger.trace("fetchNotifications", "Bad args"); throw new IllegalArgumentException("Bad args to fetch"); } if (logger.debugOn()) { logger.trace("fetchNotifications", "filter=" + filter + "; startSeq=" + startSequenceNumber + "; timeout=" + timeout + "; max=" + maxNotifications); } if (startSequenceNumber > nextSequenceNumber()) { final String msg = "Start sequence number too big: " + startSequenceNumber + " > " + nextSequenceNumber(); logger.trace("fetchNotifications", msg); throw new IllegalArgumentException(msg); } /* Determine the end time corresponding to the timeout value. Caller may legitimately supply Long.MAX_VALUE to indicate no timeout. In that case the addition will overflow and produce a negative end time. Set end time to Long.MAX_VALUE in that case. We assume System.currentTimeMillis() is positive. */ long endTime = System.currentTimeMillis() + timeout; if (endTime < 0) // overflow endTime = Long.MAX_VALUE; if (logger.debugOn()) logger.debug("fetchNotifications", "endTime=" + endTime); /* We set earliestSeq the first time through the loop. If we set it here, notifications could be dropped before we started examining them, so earliestSeq might not correspond to the earliest notification we examined. */ long earliestSeq = -1; long nextSeq = startSequenceNumber; List<TargetedNotification> notifs = new ArrayList<TargetedNotification>(); /* On exit from this loop, notifs, earliestSeq, and nextSeq must all be correct values for the returned NotificationResult. */ while (true) { logger.debug("fetchNotifications", "main loop starts"); NamedNotification candidate; /* Get the next available notification regardless of filters, or wait for one to arrive if there is none. */ synchronized (this) { /* First time through. The current earliestSequenceNumber is the first one we could have examined. */ if (earliestSeq < 0) { earliestSeq = earliestSequenceNumber(); if (logger.debugOn()) { logger.debug("fetchNotifications", "earliestSeq=" + earliestSeq); } if (nextSeq < earliestSeq) { nextSeq = earliestSeq; logger.debug("fetchNotifications", "nextSeq=earliestSeq"); } } else earliestSeq = earliestSequenceNumber(); /* If many notifications have been dropped since the last time through, nextSeq could now be earlier than the current earliest. If so, notifications may have been lost and we return now so the caller can see this next time it calls. */ if (nextSeq < earliestSeq) { logger.trace("fetchNotifications", "nextSeq=" + nextSeq + " < " + "earliestSeq=" + earliestSeq + " so may have lost notifs"); break; } if (nextSeq < nextSequenceNumber()) { candidate = notificationAt(nextSeq); if (logger.debugOn()) { logger.debug("fetchNotifications", "candidate: " + candidate); logger.debug("fetchNotifications", "nextSeq now " + nextSeq); } } else { /* nextSeq is the largest sequence number. If we already got notifications, return them now. Otherwise wait for some to arrive, with timeout. */ if (notifs.size() > 0) { logger.debug("fetchNotifications", "no more notifs but have some so don't wait"); break; } long toWait = endTime - System.currentTimeMillis(); if (toWait <= 0) { logger.debug("fetchNotifications", "timeout"); break; } /* dispose called */ if (isDisposed()) { if (logger.debugOn()) logger.debug("fetchNotifications", "dispose callled, no wait"); return new NotificationResult(earliestSequenceNumber(), nextSequenceNumber(), new TargetedNotification[0]); } if (logger.debugOn()) logger.debug("fetchNotifications", "wait(" + toWait + ")"); wait(toWait); continue; } } /* We have a candidate notification. See if it matches our filters. We do this outside the synchronized block so we don't hold up everyone accessing the buffer (including notification senders) while we evaluate potentially slow filters. */ ObjectName name = candidate.getObjectName(); Notification notif = candidate.getNotification(); List<TargetedNotification> matchedNotifs = new ArrayList<TargetedNotification>(); logger.debug("fetchNotifications", "applying filter to candidate"); filter.apply(matchedNotifs, name, notif); if (matchedNotifs.size() > 0) { /* We only check the max size now, so that our returned nextSeq is as large as possible. This prevents the caller from thinking it missed interesting notifications when in fact we knew they weren't. */ if (maxNotifications <= 0) { logger.debug("fetchNotifications", "reached maxNotifications"); break; } --maxNotifications; if (logger.debugOn()) logger.debug("fetchNotifications", "add: " + matchedNotifs); notifs.addAll(matchedNotifs); } ++nextSeq; } // end while /* Construct and return the result. */ int nnotifs = notifs.size(); TargetedNotification[] resultNotifs = new TargetedNotification[nnotifs]; notifs.toArray(resultNotifs); NotificationResult nr = new NotificationResult(earliestSeq, nextSeq, resultNotifs); if (logger.debugOn()) logger.debug("fetchNotifications", nr.toString()); logger.trace("fetchNotifications", "ends"); return nr; } synchronized long earliestSequenceNumber() { return earliestSequenceNumber; } synchronized long nextSequenceNumber() { return nextSequenceNumber; } synchronized void addNotification(NamedNotification notif) { if (logger.traceOn()) logger.trace("addNotification", notif.toString()); Lines 1–500 of 838. Load more
/* * Copyright (c) 2003, 2008, Oracle and/or its affiliates. All rights reserved. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * * This code is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License version 2 only, as * published by the Free Software Foundation. Oracle designates this * particular file as subject to the "Classpath" exception as provided * by Oracle in the LICENSE file that accompanied this code. * * This code is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License * version 2 for more details (a copy is included in the LICENSE file that * accompanied this code). * * You should have received a copy of the GNU General Public License version * 2 along with this work; if not, write to the Free Software Foundation, * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. * * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA * or visit www.oracle.com if you need additional information or have any * questions. */ 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<String, ?> 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 * return. May be 0 to indicate a wait for eligible notifications * that will return a usable <code>nextSequenceNumber</code>. The * {@link TargetedNotification} array in the returned {@link * NotificationResult} may contain more than this number of * elements but will not contain more than this number of * different notifications. */ public NotificationResult fetchNotifications(NotificationBufferFilter filter, long startSequenceNumber, long timeout, int maxNotifications) throws InterruptedException { logger.trace("fetchNotifications", "starts"); if (startSequenceNumber < 0 || isDisposed()) { synchronized(this) { return new NotificationResult(earliestSequenceNumber(), nextSequenceNumber(), new TargetedNotification[0]); } } // Check arg validity if (filter == null || startSequenceNumber < 0 || timeout < 0 || maxNotifications < 0) { logger.trace("fetchNotifications", "Bad args"); throw new IllegalArgumentException("Bad args to fetch"); } if (logger.debugOn()) { logger.trace("fetchNotifications", "filter=" + filter + "; startSeq=" + startSequenceNumber + "; timeout=" + timeout + "; max=" + maxNotifications); } if (startSequenceNumber > nextSequenceNumber()) { final String msg = "Start sequence number too big: " + startSequenceNumber + " > " + nextSequenceNumber(); logger.trace("fetchNotifications", msg); throw new IllegalArgumentException(msg); } /* Determine the end time corresponding to the timeout value. Caller may legitimately supply Long.MAX_VALUE to indicate no timeout. In that case the addition will overflow and produce a negative end time. Set end time to Long.MAX_VALUE in that case. We assume System.currentTimeMillis() is positive. */ long endTime = System.currentTimeMillis() + timeout; if (endTime < 0) // overflow endTime = Long.MAX_VALUE; if (logger.debugOn()) logger.debug("fetchNotifications", "endTime=" + endTime); /* We set earliestSeq the first time through the loop. If we set it here, notifications could be dropped before we started examining them, so earliestSeq might not correspond to the earliest notification we examined. */ long earliestSeq = -1; long nextSeq = startSequenceNumber; List<TargetedNotification> notifs = new ArrayList<TargetedNotification>(); /* On exit from this loop, notifs, earliestSeq, and nextSeq must all be correct values for the returned NotificationResult. */ while (true) { logger.debug("fetchNotifications", "main loop starts"); NamedNotification candidate; /* Get the next available notification regardless of filters, or wait for one to arrive if there is none. */ synchronized (this) { /* First time through. The current earliestSequenceNumber is the first one we could have examined. */ if (earliestSeq < 0) { earliestSeq = earliestSequenceNumber(); if (logger.debugOn()) { logger.debug("fetchNotifications", "earliestSeq=" + earliestSeq); } if (nextSeq < earliestSeq) { nextSeq = earliestSeq; logger.debug("fetchNotifications", "nextSeq=earliestSeq"); } } else earliestSeq = earliestSequenceNumber(); /* If many notifications have been dropped since the last time through, nextSeq could now be earlier than the current earliest. If so, notifications may have been lost and we return now so the caller can see this next time it calls. */ if (nextSeq < earliestSeq) { logger.trace("fetchNotifications", "nextSeq=" + nextSeq + " < " + "earliestSeq=" + earliestSeq + " so may have lost notifs"); break; } if (nextSeq < nextSequenceNumber()) { candidate = notificationAt(nextSeq); if (logger.debugOn()) { logger.debug("fetchNotifications", "candidate: " + candidate); logger.debug("fetchNotifications", "nextSeq now " + nextSeq); } } else { /* nextSeq is the largest sequence number. If we already got notifications, return them now. Otherwise wait for some to arrive, with timeout. */ if (notifs.size() > 0) { logger.debug("fetchNotifications", "no more notifs but have some so don't wait"); break; } long toWait = endTime - System.currentTimeMillis(); if (toWait <= 0) { logger.debug("fetchNotifications", "timeout"); break; } /* dispose called */ if (isDisposed()) { if (logger.debugOn()) logger.debug("fetchNotifications", "dispose callled, no wait"); return new NotificationResult(earliestSequenceNumber(), nextSequenceNumber(), new TargetedNotification[0]); } if (logger.debugOn()) logger.debug("fetchNotifications", "wait(" + toWait + ")"); wait(toWait); continue; } } /* We have a candidate notification. See if it matches our filters. We do this outside the synchronized block so we don't hold up everyone accessing the buffer (including notification senders) while we evaluate potentially slow filters. */ ObjectName name = candidate.getObjectName(); Notification notif = candidate.getNotification(); List<TargetedNotification> matchedNotifs = new ArrayList<TargetedNotification>(); logger.debug("fetchNotifications", "applying filter to candidate"); filter.apply(matchedNotifs, name, notif); if (matchedNotifs.size() > 0) { /* We only check the max size now, so that our returned nextSeq is as large as possible. This prevents the caller from thinking it missed interesting notifications when in fact we knew they weren't. */ if (maxNotifications <= 0) { logger.debug("fetchNotifications", "reached maxNotifications"); break; } --maxNotifications; if (logger.debugOn()) logger.debug("fetchNotifications", "add: " + matchedNotifs); notifs.addAll(matchedNotifs); } ++nextSeq; } // end while /* Construct and return the result. */ int nnotifs = notifs.size(); TargetedNotification[] resultNotifs = new TargetedNotification[nnotifs]; notifs.toArray(resultNotifs); NotificationResult nr = new NotificationResult(earliestSeq, nextSeq, resultNotifs); if (logger.debugOn()) logger.debug("fetchNotifications", nr.toString()); logger.trace("fetchNotifications", "ends"); return nr; } synchronized long earliestSequenceNumber() { return earliestSequenceNumber; } synchronized long nextSequenceNumber() { return nextSequenceNumber; } synchronized void addNotification(NamedNotification notif) { if (logger.traceOn()) logger.trace("addNotification", notif.toString());