EventClient.java

/*
 * Copyright 2007 Sun Microsystems, Inc.  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.  Sun designates this
 * particular file as subject to the "Classpath" exception as provided
 * by Sun 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 Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
 * CA 95054 USA or visit www.sun.com if you need additional information or
 * have any questions.
 */

package javax.management.event;

import com.sun.jmx.event.DaemonThreadFactory;
import com.sun.jmx.event.LeaseRenewer;
import com.sun.jmx.event.ReceiverBuffer;
import com.sun.jmx.event.RepeatedSingletonJob;
import com.sun.jmx.namespace.JMXNamespaceUtils;
import com.sun.jmx.mbeanserver.PerThreadGroupPool;
import com.sun.jmx.remote.util.ClassLogger;

import java.io.IOException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import javax.management.InstanceNotFoundException;
import javax.management.ListenerNotFoundException;
import javax.management.MBeanNotificationInfo;
import javax.management.MBeanServerConnection;
import javax.management.Notification;
import javax.management.NotificationBroadcasterSupport;
import javax.management.NotificationFilter;
import javax.management.NotificationListener;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.NotificationResult;
import javax.management.remote.TargetedNotification;

/**
 * <p>This class is used to manage its notification listeners on the client
 * side in the same way as on the MBean server side. This class needs to work
 * with an {@link EventClientDelegateMBean} on the server side.</p>
 *
 * <P>A user can specify an {@link EventRelay} object to specify how to receive
 * notifications forwarded by the {@link EventClientDelegateMBean}. By default,
 * the class {@link FetchingEventRelay} is used.</p>
 *
 * <p>A user can specify an {@link java.util.concurrent.Executor Executor}
 * to distribute notifications to local listeners. If no executor is
 * specified, the thread in the {@link EventRelay} which calls {@link
 * EventReceiver#receive EventReceiver.receive} will be reused to distribute
 * the notifications (in other words, to call the {@link
 * NotificationListener#handleNotification handleNotification} method of the
 * appropriate listeners). It is useful to make a separate thread do this
 * distribution in some cases. For example, if network communication is slow,
 * the forwarding thread can concentrate on communication while, locally,
 * the distributing thread distributes the received notifications. Another
 * usage is to share a thread pool between many clients, for scalability.
 * Note, though, that if the {@code Executor} can create more than one thread
 * then it is possible that listeners will see notifications in a different
 * order from the order in which they were sent.</p>
 *
 * <p>An object of this class sends notifications to listeners added with
 * {@link #addEventClientListener}.  The {@linkplain Notification#getType()
 * type} of each such notification is one of {@link #FAILED}, {@link #NONFATAL},
 * or {@link #NOTIFS_LOST}.</p>
 *
 * @since JMX 2.0
 */
public class EventClient implements EventConsumer, NotificationManager {

    /**
     * <p>A notification string type used by an {@code EventClient} object
     * to inform a listener added by {@link #addEventClientListener} that
     * it failed to get notifications from a remote server, and that it is
     * possible that no more notifications will be delivered.</p>
     *
     * @see #addEventClientListener
     * @see EventReceiver#failed
     */
    public static final String FAILED = "jmx.event.service.failed";

    /**
     * <p>Reports that an unexpected exception has been received by the {@link
     * EventRelay} object but that it is non-fatal. For example, a notification
     * received is not serializable or its class is not found.</p>
     *
     * @see #addEventClientListener
     * @see EventReceiver#nonFatal
     */
    public static final String NONFATAL = "jmx.event.service.nonfatal";

    /**
     * <p>A notification string type used by an {@code EventClient} object to
     * inform a listener added by {@code #addEventClientListener} that it
     * has detected that notifications have been lost.  The {@link
     * Notification#getUserData() userData} of the notification is a Long which
     * is an upper bound on the number of lost notifications that have just
     * been detected.</p>
     *
     * @see #addEventClientListener
     */
    public static final String NOTIFS_LOST = "jmx.event.service.notifs.lost";

    /**
     * The default lease time, {@value}, in milliseconds.
     *
     * @see EventClientDelegateMBean#lease
     */
    public static final long DEFAULT_LEASE_TIMEOUT = 300000;

