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 }