View Javadoc

    1   /***
    2    *
    3    * Copyright 2004 Protique Ltd
    4    *
    5    * Licensed under the Apache License, Version 2.0 (the "License");
    6    * you may not use this file except in compliance with the License.
    7    * You may obtain a copy of the License at
    8    *
    9    * http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   **/
    18  package org.codehaus.activemq.service.impl;
    19
    20  import EDU.oswego.cs.dl.util.concurrent.ConcurrentHashMap;
    21  import org.apache.commons.logging.Log;
    22  import org.apache.commons.logging.LogFactory;
    23  import org.codehaus.activemq.broker.BrokerClient;
    24  import org.codehaus.activemq.filter.AndFilter;
    25  import org.codehaus.activemq.filter.Filter;
    26  import org.codehaus.activemq.filter.FilterFactory;
    27  import org.codehaus.activemq.filter.FilterFactoryImpl;
    28  import org.codehaus.activemq.filter.NoLocalFilter;
    29  import org.codehaus.activemq.message.ActiveMQDestination;
    30  import org.codehaus.activemq.message.ActiveMQMessage;
    31  import org.codehaus.activemq.message.ConsumerInfo;
    32  import org.codehaus.activemq.message.MessageAck;
    33  import org.codehaus.activemq.service.Dispatcher;
    34  import org.codehaus.activemq.service.MessageContainer;
    35  import org.codehaus.activemq.service.QueueList;
    36  import org.codehaus.activemq.service.QueueListEntry;
    37  import org.codehaus.activemq.service.QueueMessageContainer;
    38  import org.codehaus.activemq.service.Subscription;
    39  import org.codehaus.activemq.service.SubscriptionContainer;
    40  import org.codehaus.activemq.store.PersistenceAdapter;
    41
    42  import javax.jms.JMSException;
    43  import java.util.Iterator;
    44  import java.util.Map;
    45
    46  /***
    47   * A default Broker used for Queue messages
    48   *
    49   * @version $Revision: 1.14 $
    50   */
    51  public class QueueMessageContainerManager extends MessageContainerManagerSupport {
    52      private static final Log log = LogFactory.getLog(QueueMessageContainerManager.class);
    53      private static final int MAX_MESSAGES_DISPATCHED_FROM_POLL = 50;
    54
    55      private PersistenceAdapter persistenceAdapter;
    56      protected SubscriptionContainer subscriptionContainer;
    57      protected FilterFactory filterFactory;
    58      protected Map activeSubscriptions = new ConcurrentHashMap();
    59      protected Map browsers = new ConcurrentHashMap();
    60      private Object subscriptionMutex = new Object();
    61
    62      public QueueMessageContainerManager(PersistenceAdapter persistenceAdapter) {
    63          this(persistenceAdapter, new SubscriptionContainerImpl(), new FilterFactoryImpl(), new DispatcherImpl());
    64      }
    65
    66      public QueueMessageContainerManager(PersistenceAdapter persistenceAdapter, SubscriptionContainer subscriptionContainer, FilterFactory filterFactory, Dispatcher dispatcher) {
    67          super(dispatcher);
    68          this.persistenceAdapter = persistenceAdapter;
    69          this.subscriptionContainer = subscriptionContainer;
    70          this.filterFactory = filterFactory;
    71      }
    72
    73      /***
    74       * @param client
    75       * @param info
    76       * @throws javax.jms.JMSException
    77       */
    78      public void addMessageConsumer(BrokerClient client, ConsumerInfo info) throws JMSException {
    79          if (log.isDebugEnabled()) {
    80              log.debug("Adding consumer: " + info);
    81          }
    82          if (info.getDestination().isQueue()) {
    83              Subscription sub = subscriptionContainer.makeSubscription(dispatcher, info, createFilter(info));
    84              dispatcher.addActiveSubscription(client, sub);
    85              updateActiveSubscriptions(sub);
    86
    87              // set active last in case we end up dispatching some messages
    88              // while recovering
    89              sub.setActive(true);
    90          }
    91      }
    92
    93      /***
    94       * @param client
    95       * @param info
    96       * @throws javax.jms.JMSException
    97       */
    98      public void removeMessageConsumer(BrokerClient client, ConsumerInfo info) throws JMSException {
    99          if (log.isDebugEnabled()) {
    100             log.debug("Removing consumer: " + info);
    101         }
    102         if (info.getDestination() != null && info.getDestination().isQueue()) {
    103             synchronized (subscriptionMutex) {
    104                 Subscription sub = (Subscription) subscriptionContainer.removeSubscription(info.getConsumerId());
    105                 if (sub != null) {
    106                     sub.setActive(false);
    107                     sub.clear();//resets entries in the QueueMessageContainer
    108                     dispatcher.removeActiveSubscription(client, sub);
    109                     //need to do wildcards for this - but for now use exact matches
    110                     for (Iterator iter = messageContainers.values().iterator(); iter.hasNext();) {
    111                         QueueMessageContainer container = (QueueMessageContainer) iter.next();
    112                         //should change this for wild cards ...
    113                         if (container.getDestinationName().equals(sub.getDestination().getPhysicalName())) {
    114                             QueueList list = getSubscriptionList(container);
    115                             list.remove(sub);
    116                             if (list.isEmpty()) {
    117                                 activeSubscriptions.remove(sub.getDestination().getPhysicalName());
    118                             }
    119                             list = getBrowserList(container);
    120                             list.remove(sub);
    121                             if (list.isEmpty()) {
    122                                 browsers.remove(sub.getDestination().getPhysicalName());
    123                             }
    124                         }
    125                     }
    126                 }
    127             }
    128         }
    129     }
    130
    131     /***
    132      * Delete a durable subscriber
    133      *
    134      * @param clientId
    135      * @param subscriberName
    136      * @throws javax.jms.JMSException if the subscriber doesn't exist or is still active
    137      */
    138     public void deleteSubscription(String clientId, String subscriberName) throws JMSException {
    139     }
    140
    141     /***
    142      * @param client
    143      * @param message
    144      * @throws javax.jms.JMSException
    145      */
    146     public void sendMessage(BrokerClient client, ActiveMQMessage message) throws JMSException {
    147         ActiveMQDestination dest = (ActiveMQDestination) message.getJMSDestination();
    148         if (dest != null && dest.isQueue()) {
    149             if (log.isDebugEnabled()) {
    150                 log.debug("Dispaching message: " + message);
    151             }
    152             QueueMessageContainer container = (QueueMessageContainer) getContainer(((ActiveMQDestination) message.getJMSDestination())
    153                     .getPhysicalName());
    154             container.addMessage(message);
    155             dispatcher.wakeup();
    156         }
    157     }
    158
    159     /***
    160      * Acknowledge a message as being read and consumed by the Consumer
    161      *
    162      * @param client
    163      * @param ack
    164      * @throws javax.jms.JMSException
    165      */
    166     public void acknowledgeMessage(BrokerClient client, MessageAck ack) throws JMSException {
    167         Subscription sub = subscriptionContainer.getSubscription(ack.getConsumerId());
    168         if (sub != null) {
    169             sub.messageConsumed(ack);
    170         }
    171     }
    172
    173     public void acknowledgeTransactedMessage(BrokerClient client, String transactionId, MessageAck ack) throws JMSException {
    174         Subscription sub = subscriptionContainer.getSubscription(ack.getConsumerId());
    175         if (sub != null) {
    176             sub.onAcknowledgeTransactedMessageBeforeCommit(ack);
    177         }
    178     }
    179
    180     public void redeliverMessage(BrokerClient client, MessageAck ack) throws JMSException {
    181         Subscription sub = subscriptionContainer.getSubscription(ack.getConsumerId());
    182         if (sub != null) {
    183             sub.redeliverMessage(null, ack);
    184         }
    185     }
    186
    187     /***
    188      * Poll for messages
    189      *
    190      * @throws javax.jms.JMSException
    191      */
    192     public void poll() throws JMSException {
    193         synchronized (subscriptionMutex) {
    194             for (Iterator iter = activeSubscriptions.keySet().iterator(); iter.hasNext();) {
    195                 QueueMessageContainer container = (QueueMessageContainer) iter.next();
    196
    197                 QueueList browserList = (QueueList) browsers.get(container);
    198                 doPeek(container, browserList);
    199                 QueueList list = (QueueList) activeSubscriptions.get(container);
    200                 doPoll(container, list);
    201             }
    202         }
    203     }
    204
    205     public void commitTransaction(BrokerClient client, String transactionId) {
    206     }
    207
    208     public void rollbackTransaction(BrokerClient client, String transactionId) {
    209     }
    210
    211     public MessageContainer getContainer(String destinationName) throws JMSException {
    212         QueueMessageContainer container = null;
    213         synchronized (subscriptionMutex) {
    214             container = (QueueMessageContainer) messageContainers.get(destinationName);
    215             if (container == null) {
    216                 container = persistenceAdapter.createQueueMessageContainer(destinationName);
    217                 container.start();
    218                 messageContainers.put(destinationName, container);
    219                 //Add any interested Subscriptions to the new Container
    220                 for (Iterator iter = subscriptionContainer.subscriptionIterator(); iter.hasNext();) {
    221                     Subscription sub = (Subscription) iter.next();
    222                     if (sub.isBrowser()) {
    223                         updateBrowsers(container, sub);
    224                     }
    225                     else {
    226                         updateActiveSubscriptions(container, sub);
    227                     }
    228                 }
    229             }
    230         }
    231         return container;
    232     }
    233
    234     // Implementation methods
    235     //-------------------------------------------------------------------------
    236
    237     private void doPeek(QueueMessageContainer container, QueueList browsers) throws JMSException {
    238         if (browsers != null && browsers.size() > 0) {
    239             for (int i = 0; i < browsers.size(); i++) {
    240                 SubscriptionImpl sub = (SubscriptionImpl) browsers.get(i);
    241                 int count = 0;
    242                 ActiveMQMessage msg = null;
    243                 do {
    244                     msg = container.peekNext(sub.getLastMessageIdentity());
    245                     if (msg != null) {
    246                         if (sub.isTarget(msg)) {
    247                             sub.addMessage(container, msg);
    248                             dispatcher.wakeup(sub);
    249                         }
    250                         else {
    251                             sub.setLastMessageIdentifier(msg.getJMSMessageIdentity());
    252                         }
    253                     }
    254                 }
    255                 while (msg != null && !sub.isAtPrefetchLimit() && count++ < MAX_MESSAGES_DISPATCHED_FROM_POLL);
    256             }
    257         }
    258     }
    259
    260     private void doPoll(QueueMessageContainer container, QueueList subList) throws JMSException {
    261         int count = 0;
    262         ActiveMQMessage msg = null;
    263         if (subList != null && subList.size() > 0) {
    264             do {
    265                 boolean dispatched = false;
    266                 msg = container.poll();
    267                 if (msg != null) {
    268                     QueueListEntry entry = subList.getFirstEntry();
    269                     boolean targeted = false;
    270                     while (entry != null) {
    271                         SubscriptionImpl sub = (SubscriptionImpl) entry.getElement();
    272                         if (sub.isTarget(msg)) {
    273                             targeted = true;
    274                             if (!sub.isAtPrefetchLimit()) {
    275                                 sub.addMessage(container, msg);
    276                                 dispatched = true;
    277                                 dispatcher.wakeup(sub);
    278                                 subList.rotate(); //round-robin the list
    279                                 break;
    280                             }
    281                         }
    282                         entry = subList.getNextEntry(entry);
    283                     }
    284                     if (!dispatched) {
    285                         if (targeted) { //ie. it can be selected by current active consumers - but they are at
    286                             // pre-fectch
    287                             // limit
    288                             container.returnMessage(msg.getJMSMessageIdentity());
    289                         }
    290                         break;
    291                     }
    292                 }
    293             }
    294             while (msg != null && count++ < MAX_MESSAGES_DISPATCHED_FROM_POLL);
    295         }
    296     }
    297
    298     private void updateActiveSubscriptions(Subscription subscription) throws JMSException {
    299         //need to do wildcards for this - but for now use exact matches
    300         synchronized (subscriptionMutex) {
    301             boolean processedSubscriptionContainer = false;
    302
    303             String subscriptionPhysicalName = subscription.getDestination().getPhysicalName();
    304             for (Iterator iter = messageContainers.entrySet().iterator(); iter.hasNext();) {
    305                 Map.Entry entry = (Map.Entry) iter.next();
    306                 String destinationName = (String) entry.getKey();
    307                 QueueMessageContainer container = (QueueMessageContainer) entry.getValue();
    308
    309                 if (destinationName.equals(subscriptionPhysicalName)) {
    310                     processedSubscriptionContainer = true;
    311                 }
    312                 processSubscription(subscription, container);
    313             }
    314             if (!processedSubscriptionContainer) {
    315                 processSubscription(subscription, (QueueMessageContainer) getContainer(subscriptionPhysicalName));
    316             }
    317         }
    318     }
    319
    320     protected void processSubscription(Subscription subscription, QueueMessageContainer container) throws JMSException {
    321         // TODO should change this for wild cards ...
    322         if (subscription.isBrowser()) {
    323             updateBrowsers(container, subscription);
    324         }
    325         else {
    326             updateActiveSubscriptions(container, subscription);
    327         }
    328     }
    329
    330     private void updateActiveSubscriptions(QueueMessageContainer container, Subscription sub) throws JMSException {
    331         //need to do wildcards for this - but for now use exact matches
    332         //should change this for wild cards ...
    333         if (container.getDestinationName().equals(sub.getDestination().getPhysicalName())) {
    334             container.reset();//reset container - flushing all filter out messages to new consumer
    335             QueueList list = getSubscriptionList(container);
    336             if (!list.contains(sub)) {
    337                 list.add(sub);
    338             }
    339         }
    340     }
    341
    342     private QueueList getSubscriptionList(QueueMessageContainer container) {
    343         QueueList list = (QueueList) activeSubscriptions.get(container);
    344         if (list == null) {
    345             list = new DefaultQueueList();
    346             activeSubscriptions.put(container, list);
    347         }
    348         return list;
    349     }
    350
    351     private void updateBrowsers(QueueMessageContainer container, Subscription sub) throws JMSException {
    352         //need to do wildcards for this - but for now use exact matches
    353         //should change this for wild cards ...
    354         if (container.getDestinationName().equals(sub.getDestination().getPhysicalName())) {
    355             container.reset();//reset container - flushing all filter out messages to new consumer
    356             QueueList list = getBrowserList(container);
    357             if (!list.contains(sub)) {
    358                 list.add(sub);
    359             }
    360         }
    361     }
    362
    363     private QueueList getBrowserList(QueueMessageContainer container) {
    364         QueueList list = (QueueList) browsers.get(container);
    365         if (list == null) {
    366             list = new DefaultQueueList();
    367             browsers.put(container, list);
    368         }
    369         return list;
    370     }
    371
    372     /***
    373      * Create filter for a Consumer
    374      *
    375      * @param info
    376      * @return the Fitler
    377      * @throws javax.jms.JMSException
    378      */
    379     protected Filter createFilter(ConsumerInfo info) throws JMSException {
    380         Filter filter = filterFactory.createFilter(info.getDestination(), info.getSelector());
    381         if (info.isNoLocal()) {
    382             filter = new AndFilter(filter, new NoLocalFilter(info.getClientId()));
    383         }
    384         return filter;
    385     }
    386
    387 }