    /**
     * <p>Constructs a default {@code EventClient} object.</p>
     *
     * <p>This object creates a {@link FetchingEventRelay} object to
     * receive notifications forwarded by the {@link EventClientDelegateMBean}.
     * The {@link EventClientDelegateMBean} that it works with is the
     * one registered with the {@linkplain EventClientDelegate#OBJECT_NAME
     * default ObjectName}.  The thread from the {@link FetchingEventRelay}
     * object that fetches the notifications is also used to distribute them.
     *
     * @param conn An {@link MBeanServerConnection} object used to communicate
     * with an {@link EventClientDelegateMBean} MBean.
     *
     * @throws IllegalArgumentException If {@code conn} is null.
     * @throws IOException If an I/O error occurs when communicating with the
     * {@code EventClientDelegateMBean}.
     */
    public EventClient(MBeanServerConnection conn) throws IOException {
        this(EventClientDelegate.getProxy(conn));
    }

    /**
     * Constructs an {@code EventClient} object with a specified
     * {@link EventClientDelegateMBean}.
     *
     * <p>This object creates a {@link FetchingEventRelay} object to receive
     * notifications forwarded by the {@link EventClientDelegateMBean}.  The
     * thread from the {@link FetchingEventRelay} object that fetches the
     * notifications is also used to distribute them.
     *
     * @param delegate An {@link EventClientDelegateMBean} object to work with.
     *
     * @throws IllegalArgumentException If {@code delegate} is null.
     * @throws IOException If an I/O error occurs when communicating with the
     * the {@link EventClientDelegateMBean}.
     */
    public EventClient(EventClientDelegateMBean delegate)
    throws IOException {
        this(delegate, null, null, null, DEFAULT_LEASE_TIMEOUT);
    }

    /**
     * Constructs an {@code EventClient} object with the specified
     * {@link EventClientDelegateMBean}, {@link EventRelay}
     * object, and distributing thread.
     *
     * @param delegate An {@link EventClientDelegateMBean} object to work with.
     * Usually, this will be a proxy constructed using
     * {@link EventClientDelegate#getProxy}.
     * @param eventRelay An object used to receive notifications
     * forwarded by the {@link EventClientDelegateMBean}. If {@code null}, a
     * {@link FetchingEventRelay} object will be used.
     * @param distributingExecutor Used to distribute notifications to local
     * listeners. If {@code null}, the thread that calls {@link
     * EventReceiver#receive EventReceiver.receive} from the {@link EventRelay}
     * object is used.
     * @param leaseScheduler An object that will be used to schedule the
     * periodic {@linkplain EventClientDelegateMBean#lease lease updates}.
     * If {@code null}, a default scheduler will be used.
     * @param requestedLeaseTime The lease time used to keep this client alive
     * in the {@link EventClientDelegateMBean}.  A value of zero is equivalent
     * to the {@linkplain #DEFAULT_LEASE_TIMEOUT default value}.
     *
     * @throws IllegalArgumentException If {@code delegate} is null.
     * @throws IOException If an I/O error occurs when communicating with the
     * {@link EventClientDelegateMBean}.
     */
    public EventClient(EventClientDelegateMBean delegate,
            EventRelay eventRelay,
            Executor distributingExecutor,
            ScheduledExecutorService leaseScheduler,
            long requestedLeaseTime)
            throws IOException {
        if (delegate == null) {
            throw new IllegalArgumentException("Null EventClientDelegateMBean");
        }

        if (requestedLeaseTime == 0)
            requestedLeaseTime = DEFAULT_LEASE_TIMEOUT;
        else if (requestedLeaseTime < 0) {
            throw new IllegalArgumentException(
                    "Negative lease time: " + requestedLeaseTime);
        }

        eventClientDelegate = delegate;

        if (eventRelay != null) {
            this.eventRelay = eventRelay;
        } else {
            try {
                this.eventRelay = new FetchingEventRelay(delegate);
            } catch (IOException ioe) {
                throw ioe;
            } catch (Exception e) {
                // impossible?
                final IOException ioee = new IOException(e.toString());
                ioee.initCause(e);
                throw ioee;
            }
        }

        if (distributingExecutor == null)
            distributingExecutor = callerExecutor;
        this.distributingExecutor = distributingExecutor;
        this.dispatchingJob = new DispatchingJob();

        clientId = this.eventRelay.getClientId();

        this.requestedLeaseTime = requestedLeaseTime;
        if (leaseScheduler == null)
            leaseScheduler = defaultLeaseScheduler();
        leaseRenewer = new LeaseRenewer(leaseScheduler, renewLease);

        if (logger.traceOn()) {
            logger.trace("init", "New EventClient: "+clientId);
        }
    }

