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/qpid@493872 13f79535-47bb-0310-9956-ffa450edef68

10 files changed, 79 insertions, 57 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/failover/FailoverSupport.java b/java/client/src/main/java/org/apache/qpid/client/failover/FailoverSupport.java
index bb7397f194..2e08d303c8 100644
--- a/java/client/src/main/java/org/apache/qpid/client/failover/FailoverSupport.java
+++ b/java/client/src/main/java/org/apache/qpid/client/failover/FailoverSupport.java
@@ -56,7 +56,7 @@ public abstract class FailoverSupport
                 }
                 catch (FailoverException e)
                 {
-                    _log.info("Failover exception caught during operation");
+                    _log.info("Failover exception caught during operation: " + e, e);
                 }
             }
         }
diff --git a/java/client/src/old_test/java/org/apache/qpid/codec/BasicDeliverTest.java b/java/client/src/old_test/java/org/apache/qpid/codec/BasicDeliverTest.java
index 892b349cea..1c616f0f02 100644
--- a/java/client/src/old_test/java/org/apache/qpid/codec/BasicDeliverTest.java
+++ b/java/client/src/old_test/java/org/apache/qpid/codec/BasicDeliverTest.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
@@ -254,7 +254,7 @@ public class BasicDeliverTest
 
     static BasicDeliverBody createBasicDeliverBody()
     {
-        BasicDeliverBody body = new BasicDeliverBody();
+        BasicDeliverBody body = new BasicDeliverBody((byte)8, (byte)0);
         body.consumerTag = "myConsumerTag";
         body.deliveryTag = 1;
         body.exchange = "myExchange";
diff --git a/java/client/src/old_test/java/org/apache/qpid/framing/FieldTableTest.java b/java/client/src/old_test/java/org/apache/qpid/framing/FieldTableTest.java
index 2a7cb8be30..e56e38c098 100644
--- a/java/client/src/old_test/java/org/apache/qpid/framing/FieldTableTest.java
+++ b/java/client/src/old_test/java/org/apache/qpid/framing/FieldTableTest.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,6 +20,7 @@
  */
 package org.apache.qpid.framing;
 
+import junit.framework.TestCase;
 import org.apache.mina.common.ByteBuffer;
 
 import java.io.BufferedReader;
@@ -29,8 +30,6 @@ import java.io.Reader;
 import java.util.Enumeration;
 import java.util.Properties;
 
-import junit.framework.TestCase;
-
 public class FieldTableTest extends TestCase
 {
 
@@ -47,7 +46,7 @@ public class FieldTableTest extends TestCase
                    EncodingUtils.encodedLongStringLength(value);
 
         assertEquals(table.getEncodedSize(), size);
-        
+
         key = "Integer";
         Integer number = new Integer(60);
         table.put(key, number);
@@ -87,7 +86,7 @@ public class FieldTableTest extends TestCase
         doTestEncoding(load("FieldTableTest2.properties"));
     }
     */
-    void doTestEncoding(FieldTable table) throws AMQFrameDecodingException
+    void doTestEncoding(FieldTable table) throws AMQFrameDecodingException, AMQProtocolVersionException
     {
         assertEquivalent(table, encodeThenDecode(table));
     }
@@ -102,7 +101,7 @@ public class FieldTableTest extends TestCase
         }
     }
 
