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/activemq/browse/AMQ-2843 - add mechanism for stores to know whether to use prioritized recovery or not 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@966145 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Bosanac Dejan 2010-07-21 08:59:42 +00:00
parent 8ed8dd41c4
commit 03f38ae402
7 changed files with 164 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
View File

@ -609,6 +609,9 @@ public abstract class BaseDestination implements Destination {
public void setPrioritizedMessages(boolean prioritizedMessages) {
this.prioritizedMessages = prioritizedMessages;
if (store != null) {
store.setPrioritizedMessages(prioritizedMessages);
}
}
/**

9
activemq-core/src/main/java/org/apache/activemq/store/AbstractMessageStore.java
View File

@ -30,6 +30,7 @@ import org.apache.activemq.usage.MemoryUsage;
abstract public class AbstractMessageStore implements MessageStore {
public static final FutureTask<Object> FUTURE;
protected final ActiveMQDestination destination;
protected boolean prioritizedMessages;
public AbstractMessageStore(ActiveMQDestination destination) {
this.destination = destination;
@ -63,6 +64,14 @@ abstract public class AbstractMessageStore implements MessageStore {
public boolean isEmpty() throws Exception {
return getMessageCount() == 0;
}
public void setPrioritizedMessages(boolean prioritizedMessages) {
this.prioritizedMessages = prioritizedMessages;
}
public boolean isPrioritizedMessages() {
return this.prioritizedMessages;
}
public Future<Object> asyncAddQueueMessage(final ConnectionContext context, final Message message) throws IOException {
addMessage(context, message);

12
activemq-core/src/main/java/org/apache/activemq/store/MessageStore.java
View File

@ -148,4 +148,16 @@ public interface MessageStore extends Service {
*/
boolean isEmpty() throws Exception;
/**
* A hint to the store to try recover messages according to priority
* @param prioritizedMessages
*/
public void setPrioritizedMessages(boolean prioritizedMessages);
/**
*
* @return true if store is trying to recover messages according to priority
*/
public boolean isPrioritizedMessages();
}

8
activemq-core/src/main/java/org/apache/activemq/store/ProxyMessageStore.java
View File

@ -113,4 +113,12 @@ public class ProxyMessageStore implements MessageStore {
public void removeAsyncMessage(ConnectionContext context, MessageAck ack) throws IOException {
delegate.removeAsyncMessage(context, ack);
}
public void setPrioritizedMessages(boolean prioritizedMessages) {
delegate.setPrioritizedMessages(prioritizedMessages);
}
public boolean isPrioritizedMessages() {
return delegate.isPrioritizedMessages();
}
}

8
activemq-core/src/main/java/org/apache/activemq/store/ProxyTopicMessageStore.java
View File

@ -153,4 +153,12 @@ public class ProxyTopicMessageStore implements TopicMessageStore {
public void removeAsyncMessage(ConnectionContext context, MessageAck ack) throws IOException {
delegate.removeAsyncMessage(context, ack);
}
public void setPrioritizedMessages(boolean prioritizedMessages) {
delegate.setPrioritizedMessages(prioritizedMessages);
}
public boolean isPrioritizedMessages() {
return delegate.isPrioritizedMessages();
}
}

88
activemq-core/src/test/java/org/apache/activemq/store/MessagePriorityTest.java Normal file
View File

@ -0,0 +1,88 @@
/**
* 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.store;
import javax.jms.Connection;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.Topic;
import junit.framework.TestCase;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
abstract public class MessagePriorityTest extends TestCase {
BrokerService broker;
PersistenceAdapter adapter;
ActiveMQConnectionFactory factory;
Connection conn;
Session sess;
abstract protected PersistenceAdapter createPersistenceAdapter(boolean delete) throws Exception;
protected void setUp() throws Exception {
broker = new BrokerService();
broker.setBrokerName("priorityTest");
adapter = createPersistenceAdapter(true);
broker.setPersistenceAdapter(adapter);
PolicyEntry policy = new PolicyEntry();
policy.setPrioritizedMessages(true);
PolicyMap policyMap = new PolicyMap();
policyMap.setDefaultEntry(policy);
broker.setDestinationPolicy(policyMap);
broker.start();
broker.waitUntilStarted();
factory = new ActiveMQConnectionFactory("vm://priorityTest");
conn = factory.createConnection();
sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
}
protected void tearDown() throws Exception {
sess.close();
conn.close();
broker.stop();
broker.waitUntilStopped();
}
public void testStoreConfigured() throws Exception {
Queue queue = sess.createQueue("TEST");
Topic topic = sess.createTopic("TEST");
MessageProducer queueProducer = sess.createProducer(queue);
MessageProducer topicProducer = sess.createProducer(topic);
Thread.sleep(100); // get it all propagated
assertTrue(broker.getRegionBroker().getDestinationMap().get(queue).getMessageStore().isPrioritizedMessages());
assertTrue(broker.getRegionBroker().getDestinationMap().get(topic).getMessageStore().isPrioritizedMessages());
queueProducer.close();
topicProducer.close();
}
}

36
activemq-core/src/test/java/org/apache/activemq/store/jdbc/JDBCMessagePriorityTest.java Normal file
View File

@ -0,0 +1,36 @@
/**
* 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.store.jdbc;
import org.apache.activemq.store.MessagePriorityTest;
import org.apache.activemq.store.PersistenceAdapter;
import org.apache.derby.jdbc.EmbeddedDataSource;
public class JDBCMessagePriorityTest extends MessagePriorityTest {
@Override
protected PersistenceAdapter createPersistenceAdapter(boolean delete) throws Exception {
JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
EmbeddedDataSource dataSource = new EmbeddedDataSource();
dataSource.setDatabaseName("derbyDb");
dataSource.setCreateDatabase("create");
jdbc.setDataSource(dataSource);
return jdbc;
}
}