    private static ScheduledExecutorService defaultLeaseScheduler() {
        // The default lease scheduler uses a ScheduledThreadPoolExecutor
        // with a maximum of 20 threads.  This means that if you have many
        // EventClient instances and some of them get blocked (because of an
        // unresponsive network, for example), then even the instances that
        // are connected to responsive servers may have their leases expire.
        // XXX check if the above is true and possibly fix.
        PerThreadGroupPool.Create<ScheduledThreadPoolExecutor> create =
                new PerThreadGroupPool.Create<ScheduledThreadPoolExecutor>() {
            public ScheduledThreadPoolExecutor createThreadPool(ThreadGroup group) {
                ThreadFactory daemonThreadFactory = new DaemonThreadFactory(
                        "JMX EventClient lease renewer %d");
                ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(
                        20, daemonThreadFactory);
                exec.setKeepAliveTime(1, TimeUnit.SECONDS);
                exec.allowCoreThreadTimeOut(true);
                exec.setRemoveOnCancelPolicy(true);
                return exec;
            }
        };
        return leaseRenewerThreadPool.getThreadPoolExecutor(create);

    }

    /**
     * <p>Closes this EventClient, removes all listeners and stops receiving
     * notifications.</p>
     *
     * <p>This method calls {@link
     * EventClientDelegateMBean#removeClient(String)} and {@link
     * EventRelay#stop}.  Both operations occur even if one of them
     * throws an {@code IOException}.
     *
     * @throws IOException if an I/O error occurs when communicating with
     * {@link EventClientDelegateMBean}, or if {@link EventRelay#stop}
     * throws an {@code IOException}.
     */
    public void close() throws IOException {
        if (logger.traceOn()) {
            logger.trace("close", clientId);
        }

        synchronized(listenerInfoMap) {
            if (closed) {
                return;
            }

            closed = true;
            listenerInfoMap.clear();
        }

        if (leaseRenewer != null)
            leaseRenewer.close();

        IOException ioe = null;
        try {
            eventRelay.stop();
        } catch (IOException e) {
            ioe = e;
            logger.debug("close", "EventRelay.stop", e);
        }

        try {
            eventClientDelegate.removeClient(clientId);
        } catch (Exception e) {
            if (e instanceof IOException)
                ioe = (IOException) e;
            else
                ioe = new IOException(e);
            logger.debug("close",
                    "Got exception when removing "+clientId, e);
        }

        if (ioe != null)
            throw ioe;
    }

    /**
     * <p>Determine if this {@code EventClient} is closed.</p>
     *
     * @return True if the {@code EventClient} is closed.
     */
    public boolean closed() {
        return closed;
    }

    /**
     * <p>Return the {@link EventRelay} associated with this
     * {@code EventClient}.</p>
     *
     * @return The {@link EventRelay} object used.
     */
    public EventRelay getEventRelay() {
        return eventRelay;
    }

    /**
     * <p>Return the lease time that this {@code EventClient} requests
     * on every lease renewal.</p>
     *
     * @return The requested lease time.
     *
     * @see EventClientDelegateMBean#lease
     */
    public long getRequestedLeaseTime() {
        return requestedLeaseTime;
    }

    /**
     * @see javax.management.MBeanServerConnection#addNotificationListener(
     * ObjectName, NotificationListener, NotificationFilter, Object).
     */
    public void addNotificationListener(ObjectName name,
            NotificationListener listener,
            NotificationFilter filter,
            Object handback)
            throws InstanceNotFoundException, IOException {
        if (logger.traceOn()) {
            logger.trace("addNotificationListener", "");
        }

        checkState();

        Integer listenerId;
        try {
            listenerId =
                    eventClientDelegate.addListener(clientId, name, filter);
        } catch (EventClientNotFoundException ecnfe) {
            final IOException ioe = new IOException();
            ioe.initCause(ecnfe);
            throw ioe;
        }

        synchronized(listenerInfoMap) {
            listenerInfoMap.put(listenerId,  new ListenerInfo(
                    name,
                    listener,
                    filter,
                    handback,
                    false));
        }

        startListening();
    }

