You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by de...@apache.org on 2011/06/28 13:27:58 UTC
svn commit: r1140531 - in /activemq/trunk/activemq-core/src:
main/java/org/apache/activemq/broker/region/policy/UniquePropertyMessageEvictionStrategy.java
test/java/org/apache/activemq/broker/region/UniquePropertyMessageEvictionStrategyTest.java
Author: dejanb
Date: Tue Jun 28 11:27:58 2011
New Revision: 1140531
URL: http://svn.apache.org/viewvc?rev=1140531&view=rev
Log:
https://issues.apache.org/jira/browse/AMQ-3379 - unique property eviction strategy
Added:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/UniquePropertyMessageEvictionStrategy.java
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/region/UniquePropertyMessageEvictionStrategyTest.java
Added: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/UniquePropertyMessageEvictionStrategy.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/UniquePropertyMessageEvictionStrategy.java?rev=1140531&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/UniquePropertyMessageEvictionStrategy.java (added)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/UniquePropertyMessageEvictionStrategy.java Tue Jun 28 11:27:58 2011
@@ -0,0 +1,77 @@
+/**
+ * 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.broker.region.policy;
+
+import org.apache.activemq.broker.region.MessageReference;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+
+
+/**
+ * An eviction strategy which evicts the oldest message with the lowest priority first.
+ *
+ * @org.apache.xbean.XBean
+ *
+ * messageEvictionStrategy
+ */
+public class UniquePropertyMessageEvictionStrategy extends MessageEvictionStrategySupport {
+
+ protected String propertyName;
+
+ public String getPropertyName() {
+ return propertyName;
+ }
+
+ public void setPropertyName(String propertyName) {
+ this.propertyName = propertyName;
+ }
+
+ @Override
+ public MessageReference[] evictMessages(LinkedList messages) throws IOException {
+ HashMap<Object, MessageReference> pivots = new HashMap<Object, MessageReference>();
+ Iterator iter = messages.iterator();
+
+ for (int i = 0; iter.hasNext(); i++) {
+ MessageReference reference = (MessageReference) iter.next();
+ if (propertyName != null && reference.getMessage().getProperty(propertyName) != null) {
+ Object key = reference.getMessage().getProperty(propertyName);
+ if (pivots.containsKey(key)) {
+ MessageReference pivot = pivots.get(key);
+ if (reference.getMessage().getTimestamp() > pivot.getMessage().getTimestamp()) {
+ pivots.put(key, reference);
+ }
+ } else {
+ pivots.put(key, reference);
+ }
+ }
+ }
+
+ if (!pivots.isEmpty()) {
+ for (MessageReference ref : pivots.values()) {
+ messages.remove(ref);
+ }
+ return (MessageReference[])messages.toArray(new MessageReference[messages.size()]);
+ } else {
+ return new MessageReference[] {(MessageReference) messages.removeFirst()};
+ }
+
+ }
+}
Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/region/UniquePropertyMessageEvictionStrategyTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/region/UniquePropertyMessageEvictionStrategyTest.java?rev=1140531&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/region/UniquePropertyMessageEvictionStrategyTest.java (added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/region/UniquePropertyMessageEvictionStrategyTest.java Tue Jun 28 11:27:58 2011
@@ -0,0 +1,105 @@
+/**
+ * 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.broker.region;
+
+import org.apache.activemq.EmbeddedBrokerTestSupport;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.policy.ConstantPendingMessageLimitStrategy;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.broker.region.policy.UniquePropertyMessageEvictionStrategy;
+
+import javax.jms.*;
+import java.util.ArrayList;
+import java.util.List;
+
+public class UniquePropertyMessageEvictionStrategyTest extends EmbeddedBrokerTestSupport {
+
+
+
+ @Override
+ protected BrokerService createBroker() throws Exception {
+ BrokerService broker = super.createBroker();
+ final List<PolicyEntry> policyEntries = new ArrayList<PolicyEntry>();
+ final PolicyEntry entry = new PolicyEntry();
+ entry.setTopic(">");
+
+ entry.setAdvisoryForDiscardingMessages(true);
+ entry.setTopicPrefetch(1);
+
+ ConstantPendingMessageLimitStrategy pendingMessageLimitStrategy = new ConstantPendingMessageLimitStrategy();
+ pendingMessageLimitStrategy.setLimit(10);
+ entry.setPendingMessageLimitStrategy(pendingMessageLimitStrategy);
+
+
+ UniquePropertyMessageEvictionStrategy messageEvictionStrategy = new UniquePropertyMessageEvictionStrategy();
+ messageEvictionStrategy.setPropertyName("sequenceI");
+ entry.setMessageEvictionStrategy(messageEvictionStrategy);
+
+ // let evicted messages disappear
+ entry.setDeadLetterStrategy(null);
+ policyEntries.add(entry);
+
+ final PolicyMap policyMap = new PolicyMap();
+ policyMap.setPolicyEntries(policyEntries);
+ broker.setDestinationPolicy(policyMap);
+
+ return broker;
+ }
+
+ public void testEviction() throws Exception {
+ Connection conn = connectionFactory.createConnection();
+ conn.start();
+ Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ javax.jms.Topic destination = session.createTopic("TEST");
+
+ MessageProducer producer = session.createProducer(destination);
+ MessageConsumer consumer = session.createConsumer(destination);
+
+ for (int i = 0; i < 10; i++) {
+ for (int j = 0; j < 10; j++) {
+ TextMessage msg = session.createTextMessage("message " + i + j);
+ msg.setIntProperty("sequenceI", i);
+ msg.setIntProperty("sequenceJ", j);
+ producer.send(msg);
+ Thread.sleep(100);
+ }
+ }
+
+
+ for (int i = 0; i < 11; i++) {
+ javax.jms.Message msg = consumer.receive(1000);
+ assertNotNull(msg);
+ int seqI = msg.getIntProperty("sequenceI");
+ int seqJ = msg.getIntProperty("sequenceJ");
+ if (i ==0 ) {
+ assertEquals(0, seqI);
+ assertEquals(0, seqJ);
+ } else {
+ assertEquals(9, seqJ);
+ assertEquals(i - 1, seqI);
+ }
+ //System.out.println(msg.getIntProperty("sequenceI") + " " + msg.getIntProperty("sequenceJ"));
+ }
+
+ javax.jms.Message msg = consumer.receive(1000);
+ assertNull(msg);
+
+ }
+
+}