Author: rajdavies
Date: Tue Nov 17 14:55:45 2009
New Revision: 881313
URL: http://svn.apache.org/viewvc?rev=881313&view=rev
Log:
Fix for https://issues.apache.org/activemq/browse/AMQ-2475
Added:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/TempStorageBlockedBrokerTest.java
(with props)
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/store/temp/
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/store/temp/TempStoreTest.java
(with props)
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java?rev=881313&r1=881312&r2=881313&view=diff
==============================================================================
---
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
(original)
+++
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
Tue Nov 17 14:55:45 2009
@@ -96,8 +96,14 @@
}
}
if (maximumPendingMessages != 0) {
+ synchronized(matchedListMutex){
+ while (matched.isFull()){
+ matchedListMutex.wait(20);
+ }
+ matched.addMessageLast(node);
+ }
synchronized (matchedListMutex) {
- matched.addMessageLast(node);
+
// NOTE - be careful about the slaveBroker!
if (maximumPendingMessages > 0) {
// calculate the high water mark from which point we
@@ -115,7 +121,7 @@
// only page in a 1000 at a time - else we could
// blow da memory
pageInSize = Math.max(1000, pageInSize);
- LinkedList list = null;
+ LinkedList<MessageReference> list = null;
MessageReference[] oldMessages=null;
synchronized(matched){
list = matched.pageInList(pageInSize);
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java?rev=881313&r1=881312&r2=881313&view=diff
==============================================================================
---
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
(original)
+++
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
Tue Nov 17 14:55:45 2009
@@ -141,7 +141,7 @@
destroyDiskList();
}
- private void destroyDiskList() {
+ private void destroyDiskList() throws Exception {
if (!isDiskListEmpty()) {
Iterator<MessageReference> iterator = diskList.iterator();
while (iterator.hasNext()) {
@@ -149,7 +149,8 @@
iterator.remove();
}
diskList.clear();
- }
+ }
+ store.deleteListContainer(name, "TopicSubscription");
}
public synchronized LinkedList<MessageReference> pageInList(int maxItems) {
@@ -318,10 +319,12 @@
last=null;
}
- public synchronized boolean isFull() {
- // we always have space - as we can persist to disk
- return false;
- }
+ public synchronized boolean isFull() {
+
+ return super.isFull()
+ || (systemUsage != null &&
systemUsage.getTempUsage().isFull());
+
+ }
public boolean hasMessagesBufferedToDeliver() {
return !isEmpty();
Added:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/TempStorageBlockedBrokerTest.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/TempStorageBlockedBrokerTest.java?rev=881313&view=auto
==============================================================================
---
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/TempStorageBlockedBrokerTest.java
(added)
+++
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/TempStorageBlockedBrokerTest.java
Tue Nov 17 14:55:45 2009
@@ -0,0 +1,221 @@
+/**
+ * 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.bugs;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import java.io.File;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.jms.Connection;
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.ActiveMQPrefetchPolicy;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.kaha.Store;
+import org.apache.activemq.store.amq.AMQPersistenceAdapter;
+import org.apache.activemq.usage.MemoryUsage;
+import org.apache.activemq.usage.StoreUsage;
+import org.apache.activemq.usage.SystemUsage;
+import org.apache.activemq.usage.TempUsage;
+import org.apache.activemq.util.IOHelper;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TempStorageBlockedBrokerTest {
+
+ public boolean consumeAll = false;
+ public int deliveryMode = DeliveryMode.PERSISTENT;
+
+ private static final Log LOG =
LogFactory.getLog(TempStorageBlockedBrokerTest.class);
+ private static final int MESSAGES_COUNT = 1000;
+ private static byte[] buf = new byte[4 * 1024];
+ private BrokerService broker;
+ AtomicInteger messagesSent = new AtomicInteger(0);
+ AtomicInteger messagesConsumed = new AtomicInteger(0);
+
+ protected long messageReceiveTimeout = 10L;
+
+ Destination destination = new ActiveMQTopic("FooTwo");
+
+ @Test
+ public void runProducerWithHungConsumer() throws Exception {
+
+ final long origTempUsage =
broker.getSystemUsage().getTempUsage().getUsage();
+
+ ActiveMQConnectionFactory factory = new
ActiveMQConnectionFactory("tcp://localhost:61618");
+ // ensure messages are spooled to disk for this consumer
+ ActiveMQPrefetchPolicy prefetch = new ActiveMQPrefetchPolicy();
+ prefetch.setTopicPrefetch(10);
+ factory.setPrefetchPolicy(prefetch);
+ Connection consumerConnection = factory.createConnection();
+ consumerConnection.start();
+
+ Session consumerSession = consumerConnection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ MessageConsumer consumer = consumerSession.createConsumer(destination);
+
+ final Connection producerConnection = factory.createConnection();
+ producerConnection.start();
+
+ Thread producingThread = new Thread("Producing thread") {
+ public void run() {
+ try {
+ Session session = producerConnection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer =
session.createProducer(destination);
+ producer.setDeliveryMode(deliveryMode);
+ for (int idx = 0; idx < MESSAGES_COUNT; ++idx) {
+ Message message = session.createTextMessage(new
String(buf) + idx);
+
+ producer.send(message);
+ messagesSent.incrementAndGet();
+ Thread.sleep(10);
+ LOG.info("Sent Message " + idx);
+ LOG.info("Temp Store Usage " +
broker.getSystemUsage().getTempUsage().getUsage());
+
+ }
+ producer.close();
+ session.close();
+ } catch (Throwable ex) {
+ ex.printStackTrace();
+ }
+ }
+ };
+ producingThread.start();
+
+ int count = 0;
+
+ Message m = null;
+ while ((m = consumer.receive(messageReceiveTimeout)) != null) {
+ count++;
+ LOG.info("Recieved Message (" + count + "):" + m);
+ messagesConsumed.incrementAndGet();
+ try {
+ Thread.sleep(100);
+ } catch (Exception e) {
+ LOG.info("error sleeping");
+ }
+ }
+
+ LOG.info("Connection Timeout: Retrying");
+
+ // session.close();
+ // consumerConnection.close();
+ //
+ // consumerConnection2.start();
+ // session2 = consumerConnection2.createSession(false,
+ // Session.AUTO_ACKNOWLEDGE);
+ // consumer = session2.createConsumer(destination);
+
+ while ((m = consumer.receive(messageReceiveTimeout)) != null) {
+ count++;
+ LOG.info("Recieved Message (" + count + "):" + m);
+ messagesConsumed.incrementAndGet();
+ try {
+ Thread.sleep(100);
+ } catch (Exception e) {
+ LOG.info("error sleeping");
+ }
+ }
+
+ LOG.info("consumer session closing: consumed count: " + count);
+
+ consumerSession.close();
+
+ producingThread.join();
+
+ final long tempUsageBySubscription =
broker.getSystemUsage().getTempUsage().getUsage();
+ LOG.info("Orig Usage: " + origTempUsage + ", currentUsage: " +
tempUsageBySubscription);
+ // assertTrue("some temp store has been used", tempUsageBySubscription
+ // != origTempUsage);
+
+ producerConnection.close();
+ consumerConnection.close();
+
+ LOG.info("Subscrition Usage: " + tempUsageBySubscription + ",
endUsage: "
+ + broker.getSystemUsage().getTempUsage().getUsage());
+
+
+ assertEquals("Incorrect number of Messages Sent: " +
messagesSent.get(), messagesSent.get(), MESSAGES_COUNT);
+ assertEquals("Incorrect number of Messages Consumed: " +
messagesConsumed.get(), messagesConsumed.get(),
+ MESSAGES_COUNT);
+ }
+
+ @Before
+ public void setUp() throws Exception {
+
+ broker = new BrokerService();
+ broker.setDataDirectory("target" + File.separator + "activemq-data");
+ broker.setPersistent(true);
+ broker.setUseJmx(true);
+ broker.setAdvisorySupport(false);
+ broker.setDeleteAllMessagesOnStartup(true);
+
+ AMQPersistenceAdapter persistence = new AMQPersistenceAdapter();
+ persistence.setSyncOnWrite(false);
+ File directory = new File("target" + File.separator + "activemq-data");
+ persistence.setDirectory(directory);
+ File tmpDir = new File(directory, "tmp");
+ IOHelper.deleteChildren(tmpDir);
+ Store tempStore = new org.apache.activemq.kaha.impl.KahaStore(tmpDir,
"rw");
+
+ SystemUsage sysUsage = new SystemUsage("mySysUsage", persistence,
tempStore);
+ MemoryUsage memUsage = new MemoryUsage();
+ memUsage.setLimit((1024 * 1024));
+ StoreUsage storeUsage = new StoreUsage();
+ storeUsage.setLimit((1024 * 1024) * 38);
+ TempUsage tmpUsage = new TempUsage();
+ tmpUsage.setLimit((1024 * 1024) * 38);
+
+ PolicyEntry defaultPolicy = new PolicyEntry();
+ // defaultPolicy.setTopic("FooTwo");
+ defaultPolicy.setProducerFlowControl(false);
+ defaultPolicy.setMemoryLimit(10 * 1024);
+
+ PolicyMap policyMap = new PolicyMap();
+ policyMap.setDefaultEntry(defaultPolicy);
+
+ sysUsage.setMemoryUsage(memUsage);
+ sysUsage.setStoreUsage(storeUsage);
+ sysUsage.setTempUsage(tmpUsage);
+
+ broker.setDestinationPolicy(policyMap);
+ broker.setSystemUsage(sysUsage);
+ broker.setTempDataStore(tempStore);
+ broker.setPersistenceAdapter(persistence);
+
+ broker.addConnector("tcp://localhost:61618").setName("Default");
+ broker.start();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (broker != null) {
+ broker.stop();
+ }
+ }
+
+}
Propchange:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/TempStorageBlockedBrokerTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/TempStorageBlockedBrokerTest.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/store/temp/TempStoreTest.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/store/temp/TempStoreTest.java?rev=881313&view=auto
==============================================================================
---
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/store/temp/TempStoreTest.java
(added)
+++
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/store/temp/TempStoreTest.java
Tue Nov 17 14:55:45 2009
@@ -0,0 +1,106 @@
+/**
+ * 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.temp;
+
+import java.io.File;
+import java.util.LinkedList;
+import java.util.List;
+import junit.framework.TestCase;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.Message;
+import org.apache.activemq.command.MessageId;
+import org.apache.kahadb.util.IOHelper;
+
+public class TempStoreTest extends TestCase {
+ protected TempStore ts;
+ protected final LinkedList<Message> testList = new LinkedList<Message>();
+ protected List<Message> list;
+
+ public void testSize() throws Exception {
+
+ this.list.addAll(testList);
+
+ assertEquals(this.list.size(), this.testList.size());
+
+ }
+
+ public void testAddFirst() throws Exception {
+ this.list.addAll(testList);
+ assertEquals(this.list.size(), this.testList.size());
+ Message first = createMessage(10000);
+ this.list.add(0, first);
+ assertEquals(first, this.list.get(0));
+ assertEquals(this.list.size(), this.testList.size() + 1);
+ }
+
+ public void testAddLast() throws Exception {
+ this.list.addAll(testList);
+ assertEquals(this.list.size(), this.testList.size());
+ Message last = createMessage(10000);
+ this.list.add(last);
+ assertEquals(last, this.list.get(this.list.size()-1));
+ assertEquals(this.list.size(), this.testList.size() + 1);
+ }
+
+ public void testRemoveFirst() throws Exception {
+ this.list.addAll(testList);
+ assertEquals(testList.get(0), this.list.remove(0));
+ assertEquals(this.list.size(), testList.size() - 1);
+ System.err.println(this.list.get(0).getMessageId());
+ for (int i =0; i < testList.size();i++) {
+ System.err.println(testList.get(i).getMessageId());
+ }
+ for (int i = 1; i < testList.size(); i++) {
+ assertEquals(testList.get(i), this.list.get(i - 1));
+ }
+ }
+
+ private Message createMessage(int seq) {
+ ActiveMQMessage message = new ActiveMQMessage();
+ message.setCommandId((short) 1);
+ message.setDestination(new ActiveMQQueue("queue"));
+
+ message.setMessageId(new MessageId("c1:1:1", seq));
+ return message;
+
+ }
+
+ protected void setUp() throws Exception {
+ super.setUp();
+ this.ts = new TempStore();
+ File directory = new File("target/test");
+ this.ts.setDirectory(directory);
+ IOHelper.mkdirs(directory);
+ IOHelper.deleteChildren(directory);
+
+ ts.start();
+ for (int i = 0; i < 10; i++) {
+
+ Message msg = createMessage(i);
+ testList.add(msg);
+ }
+ this.list = ts.getList(getName());
+ }
+
+ protected void tearDown() throws Exception {
+ testList.clear();
+ if (ts != null) {
+ ts.stop();
+ }
+ }
+}
Propchange:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/store/temp/TempStoreTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/store/temp/TempStoreTest.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