    /**
     * @see javax.management.MBeanServerConnection#removeNotificationListener(
     * ObjectName, NotificationListener).
     */
    public void removeNotificationListener(ObjectName name,
            NotificationListener listener)
            throws InstanceNotFoundException,
            ListenerNotFoundException,
            IOException {
        if (logger.traceOn()) {
            logger.trace("removeNotificationListener", "");
        }
        checkState();

        for (Integer id : getListenerInfo(name, listener, false)) {
            removeListener(id);
        }
    }

    /**
     * @see javax.management.MBeanServerConnection#removeNotificationListener(
     * ObjectName, NotificationListener, NotificationFilter, Object).
     */
    public void removeNotificationListener(ObjectName name,
            NotificationListener listener,
            NotificationFilter filter,
            Object handback)
            throws InstanceNotFoundException,
            ListenerNotFoundException,
            IOException {
        if (logger.traceOn()) {
            logger.trace("removeNotificationListener", "with all arguments.");
        }
        checkState();
        final Integer listenerId =
                getListenerInfo(name, listener, filter, handback, false);

        removeListener(listenerId);
    }

    /**
     * @see javax.management.event.EventConsumer#unsubscribe(
     * ObjectName, NotificationListener).
     */
    public void unsubscribe(ObjectName name,
            NotificationListener listener)
            throws ListenerNotFoundException, IOException {
        if (logger.traceOn()) {
            logger.trace("unsubscribe", "");
        }
        checkState();
        final Integer listenerId =
                getMatchedListenerInfo(name, listener, true);

        synchronized(listenerInfoMap) {
            if (listenerInfoMap.remove(listenerId) == null) {
                throw new ListenerNotFoundException();
            }
        }

        stopListening();

        try {
            eventClientDelegate.removeListenerOrSubscriber(clientId, listenerId);
        } catch (InstanceNotFoundException e) {
            logger.trace("unsubscribe", "removeSubscriber", e);
        } catch (EventClientNotFoundException cnfe) {
            logger.trace("unsubscribe", "removeSubscriber", cnfe);
        }
    }

    /**
     * @see javax.management.event.EventConsumer#subscribe(
     * ObjectName, NotificationListener, NotificationFilter, Object).
     */
    public void subscribe(ObjectName name,
            NotificationListener listener,
            NotificationFilter filter,
            Object handback) throws IOException {
        if (logger.traceOn()) {
            logger.trace("subscribe", "");
        }

        checkState();

        Integer listenerId;
        try {
            listenerId =
                    eventClientDelegate.addSubscriber(clientId, name, filter);
        } catch (EventClientNotFoundException ecnfe) {
            final IOException ioe = new IOException();
            ioe.initCause(ecnfe);
            throw ioe;
        }

        synchronized(listenerInfoMap) {
            listenerInfoMap.put(listenerId,  new ListenerInfo(
                    name,
                    listener,
                    filter,
                    handback,
                    true));
        }

        startListening();
    }

