QPID-32: new model for holding and processing message in memory to support new persistent stores

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@493872 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 23 insertions, 17 deletions

diff --git a/qpid/java/systests/src/old_test/java/org/apache/qpid/server/queue/SendPerfTest.java b/qpid/java/systests/src/old_test/java/org/apache/qpid/server/queue/SendPerfTest.java
index 6490b9f270..2c5712fd35 100644
--- a/qpid/java/systests/src/old_test/java/org/apache/qpid/server/queue/SendPerfTest.java
+++ b/qpid/java/systests/src/old_test/java/org/apache/qpid/server/queue/SendPerfTest.java
@@ -7,9 +7,9 @@
  * 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
@@ -27,6 +27,9 @@ import org.apache.qpid.framing.ContentBody;
 import org.apache.qpid.framing.ContentHeaderBody;
 import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.RequiredDeliveryException;
+import org.apache.qpid.server.txn.TransactionalContext;
+import org.apache.qpid.server.txn.NonTransactionalContext;
 import org.apache.qpid.server.exchange.AbstractExchange;
 import org.apache.qpid.server.exchange.Exchange;
 import org.apache.qpid.server.handler.OnCurrentThreadExecutor;
@@ -43,6 +46,7 @@ import org.apache.qpid.server.util.TimedRun;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.LinkedList;
 
 public class SendPerfTest extends TimedRun
 {
@@ -101,13 +105,16 @@ public class SendPerfTest extends TimedRun
         ContentHeaderBody header = new ContentHeaderBody();
         List<ContentBody> body = new ArrayList<ContentBody>();
         MessageStore messageStore = new SkeletonMessageStore();
+        // channel can be null since it is only used in ack processing which does not apply to this test
+        TransactionalContext txContext = new NonTransactionalContext(messageStore, null,
+                                                                     new LinkedList<RequiredDeliveryException>());
         body.add(new ContentBody());
+        MessageHandleFactory factory = new MessageHandleFactory();
         for (int i = 0; i < count; i++)
         {
-            for (AMQQueue q : queues)
-            {
-                q.deliver(new AMQMessage(messageStore, i, publish, header, body));
-            }
+            // this routes and delivers the message
+            AMQMessage msg = new AMQMessage(i, publish, txContext, header, queues, body, messageStore,
+                                            factory);
         }
     }
 
diff --git a/qpid/java/systests/src/old_test/java/org/apache/qpid/test/unit/ack/DisconnectAndRedeliverTest.java b/qpid/java/systests/src/old_test/java/org/apache/qpid/test/unit/ack/DisconnectAndRedeliverTest.java
index a3e555aac9..3e35e3c85b 100644
--- a/qpid/java/systests/src/old_test/java/org/apache/qpid/test/unit/ack/DisconnectAndRedeliverTest.java
+++ b/qpid/java/systests/src/old_test/java/org/apache/qpid/test/unit/ack/DisconnectAndRedeliverTest.java
@@ -7,9 +7,9 @@
  * 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
@@ -20,22 +20,19 @@
  */
 package org.apache.qpid.test.unit.ack;
 
+import junit.framework.TestCase;
 import org.apache.log4j.Logger;
 import org.apache.log4j.xml.DOMConfigurator;
 import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQQueue;
 import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
 import org.apache.qpid.client.transport.TransportConnection;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.store.TestableMemoryMessageStore;
 import org.apache.qpid.server.util.TestApplicationRegistry;
-import org.apache.qpid.test.VMBrokerSetup;
 
 import javax.jms.*;
 
-import junit.framework.TestCase;
-
 public class DisconnectAndRedeliverTest extends TestCase
 {
     private static final Logger _logger = Logger.getLogger(DisconnectAndRedeliverTest.class);
@@ -55,12 +52,14 @@ public class DisconnectAndRedeliverTest extends TestCase
     protected void setUp() throws Exception
     {
         super.setUp();
+        TransportConnection.createVMBroker(1);
         ApplicationRegistry.initialise(new TestApplicationRegistry(), 1);
     }
 
     protected void tearDown() throws Exception
     {
         super.tearDown();
+        TransportConnection.killAllVMBrokers();
     }
 
     /**
@@ -82,7 +81,7 @@ public class DisconnectAndRedeliverTest extends TestCase
         ((AMQSession) consumerSession).declareExchangeSynch("amq.direct", "direct");
 
         Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "/test");
-                 
+
 
         Session producerSession = con2.createSession(false, Session.CLIENT_ACKNOWLEDGE);
         MessageProducer producer = producerSession.createProducer(queue);
@@ -149,7 +148,7 @@ public class DisconnectAndRedeliverTest extends TestCase
         _logger.info("No messages redelivered as is expected");
         con.close();
 
-        _logger.info("Actually:" + store.getMessageMap().size());
+        _logger.info("Actually:" + store.getMessageMetaDataMap().size());
         //  assertTrue(store.getMessageMap().size() == 0);
     }
 
@@ -204,13 +203,13 @@ public class DisconnectAndRedeliverTest extends TestCase
         assertNull(tm);
         _logger.info("No messages redelivered as is expected");
 
-        _logger.info("Actually:" + store.getMessageMap().size());
-        assertTrue(store.getMessageMap().size() == 0);
+        _logger.info("Actually:" + store.getMessageMetaDataMap().size());
+        assertTrue(store.getMessageMetaDataMap().size() == 0);
         con.close();
     }
 
     public static junit.framework.Test suite()
     {
-        return new VMBrokerSetup(new junit.framework.TestSuite(DisconnectAndRedeliverTest.class));
+        return new junit.framework.TestSuite(DisconnectAndRedeliverTest.class);
     }
 }