merge bug22628 into default

Author: Matthias Radestock

test/src/com/rabbitmq/client/test/BrokerTestCase.java

@@ -159,19 +159,36 @@ protected GetResponse basicGet(String q) throws IOException {
   }
 
   protected void basicPublishPersistent(String q) throws IOException {
-    channel.basicPublish("", q, MessageProperties.PERSISTENT_TEXT_PLAIN, "persistent message".getBytes());
+      basicPublishPersistent("persistent message".getBytes(), q);
   }
 
-  protected void basicPublishPersistent(String x, String routingKey) throws IOException {
-    channel.basicPublish(x, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, "persistent message".getBytes());
+  protected void basicPublishPersistent(byte[] msg, String q) throws IOException {
+      basicPublishPersistent(msg, "", q);
   }
 
+  protected void basicPublishPersistent(String x, String routingKey) throws IOException {
+      basicPublishPersistent("persistent message".getBytes(), x, routingKey);
+    }
+
+
+  protected void basicPublishPersistent(byte[] msg, String x, String routingKey) throws IOException {
+      channel.basicPublish(x, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, msg);
+    }
+
   protected void basicPublishVolatile(String q) throws IOException {
-    channel.basicPublish("", q, MessageProperties.TEXT_PLAIN, "not persistent message".getBytes());
+      basicPublishVolatile("not persistent message".getBytes(), q);
+  }
+
+  protected void basicPublishVolatile(byte[] msg, String q) throws IOException {
+      basicPublishVolatile(msg, "", q);
   }
 
   protected void basicPublishVolatile(String x, String routingKey) throws IOException {
-    channel.basicPublish(x, routingKey, MessageProperties.TEXT_PLAIN, "not persistent message".getBytes());
+      basicPublishVolatile("not persistent message".getBytes(), x, routingKey);
+  }
+
+  protected void basicPublishVolatile(byte[] msg, String x, String routingKey) throws IOException {
+      channel.basicPublish(x, routingKey, MessageProperties.TEXT_PLAIN, msg);
   }
 
   protected void declareAndBindDurableQueue(String q, String x, String r) throws IOException {

test/src/com/rabbitmq/client/test/server/PersisterRestart6.java

@@ -0,0 +1,25 @@
+package com.rabbitmq.client.test.server;
+
+import java.io.IOException;
+
+public class PersisterRestart6 extends RestartBase {
+
+    private static final String q = "Restart6";
+    
+    public void testRestart() throws IOException, InterruptedException {
+        declareDurableQueue(q);
+        basicPublishPersistent("a".getBytes(), q);
+        basicPublishPersistent("b".getBytes(), q);
+        basicPublishPersistent("c".getBytes(), q);
+        forceSnapshot();
+        restart();
+        assertTrue(new String(basicGet(q).getBody()).equals("a"));
+        forceSnapshot();
+        restart();
+        restart();
+        assertTrue(new String(basicGet(q).getBody()).equals("b"));
+        assertTrue(new String(basicGet(q).getBody()).equals("c"));
+        deleteQueue(q);
+    }
+    
+}

test/src/com/rabbitmq/client/test/server/PersisterRestartTests.java

@@ -42,6 +42,7 @@ public static TestSuite suite() {
         suite.addTestSuite(PersisterRestart3.class);
         suite.addTestSuite(PersisterRestart4.class);
         suite.addTestSuite(PersisterRestart5.class);
+        suite.addTestSuite(PersisterRestart6.class);
         return suite;
     }