    /**
     * <p>Adds a set of listeners to the remote MBeanServer.  This method can
     * be used to copy the listeners from one {@code EventClient} to another.</p>
     *
     * <p>A listener is represented by a {@link ListenerInfo} object. The listener
     * is added by calling {@link #subscribe(ObjectName,
     * NotificationListener, NotificationFilter, Object)} if the method
     * {@link ListenerInfo#isSubscription() isSubscription}
     * returns {@code true}; otherwise it is added by calling
     * {@link #addNotificationListener(ObjectName, NotificationListener,
     * NotificationFilter, Object)}.</p>
     *
     * <P>The method returns the listeners which were added successfully. The
     * elements in the returned collection are a subset of the elements in
     * {@code infoList}. If all listeners were added successfully, the two
     * collections are the same. If no listener was added successfully, the
     * returned collection is empty.</p>
     *
     * @param listeners the listeners to add.
     *
     * @return The listeners that were added successfully.
     *
     * @throws IOException If an I/O error occurs.
     *
     * @see #getListeners()
     */
    public Collection<ListenerInfo> addListeners(Collection<ListenerInfo> listeners)
    throws IOException {
        if (logger.traceOn()) {
            logger.trace("addListeners", "");
        }

        checkState();

        if (listeners == null || listeners.isEmpty())
            return Collections.emptySet();

        final List<ListenerInfo> list = new ArrayList<ListenerInfo>();
        for (ListenerInfo l : listeners) {
            try {
                if (l.isSubscription()) {
                    subscribe(l.getObjectName(),
                            l.getListener(),
                            l.getFilter(),
                            l.getHandback());
                } else {
                    addNotificationListener(l.getObjectName(),
                            l.getListener(),
                            l.getFilter(),
                            l.getHandback());
                }

                list.add(l);
            } catch (Exception e) {
                if (logger.traceOn()) {
                    logger.trace("addListeners", "failed to add: "+l, e);
                }
            }
        }

        return list;
    }

    /**
     * Returns the set of listeners that have been added through
     * this {@code EventClient} and not subsequently removed.
     *
     * @return A collection of listener information. Empty if there are no
     * current listeners or if this {@code EventClient} has been {@linkplain
     * #close closed}.
     *
     * @see #addListeners
     */
    public Collection<ListenerInfo> getListeners() {
        if (logger.traceOn()) {
            logger.trace("getListeners", "");
        }

        synchronized(listenerInfoMap) {
            return Collections.unmodifiableCollection(listenerInfoMap.values());
        }
    }

    /**
     * Adds a listener to receive the {@code EventClient} notifications specified in
     * {@link #getEventClientNotificationInfo}.
     *
     * @param listener A listener to receive {@code EventClient} notifications.
     * @param filter A filter to select which notifications are to be delivered
     * to the listener, or {@code null} if all notifications are to be delivered.
     * @param handback An object to be given to the listener along with each
     * notification. Can be null.
     * @throws NullPointerException If listener is null.
     * @see #removeEventClientListener
     */
    public void addEventClientListener(NotificationListener listener,
            NotificationFilter filter,
            Object handback) {
        if (logger.traceOn()) {
            logger.trace("addEventClientListener", "");
        }
        broadcaster.addNotificationListener(listener, filter, handback);
    }

    /**
     * Removes a listener added to receive {@code EventClient} notifications specified in
     * {@link #getEventClientNotificationInfo}.
     *
     * @param listener A listener to receive {@code EventClient} notifications.
     * @throws NullPointerException If listener is null.
     * @throws ListenerNotFoundException If the listener is not added to
     * this {@code EventClient}.
     */
    public void removeEventClientListener(NotificationListener listener)
    throws ListenerNotFoundException {
        if (logger.traceOn()) {
            logger.trace("removeEventClientListener", "");
        }
        broadcaster.removeNotificationListener(listener);
    }

    /**
     * <p>Get the types of notification that an {@code EventClient} can send
     * to listeners added with {@link #addEventClientListener
     * addEventClientListener}.</p>
     *
     * @return Types of notification emitted by this {@code EventClient}.
     *
     * @see #FAILED
     * @see #NONFATAL
     * @see #NOTIFS_LOST
     */
    public MBeanNotificationInfo[] getEventClientNotificationInfo() {
        return myInfo.clone();
    }

    private static boolean match(ListenerInfo li,
            ObjectName name,
            NotificationListener listener,
            boolean subscribed) {
        return li.getObjectName().equals(name) &&
                li.getListener() == listener &&
                li.isSubscription() == subscribed;
    }

    private static boolean match(ListenerInfo li,
            ObjectName name,
            NotificationListener listener,
            NotificationFilter filter,
            Object handback,
            boolean subscribed) {
        return li.getObjectName().equals(name) &&
                li.getFilter() == filter &&
                li.getListener() == listener &&
                li.getHandback() == handback &&
                li.isSubscription() == subscribed;
    }

// ---------------------------------------------------
// private classes
// ---------------------------------------------------
    private class DispatchingJob extends RepeatedSingletonJob {
        public DispatchingJob() {
            super(distributingExecutor);
        }

