consumermanagerimpl.java

来自「OpenJMS是一个开源的Java Message Service API 1.」· Java 代码 · 共 919 行 · 第 1/3 页

     * @throws JMSException for any JMS error
     * @throws PersistenceException for any persistence error
     */
    private DurableConsumerEndpoint addDurableConsumer(JmsTopic topic,
                                                       String name,
                                                       String clientID)
            throws JMSException, PersistenceException {
        DurableConsumerEndpoint consumer;
        // cache the consumer locally
        addConsumerEntry(name, topic, clientID, true);

        long consumerId = getNextConsumerId();
        consumer = new DurableConsumerEndpoint(consumerId, topic, name,
                                               _destinations);
        _endpoints.put(consumer.getPersistentId(), consumer);
        return consumer;
    }

    /**
     * Add a consumer entry.
     *
     * @param key         a key to identify the entry.
     * @param destination the destination it is subscribed to. It can be a
     *                    wildcard
     * @param clientID    the client identifier. May be <code>null</code>
     * @param durable     indicates whether it is a durable subscription
     * @throws JMSException if key specifies a duplicate entry.
     */
    private void addConsumerEntry(Object key, JmsDestination destination,
                                  String clientID,
                                  boolean durable)
            throws JMSException {
        if (_log.isDebugEnabled()) {
            _log.debug("addConsumerEntry(key=" + key + ", destination="
                       + destination + ", clientID=" + clientID
                       + ", durable=" + durable + ")");
        }

        if (_consumers.containsKey(key)) {
            throw new JMSException("Duplicate consumer key:" + key);
        }

        ConsumerEntry entry = new ConsumerEntry(key, destination, clientID,
                                                durable);
        _consumers.put(key, entry);

        if (destination instanceof JmsTopic
                && ((JmsTopic) destination).isWildCard()) {
            // if the specified destination is a JmsTopic and also a wildcard
            // then we need to add it to all matching destinations
            _wildcardConsumers.put(entry, destination);
        } else {
            // we also need to add the reverse mapping
            List consumers = (List) _destToConsumerMap.get(destination);
            if (consumers == null) {
                consumers = new ArrayList();
                _destToConsumerMap.put(destination, consumers);
            }

            // add the mapping
            consumers.add(entry);
        }
    }

    /**
     * Remove the specified consumer from the cache.
     *
     * @param key the consumer key
     */
    private void removeConsumerEntry(Object key) {
        if (_log.isDebugEnabled()) {
            _log.debug("removeConsumerEntry(key=" + key + ")");
        }

        ConsumerEntry entry = (ConsumerEntry) _consumers.remove(key);
        if (entry != null) {
            JmsDestination dest = entry.getDestination();

            if (dest instanceof JmsTopic && ((JmsTopic) dest).isWildCard()) {
                // remove it from the wildcard cache.
                _wildcardConsumers.remove(entry);
            } else {
                // remove it from the specified destination
                List consumers = (List) _destToConsumerMap.get(dest);
                if (consumers != null) {
                    consumers.remove(entry);

                    // if consumers is of size 0 then remove it
                    if (consumers.isEmpty()) {
                        _destToConsumerMap.remove(dest);
                    }
                }
            }
        } else if (_log.isDebugEnabled()) {
            _log.debug("removeConsumerEntry(key=" + key
                       + "): consumer not found");
        }
    }

    /**
     * Remove all the consumers for the specified destination from the cache.
     *
     * @param destination the destination to remove
     */
    private void removeFromConsumerCache(JmsDestination destination) {
        _destToConsumerMap.remove(destination);
    }

    /**
     * Returns the next seed value to be allocated to a new consumer.
     *
     * @return a unique identifier for a consumer
     */
    private long getNextConsumerId() {
        return ++_consumerIdSeed;
    }

    /**
     * Returns the destination managed by {@link DestinationManager}
     * corresponding to that supplied, creating it if needed.
     *
     * @param destination the destination to look up
     * @param create      if <code>true</code> the destination may be created if
     *                    it doesn't exist
     * @return the destination managed by {@link DestinationManager}
     *         corresponding to <code>destination</code>.
     * @throws InvalidDestinationException if the destination doesn't exist and
     *                                     <code>create</code> is false; or the
     *                                     destination's properties don't match
     *                                     the existing destination
     * @throws JMSException                if the destination can't be created
     */
    private JmsDestination getDestination(JmsDestination destination,
                                          boolean create)
            throws InvalidDestinationException, JMSException {
        final String name = destination.getName();
        JmsDestination result;
        JmsDestination existing = _destinations.getDestination(name);

        if (existing == null) {
            if (!create) {
                throw new InvalidDestinationException(
                        "No destination with name=" + name + " exists");
            }
            // register the destination dynamically.
            _destinations.createDestination(destination);
            result = _destinations.getDestination(destination.getName());
        } else {
            // make sure the supplied destination has the same properties
            // as the existing one
            if (!destination.getClass().getName().equals(
                    existing.getClass().getName())) {
                throw new InvalidDestinationException(
                        "Mismatched destination properties for destination"
                        + " with name=" + name);
            }
            if (existing.getPersistent() != destination.getPersistent()) {
                throw new InvalidDestinationException(
                        "Mismatched destination properties for destination"
                        + " with name=" + name);
            }
            result = existing;
        }
        return result;
    }

    /**
     * Returns the consumers managed by this.
     *
     * @return an array of consumers
     */
    private ConsumerEndpoint[] getConsumers() {
        return (ConsumerEndpoint[]) _endpoints.values().toArray(
                new ConsumerEndpoint[0]);
    }

    /**
     * Rollback any transaction.
     */
    private void rollback() {
        try {
            if (_database.isTransacted()) {
                _database.rollback();
            }
        } catch (PersistenceException error) {
            _log.warn("Failed to rollback after error", error);
        }
    }

    /**
     * Helper to clean up after a failed call, and rethrow.
     * Any transaction will be rolled back.
     *
     * @param message   the message to log
     * @param exception the exception
     * @throws JMSException the original exception adapted to a
     *                      <code>JMSException</code> if necessary
     */
    private void rethrow(String message, Exception exception)
            throws JMSException {
        rollback();

        if (exception instanceof JMSException) {
            _log.debug(message, exception);
            throw (JMSException) exception;
        }
        // need to adapt the exception, so log as an error before rethrow
        _log.error(message, exception);
        throw new JMSException(exception.getMessage());
    }

    /**
     * Helper class used to maintain consumer information
     */
    private static final class ConsumerEntry {

        /**
         * An identifier for the consumer.
         */
        private final Object _key;

        /**
         * The destination that the consumer is subscribed to.
         */
        private final JmsDestination _destination;

        /**
         * The client identifier. May be <code>null</code>.
         */
        private final String _clientID;

        /**
         * Indicated whether this entry is for a durable subscriber
         */
        private final boolean _durable;


        /**
         * Construct a new <code>ConsumerEntry</code>.
         *
         * @param key         an identifier for the consumer
         * @param destination the destination consumer is subscribed to
         * @param clientID    the client identifier. May be <code>null</code>
         * @param durable     indicates whether it is a durable subscription
         */
        public ConsumerEntry(Object key, JmsDestination destination,
                             String clientID, boolean durable) {
            _key = key;
            _destination = destination;
            _clientID = clientID;
            _durable = durable;
        }

        public boolean equals(Object obj) {
            boolean result = false;
            if (obj instanceof ConsumerEntry) {
                result = ((ConsumerEntry) obj)._key.equals(_key);
            }

            return result;
        }

        public Object getKey() {
            return _key;
        }

        public String getName() {
            return (_key instanceof String) ? (String) _key : null;
        }

        public JmsDestination getDestination() {
            return _destination;
        }

        public String getClientID() {
            return _clientID;
        }

        public boolean isDurable() {
            return _durable;
        }

        /**
         * Helper to return a key for identifying {@link ConsumerEndpoint}
         * instances. This returns the consumers persistent identifier if it has
         * one; if not, it returns its transient identifier.
         *
         * @param consumer the consumer
         * @return a key for identifying <code>consumer</code>
         */
        public static Object getConsumerKey(ConsumerEndpoint consumer) {
            Object key = null;
            String id = consumer.getPersistentId();
            if (id != null) {
                key = id;
            } else {
                key = new Long(consumer.getId());
            }
            return key;
        }
    }

}