mirror of https://github.com/apache/activemq.git
AMQ-6901 - Make sure proper policy is used to configure a destination
When multiple wildcard policies exist in a hierarchy it was possible for the wrong policy to be selected when configuring a destination
This commit is contained in:
parent
0785029b88
commit
2ce1c1352f
|
@ -23,6 +23,8 @@ import java.util.List;
|
|||
import java.util.Set;
|
||||
import java.util.SortedSet;
|
||||
import java.util.TreeSet;
|
||||
import java.util.function.Supplier;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.apache.activemq.command.ActiveMQDestination;
|
||||
|
||||
|
@ -46,6 +48,7 @@ public class DestinationMap {
|
|||
private DestinationMapNode topicRootNode = new DestinationMapNode(null);
|
||||
private DestinationMapNode tempTopicRootNode = new DestinationMapNode(null);
|
||||
|
||||
|
||||
/**
|
||||
* Looks up the value(s) matching the given Destination key. For simple
|
||||
* destinations this is typically a List of one single value, for wildcards
|
||||
|
@ -202,19 +205,34 @@ public class DestinationMap {
|
|||
* @return the largest matching value or null if no value matches
|
||||
*/
|
||||
@SuppressWarnings({"rawtypes", "unchecked"})
|
||||
public Object chooseValue(final ActiveMQDestination destination) {
|
||||
Set set = get(destination);
|
||||
public DestinationMapEntry chooseValue(final ActiveMQDestination destination) {
|
||||
Set<DestinationMapEntry> set = get(destination);
|
||||
if (set == null || set.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
SortedSet sortedSet = new TreeSet(new Comparator<DestinationMapEntry>() {
|
||||
|
||||
//Comparator to sort in order - we want to pick the exact match by destination or the
|
||||
//closest parent that applies
|
||||
final Comparator<DestinationMapEntry> comparator = new Comparator<DestinationMapEntry>() {
|
||||
@Override
|
||||
public int compare(DestinationMapEntry entry1, DestinationMapEntry entry2) {
|
||||
return destination.equals(entry1.destination) ? -1 : (destination.equals(entry2.destination) ? 1 : entry1.compareTo(entry2));
|
||||
}
|
||||
});
|
||||
sortedSet.addAll(set);
|
||||
return sortedSet.first();
|
||||
};
|
||||
|
||||
//Sort and filter out any children and non matching entries
|
||||
final SortedSet<DestinationMapEntry> sortedSet = set.stream()
|
||||
.filter(entry -> isMatchOrParent(destination, (DestinationMapEntry)entry))
|
||||
.collect(Collectors.toCollection(() -> new TreeSet<DestinationMapEntry>(comparator)));
|
||||
|
||||
return sortedSet.size() > 0 ? sortedSet.first() : null;
|
||||
}
|
||||
|
||||
@SuppressWarnings("rawtypes")
|
||||
//Used to filter out any child/unmatching entries
|
||||
private boolean isMatchOrParent(final ActiveMQDestination destination, final DestinationMapEntry entry) {
|
||||
final DestinationFilter filter = DestinationFilter.parseFilter(entry.getDestination());
|
||||
return destination.equals(entry.getDestination()) || filter.matches(destination);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
@ -457,6 +457,54 @@ public class JavaPolicyEntryTest extends RuntimeConfigTestSupport {
|
|||
verifyQueueLimit("queue.test", 1024);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testModWithChildWildcardPolicies() throws Exception {
|
||||
BrokerService brokerService = new BrokerService();
|
||||
PolicyMap policyMap = new PolicyMap();
|
||||
PolicyEntry entry = new PolicyEntry();
|
||||
entry.setQueue(">");
|
||||
entry.setMemoryLimit(1024);
|
||||
PolicyEntry entry2 = new PolicyEntry();
|
||||
entry2.setQueue("queue.child.>");
|
||||
entry2.setMemoryLimit(2048);
|
||||
PolicyEntry entry3 = new PolicyEntry();
|
||||
entry3.setQueue("queue.child.one.>");
|
||||
entry3.setMemoryLimit(4096);
|
||||
|
||||
policyMap.setPolicyEntries(Arrays.asList(entry, entry2, entry3));
|
||||
brokerService.setDestinationPolicy(policyMap);
|
||||
|
||||
startBroker(brokerService);
|
||||
assertTrue("broker alive", brokerService.isStarted());
|
||||
|
||||
brokerService.getBroker().addDestination(
|
||||
brokerService.getAdminConnectionContext(), new ActiveMQQueue("queue.>"), false);
|
||||
brokerService.getBroker().addDestination(
|
||||
brokerService.getAdminConnectionContext(), new ActiveMQQueue("queue.child.>"), false);
|
||||
brokerService.getBroker().addDestination(
|
||||
brokerService.getAdminConnectionContext(), new ActiveMQQueue("queue.child.one.>"), false);
|
||||
brokerService.getBroker().addDestination(
|
||||
brokerService.getAdminConnectionContext(), new ActiveMQQueue("queue.child.one"), false);
|
||||
|
||||
//check destinations before policy updates
|
||||
verifyQueueLimit("queue.>", 1024);
|
||||
verifyQueueLimit("queue.child.>", 2048);
|
||||
verifyQueueLimit("queue.child.one", 4096);
|
||||
|
||||
//Reapply new limit to policy 2
|
||||
entry2.setMemoryLimit(4194304);
|
||||
javaConfigBroker.modifyPolicyEntry(entry2);
|
||||
TimeUnit.SECONDS.sleep(SLEEP);
|
||||
|
||||
//verify that destination at a higher level policy is not affected
|
||||
verifyQueueLimit("queue.>", 1024);
|
||||
|
||||
verifyQueueLimit("queue.child.>", 4194304);
|
||||
|
||||
verifyQueueLimit("queue.child.one.>", 4096);
|
||||
verifyQueueLimit("queue.child.one", 4096);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testModParentPolicy() throws Exception {
|
||||
BrokerService brokerService = new BrokerService();
|
||||
|
|
Loading…
Reference in New Issue