You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cs...@apache.org on 2015/12/14 20:09:41 UTC
[1/5] activemq git commit:
https://issues.apache.org/jira/browse/AMQ-6077 - test
Repository: activemq
Updated Branches:
refs/heads/activemq-5.13.x f8488916c -> 27b08b03c
https://issues.apache.org/jira/browse/AMQ-6077 - test
(cherry picked from commit 7951037f20e8d6752ab0bbd90189a8276bc61a55)
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/f3aedc75
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/f3aedc75
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/f3aedc75
Branch: refs/heads/activemq-5.13.x
Commit: f3aedc753c877cd947c6eb968f1ca8d8d1809f8a
Parents: f848891
Author: Dejan Bosanac <de...@nighttale.net>
Authored: Mon Dec 14 14:02:31 2015 +0100
Committer: Christopher L. Shannon (cshannon) <ch...@gmail.com>
Committed: Mon Dec 14 19:09:14 2015 +0000
----------------------------------------------------------------------
.../activemq/camel/ObjectMessageTest.java | 83 +++++++++++++++++
.../apache/activemq/camel/ObjectPayload.java | 25 ++++++
.../activemq/camel/jms-object-message.xml | 93 ++++++++++++++++++++
3 files changed, 201 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/f3aedc75/activemq-camel/src/test/java/org/apache/activemq/camel/ObjectMessageTest.java
----------------------------------------------------------------------
diff --git a/activemq-camel/src/test/java/org/apache/activemq/camel/ObjectMessageTest.java b/activemq-camel/src/test/java/org/apache/activemq/camel/ObjectMessageTest.java
new file mode 100644
index 0000000..226cfd3
--- /dev/null
+++ b/activemq-camel/src/test/java/org/apache/activemq/camel/ObjectMessageTest.java
@@ -0,0 +1,83 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.activemq.camel;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.jms.JmsBinding;
+import org.apache.camel.component.jms.JmsMessage;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.spring.CamelSpringTestSupport;
+import org.apache.camel.util.ExchangeHelper;
+import org.apache.xbean.spring.context.ClassPathXmlApplicationContext;
+import org.junit.Test;
+import org.springframework.context.support.AbstractApplicationContext;
+
+import javax.jms.*;
+import java.util.concurrent.TimeUnit;
+
+public class ObjectMessageTest extends CamelSpringTestSupport {
+
+ @Test
+ public void testUntrusted() throws Exception {
+ ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");
+ Connection conn = factory.createConnection();
+ conn.start();
+ Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = sess.createProducer(sess.createTopic("foo"));
+ ObjectMessage msg = sess.createObjectMessage();
+ ObjectPayload payload = new ObjectPayload();
+ payload.payload = "test";
+ msg.setObject(payload);
+ producer.send(msg);
+
+ Thread.sleep(1000);
+
+ MockEndpoint resultActiveMQ = resolveMandatoryEndpoint("mock:result-activemq", MockEndpoint.class);
+ resultActiveMQ.expectedMessageCount(1);
+ resultActiveMQ.assertIsSatisfied();
+ assertCorrectObjectReceived(resultActiveMQ);
+
+ MockEndpoint resultTrusted = resolveMandatoryEndpoint("mock:result-trusted", MockEndpoint.class);
+ resultTrusted.expectedMessageCount(1);
+ resultTrusted.assertIsSatisfied();
+ assertCorrectObjectReceived(resultTrusted);
+
+ MockEndpoint resultCamel = resolveMandatoryEndpoint("mock:result-camel", MockEndpoint.class);
+ resultCamel.expectedMessageCount(0);
+ resultCamel.assertIsSatisfied(1, TimeUnit.SECONDS);
+
+ }
+
+ protected void assertCorrectObjectReceived(MockEndpoint result) {
+ Exchange exchange = result.getReceivedExchanges().get(0);
+ // This should be a JMS Exchange
+ assertNotNull(ExchangeHelper.getBinding(exchange, JmsBinding.class));
+ JmsMessage in = (JmsMessage) exchange.getIn();
+ assertNotNull(in);
+ assertIsInstanceOf(ObjectMessage.class, in.getJmsMessage());
+
+ ObjectPayload received = exchange.getIn().getBody(ObjectPayload.class);
+ assertEquals("test", received.payload);
+ }
+
+ @Override
+ protected AbstractApplicationContext createApplicationContext() {
+ AbstractApplicationContext context = new ClassPathXmlApplicationContext("org/apache/activemq/camel/jms-object-message.xml");
+ return context;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq/blob/f3aedc75/activemq-camel/src/test/java/org/apache/activemq/camel/ObjectPayload.java
----------------------------------------------------------------------
diff --git a/activemq-camel/src/test/java/org/apache/activemq/camel/ObjectPayload.java b/activemq-camel/src/test/java/org/apache/activemq/camel/ObjectPayload.java
new file mode 100644
index 0000000..301d03c
--- /dev/null
+++ b/activemq-camel/src/test/java/org/apache/activemq/camel/ObjectPayload.java
@@ -0,0 +1,25 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.activemq.camel;
+
+import java.io.Serializable;
+
+public class ObjectPayload implements Serializable {
+ private static final long serialVersionUID = 121277L;
+
+ public String payload;
+}
http://git-wip-us.apache.org/repos/asf/activemq/blob/f3aedc75/activemq-camel/src/test/resources/org/apache/activemq/camel/jms-object-message.xml
----------------------------------------------------------------------
diff --git a/activemq-camel/src/test/resources/org/apache/activemq/camel/jms-object-message.xml b/activemq-camel/src/test/resources/org/apache/activemq/camel/jms-object-message.xml
new file mode 100644
index 0000000..a4534c0
--- /dev/null
+++ b/activemq-camel/src/test/resources/org/apache/activemq/camel/jms-object-message.xml
@@ -0,0 +1,93 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You 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.
+-->
+<beans xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="
+ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
+ http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
+ ">
+
+ <camelContext xmlns="http://camel.apache.org/schema/spring">
+ <route>
+ <from uri="activemq-activemq:topic:foo"/>
+ <to uri="mock:result-activemq"/>
+ </route>
+ <route>
+ <from uri="activemq-camel:topic:foo"/>
+ <to uri="mock:result-camel"/>
+ </route>
+ <route>
+ <from uri="activemq-trusted:topic:foo"/>
+ <to uri="mock:result-trusted"/>
+ </route>
+ </camelContext>
+
+ <!-- configuration for activemq-camel endpoint -->
+
+ <bean id="camelConnectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
+ <property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
+ <property name="trustedPackages">
+ <list>
+ <value>org.apache.camel</value>
+ </list>
+ </property>
+ </bean>
+
+ <bean id="camelConfig" class="org.apache.camel.component.jms.JmsConfiguration">
+ <property name="connectionFactory" ref="camelConnectionFactory"/>
+ </bean>
+
+ <bean id="activemq-camel" class="org.apache.activemq.camel.component.ActiveMQComponent">
+ <property name="configuration" ref="camelConfig"/>
+ </bean>
+
+ <!-- configuration for activemq-activemq endpoint -->
+
+ <bean id="activemqConnectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
+ <property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
+ <property name="trustedPackages">
+ <list>
+ <value>org.apache.activemq</value>
+ </list>
+ </property>
+ </bean>
+
+ <bean id="activemqConfig" class="org.apache.camel.component.jms.JmsConfiguration">
+ <property name="connectionFactory" ref="activemqConnectionFactory"/>
+ </bean>
+
+ <bean id="activemq-activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
+ <property name="configuration" ref="activemqConfig"/>
+ </bean>
+
+ <!-- configuration for activemq-trusted endpoint -->
+
+ <bean id="trustedConnectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
+ <property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
+ <property name="trustAllPackages" value="true"/>
+ </bean>
+
+ <bean id="trustedConfig" class="org.apache.camel.component.jms.JmsConfiguration">
+ <property name="connectionFactory" ref="trustedConnectionFactory"/>
+ </bean>
+
+ <bean id="activemq-trusted" class="org.apache.activemq.camel.component.ActiveMQComponent">
+ <property name="configuration" ref="trustedConfig"/>
+ </bean>
+
+</beans>
[5/5] activemq git commit:
https://issues.apache.org/jira/browse/AMQ-6088
Posted by cs...@apache.org.
https://issues.apache.org/jira/browse/AMQ-6088
The runtime plugins will now find the exact policy to update which means
that a destination can match more than one policy and the policy can
still be updated at runtime.
The java runtime broker also supports the ability to replace or add a
policy entry based on a flag on a new method call.
(cherry picked from commit 9e7fae0d83c584f98e99024ba6d20e53f14b81f7)
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/27b08b03
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/27b08b03
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/27b08b03
Branch: refs/heads/activemq-5.13.x
Commit: 27b08b03c58fcaaf11c278ac28a4da472e55d274
Parents: fc4120e
Author: Christopher L. Shannon (cshannon) <ch...@gmail.com>
Authored: Mon Dec 14 18:51:11 2015 +0000
Committer: Christopher L. Shannon (cshannon) <ch...@gmail.com>
Committed: Mon Dec 14 19:09:16 2015 +0000
----------------------------------------------------------------------
.../AbstractRuntimeConfigurationBroker.java | 6 +-
.../activemq/plugin/PolicyEntryProcessor.java | 33 +--
.../java/JavaRuntimeConfigurationBroker.java | 86 +++++--
.../activemq/plugin/util/PolicyEntryUtil.java | 93 +++++++
.../org/apache/activemq/PolicyEntryTest.java | 38 +++
.../activemq/java/JavaPolicyEntryTest.java | 256 +++++++++++++++++++
.../policyEntryTest-policy-ml-child-mod.xml | 37 +++
.../policyEntryTest-policy-ml-parent-mod.xml | 37 +++
.../policyEntryTest-policy-ml-parent.xml | 37 +++
9 files changed, 573 insertions(+), 50 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/27b08b03/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/AbstractRuntimeConfigurationBroker.java
----------------------------------------------------------------------
diff --git a/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/AbstractRuntimeConfigurationBroker.java b/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/AbstractRuntimeConfigurationBroker.java
index a430339..c672579 100644
--- a/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/AbstractRuntimeConfigurationBroker.java
+++ b/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/AbstractRuntimeConfigurationBroker.java
@@ -143,11 +143,11 @@ public class AbstractRuntimeConfigurationBroker extends BrokerFilter {
}
}
- protected void debug(String s) {
+ public void debug(String s) {
LOG.debug(s);
}
- protected void info(String s) {
+ public void info(String s) {
LOG.info(filterPasswords(s));
if (infoString != null) {
infoString += s;
@@ -155,7 +155,7 @@ public class AbstractRuntimeConfigurationBroker extends BrokerFilter {
}
}
- protected void info(String s, Throwable t) {
+ public void info(String s, Throwable t) {
LOG.info(filterPasswords(s), t);
if (infoString != null) {
infoString += s;
http://git-wip-us.apache.org/repos/asf/activemq/blob/27b08b03/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/PolicyEntryProcessor.java
----------------------------------------------------------------------
diff --git a/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/PolicyEntryProcessor.java b/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/PolicyEntryProcessor.java
index 4d0dc42..2bf7535 100644
--- a/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/PolicyEntryProcessor.java
+++ b/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/PolicyEntryProcessor.java
@@ -16,15 +16,13 @@
*/
package org.apache.activemq.plugin;
-import org.apache.activemq.broker.region.*;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
-
-import java.util.Set;
+import org.apache.activemq.plugin.util.PolicyEntryUtil;
public class PolicyEntryProcessor extends DefaultConfigurationProcessor {
- public PolicyEntryProcessor(RuntimeConfigurationBroker plugin, Class configurationClass) {
+ public PolicyEntryProcessor(RuntimeConfigurationBroker plugin, Class<?> configurationClass) {
super(plugin, configurationClass);
}
@@ -39,33 +37,22 @@ public class PolicyEntryProcessor extends DefaultConfigurationProcessor {
@Override
public void modify(Object existing, Object candidate) {
- PolicyMap existingMap = plugin.getBrokerService().getDestinationPolicy();
-
PolicyEntry updatedEntry = fromDto(candidate, new PolicyEntry());
- Set existingEntry = existingMap.get(updatedEntry.getDestination());
- if (existingEntry.size() == 1) {
- updatedEntry = fromDto(candidate, (PolicyEntry) existingEntry.iterator().next());
+ //Look up an existing entry that matches the candidate
+ //First just look up by the destination type to see if anything matches
+ PolicyEntry existingEntry = PolicyEntryUtil.findEntryByDestination(plugin, updatedEntry);
+ if (existingEntry != null) {
+ //if found, update the policy and apply the updates to existing destinations
+ updatedEntry = fromDto(candidate, existingEntry);
applyRetrospectively(updatedEntry);
plugin.info("updated policy for: " + updatedEntry.getDestination());
} else {
- plugin.info("cannot modify policy matching multiple destinations: " + existingEntry + ", destination:" + updatedEntry.getDestination());
+ plugin.info("cannot find policy entry candidate to update: " + updatedEntry + ", destination:" + updatedEntry.getDestination());
}
}
protected void applyRetrospectively(PolicyEntry updatedEntry) {
- RegionBroker regionBroker = (RegionBroker) plugin.getBrokerService().getRegionBroker();
- for (Destination destination : regionBroker.getDestinations(updatedEntry.getDestination())) {
- Destination target = destination;
- if (destination instanceof DestinationFilter) {
- target = ((DestinationFilter)destination).getNext();
- }
- if (target.getActiveMQDestination().isQueue()) {
- updatedEntry.update((Queue) target);
- } else if (target.getActiveMQDestination().isTopic()) {
- updatedEntry.update((Topic) target);
- }
- plugin.debug("applied update to:" + target);
- }
+ PolicyEntryUtil.applyRetrospectively(plugin, updatedEntry);
}
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/27b08b03/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/java/JavaRuntimeConfigurationBroker.java
----------------------------------------------------------------------
diff --git a/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/java/JavaRuntimeConfigurationBroker.java b/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/java/JavaRuntimeConfigurationBroker.java
index ca4fd50..8e4ed29 100644
--- a/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/java/JavaRuntimeConfigurationBroker.java
+++ b/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/java/JavaRuntimeConfigurationBroker.java
@@ -17,14 +17,8 @@
package org.apache.activemq.plugin.java;
import java.util.Arrays;
-import java.util.Set;
import org.apache.activemq.broker.Broker;
-import org.apache.activemq.broker.region.Destination;
-import org.apache.activemq.broker.region.DestinationFilter;
-import org.apache.activemq.broker.region.Queue;
-import org.apache.activemq.broker.region.RegionBroker;
-import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.broker.region.virtual.VirtualDestination;
@@ -32,6 +26,7 @@ import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.network.DiscoveryNetworkConnector;
import org.apache.activemq.plugin.AbstractRuntimeConfigurationBroker;
import org.apache.activemq.plugin.UpdateVirtualDestinationsTask;
+import org.apache.activemq.plugin.util.PolicyEntryUtil;
import org.apache.activemq.security.AuthorizationBroker;
import org.apache.activemq.security.AuthorizationMap;
import org.apache.activemq.security.SimpleAuthenticationBroker;
@@ -142,32 +137,75 @@ public class JavaRuntimeConfigurationBroker extends AbstractRuntimeConfiguration
info("added policy for: " + addition.getDestination());
}
+
+ /**
+ * This method will modify an existing policy entry that matches the destination
+ * set on the PolicyEntry passed in.
+ *
+ * The PolicyEntry reference must already be in the PolicyMap or it won't be updated.
+ * To modify the entry the best way is to look up the existing PolicyEntry from the
+ * PolicyMap, make changes to it, and pass it to this method to apply.
+ *
+ * To create or replace an existing entry (if the destination matches), see
+ * {@link #modifyPolicyEntry(PolicyEntry, boolean)
+ *
+ *
+ * @param existing
+ */
public void modifyPolicyEntry(PolicyEntry existing) {
+ modifyPolicyEntry(existing, false);
+ }
+
+ /**
+ * This method will modify an existing policy entry that matches the destination
+ * set on the PolicyEntry passed in. If createOrReplace is true, a new policy
+ * will be created if it doesn't exist and a policy will be replaced in the PolicyMap,
+ * versus modified, if it is a different reference but the destinations for the Policy match.
+ *
+ * If createOrReplace is false, the policy update will only be applied if
+ * the PolicyEntry reference already exists in the PolicyMap.
+ *
+ * @param existing
+ * @param createIfAbsent
+ */
+ public void modifyPolicyEntry(PolicyEntry existing, boolean createOrReplace) {
PolicyMap existingMap = this.getBrokerService().getDestinationPolicy();
- Set<?> existingEntry = existingMap.get(existing.getDestination());
- if (existingEntry.size() == 1) {
- applyRetrospectively(existing);
- this.info("updated policy for: " + existing.getDestination());
+ //First just look up by the destination type to see if anything matches
+ PolicyEntry existingEntry = PolicyEntryUtil.findEntryByDestination(this, existing);
+
+ //handle createOrReplace
+ if (createOrReplace) {
+ //if not found at all, go ahead and insert the policy entry
+ if (existingEntry == null) {
+ existingMap.put(existing.getDestination(), existing);
+ existingEntry = existing;
+ //If found but the objects are different, remove the old policy entry
+ //and replace it with the new one
+ } else if (!existing.equals(existingEntry)) {
+ synchronized(existingMap) {
+ existingMap.remove(existingEntry.getDestination(), existingEntry);
+ existingMap.put(existing.getDestination(), existing);
+ }
+ existingEntry = existing;
+ }
+ }
+
+ //Make sure that at this point the passed in object and the entry in
+ //the map are the same
+ if (existingEntry != null && existingEntry.equals(existing)) {
+ applyRetrospectively(existingEntry);
+ this.info("updated policy for: " + existingEntry.getDestination());
} else {
- this.info("cannot modify policy matching multiple destinations: " + existingEntry + ", destination:" + existing.getDestination());
+ throw new IllegalArgumentException("The policy can not be updated because it either does not exist or the PolicyEntry"
+ + " reference does not match an existing PolicyEntry in the PolicyMap. To replace an"
+ + " entry (versus modifying) or add, set createOrReplace to true. "
+ + existing + ", destination:" + existing.getDestination());
}
}
protected void applyRetrospectively(PolicyEntry updatedEntry) {
- RegionBroker regionBroker = (RegionBroker) this.getBrokerService().getRegionBroker();
- for (Destination destination : regionBroker.getDestinations(updatedEntry.getDestination())) {
- Destination target = destination;
- if (destination instanceof DestinationFilter) {
- target = ((DestinationFilter)destination).getNext();
- }
- if (target.getActiveMQDestination().isQueue()) {
- updatedEntry.update((Queue) target);
- } else if (target.getActiveMQDestination().isTopic()) {
- updatedEntry.update((Topic) target);
- }
- this.debug("applied update to:" + target);
- }
+ PolicyEntryUtil.applyRetrospectively(this, updatedEntry);
}
//authentication plugin
http://git-wip-us.apache.org/repos/asf/activemq/blob/27b08b03/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/util/PolicyEntryUtil.java
----------------------------------------------------------------------
diff --git a/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/util/PolicyEntryUtil.java b/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/util/PolicyEntryUtil.java
new file mode 100644
index 0000000..7f38f5b
--- /dev/null
+++ b/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/util/PolicyEntryUtil.java
@@ -0,0 +1,93 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.activemq.plugin.util;
+
+import java.util.Set;
+
+import org.apache.activemq.broker.region.Destination;
+import org.apache.activemq.broker.region.DestinationFilter;
+import org.apache.activemq.broker.region.Queue;
+import org.apache.activemq.broker.region.RegionBroker;
+import org.apache.activemq.broker.region.Topic;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.plugin.AbstractRuntimeConfigurationBroker;
+
+
+public class PolicyEntryUtil {
+
+
+ /**
+ * Find a matching PolicyEntry by looking up the Set of entries from the map and
+ * then comparing the destination to find the exact match. This lets us be able to
+ * find the correct policy entry to update even though there might be multiple that
+ * are returned from the get method of the PolicyMap.
+ *
+ * @param runtimeBroker
+ * @param entry
+ * @return
+ */
+ public static PolicyEntry findEntryByDestination(AbstractRuntimeConfigurationBroker runtimeBroker,
+ PolicyEntry entry) {
+
+ PolicyMap existingMap = runtimeBroker.getBrokerService().getDestinationPolicy();
+ @SuppressWarnings("unchecked")
+ Set<PolicyEntry> existingEntries = existingMap.get(entry.getDestination());
+
+ //First just look up by the destination type to see if anything matches
+ PolicyEntry existingEntry = null;
+ for (PolicyEntry ee: existingEntries) {
+ if (ee.getDestination().equals(entry.getDestination())) {
+ existingEntry = ee;
+ break;
+ }
+ }
+ return existingEntry;
+ }
+
+ /**
+ * Utility to properly apply an updated policy entry to all existing destinations that
+ * match this entry. The destination will only be updated if the policy is the exact
+ * policy (most specific) that matches the destination.
+ *
+ * @param runtimeBroker
+ * @param updatedEntry
+ */
+ public static void applyRetrospectively(AbstractRuntimeConfigurationBroker runtimeBroker, PolicyEntry updatedEntry) {
+ RegionBroker regionBroker = (RegionBroker) runtimeBroker.getBrokerService().getRegionBroker();
+ for (Destination destination : regionBroker.getDestinations(updatedEntry.getDestination())) {
+ //Look up the policy that applies to the destination
+ PolicyEntry specificyPolicy = regionBroker.getDestinationPolicy().getEntryFor(
+ destination.getActiveMQDestination());
+
+ //only update the destination if it matches the specific policy being updated
+ //currently just an identity check which is what we want
+ if (updatedEntry.equals(specificyPolicy)){
+ Destination target = destination;
+ if (destination instanceof DestinationFilter) {
+ target = ((DestinationFilter)destination).getNext();
+ }
+ if (target.getActiveMQDestination().isQueue()) {
+ updatedEntry.update((Queue) target);
+ } else if (target.getActiveMQDestination().isTopic()) {
+ updatedEntry.update((Topic) target);
+ }
+ runtimeBroker.debug("applied update to:" + target);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq/blob/27b08b03/activemq-runtime-config/src/test/java/org/apache/activemq/PolicyEntryTest.java
----------------------------------------------------------------------
diff --git a/activemq-runtime-config/src/test/java/org/apache/activemq/PolicyEntryTest.java b/activemq-runtime-config/src/test/java/org/apache/activemq/PolicyEntryTest.java
index 0c14452..120a4c6 100644
--- a/activemq-runtime-config/src/test/java/org/apache/activemq/PolicyEntryTest.java
+++ b/activemq-runtime-config/src/test/java/org/apache/activemq/PolicyEntryTest.java
@@ -17,6 +17,7 @@
package org.apache.activemq;
import javax.jms.Session;
+
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.junit.Test;
@@ -62,6 +63,43 @@ public class PolicyEntryTest extends RuntimeConfigTestSupport {
verifyTopicLimit("Before", 2048l);
}
+ @Test
+ public void testModParentPolicy() throws Exception {
+ final String brokerConfig = configurationSeed + "-policy-ml-broker";
+ applyNewConfig(brokerConfig, configurationSeed + "-policy-ml-parent");
+ startBroker(brokerConfig);
+ assertTrue("broker alive", brokerService.isStarted());
+
+ verifyQueueLimit("queue.test", 1024);
+ verifyQueueLimit("queue.child.test", 2048);
+ applyNewConfig(brokerConfig, configurationSeed + "-policy-ml-parent-mod", SLEEP);
+ verifyQueueLimit("queue.test2", 4194304);
+
+ // change to existing dest
+ verifyQueueLimit("queue.test", 4194304);
+ //verify no change
+ verifyQueueLimit("queue.child.test", 2048);
+ }
+
+ @Test
+ public void testModChildPolicy() throws Exception {
+ final String brokerConfig = configurationSeed + "-policy-ml-broker";
+ applyNewConfig(brokerConfig, configurationSeed + "-policy-ml-parent");
+ startBroker(brokerConfig);
+ assertTrue("broker alive", brokerService.isStarted());
+
+ verifyQueueLimit("queue.test", 1024);
+ verifyQueueLimit("queue.child.test", 2048);
+ applyNewConfig(brokerConfig, configurationSeed + "-policy-ml-child-mod", SLEEP);
+ //verify no change
+ verifyQueueLimit("queue.test", 1024);
+
+ // change to existing dest
+ verifyQueueLimit("queue.child.test", 4194304);
+ //new dest change
+ verifyQueueLimit("queue.child.test2", 4194304);
+ }
+
private void verifyQueueLimit(String dest, int memoryLimit) throws Exception {
ActiveMQConnection connection = new ActiveMQConnectionFactory("vm://localhost").createActiveMQConnection();
try {
http://git-wip-us.apache.org/repos/asf/activemq/blob/27b08b03/activemq-runtime-config/src/test/java/org/apache/activemq/java/JavaPolicyEntryTest.java
----------------------------------------------------------------------
diff --git a/activemq-runtime-config/src/test/java/org/apache/activemq/java/JavaPolicyEntryTest.java b/activemq-runtime-config/src/test/java/org/apache/activemq/java/JavaPolicyEntryTest.java
index 94a5496..accae06 100644
--- a/activemq-runtime-config/src/test/java/org/apache/activemq/java/JavaPolicyEntryTest.java
+++ b/activemq-runtime-config/src/test/java/org/apache/activemq/java/JavaPolicyEntryTest.java
@@ -81,6 +81,215 @@ public class JavaPolicyEntryTest extends RuntimeConfigTestSupport {
}
@Test
+ public void testModNewPolicyObject() throws Exception {
+ BrokerService brokerService = new BrokerService();
+ PolicyMap policyMap = new PolicyMap();
+ PolicyEntry entry = new PolicyEntry();
+ entry.setQueue(">");
+ entry.setMemoryLimit(1024);
+ policyMap.setPolicyEntries(Arrays.asList(entry));
+ brokerService.setDestinationPolicy(policyMap);
+
+
+ startBroker(brokerService);
+ assertTrue("broker alive", brokerService.isStarted());
+
+ verifyQueueLimit("Before", 1024);
+
+ //Reapply new limit with new object that matches
+ //the same destination, so it should still apply
+ PolicyEntry entry2 = new PolicyEntry();
+ entry2.setQueue(">");
+ entry2.setMemoryLimit(4194304);
+ javaConfigBroker.modifyPolicyEntry(entry2, true);
+ TimeUnit.SECONDS.sleep(SLEEP);
+
+ // These should change because the policy entry passed in
+ //matched an existing entry but was not the same reference.
+ //Since createOrReplace is true, we replace the entry with
+ //this new entry and apply
+ verifyQueueLimit("Before", 4194304);
+ verifyQueueLimit("After", 4194304);
+ }
+
+ /**
+ * Test that a new policy is added and applied
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testCreate() throws Exception {
+ BrokerService brokerService = new BrokerService();
+ PolicyMap policyMap = new PolicyMap();
+ policyMap.setPolicyEntries(Arrays.asList());
+ brokerService.setDestinationPolicy(policyMap);
+
+ startBroker(brokerService);
+ assertTrue("broker alive", brokerService.isStarted());
+ verifyQueueLimit("Before", (int)brokerService.getSystemUsage().getMemoryUsage().getLimit());
+
+ PolicyEntry entry = new PolicyEntry();
+ entry.setQueue(">");
+ entry.setMemoryLimit(1024);
+
+ //The true flag should add the new policy
+ javaConfigBroker.modifyPolicyEntry(entry, true);
+ TimeUnit.SECONDS.sleep(SLEEP);
+
+ //Make sure the new policy is added and applied
+ verifyQueueLimit("Before", 1024);
+ verifyQueueLimit("After", 1024);
+ }
+
+ /**
+ * Test that a new policy is not added
+ * @throws Exception
+ */
+ @Test
+ public void testCreateFalse() throws Exception {
+ BrokerService brokerService = new BrokerService();
+ PolicyMap policyMap = new PolicyMap();
+ policyMap.setPolicyEntries(Arrays.asList());
+ brokerService.setDestinationPolicy(policyMap);
+
+ startBroker(brokerService);
+ assertTrue("broker alive", brokerService.isStarted());
+ verifyQueueLimit("Before", (int)brokerService.getSystemUsage().getMemoryUsage().getLimit());
+
+ PolicyEntry entry = new PolicyEntry();
+ entry.setQueue(">");
+ entry.setMemoryLimit(1024);
+ //The default should NOT add this policy since it won't match an existing policy to modify
+ boolean caughtException = false;
+ try {
+ javaConfigBroker.modifyPolicyEntry(entry);
+ } catch (IllegalArgumentException e) {
+ caughtException = true;
+ }
+ assertTrue(caughtException);
+ TimeUnit.SECONDS.sleep(SLEEP);
+
+ //Make sure there was no change
+ verifyQueueLimit("Before", (int)brokerService.getSystemUsage().getMemoryUsage().getLimit());
+ verifyQueueLimit("After", (int)brokerService.getSystemUsage().getMemoryUsage().getLimit());
+ }
+
+
+ @Test
+ public void testModNewPolicyObjectCreateOrReplaceFalse() throws Exception {
+ BrokerService brokerService = new BrokerService();
+ PolicyMap policyMap = new PolicyMap();
+ PolicyEntry entry = new PolicyEntry();
+ entry.setQueue(">");
+ entry.setMemoryLimit(1024);
+ policyMap.setPolicyEntries(Arrays.asList(entry));
+ brokerService.setDestinationPolicy(policyMap);
+
+
+ startBroker(brokerService);
+ assertTrue("broker alive", brokerService.isStarted());
+
+ verifyQueueLimit("Before", 1024);
+
+ //Reapply new limit with new object that matches
+ //the same destination, but createOrReplace is false
+ PolicyEntry entry2 = new PolicyEntry();
+ entry2.setQueue(">");
+ entry2.setMemoryLimit(4194304);
+ boolean caughtException = false;
+ try {
+ javaConfigBroker.modifyPolicyEntry(entry2, false);
+ } catch (IllegalArgumentException e) {
+ caughtException = true;
+ }
+ assertTrue(caughtException);
+ TimeUnit.SECONDS.sleep(SLEEP);
+
+ // These should not change because the policy entry passed in
+ //matched an existing entry but was not the same reference.
+ //Since createOrReplace is false, it should noo be updated
+ verifyQueueLimit("Before", 1024);
+ verifyQueueLimit("After", 1024);
+ }
+
+ @Test
+ public void testModWithChildPolicy() throws Exception {
+ BrokerService brokerService = new BrokerService();
+ PolicyMap policyMap = new PolicyMap();
+ PolicyEntry entry = new PolicyEntry();
+ entry.setQueue("queue.>");
+ entry.setMemoryLimit(1024);
+ PolicyEntry entry2 = new PolicyEntry();
+ entry2.setQueue("queue.child.>");
+ entry2.setMemoryLimit(2048);
+ policyMap.setPolicyEntries(Arrays.asList(entry, entry2));
+ brokerService.setDestinationPolicy(policyMap);
+
+ startBroker(brokerService);
+ assertTrue("broker alive", brokerService.isStarted());
+
+ brokerService.getBroker().addDestination(
+ brokerService.getAdminConnectionContext(), new ActiveMQQueue("queue.test"), false);
+ brokerService.getBroker().addDestination(
+ brokerService.getAdminConnectionContext(), new ActiveMQQueue("queue.child.test"), false);
+
+ //check destinations before policy updates
+ verifyQueueLimit("queue.test", 1024);
+ verifyQueueLimit("queue.child.test", 2048);
+
+ //Reapply new limit to policy 2
+ entry2.setMemoryLimit(4194304);
+ javaConfigBroker.modifyPolicyEntry(entry2);
+ TimeUnit.SECONDS.sleep(SLEEP);
+
+ //verify new dest and existing are changed
+ verifyQueueLimit("queue.child.test", 4194304);
+ verifyQueueLimit("queue.child.test2", 4194304);
+
+ //verify that destination at a higher level policy is not affected
+ verifyQueueLimit("queue.test", 1024);
+ }
+
+ @Test
+ public void testModParentPolicy() throws Exception {
+ BrokerService brokerService = new BrokerService();
+ PolicyMap policyMap = new PolicyMap();
+
+ PolicyEntry entry = new PolicyEntry();
+ entry.setQueue("queue.>");
+ entry.setMemoryLimit(1024);
+ PolicyEntry entry2 = new PolicyEntry();
+ entry2.setQueue("queue.child.>");
+ entry2.setMemoryLimit(2048);
+ policyMap.setPolicyEntries(Arrays.asList(entry, entry2));
+ brokerService.setDestinationPolicy(policyMap);
+
+ startBroker(brokerService);
+ assertTrue("broker alive", brokerService.isStarted());
+
+ brokerService.getBroker().addDestination(
+ brokerService.getAdminConnectionContext(), new ActiveMQQueue("queue.test"), false);
+ brokerService.getBroker().addDestination(
+ brokerService.getAdminConnectionContext(), new ActiveMQQueue("queue.child.test"), false);
+
+ //check destinations before policy updates
+ verifyQueueLimit("queue.test", 1024);
+ verifyQueueLimit("queue.child.test", 2048);
+
+ //Reapply new limit to policy
+ entry.setMemoryLimit(4194304);
+ javaConfigBroker.modifyPolicyEntry(entry);
+ TimeUnit.SECONDS.sleep(SLEEP);
+
+ //verify new dest and existing are not changed
+ verifyQueueLimit("queue.child.test", 2048);
+ verifyQueueLimit("queue.child.test2", 2048);
+
+ //verify that destination at a higher level policy is changed
+ verifyQueueLimit("queue.test", 4194304);
+ }
+
+ @Test
public void testAddNdMod() throws Exception {
BrokerService brokerService = new BrokerService();
PolicyMap policyMap = new PolicyMap();
@@ -113,6 +322,53 @@ public class JavaPolicyEntryTest extends RuntimeConfigTestSupport {
verifyTopicLimit("Before", 2048l);
}
+ @Test
+ public void testAddNdModWithMultiplePolicies() throws Exception {
+ BrokerService brokerService = new BrokerService();
+ PolicyMap policyMap = new PolicyMap();
+ PolicyEntry entry = new PolicyEntry();
+ entry.setQueue(">");
+ entry.setMemoryLimit(1024);
+ policyMap.setPolicyEntries(Arrays.asList(entry));
+ brokerService.setDestinationPolicy(policyMap);
+
+ startBroker(brokerService);
+ assertTrue("broker alive", brokerService.isStarted());
+
+ verifyQueueLimit("Before", 1024);
+ verifyTopicLimit("Before", brokerService.getSystemUsage().getMemoryUsage().getLimit());
+
+ entry.setMemoryLimit(2048);
+ javaConfigBroker.modifyPolicyEntry(entry);
+ TimeUnit.SECONDS.sleep(SLEEP);
+
+ PolicyEntry newEntry = new PolicyEntry();
+ newEntry.setTopic("test2.>");
+ newEntry.setMemoryLimit(2048);
+ PolicyEntry newEntry2 = new PolicyEntry();
+ newEntry2.setTopic("test2.test.>");
+ newEntry2.setMemoryLimit(4000);
+ javaConfigBroker.addNewPolicyEntry(newEntry);
+ javaConfigBroker.addNewPolicyEntry(newEntry2);
+ TimeUnit.SECONDS.sleep(SLEEP);
+
+ verifyTopicLimit("test2.after", 2048l);
+ verifyTopicLimit("test2.test.after", 4000l);
+ //check existing modified entry
+ verifyQueueLimit("After", 2048);
+
+ // change to existing dest
+ PolicyEntry newEntry3 = new PolicyEntry();
+ newEntry3.setTopic(">");
+ newEntry3.setMemoryLimit(5000);
+ javaConfigBroker.addNewPolicyEntry(newEntry3);
+ verifyTopicLimit("Before", 5000l);
+
+ //reverify children
+ verifyTopicLimit("test2.after", 2048l);
+ verifyTopicLimit("test2.test.after", 4000l);
+ }
+
private void verifyQueueLimit(String dest, int memoryLimit) throws Exception {
ActiveMQConnection connection = (ActiveMQConnection) new ActiveMQConnectionFactory("vm://localhost").createConnection();
try {
http://git-wip-us.apache.org/repos/asf/activemq/blob/27b08b03/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-ml-child-mod.xml
----------------------------------------------------------------------
diff --git a/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-ml-child-mod.xml b/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-ml-child-mod.xml
new file mode 100644
index 0000000..86b5249
--- /dev/null
+++ b/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-ml-child-mod.xml
@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You 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.
+-->
+<beans
+ xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
+ http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
+
+ <broker xmlns="http://activemq.apache.org/schema/core" start="false" persistent="false">
+ <plugins>
+ <runtimeConfigurationPlugin checkPeriod="1000" />
+ </plugins>
+ <destinationPolicy>
+ <policyMap>
+ <policyEntries>
+ <policyEntry queue="queue.>" memoryLimit="1024b"/>
+ <policyEntry queue="queue.child.>" memoryLimit="4mb"/>
+ </policyEntries>
+ </policyMap>
+ </destinationPolicy>
+ </broker>
+</beans>
http://git-wip-us.apache.org/repos/asf/activemq/blob/27b08b03/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-ml-parent-mod.xml
----------------------------------------------------------------------
diff --git a/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-ml-parent-mod.xml b/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-ml-parent-mod.xml
new file mode 100644
index 0000000..5a5b784
--- /dev/null
+++ b/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-ml-parent-mod.xml
@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You 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.
+-->
+<beans
+ xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
+ http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
+
+ <broker xmlns="http://activemq.apache.org/schema/core" start="false" persistent="false">
+ <plugins>
+ <runtimeConfigurationPlugin checkPeriod="1000" />
+ </plugins>
+ <destinationPolicy>
+ <policyMap>
+ <policyEntries>
+ <policyEntry queue="queue.>" memoryLimit="4mb"/>
+ <policyEntry queue="queue.child.>" memoryLimit="2048b"/>
+ </policyEntries>
+ </policyMap>
+ </destinationPolicy>
+ </broker>
+</beans>
http://git-wip-us.apache.org/repos/asf/activemq/blob/27b08b03/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-ml-parent.xml
----------------------------------------------------------------------
diff --git a/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-ml-parent.xml b/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-ml-parent.xml
new file mode 100644
index 0000000..7756f4f
--- /dev/null
+++ b/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-ml-parent.xml
@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You 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.
+-->
+<beans
+ xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
+ http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
+
+ <broker xmlns="http://activemq.apache.org/schema/core" start="false" persistent="false">
+ <plugins>
+ <runtimeConfigurationPlugin checkPeriod="1000" />
+ </plugins>
+ <destinationPolicy>
+ <policyMap>
+ <policyEntries>
+ <policyEntry queue="queue.>" memoryLimit="1024b"/>
+ <policyEntry queue="queue.child.>" memoryLimit="2048b"/>
+ </policyEntries>
+ </policyMap>
+ </destinationPolicy>
+ </broker>
+</beans>
[3/5] activemq git commit:
https://issues.apache.org/jira/browse/AMQ-6086 - start exception can prevent
stop from exiting fully, need to be more selective in creation on stop.
Posted by cs...@apache.org.
https://issues.apache.org/jira/browse/AMQ-6086 - start exception can prevent stop from exiting fully, need to be more selective in creation on stop.
(cherry picked from commit 35df815fb86d201e63b4f0f2d53bee3bae5c0752)
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/2ffd7498
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/2ffd7498
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/2ffd7498
Branch: refs/heads/activemq-5.13.x
Commit: 2ffd7498a835be85b08e5ca9966a26a2d4ac0cd3
Parents: f3aedc7
Author: gtully <ga...@gmail.com>
Authored: Mon Dec 14 15:39:37 2015 +0000
Committer: Christopher L. Shannon (cshannon) <ch...@gmail.com>
Committed: Mon Dec 14 19:09:15 2015 +0000
----------------------------------------------------------------------
.../apache/activemq/broker/BrokerService.java | 13 ++--
.../StartAndConcurrentStopBrokerTest.java | 75 ++++++++++++++++++--
2 files changed, 77 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/2ffd7498/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java b/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
index 62af182..d028078 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
@@ -785,9 +785,7 @@ public class BrokerService implements Service {
return;
}
- if (started.get()) {
- setStartException(new BrokerStoppedException("Stop invoked"));
- }
+ setStartException(new BrokerStoppedException("Stop invoked"));
MDC.put("activemq.broker", brokerName);
if (systemExitOnShutdown) {
@@ -836,7 +834,7 @@ public class BrokerService implements Service {
stopper.stop(getPersistenceAdapter());
persistenceAdapter = null;
if (isUseJmx()) {
- stopper.stop(getManagementContext());
+ stopper.stop(managementContext);
managementContext = null;
}
// Clear SelectorParser cache to free memory
@@ -1229,8 +1227,7 @@ public class BrokerService implements Service {
}
public synchronized PersistenceAdapter getPersistenceAdapter() throws IOException {
- if (persistenceAdapter == null) {
- checkStartException();
+ if (persistenceAdapter == null && !hasStartException()) {
persistenceAdapter = createPersistenceAdapter();
configureService(persistenceAdapter);
this.persistenceAdapter = registerPersistenceAdapterMBean(persistenceAdapter);
@@ -1330,6 +1327,10 @@ public class BrokerService implements Service {
}
}
+ synchronized private boolean hasStartException() {
+ return startException != null;
+ }
+
synchronized private void setStartException(Throwable t) {
startException = t;
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/2ffd7498/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java
index b2ad1cd..4d194ca 100755
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java
@@ -23,6 +23,8 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import javax.management.Attribute;
import javax.management.AttributeList;
import javax.management.AttributeNotFoundException;
@@ -51,6 +53,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@@ -61,7 +64,11 @@ public class StartAndConcurrentStopBrokerTest {
@Test(timeout = 30000)
public void testConcurrentStop() throws Exception {
+ final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
final CountDownLatch gotBrokerMbean = new CountDownLatch(1);
+ final CountDownLatch gotPaMBean = new CountDownLatch(1);
+ final AtomicBoolean checkPaMBean = new AtomicBoolean(false);
+
final HashMap mbeans = new HashMap();
final MBeanServer mBeanServer = new MBeanServer() {
@Override
@@ -87,7 +94,7 @@ public class StartAndConcurrentStopBrokerTest {
@Override
public ObjectInstance registerMBean(Object object, ObjectName name) throws InstanceAlreadyExistsException, MBeanRegistrationException, NotCompliantMBeanException {
if (mbeans.containsKey(name)) {
- throw new InstanceAlreadyExistsException("Got one already");
+ throw new InstanceAlreadyExistsException("Got one already: " + name);
}
LOG.info("register:" + name);
@@ -95,8 +102,16 @@ public class StartAndConcurrentStopBrokerTest {
if (name.compareTo(new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost")) == 0) {
gotBrokerMbean.countDown();
}
+
+ if (checkPaMBean.get()) {
+ if (new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost,service=PersistenceAdapter,instanceName=*").apply(name)) {
+ gotPaMBean.countDown();
+ }
+ }
+
} catch (Exception e) {
e.printStackTrace();
+ error.set(e);
}
mbeans.put(name, object);
return new ObjectInstance(name, object.getClass().getName());
@@ -261,6 +276,7 @@ public class StartAndConcurrentStopBrokerTest {
final BrokerService broker = new BrokerService();
+ broker.setDeleteAllMessagesOnStartup(true);
ExecutorService executor = Executors.newFixedThreadPool(4);
executor.execute(new Runnable() {
@@ -272,6 +288,7 @@ public class StartAndConcurrentStopBrokerTest {
} catch (BrokerStoppedException expected) {
} catch (Exception e) {
e.printStackTrace();
+ error.set(e);
}
}
});
@@ -285,6 +302,7 @@ public class StartAndConcurrentStopBrokerTest {
broker.stop();
} catch (Exception e) {
e.printStackTrace();
+ error.set(e);
}
}
});
@@ -292,10 +310,57 @@ public class StartAndConcurrentStopBrokerTest {
executor.shutdown();
assertTrue("stop tasks done", executor.awaitTermination(20, TimeUnit.SECONDS));
- BrokerService second = new BrokerService();
- second.getManagementContext().setMBeanServer(mBeanServer);
- second.start();
- second.stop();
+ BrokerService sanityBroker = new BrokerService();
+ sanityBroker.getManagementContext().setMBeanServer(mBeanServer);
+ sanityBroker.start();
+ sanityBroker.stop();
+
+ assertNull("No error", error.get());
+
+ // again, after Persistence adapter mbean
+ final BrokerService brokerTwo = new BrokerService();
+ broker.setDeleteAllMessagesOnStartup(true);
+
+ checkPaMBean.set(true);
+ executor = Executors.newFixedThreadPool(4);
+ executor.execute(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ brokerTwo.getManagementContext().setMBeanServer(mBeanServer);
+ brokerTwo.start();
+ } catch (BrokerStoppedException expected) {
+ } catch (Exception e) {
+ e.printStackTrace();
+ error.set(e);
+ }
+ }
+ });
+
+ executor.execute(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ assertTrue("broker has registered persistence adapter mbean", gotPaMBean.await(10, TimeUnit.SECONDS));
+ brokerTwo.stop();
+ } catch (Exception e) {
+ e.printStackTrace();
+ error.set(e);
+ }
+ }
+ });
+
+ executor.shutdown();
+ assertTrue("stop tasks done", executor.awaitTermination(20, TimeUnit.SECONDS));
+
+ assertTrue("broker has registered persistence adapter mbean", gotPaMBean.await(0, TimeUnit.SECONDS));
+
+ sanityBroker = new BrokerService();
+ sanityBroker.getManagementContext().setMBeanServer(mBeanServer);
+ sanityBroker.start();
+ sanityBroker.stop();
+
+ assertNull("No error", error.get());
}
[4/5] activemq git commit:
https://issues.apache.org/jira/browse/AMQ-6086 - test cannot validate all
start failures scenarios at this time, suffice that start and stop complete ok
Posted by cs...@apache.org.
https://issues.apache.org/jira/browse/AMQ-6086 - test cannot validate all start failures scenarios at this time, suffice that start and stop complete ok
(cherry picked from commit b7787bf6fbbb33b5d16b34dab07de52b76044a4c)
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/fc4120e8
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/fc4120e8
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/fc4120e8
Branch: refs/heads/activemq-5.13.x
Commit: fc4120e85016e094eee6d7a758c73f49f3db191b
Parents: b4360c6
Author: gtully <ga...@gmail.com>
Authored: Mon Dec 14 16:46:22 2015 +0000
Committer: Christopher L. Shannon (cshannon) <ch...@gmail.com>
Committed: Mon Dec 14 19:09:15 2015 +0000
----------------------------------------------------------------------
.../activemq/usecases/StartAndConcurrentStopBrokerTest.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/fc4120e8/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java
index 8ea0af8..06b69ca 100755
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java
@@ -290,7 +290,7 @@ public class StartAndConcurrentStopBrokerTest {
} catch (ConfigurationException expected) {
} catch (Exception e) {
e.printStackTrace();
- error.set(e);
+ // lots of possible errors depending on progress
}
}
});
@@ -335,7 +335,7 @@ public class StartAndConcurrentStopBrokerTest {
} catch (ConfigurationException expected) {
} catch (Exception e) {
e.printStackTrace();
- error.set(e);
+ // lots of possible errors depending on progress
}
}
});
[2/5] activemq git commit:
https://issues.apache.org/jira/browse/AMQ-6086 - avoid logging npe on
attempted start of persistence adapter post stop
Posted by cs...@apache.org.
https://issues.apache.org/jira/browse/AMQ-6086 - avoid logging npe on attempted start of persistence adapter post stop
(cherry picked from commit 455a628305dc49476d7033701901d119db3fffdb)
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/b4360c6e
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/b4360c6e
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/b4360c6e
Branch: refs/heads/activemq-5.13.x
Commit: b4360c6e0e285aa97851d5bda9083d376709b0b4
Parents: 2ffd749
Author: gtully <ga...@gmail.com>
Authored: Mon Dec 14 16:32:10 2015 +0000
Committer: Christopher L. Shannon (cshannon) <ch...@gmail.com>
Committed: Mon Dec 14 19:09:15 2015 +0000
----------------------------------------------------------------------
.../java/org/apache/activemq/broker/BrokerService.java | 13 +++++++++----
.../usecases/StartAndConcurrentStopBrokerTest.java | 3 +++
2 files changed, 12 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/b4360c6e/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java b/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
index d028078..cef2520 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
@@ -658,13 +658,18 @@ public class BrokerService implements Service {
}
private void doStartPersistenceAdapter() throws Exception {
- getPersistenceAdapter().setUsageManager(getProducerSystemUsage());
- getPersistenceAdapter().setBrokerName(getBrokerName());
- LOG.info("Using Persistence Adapter: {}", getPersistenceAdapter());
+ PersistenceAdapter persistenceAdapterToStart = getPersistenceAdapter();
+ if (persistenceAdapterToStart == null) {
+ checkStartException();
+ throw new ConfigurationException("Cannot start null persistence adapter");
+ }
+ persistenceAdapterToStart.setUsageManager(getProducerSystemUsage());
+ persistenceAdapterToStart.setBrokerName(getBrokerName());
+ LOG.info("Using Persistence Adapter: {}", persistenceAdapterToStart);
if (deleteAllMessagesOnStartup) {
deleteAllMessages();
}
- getPersistenceAdapter().start();
+ persistenceAdapterToStart.start();
getTempDataStore();
if (tempDataStore != null) {
http://git-wip-us.apache.org/repos/asf/activemq/blob/b4360c6e/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java
index 4d194ca..8ea0af8 100755
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/StartAndConcurrentStopBrokerTest.java
@@ -46,6 +46,7 @@ import javax.management.OperationsException;
import javax.management.QueryExp;
import javax.management.ReflectionException;
import javax.management.loading.ClassLoaderRepository;
+import org.apache.activemq.ConfigurationException;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.BrokerStoppedException;
import org.junit.Test;
@@ -286,6 +287,7 @@ public class StartAndConcurrentStopBrokerTest {
broker.getManagementContext().setMBeanServer(mBeanServer);
broker.start();
} catch (BrokerStoppedException expected) {
+ } catch (ConfigurationException expected) {
} catch (Exception e) {
e.printStackTrace();
error.set(e);
@@ -330,6 +332,7 @@ public class StartAndConcurrentStopBrokerTest {
brokerTwo.getManagementContext().setMBeanServer(mBeanServer);
brokerTwo.start();
} catch (BrokerStoppedException expected) {
+ } catch (ConfigurationException expected) {
} catch (Exception e) {
e.printStackTrace();
error.set(e);