        public boolean isSuspended() {
            return closed || buffer.size() == 0;
        }

        public void task() {
            TargetedNotification[] tns ;
            int lost = 0;

            synchronized(buffer) {
                tns = buffer.removeNotifs();
                lost = buffer.removeLost();
            }

            if ((tns == null || tns.length == 0)
            && lost == 0) {
                return;
            }

            // forwarding
            if (tns != null && tns.length > 0) {
                if (logger.traceOn()) {
                    logger.trace("DispatchingJob-task",
                            "Forwarding: "+tns.length);
                }
                for (TargetedNotification tn : tns) {
                    final ListenerInfo li = listenerInfoMap.get(tn.getListenerID());
                    try {
                        li.getListener().handleNotification(tn.getNotification(),
                                li.getHandback());
                    } catch (Exception e) {
                        logger.fine(
                                "DispatchingJob.task", "listener got exception", e);
                    }
                }
            }

            if (lost > 0) {
                if (logger.traceOn()) {
                    logger.trace("DispatchingJob-task",
                            "lost: "+lost);
                }
                final Notification n = new Notification(NOTIFS_LOST,
                        EventClient.this,
                        myNotifCounter.getAndIncrement(),
                        System.currentTimeMillis(),
                        "Lost notifications.");
                n.setUserData(new Long(lost));
                broadcaster.sendNotification(n);
            }
        }
    }


    private class EventReceiverImpl implements EventReceiver {
        public void receive(NotificationResult nr) {
            if (logger.traceOn()) {
                logger.trace("MyEventReceiver-receive", "");
            }

            synchronized(buffer) {
                buffer.addNotifs(nr);

                dispatchingJob.resume();
            }
        }

        public void failed(Throwable t) {
            if (logger.traceOn()) {
                logger.trace("MyEventReceiver-failed", "", t);
            }
            final Notification n = new Notification(FAILED,
                    this,
                    myNotifCounter.getAndIncrement(),
                    System.currentTimeMillis());
            n.setSource(t);
            broadcaster.sendNotification(n);
        }

        public void nonFatal(Exception e) {
            if (logger.traceOn()) {
                logger.trace("MyEventReceiver-nonFatal", "", e);
            }

            final Notification n = new Notification(NONFATAL,
                    this,
                    myNotifCounter.getAndIncrement(),
                    System.currentTimeMillis());
            n.setSource(e);
            broadcaster.sendNotification(n);
        }
    }

// ----------------------------------------------------
// private class
// ----------------------------------------------------


// ----------------------------------------------------
// private methods
// ----------------------------------------------------
    private Integer getListenerInfo(ObjectName name,
            NotificationListener listener,
            NotificationFilter filter,
            Object handback,
            boolean subscribed) throws ListenerNotFoundException {

        synchronized(listenerInfoMap) {
            for (Map.Entry<Integer, ListenerInfo> entry :
                    listenerInfoMap.entrySet()) {
                ListenerInfo li = entry.getValue();
                if (match(li, name, listener, filter, handback, subscribed)) {
                    return entry.getKey();
                }
            }
        }

        throw new ListenerNotFoundException();
    }

    private Integer getMatchedListenerInfo(ObjectName name,
            NotificationListener listener,
            boolean subscribed) throws ListenerNotFoundException {

        synchronized(listenerInfoMap) {
            for (Map.Entry<Integer, ListenerInfo> entry :
                    listenerInfoMap.entrySet()) {
                ListenerInfo li = entry.getValue();
                if (li.getObjectName().equals(name) &&
                        li.getListener() == listener &&
                        li.isSubscription() == subscribed) {
                    return entry.getKey();
                }
            }
        }

        throw new ListenerNotFoundException();
    }

    private Collection<Integer> getListenerInfo(ObjectName name,
            NotificationListener listener,
            boolean subscribed) throws ListenerNotFoundException {

        final ArrayList<Integer> ids = new ArrayList<Integer>();
        synchronized(listenerInfoMap) {
            for (Map.Entry<Integer, ListenerInfo> entry :
                    listenerInfoMap.entrySet()) {
                ListenerInfo li = entry.getValue();
                if (match(li, name, listener, subscribed)) {
                    ids.add(entry.getKey());
                }
            }
        }

        if (ids.isEmpty()) {
            throw new ListenerNotFoundException();
        }

        return ids;
    }

