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 org.apache.commons.logging.Log;
    21  import org.apache.commons.logging.LogFactory;
    22  import org.codehaus.activemq.broker.BrokerClient;
    23  import org.codehaus.activemq.filter.Filter;
    24  import org.codehaus.activemq.message.ConsumerInfo;
    25  import org.codehaus.activemq.message.MessageAck;
    26  import org.codehaus.activemq.service.Dispatcher;
    27  import org.codehaus.activemq.service.MessageContainer;
    28  import org.codehaus.activemq.service.QueueListEntry;
    29  import org.codehaus.activemq.service.TopicMessageContainer;
    30  import org.codehaus.activemq.service.RedeliveryPolicy;
    31
    32  import javax.jms.JMSException;
    33  import java.util.HashMap;
    34  import java.util.Iterator;
    35  import java.util.Map;
    36
    37  /***
    38   * Represents a durable topic subscription where the consumer has a unique
    39   * clientID used to persist the messages across both Broker restarts and
    40   * JMS client restarts
    41   *
    42   * @version $Revision: 1.12 $
    43   */
    44  public class DurableTopicSubscription extends SubscriptionImpl {
    45
    46      private static final Log log = LogFactory.getLog(DurableTopicSubscription.class);
    47
    48      private String persistentKey;
    49
    50      public DurableTopicSubscription(Dispatcher dispatcher, BrokerClient client, ConsumerInfo info, Filter filter, RedeliveryPolicy redeliveryPolicy) {
    51          super(dispatcher, client, info, filter, redeliveryPolicy);
    52      }
    53
    54      public synchronized void messageConsumed(MessageAck ack) throws JMSException {
    55          if (!ack.isMessageRead() && !isBrowser()) {
    56              super.messageConsumed(ack);
    57          }
    58          else {
    59              Map lastMessagePointersPerContainer = new HashMap();
    60
    61              //remove up to this message
    62              boolean found = false;
    63              QueueListEntry queueEntry = messagePtrs.getFirstEntry();
    64              while (queueEntry != null) {
    65                  MessagePointer pointer = (MessagePointer) queueEntry.getElement();
    66
    67                  messagePtrs.remove(queueEntry);
    68                  lastMessagePointersPerContainer.put(pointer.getContainer(), pointer);
    69                  unconsumedMessagesDispatched.decrement();
    70
    71                  if (pointer.getMessageIdentity().equals(ack.getMessageIdentity())) {
    72                      found = true;
    73                      break;
    74                  }
    75                  queueEntry = messagePtrs.getNextEntry(queueEntry);
    76              }
    77              if (!found) {
    78                  log.warn("Did not find a matching message for identity: " + ack.getMessageIdentity());
    79              }
    80
    81              // now lets tell each container to update its lastAcknowlegedMessageID
    82              for (Iterator iter = lastMessagePointersPerContainer.entrySet().iterator(); iter.hasNext();) {
    83                  Map.Entry entry = (Map.Entry) iter.next();
    84                  TopicMessageContainer container = (TopicMessageContainer) entry.getKey();
    85                  MessagePointer pointer = (MessagePointer) entry.getValue();
    86                  container.setLastAcknowledgedMessageID(this, pointer.getMessageIdentity());
    87              }
    88
    89              //System.out.println("Message consumed. Remaining: " + messagePtrs.size() + " unconsumedMessagesDispatched: " + unconsumedMessagesDispatched.get());
    90              dispatch.wakeup(this);
    91          }
    92      }
    93
    94      public synchronized void redeliverMessage(MessageContainer container, MessageAck ack) throws JMSException {
    95          // rollback is rare so we don't have to be super efficient here
    96          MessagePointer pointer = new MessagePointer(container, ack.getMessageIdentity());
    97          pointer.setRedelivered(true);
    98          messagePtrs.add(pointer);
    99          unconsumedMessagesDispatched.increment();
    100     }
    101
    102
    103     public String getPersistentKey() {
    104         if (persistentKey == null) {
    105             persistentKey = "[" + getClientId() + ":" + getSubscriberName() + "]";
    106         }
    107         return persistentKey;
    108     }
    109 }