destinationmanager.java

一个java方面的消息订阅发送的源码

    // implement of GarbageCollectable.collectGarbage
    public synchronized void collectGarbage(boolean aggressive) {
        // before continuing we should change the priority of the thread
        // to the lowest value.
        int gc_caches = 0;
        int gc_destinations = 0;

        Object[] caches = _caches.values().toArray();
        for (int index = 0; index < caches.length; index++) {
            DestinationCache cache = (DestinationCache) caches[index];
            if (cache.canDestroy()) {
                if (_log.isDebugEnabled()) {
                    _log.debug("Garbage collecting destination cache="
                               + cache);
                }
                destroyDestinationCache(cache);
                gc_caches++;
            } else {
                // the cache is active, so issue a garbage collection
                // request on it
                cache.collectGarbage(aggressive);
            }
        }

        // get rid of non-administered destinations, without
        // associated caches.
        Iterator destinations = _destinationCache.values().iterator();
        Vector to_delete = new Vector();
        while (destinations.hasNext()) {
            JmsDestination dest = (JmsDestination) destinations.next();
            if (!(dest.getPersistent()) &&
                    (!_caches.containsKey(dest))) {
                to_delete.add(dest);
                gc_destinations++;
            }
        }

        // now delete the actual destinations
        Enumeration todel = to_delete.elements();
        while (todel.hasMoreElements()) {
            _destinationCache.remove(
                    ((JmsDestination) todel.nextElement()).getName());
        }

        // log the information
        _log.info("DMGC Collected " + gc_caches + " caches, " + _caches.size()
                  + " remaining.");
        _log.info("DMGC Collected " + gc_destinations + " destinations, "
                  + _destinationCache.size() + " remaining.");
    }

    /**
     * Return a HashMap of all destinations that match the specified topic If
     * the topic represents a wildcard then it may match none, one or more
     * destinations. The results are returns as a mapping of destination and the
     * corresponding cache
     * <p/>
     * The topic maybe a straight topic name or a wildcard
     *
     * @param topic - topic to query
     * @return HashMap
     */
    synchronized HashMap getTopicDestinationCaches(JmsTopic topic) {
        HashMap result = new HashMap();

        Iterator iter = _caches.keySet().iterator();
        while (iter.hasNext()) {
            JmsDestination dest = (JmsDestination) iter.next();
            if ((dest instanceof JmsTopic) &&
                    (topic.match((JmsTopic) dest))) {
                result.put(dest, _caches.get(dest));
            }
        }

        return result;
    }

    /**
     * Destroy this manager. This is brutal and final
     */
    public synchronized void destroy() {
        // clean up all the destinations
        Object[] dests = _caches.keySet().toArray();
        for (int index = 0; index < dests.length; index++) {
            destroyDestinationCache((JmsDestination) dests[index]);
        }

        _caches.clear();
        _caches = null;

        _destinationCache.clear();
        _destinationCache = null;

        // remove all the listeners
        _listeners.clear();
        _listeners = null;

        // reset the singleton
        _instance = null;
    }

    /**
     * Determines if a destination is persistent
     *
     * @param name the name of the destination
     * @return <code>true</code> if the named destination is persistent;
     *         <code>false</code> if it is temporary or doesn't exist
     */
    public boolean isPersistent(String name) {
        boolean result = false;
        JmsDestination destination =
                (JmsDestination) _destinationCache.get(name);
        if (destination != null) {
            result = destination.getPersistent();
        }

        return result;
    }

    /**
     * Determines if a destination is persistent
     *
     * @param destination the destination
     * @return <code>true</code> if the named destination is persistent;
     *         <code>false</code> if it is temporary or doesn't exist
     */
    public boolean isPersistent(JmsDestination destination) {
        return isPersistent(destination.getName());
    }

    /**
     * Test whether the specified message is for an administered destination.
     * This would be the case if the destination is administered or if there is
     * an administered wildcard destination that is satisfied by the message
     * destination
     *
     * @param msg - the message to check
     * @return boolean
     */
    public boolean isMessageForAdministeredDestination(MessageImpl msg) {
        boolean result = false;
        try {
            JmsDestination mdest = (JmsDestination) msg.getJMSDestination();
            JmsDestination dest = (JmsDestination) _destinationCache.get(
                    mdest.getName());

            if (dest != null) {
                if (dest.getPersistent()) {
                    result = true;
                } else if (mdest instanceof JmsTopic) {
                    // check if any of the wildcards are administered
                    Object[] dests = _wildcardDestinations.toArray();
                    for (int index = 0; index < dests.length; index++) {
                        JmsTopic adest = (JmsTopic) dests[index];
                        if ((adest.match((JmsTopic) mdest)) &&
                                (adest.getPersistent())) {
                            result = true;
                            break;
                        }
                    }

                }
            }
        } catch (JMSException ignore) {
        }

        return result;
    }

    /**
     * Add the specified entry to the destination cache, if it doesn't already
     * exist.
     *
     * @param destination - destination to add
     */
    void addToDestinationCache(JmsDestination destination) {
        synchronized (_destinationCache) {
            if (!_destinationCache.containsKey(destination.getName())) {
                _destinationCache.put(destination.getName(), destination);

                // check whether it is a wildcard destination
                if (((destination instanceof JmsTopic) &&
                        (((JmsTopic) destination).isWildCard()))) {
                    _wildcardDestinations.add(destination);
                }
            }
        }
    }

    /**
     * Remove the specified destination from the cache
     *
     * @param destination - the destination to remove
     */
    void removeFromDestinationCache(JmsDestination destination) {
        synchronized (_destinationCache) {
            if (_destinationCache.remove(destination.getName()) != null) {

                // check whether we also need to delete it from the
                // list of wildcard subscriptions
                if (((destination instanceof JmsTopic) &&
                        (((JmsTopic) destination).isWildCard()))) {
                    _wildcardDestinations.remove(destination);
                }
            }

        }
    }

    /**
     * Check if the specified destination exists.
     *
     * @param destination - the destination to check
     * @return boolean - true if it exists and false otherwise
     */
    public boolean destinationExists(JmsDestination destination) {
        return _destinationCache.containsKey(destination.getName());
    }

    /**
     * Initialises the destination manager.
     *
     * @throws ServiceException if the service cannot be initialised
     */
    protected void init() throws ServiceException {
        Connection connection = null;
        try {
            connection = DatabaseService.getConnection();

            // return a list of JmsDestination objects.
            Enumeration iter =
                    DatabaseService.getAdapter().getAllDestinations(connection);
            connection.commit();

            while (iter.hasMoreElements()) {
                // add each destination to the cache and also bind
                // it to the context
                JmsDestination dest = (JmsDestination) iter.nextElement();
                addToDestinationCache(dest);
                try {
                    // for each of the administered destinations rebind it to
                    // the jndi context
                    ContextHelper.rebind(getContext(), dest.getName(), dest);
                } catch (NamingException error) {
                    throw new ServiceException("Failed to add destination "
                                               + dest.getName()
                                               + " to JNDI", error);
                }
            }
        } catch (PersistenceException exception) {
            SQLHelper.rollback(connection);
            String msg = "Failed to initialise DestinationManager";
            _log.error(msg, exception);
            throw exception;
        } catch (SQLException exception) {
            SQLHelper.rollback(connection);
            String msg = "Failed to initialise DestinationManager";
            _log.error(msg, exception);
            throw new ServiceException(msg, exception);
        } finally {
            SQLHelper.close(connection);
        }
    }

    /**
     * Notyify the list of DestinationEventListener objects that the specified
     * destination has been added
     *
     * @param dest  - the destination that was added
     * @param cache - the corresponding cache
     */
    private void notifyDestinationAdded(JmsDestination dest,
                                        DestinationCache cache) {
        synchronized (_listeners) {
            Iterator iter = _listeners.iterator();
            while (iter.hasNext()) {
                ((DestinationEventListener) iter.next()).destinationAdded(dest,
                                                                          cache);
            }
        }
    }

    /**
     * Notyify the list of DestinationEventListener objects that the specified
     * destination has been removed
     *
     * @param dest  - the destination that was removed
     * @param cache - the corresponding cache
     */
    private void notifyDestinationRemoved(JmsDestination dest,
                                          DestinationCache cache) {
        synchronized (_listeners) {
            Iterator iter = _listeners.iterator();
            while (iter.hasNext()) {
                ((DestinationEventListener) iter.next()).destinationRemoved(
                        dest, cache);
            }
        }
    }

    /**
     * Return the initial context using the JndiHelper. It assumes that the
     * configuration manager has been successfully initialized. If the context
     * has been retrieved it is cached so subsequent gets will return the cached
     * instance
     *
     * @return the root context
     */
    private static Context getContext() throws NamingException {
        return NamingHelper.getInitialContext(ConfigurationManager.getConfig());
    }

}