/**
*
* Copyright 2004 Protique Ltd
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
**/
package org.codehaus.activemq.service.impl;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.activemq.broker.BrokerClient;
import org.codehaus.activemq.filter.Filter;
import org.codehaus.activemq.message.ConsumerInfo;
import org.codehaus.activemq.message.MessageAck;
import org.codehaus.activemq.service.Dispatcher;
import org.codehaus.activemq.service.MessageContainer;
import org.codehaus.activemq.service.QueueListEntry;
import org.codehaus.activemq.service.TopicMessageContainer;
import org.codehaus.activemq.service.RedeliveryPolicy;
import javax.jms.JMSException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
/**
* Represents a durable topic subscription where the consumer has a unique
* clientID used to persist the messages across both Broker restarts and
* JMS client restarts
*
* @version $Revision: 1.12 $
*/
public class DurableTopicSubscription extends SubscriptionImpl {
private static final Log log = LogFactory.getLog(DurableTopicSubscription.class);
private String persistentKey;
public DurableTopicSubscription(Dispatcher dispatcher, BrokerClient client, ConsumerInfo info, Filter filter, RedeliveryPolicy redeliveryPolicy) {
super(dispatcher, client, info, filter, redeliveryPolicy);
}
public synchronized void messageConsumed(MessageAck ack) throws JMSException {
if (!ack.isMessageRead() && !isBrowser()) {
super.messageConsumed(ack);
}
else {
Map lastMessagePointersPerContainer = new HashMap();
//remove up to this message
boolean found = false;
QueueListEntry queueEntry = messagePtrs.getFirstEntry();
while (queueEntry != null) {
MessagePointer pointer = (MessagePointer) queueEntry.getElement();
messagePtrs.remove(queueEntry);
lastMessagePointersPerContainer.put(pointer.getContainer(), pointer);
unconsumedMessagesDispatched.decrement();
if (pointer.getMessageIdentity().equals(ack.getMessageIdentity())) {
found = true;
break;
}
queueEntry = messagePtrs.getNextEntry(queueEntry);
}
if (!found) {
log.warn("Did not find a matching message for identity: " + ack.getMessageIdentity());
}
// now lets tell each container to update its lastAcknowlegedMessageID
for (Iterator iter = lastMessagePointersPerContainer.entrySet().iterator(); iter.hasNext();) {
Map.Entry entry = (Map.Entry) iter.next();
TopicMessageContainer container = (TopicMessageContainer) entry.getKey();
MessagePointer pointer = (MessagePointer) entry.getValue();
container.setLastAcknowledgedMessageID(this, pointer.getMessageIdentity());
}
//System.out.println("Message consumed. Remaining: " + messagePtrs.size() + " unconsumedMessagesDispatched: " + unconsumedMessagesDispatched.get());
dispatch.wakeup(this);
}
}
public synchronized void redeliverMessage(MessageContainer container, MessageAck ack) throws JMSException {
// rollback is rare so we don't have to be super efficient here
MessagePointer pointer = new MessagePointer(container, ack.getMessageIdentity());
pointer.setRedelivered(true);
messagePtrs.add(pointer);
unconsumedMessagesDispatched.increment();
}
public String getPersistentKey() {
if (persistentKey == null) {
persistentKey = "[" + getClientId() + ":" + getSubscriberName() + "]";
}
return persistentKey;
}
}