Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-4995 - reapply destination interceptor after change

This commit is contained in:
Dejan Bosanac 2014-01-28 14:53:18 +01:00
parent ad3041034b
commit 55da9bc821
9 changed files with 59 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
activemq-broker/src/main/java/org/apache/activemq/broker/BrokerFilter.java
View File

@ -365,6 +365,11 @@ public class BrokerFilter implements Broker {
next.processConsumerControl(consumerExchange, control);
}
@Override
public void reapplyInterceptor() {
next.reapplyInterceptor();
}
@Override
public Scheduler getScheduler() {
return next.getScheduler();

5
activemq-broker/src/main/java/org/apache/activemq/broker/EmptyBroker.java
View File

@ -357,6 +357,11 @@ public class EmptyBroker implements Broker {
ConsumerControl control) {
}
@Override
public void reapplyInterceptor() {
}
@Override
public Scheduler getScheduler() {
return null;

5
activemq-broker/src/main/java/org/apache/activemq/broker/ErrorBroker.java
View File

@ -367,6 +367,11 @@ public class ErrorBroker implements Broker {
throw new BrokerStoppedException(this.message);
}
@Override
public void reapplyInterceptor() {
throw new BrokerStoppedException(this.message);
}
@Override
public Scheduler getScheduler() {
throw new BrokerStoppedException(this.message);

5
activemq-broker/src/main/java/org/apache/activemq/broker/MutableBrokerFilter.java
View File

@ -376,6 +376,11 @@ public class MutableBrokerFilter implements Broker {
getNext().processConsumerControl(consumerExchange, control);
}
@Override
public void reapplyInterceptor() {
getNext().reapplyInterceptor();
}
@Override
public Scheduler getScheduler() {
return getNext().getScheduler();

15
activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
View File

@ -31,6 +31,7 @@ import org.apache.activemq.broker.ConsumerBrokerExchange;
import org.apache.activemq.DestinationDoesNotExistException;
import org.apache.activemq.broker.ProducerBrokerExchange;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.virtual.CompositeDestinationFilter;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerControl;
import org.apache.activemq.command.ConsumerId;
@ -580,4 +581,18 @@ public abstract class AbstractRegion implements Region {
}
}
}
public void reapplyInterceptor() {
DestinationInterceptor destinationInterceptor = broker.getDestinationInterceptor();
Map<ActiveMQDestination, Destination> map = getDestinationMap();
for (ActiveMQDestination key : map.keySet()) {
Destination destination = map.get(key);
if (destination instanceof CompositeDestinationFilter) {
destination = ((CompositeDestinationFilter)destination).next;
}
destination = destinationInterceptor.intercept(destination);
getDestinationMap().put(key, destination);
destinations.put(key, destination);
}
}
}

2
activemq-broker/src/main/java/org/apache/activemq/broker/region/Region.java
View File

@ -149,5 +149,7 @@ public interface Region extends Service {
Set <Destination>getDestinations(ActiveMQDestination destination);
void processConsumerControl(ConsumerBrokerExchange consumerExchange, ConsumerControl control);
void reapplyInterceptor();
}

7
activemq-broker/src/main/java/org/apache/activemq/broker/region/RegionBroker.java
View File

@ -859,4 +859,11 @@ public class RegionBroker extends EmptyBroker {
public void setAllowTempAutoCreationOnSend(boolean allowTempAutoCreationOnSend) {
this.allowTempAutoCreationOnSend = allowTempAutoCreationOnSend;
}
public void reapplyInterceptor() {
queueRegion.reapplyInterceptor();
topicRegion.reapplyInterceptor();
tempQueueRegion.reapplyInterceptor();
tempTopicRegion.reapplyInterceptor();
}
}

1
activemq-runtime-config/src/main/java/org/apache/activemq/plugin/RuntimeConfigurationBroker.java
View File

@ -580,6 +580,7 @@ public class RuntimeConfigurationBroker extends BrokerFilter {
RegionBroker regionBroker = (RegionBroker) getBrokerService().getRegionBroker();
((CompositeDestinationInterceptor) regionBroker.getDestinationInterceptor()).setInterceptors(destinationInterceptors);
info("applied new: " + interceptorsList);
regionBroker.reapplyInterceptor();
}
}
});

14
activemq-runtime-config/src/test/java/org/apache/activemq/VirtualDestTest.java
View File

@ -80,6 +80,20 @@ public class VirtualDestTest extends RuntimeConfigTestSupport {
exerciseCompositeQueue("VirtualDestination.CompositeQueue", "VirtualDestination.QueueConsumer");
}
@Test
public void testModComposite() throws Exception {
final String brokerConfig = configurationSeed + "-mod-composite-vd-broker";
applyNewConfig(brokerConfig, configurationSeed + "-add-composite-vd");
startBroker(brokerConfig);
assertTrue("broker alive", brokerService.isStarted());
exerciseCompositeQueue("VirtualDestination.CompositeQueue", "VirtualDestination.QueueConsumer");
applyNewConfig(brokerConfig, configurationSeed + "-mod-composite-vd", SLEEP);
exerciseCompositeQueue("VirtualDestination.CompositeQueue", "VirtualDestination.QueueConsumer");
exerciseCompositeQueue("VirtualDestination.CompositeQueue", "VirtualDestination.CompositeQueue");
}
@Test
public void testNewNoDefaultVirtualTopicSupport() throws Exception {
final String brokerConfig = configurationSeed + "-no-vd-vt-broker";