    private void checkState() throws IOException {
        synchronized(listenerInfoMap) {
            if (closed) {
                throw new IOException("Ended!");
            }
        }
    }

    private void startListening() throws IOException {
        synchronized(listenerInfoMap) {
            if (!startedListening && listenerInfoMap.size() > 0) {
                eventRelay.setEventReceiver(myReceiver);
            }

            startedListening = true;

            if (logger.traceOn()) {
                logger.trace("startListening", "listening");
            }
        }
    }

    private void stopListening() throws IOException {
        synchronized(listenerInfoMap) {
            if (listenerInfoMap.size() == 0 && startedListening) {
                eventRelay.setEventReceiver(null);

                startedListening = false;

                if (logger.traceOn()) {
                    logger.trace("stopListening", "non listening");
                }
            }
        }
    }

    private void removeListener(Integer id)
    throws InstanceNotFoundException,
            ListenerNotFoundException,
            IOException {
        synchronized(listenerInfoMap) {
            if (listenerInfoMap.remove(id) == null) {
                throw new ListenerNotFoundException();
            }

            stopListening();
        }

        try {
            eventClientDelegate.removeListenerOrSubscriber(clientId, id);
        } catch (EventClientNotFoundException cnfe) {
            logger.trace("removeListener", "ecd.removeListener", cnfe);
        }
    }


// ----------------------------------------------------
// private variables
// ----------------------------------------------------
    private static final ClassLogger logger =
            new ClassLogger("javax.management.event", "EventClient");

    private final Executor distributingExecutor;
    private final EventClientDelegateMBean eventClientDelegate;
    private final EventRelay eventRelay;
    private volatile String clientId = null;
    private final long requestedLeaseTime;

    private final ReceiverBuffer buffer = new ReceiverBuffer();

    private final EventReceiverImpl myReceiver =
            new EventReceiverImpl();
    private final DispatchingJob dispatchingJob;

    private final HashMap<Integer, ListenerInfo> listenerInfoMap =
            new HashMap<Integer, ListenerInfo>();

    private volatile boolean closed = false;

    private volatile boolean startedListening = false;

    // Could change synchronization here. But at worst a race will mean
    // sequence numbers are not contiguous, which may not matter much.
    private final AtomicLong myNotifCounter = new AtomicLong();

    private final static MBeanNotificationInfo[] myInfo =
            new MBeanNotificationInfo[] {
        new MBeanNotificationInfo(
                new String[] {FAILED, NOTIFS_LOST},
                Notification.class.getName(), "")};

    private final NotificationBroadcasterSupport broadcaster =
            new NotificationBroadcasterSupport();

    private final static Executor callerExecutor = new Executor() {
        // DirectExecutor using caller thread
        public void execute(Runnable r) {
            r.run();
        }
    };

    private static void checkInit(final MBeanServerConnection conn,
            final ObjectName delegateName)
            throws IOException {
        if (conn == null) {
            throw new IllegalArgumentException("No connection specified");
        }
        if (delegateName != null &&
                (!conn.isRegistered(delegateName))) {
            throw new IllegalArgumentException(
                    delegateName +
                    ": not found");
        }
        if (delegateName == null &&
                (!conn.isRegistered(
                EventClientDelegate.OBJECT_NAME))) {
            throw new IllegalArgumentException(
                    EventClientDelegate.OBJECT_NAME +
                    ": not found");
        }
    }

// ----------------------------------------------------
// private event lease issues
// ----------------------------------------------------
    private Callable<Long> renewLease = new Callable<Long>() {
        public Long call() throws IOException, EventClientNotFoundException {
            return eventClientDelegate.lease(clientId, requestedLeaseTime);
        }
    };

