protected void doAddMessageConsumer(BrokerClient client, ConsumerInfo info) throws JMSException {
boolean shouldRecover = false;
if (info.getConsumerName() != null && info.getClientId() != null) {
for (Iterator iter = activeSubscriptions.values().iterator();iter.hasNext();) {
Subscription subscription = (Subscription) iter.next();
if (subscription.isSameDurableSubscription(info)) {
throw new DuplicateDurableSubscriptionException(info);
}
}
}
Subscription subscription = subscriptionContainer.getSubscription(info.getConsumerId());
if (subscription != null && subscription.isDurableTopic()) {
//check the subscription hasn't changed
if (!subscription.getDestination().equals(subscription.getDestination())
|| !subscription.getSelector().equals(info.getSelector())) {
subscriptionContainer.removeSubscription(info.getConsumerId());
subscription.clear();
subscription = subscriptionContainer.makeSubscription(dispatcher, client,info, createFilter(info));
}
}
else {
subscription = subscriptionContainer.makeSubscription(dispatcher, client,info, createFilter(info));
shouldRecover = true;
}
subscription.setActiveConsumer(client,info);
activeSubscriptions.put(info.getConsumerId(), subscription);
dispatcher.addActiveSubscription(client, subscription);
if (subscription.isWildcard()) {
synchronized (this) {
if (!loadedMessageContainers) {
loadAllMessageContainers();
loadedMessageContainers = true;
}
}
}
else {
// load the container
getContainer(subscription.getDestination().getPhysicalName());
}
Set containers = destinationMap.get(subscription.getDestination());
for (Iterator iter = containers.iterator();iter.hasNext();) {
TopicMessageContainer container = (TopicMessageContainer) iter.next();
if (container instanceof DurableTopicMessageContainer) {
((DurableTopicMessageContainer) container).storeSubscription(info, subscription);
}
}
if (shouldRecover) {
recoverSubscriptions(subscription);
}
// lets not make the subscription active until later
// as we can't start dispatching until we've sent back the receipt
// TODO we might wish to register a post-receipt action here
// to perform the wakeup
subscription.setActive(true);
//dispatcher.wakeup(subscription);
}