mirror of https://github.com/apache/activemq.git
minor refactor of a property and exposed the selector as an MBean attribute. the selector can actually be updated in real time via JMX for non-durable topic subscriptions for AMQ-625
git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@384486 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
b86da466e6
commit
5371cf5e64
|
@ -90,7 +90,7 @@ public class ActiveMQQueueBrowser implements
|
|||
/**
|
||||
* @param session
|
||||
* @param originalDestination
|
||||
* @param selector
|
||||
* @param selectorExpression
|
||||
* @param cnum
|
||||
* @return
|
||||
* @throws JMSException
|
||||
|
|
|
@ -20,6 +20,8 @@ import org.apache.activemq.broker.region.Subscription;
|
|||
import org.apache.activemq.command.ActiveMQDestination;
|
||||
import org.apache.activemq.command.ConsumerInfo;
|
||||
|
||||
import javax.jms.InvalidSelectorException;
|
||||
|
||||
|
||||
|
||||
/**
|
||||
|
@ -91,7 +93,22 @@ public class SubscriptionView implements SubscriptionViewMBean {
|
|||
return dest.getPhysicalName();
|
||||
}
|
||||
return "NOTSET";
|
||||
|
||||
}
|
||||
|
||||
public String getSelector() {
|
||||
if (subscription != null) {
|
||||
return subscription.getSelector();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
public void setSelector(String selector) throws InvalidSelectorException, UnsupportedOperationException {
|
||||
if (subscription != null) {
|
||||
subscription.setSelector(selector);
|
||||
}
|
||||
else {
|
||||
throw new UnsupportedOperationException("No subscription object");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
@ -13,6 +13,8 @@
|
|||
*/
|
||||
package org.apache.activemq.broker.jmx;
|
||||
|
||||
import javax.jms.InvalidSelectorException;
|
||||
|
||||
/**
|
||||
* @version $Revision: 1.5 $
|
||||
*/
|
||||
|
@ -42,6 +44,18 @@ public interface SubscriptionViewMBean{
|
|||
*/
|
||||
public String getDestinationName();
|
||||
|
||||
|
||||
/**
|
||||
* @return the JMS selector on the current subscription
|
||||
*/
|
||||
public String getSelector();
|
||||
|
||||
/**
|
||||
* Attempts to change the current active selector on the subscription.
|
||||
* This operation is not supported for persistent topics.
|
||||
*/
|
||||
public void setSelector(String selector) throws InvalidSelectorException, UnsupportedOperationException;
|
||||
|
||||
/**
|
||||
* @return true if the destination is a Queue
|
||||
*/
|
||||
|
|
|
@ -45,7 +45,7 @@ abstract public class AbstractSubscription implements Subscription {
|
|||
protected ConnectionContext context;
|
||||
protected ConsumerInfo info;
|
||||
final protected DestinationFilter destinationFilter;
|
||||
final protected BooleanExpression selector;
|
||||
private BooleanExpression selectorExpression;
|
||||
|
||||
final protected CopyOnWriteArrayList destinations = new CopyOnWriteArrayList();
|
||||
|
||||
|
@ -54,7 +54,7 @@ abstract public class AbstractSubscription implements Subscription {
|
|||
this.context = context;
|
||||
this.info = info;
|
||||
this.destinationFilter = DestinationFilter.parseFilter(info.getDestination());
|
||||
this.selector = parseSelector(info);
|
||||
this.selectorExpression = parseSelector(info);
|
||||
}
|
||||
|
||||
static private BooleanExpression parseSelector(ConsumerInfo info) throws InvalidSelectorException {
|
||||
|
@ -86,7 +86,7 @@ abstract public class AbstractSubscription implements Subscription {
|
|||
return false;
|
||||
}
|
||||
try {
|
||||
return (selector == null || selector.matches(context)) && this.context.isAllowedToConsume(node);
|
||||
return (selectorExpression == null || selectorExpression.matches(context)) && this.context.isAllowedToConsume(node);
|
||||
} catch (JMSException e) {
|
||||
log.info("Selector failed to evaluate: " + e.getMessage(), e);
|
||||
return false;
|
||||
|
@ -124,7 +124,20 @@ abstract public class AbstractSubscription implements Subscription {
|
|||
return info;
|
||||
}
|
||||
|
||||
public BooleanExpression getSelector() {
|
||||
return selector;
|
||||
public BooleanExpression getSelectorExpression() {
|
||||
return selectorExpression;
|
||||
}
|
||||
|
||||
public String getSelector() {
|
||||
return info.getSelector();
|
||||
}
|
||||
|
||||
public void setSelector(String selector) throws InvalidSelectorException {
|
||||
ConsumerInfo copy = info.copy();
|
||||
copy.setSelector(selector);
|
||||
BooleanExpression newSelector = parseSelector(copy);
|
||||
// its valid so lets actually update it now
|
||||
info.setSelector(selector);
|
||||
this.selectorExpression = newSelector;
|
||||
}
|
||||
}
|
||||
|
|
|
@ -16,10 +16,7 @@
|
|||
*/
|
||||
package org.apache.activemq.broker.region;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Iterator;
|
||||
|
||||
import javax.jms.InvalidSelectorException;
|
||||
import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import org.apache.activemq.broker.Broker;
|
||||
import org.apache.activemq.broker.ConnectionContext;
|
||||
|
@ -29,7 +26,10 @@ import org.apache.activemq.command.MessageAck;
|
|||
import org.apache.activemq.command.MessageDispatch;
|
||||
import org.apache.activemq.util.SubscriptionKey;
|
||||
|
||||
import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
|
||||
import javax.jms.InvalidSelectorException;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Iterator;
|
||||
|
||||
public class DurableTopicSubscription extends PrefetchSubscription {
|
||||
|
||||
|
@ -131,6 +131,9 @@ public class DurableTopicSubscription extends PrefetchSubscription {
|
|||
return 0;
|
||||
}
|
||||
|
||||
public void setSelector(String selector) throws InvalidSelectorException {
|
||||
throw new UnsupportedOperationException("You cannot dynamically change the selector for durable topic subscriptions");
|
||||
}
|
||||
|
||||
protected boolean canDispatch(MessageReference node) {
|
||||
return active;
|
||||
|
|
|
@ -25,6 +25,8 @@ import org.apache.activemq.command.MessageAck;
|
|||
import org.apache.activemq.command.MessageDispatchNotification;
|
||||
import org.apache.activemq.filter.MessageEvaluationContext;
|
||||
|
||||
import javax.jms.InvalidSelectorException;
|
||||
|
||||
/**
|
||||
* @version $Revision: 1.5 $
|
||||
*/
|
||||
|
@ -125,4 +127,14 @@ public interface Subscription {
|
|||
*/
|
||||
long getDequeueCounter();
|
||||
|
||||
/**
|
||||
* @return the JMS selector on the current subscription
|
||||
*/
|
||||
public String getSelector();
|
||||
|
||||
/**
|
||||
* Attempts to change the current active selector on the subscription.
|
||||
* This operation is not supported for persistent topics.
|
||||
*/
|
||||
public void setSelector(String selector) throws InvalidSelectorException, UnsupportedOperationException;
|
||||
}
|
Loading…
Reference in New Issue