-    FieldTable encodeThenDecode(FieldTable table) throws AMQFrameDecodingException
+    FieldTable encodeThenDecode(FieldTable table) throws AMQFrameDecodingException, AMQProtocolVersionException
     {
         ContentHeaderBody header = new ContentHeaderBody();
         header.classId = 6;
diff --git a/java/client/src/old_test/java/org/apache/qpid/requestreply1/VmRequestReply.java b/java/client/src/old_test/java/org/apache/qpid/requestreply1/VmRequestReply.java
index 56d1ce9b6d..a1577cffa5 100644
--- a/java/client/src/old_test/java/org/apache/qpid/requestreply1/VmRequestReply.java
+++ b/java/client/src/old_test/java/org/apache/qpid/requestreply1/VmRequestReply.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,12 +20,9 @@
  */
 package org.apache.qpid.requestreply1;
 
-import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
-import org.apache.qpid.client.transport.TransportConnection;
-import org.apache.qpid.test.VMBrokerSetup;
-import org.apache.log4j.Logger;
-
 import junit.framework.TestCase;
+import org.apache.log4j.Logger;
+import org.apache.qpid.testutil.VMBrokerSetup;
 
 public class VmRequestReply extends TestCase
 {
diff --git a/java/client/src/old_test/java/org/apache/qpid/test/unit/jndi/referenceabletest/JNDIReferenceableTest.java b/java/client/src/old_test/java/org/apache/qpid/test/unit/jndi/referenceabletest/JNDIReferenceableTest.java
index 4731caca98..8272b13c1d 100644
--- a/java/client/src/old_test/java/org/apache/qpid/test/unit/jndi/referenceabletest/JNDIReferenceableTest.java
+++ b/java/client/src/old_test/java/org/apache/qpid/test/unit/jndi/referenceabletest/JNDIReferenceableTest.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,15 +20,12 @@
  */
 package org.apache.qpid.test.unit.jndi.referenceabletest;
 
-import org.apache.qpid.client.transport.TransportConnection;
-import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
-import org.apache.qpid.test.VMBrokerSetup;
+import junit.framework.TestCase;
+import org.apache.qpid.testutil.VMBrokerSetup;
 
 import javax.naming.NameAlreadyBoundException;
 import javax.naming.NoInitialContextException;
 
-import junit.framework.TestCase;
-
 /**
  * Usage: To run these you need to have the sun JNDI SPI for the FileSystem.
  * This can be downloaded from sun here:
diff --git a/java/client/src/old_test/java/org/apache/qpid/weblogic/ServiceRequestingClient.java b/java/client/src/old_test/java/org/apache/qpid/weblogic/ServiceRequestingClient.java
index a1e15258c3..2f64a1dde5 100644
--- a/java/client/src/old_test/java/org/apache/qpid/weblogic/ServiceRequestingClient.java
+++ b/java/client/src/old_test/java/org/apache/qpid/weblogic/ServiceRequestingClient.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,20 +20,13 @@
  */
 package org.apache.qpid.weblogic;
 
-import org.apache.qpid.jms.*;
 import org.apache.log4j.Logger;
 
-import javax.naming.NamingException;
-import javax.naming.InitialContext;
-import javax.naming.Context;
 import javax.jms.*;
-import javax.jms.MessageConsumer;
-import javax.jms.Session;
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import javax.naming.NamingException;
 import java.util.Hashtable;
-import java.io.File;
-import java.io.FilenameFilter;
-import java.io.Reader;
-import java.io.FileReader;
 
 /**
  * Created by IntelliJ IDEA.
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/client/AMQConnectionTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/client/AMQConnectionTest.java
index 68bdc6ddf2..c4b60be1d1 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/client/AMQConnectionTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/client/AMQConnectionTest.java
@@ -24,7 +24,7 @@ import org.apache.qpid.client.AMQTopic;
 import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQQueue;
 import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.testutil.VMBrokerSetup;
+import org.apache.qpid.client.transport.TransportConnection;
 
 import javax.jms.*;
 
@@ -41,6 +41,7 @@ public class AMQConnectionTest extends TestCase
     protected void setUp() throws Exception
     {
         super.setUp();
+        TransportConnection.createVMBroker(1);
         _connection = new AMQConnection("vm://:1", "guest", "guest", "fred", "/test");
         _topic = new AMQTopic("mytopic");
         _queue = new AMQQueue("myqueue");
@@ -48,6 +49,7 @@ public class AMQConnectionTest extends TestCase
 
     protected void tearDown() throws Exception
     {
+        super.tearDown();
         try
         {
             _connection.close();
@@ -55,8 +57,8 @@ public class AMQConnectionTest extends TestCase
         catch (JMSException e)
         {
             //ignore 
-        }
-        super.tearDown();
+        }        
+        TransportConnection.killAllVMBrokers();
     }
 
     /**
@@ -195,6 +197,6 @@ public class AMQConnectionTest extends TestCase
 
     public static junit.framework.Test suite()
     {
-        return new VMBrokerSetup(new junit.framework.TestSuite(AMQConnectionTest.class));
+        return new junit.framework.TestSuite(AMQConnectionTest.class);
     }
 }
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java
index ac789eb915..d84d66e26d 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java
@@ -20,17 +20,17 @@
  */
 package org.apache.qpid.test.unit.client.channelclose;
 
+import junit.framework.TestCase;
+import junit.textui.TestRunner;
+import org.apache.log4j.Logger;
 import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.testutil.VMBrokerSetup;
-import org.apache.log4j.Logger;
+import org.apache.qpid.client.transport.TransportConnection;
 
 import javax.jms.*;
 import java.util.ArrayList;
 import java.util.List;
 
-import junit.framework.TestCase;
-import junit.textui.TestRunner;
 
 /**
  * Due to bizarre exception handling all sessions are closed if you get
@@ -64,6 +64,7 @@ public class ChannelCloseOkTest extends TestCase
     {
         super.setUp();
 
+        TransportConnection.createVMBroker(1);
         _connection = new AMQConnection(_connectionString, "guest", "guest", randomize("Client"), "/test_path");
 
         _destination1 = new AMQQueue("q1", true);
@@ -192,7 +193,15 @@ public class ChannelCloseOkTest extends TestCase
         {
             while (received.size() < count)
             {
-                received.wait();
+                try
+                {
+                    received.wait();
+                }
+                catch (InterruptedException e)
+                {
+                    _log.info("Interrupted: " + e);
+                    throw e;
+                }
             }
         }
     }
@@ -209,6 +218,6 @@ public class ChannelCloseOkTest extends TestCase
 
     public static junit.framework.Test suite()
     {
-        return new VMBrokerSetup(new junit.framework.TestSuite(ChannelCloseOkTest.class));
+        return new junit.framework.TestSuite(ChannelCloseOkTest.class);
     }
 }
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java
index 184d7cb015..4f0ca6d3aa 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java
@@ -1,11 +1,23 @@
-/**
- * User: Robert Greig
- * Date: 12-Dec-2006
- ******************************************************************************
- * (c) Copyright JP Morgan Chase Ltd 2006. All rights reserved. No part of
- * this program may be photocopied reproduced or translated to another
- * program language without prior written consent of JP Morgan Chase Ltd
- ******************************************************************************/
+/*
+ *
+ * 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.qpid.test.unit.message;
 
 import junit.framework.TestCase;
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
index 5ded8aaeef..8263e7f126 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
@@ -25,7 +25,7 @@ import org.apache.qpid.url.URLSyntaxException;
 import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQSession;
 import org.apache.qpid.client.AMQTopic;
-import org.apache.qpid.testutil.VMBrokerSetup;
+import org.apache.qpid.client.transport.TransportConnection;
 
 import javax.jms.JMSException;
 import javax.jms.Message;
@@ -39,6 +39,19 @@ import junit.framework.TestCase;
 
 public class DurableSubscriptionTest extends TestCase
 {
+
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        TransportConnection.createVMBroker(1);
+    }
+
+    protected void tearDown() throws Exception
+    {
+        super.tearDown();
+        TransportConnection.killAllVMBrokers();
+    }
+
     public void testUnsubscribe() throws AMQException, JMSException, URLSyntaxException
     {
         AMQTopic topic = new AMQTopic("MyTopic");
@@ -128,6 +141,6 @@ public class DurableSubscriptionTest extends TestCase
 
     public static junit.framework.Test suite()
     {
-        return new VMBrokerSetup(new junit.framework.TestSuite(DurableSubscriptionTest.class));
+        return new junit.framework.TestSuite(DurableSubscriptionTest.class);
     }
 }