    private final LeaseRenewer leaseRenewer;

// ------------------------------------------------------------------------
    /**
     * Constructs an {@code MBeanServerConnection} that uses an {@code EventClient} object,
     * if the underlying connection has an {@link EventClientDelegateMBean}.
     * <P> The {@code EventClient} object creates a default
     * {@link FetchingEventRelay} object to
     * receive notifications forwarded by the {@link EventClientDelegateMBean}.
     * The {@link EventClientDelegateMBean} it works with is the
     * default one registered with the ObjectName
     * {@link EventClientDelegate#OBJECT_NAME
     * OBJECT_NAME}.
     * The thread from the {@link FetchingEventRelay} object that fetches the
     * notifications is also used to distribute them.
     *
     * @param conn An {@link MBeanServerConnection} object used to communicate
     * with an {@link EventClientDelegateMBean}.
     * @throws IllegalArgumentException If the value of {@code conn} is null,
     *         or the default {@link EventClientDelegateMBean} is not registered.
     * @throws IOException If an I/O error occurs.
     */
    public static MBeanServerConnection getEventClientConnection(
            final MBeanServerConnection conn)
            throws IOException {
        return getEventClientConnection(conn, null);
    }

    /**
     * Constructs an MBeanServerConnection that uses an {@code EventClient}
     * object with a user-specific {@link EventRelay}
     * object.
     * <P>
     * The {@link EventClientDelegateMBean} which it works with is the
     * default one registered with the ObjectName
     * {@link EventClientDelegate#OBJECT_NAME
     * OBJECT_NAME}
     * The thread that calls {@link EventReceiver#receive
     * EventReceiver.receive} from the {@link EventRelay} object is used
     * to distribute notifications to their listeners.
     *
     * @param conn An {@link MBeanServerConnection} object used to communicate
     * with an {@link EventClientDelegateMBean}.
     * @param eventRelay A user-specific object used to receive notifications
     * forwarded by the {@link EventClientDelegateMBean}. If null, the default
     * {@link FetchingEventRelay} object is used.
     * @throws IllegalArgumentException If the value of {@code conn} is null,
     *         or the default {@link EventClientDelegateMBean} is not registered.
     * @throws IOException If an I/O error occurs.
     */
    public static MBeanServerConnection getEventClientConnection(
            final MBeanServerConnection conn,
            final EventRelay eventRelay)
            throws IOException {

        if (newEventConn == null) {
            throw new IllegalArgumentException(
                    "Class not found: EventClientConnection");
        }

        checkInit(conn,null);
        final Callable<EventClient> factory = new Callable<EventClient>() {
            final public EventClient call() throws Exception {
                EventClientDelegateMBean ecd = EventClientDelegate.getProxy(conn);
                return new EventClient(ecd, eventRelay, null, null,
                        DEFAULT_LEASE_TIMEOUT);
            }
        };

        try {
            return (MBeanServerConnection)newEventConn.invoke(null,
                    conn, factory);
        } catch (Exception e) {
            throw new IllegalArgumentException(e);
        }
    }

    private static Method newEventConn = null;
    static {
        try {
            Class<?> c = Class.forName(
                    "com.sun.jmx.remote.util.EventClientConnection",
                    false, Thread.currentThread().getContextClassLoader());
            newEventConn = c.getMethod("getEventConnectionFor",
                    MBeanServerConnection.class, Callable.class);
        } catch (Exception e) {
            // OK: we're running in a subset of our classes
        }
    }

    /**
     * <p>Get the client id of this {@code EventClient} in the
     * {@link EventClientDelegateMBean}.
     *
     * @return the client id.
     *
     * @see EventClientDelegateMBean#addClient(String, Object[], String[])
     * EventClientDelegateMBean.addClient
     */
    public String getClientId() {
        return clientId;
    }

    /**
     * Returns a JMX Connector that will use an {@link EventClient}
     * to subscribe for notifications. If the server doesn't have
     * an {@link EventClientDelegateMBean}, then the connector will
     * use the legacy notification mechanism instead.
     *
     * @param wrapped The underlying JMX Connector wrapped by the returned
     *               connector.
     *
     * @return A JMX Connector that will uses an {@link EventClient}, if
     *         available.
     *
     * @see EventClient#getEventClientConnection(MBeanServerConnection)
     */
    public static JMXConnector withEventClient(final JMXConnector wrapped) {
        return JMXNamespaceUtils.withEventClient(wrapped);
    }

    private static final PerThreadGroupPool<ScheduledThreadPoolExecutor>
            leaseRenewerThreadPool = PerThreadGroupPool.make();
}