fixed up the Stomp test cases

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

activemq-core/src/main/java/org/apache/activemq/transport/stomp/Abort.java

@@ -44,6 +44,7 @@ class Abort implements StompCommand {
 
         TransactionId txnId = format.getTransactionId(user_tx_id);
         TransactionInfo tx = new TransactionInfo();
+        tx.setConnectionId(format.getConnectionId());
         tx.setTransactionId(txnId);
         tx.setType(TransactionInfo.ROLLBACK);
         format.clearTransactionId(user_tx_id);

activemq-core/src/main/java/org/apache/activemq/transport/stomp/Begin.java

@@ -44,6 +44,7 @@ public class Begin implements StompCommand {
         }
         int tx_id = StompWireFormat.generateTransactionId();
         TransactionId transactionId = format.registerTransactionId(user_tx_id, tx_id);
+        tx.setConnectionId(format.getConnectionId());
         tx.setTransactionId(transactionId);
         tx.setType(TransactionInfo.BEGIN);
         return new CommandEnvelope(tx, headers);

activemq-core/src/main/java/org/apache/activemq/transport/stomp/Commit.java

@@ -39,12 +39,15 @@ class Commit implements StompCommand {
 
         String user_tx_id = headers.getProperty(Stomp.Headers.TRANSACTION);
 
-        if (!headers.containsKey(Stomp.Headers.TRANSACTION)) {
+        if (user_tx_id == null) {
             throw new ProtocolException("Must specify the transaction you are committing");
         }
 
         TransactionId tx_id = format.getTransactionId(user_tx_id);
+        if (tx_id == null)
+            throw new ProtocolException(user_tx_id + " is an invalid transaction id");
         TransactionInfo tx = new TransactionInfo();
+        tx.setConnectionId(format.getConnectionId());
         tx.setTransactionId(tx_id);
         tx.setType(TransactionInfo.COMMIT_ONE_PHASE);
         format.clearTransactionId(user_tx_id);

activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java

@@ -314,7 +314,7 @@ public class StompTest extends CombinationTestSupport {
 
 
         //receive message from socket
-        frame = receiveFrame(10000);
+        frame = receiveFrame(1000);
         assertTrue(frame.startsWith("MESSAGE"));
 
         //remove suscription
@@ -325,8 +325,7 @@ public class StompTest extends CombinationTestSupport {
             Stomp.NULL;
         sendFrame(frame);
 
-        // lets wait for the unsubscribe to take effect
-        Thread.sleep(1000);
+        waitForFrameToTakeEffect();
         
         //send a message to our queue
         sendMessage("second message");
@@ -344,6 +343,8 @@ public class StompTest extends CombinationTestSupport {
 
 
     public void testTransactionCommit() throws Exception {
+        MessageConsumer consumer = session.createConsumer(queue);
+
         String frame =
             "CONNECT\n" +
             "login: brianm\n" +
@@ -353,7 +354,7 @@ public class StompTest extends CombinationTestSupport {
 
         String f = receiveFrame(1000);
         assertTrue(f.startsWith("CONNECTED"));
-
+        
         frame =
             "BEGIN\n" +
             "transaction: tx1\n" +
@@ -370,7 +371,6 @@ public class StompTest extends CombinationTestSupport {
             Stomp.NULL;
         sendFrame(frame);
 
-
         frame =
             "COMMIT\n" +
             "transaction: tx1\n" +
@@ -378,16 +378,15 @@ public class StompTest extends CombinationTestSupport {
             Stomp.NULL;
         sendFrame(frame);
 
-        // This test case is currently failing
-
-        MessageConsumer consumer = session.createConsumer(queue);
+        waitForFrameToTakeEffect();
+        
         TextMessage message = (TextMessage) consumer.receive(1000);
-        assertNotNull(message);
-
-
+        assertNotNull("Should have received a message", message);
     }
 
     public void testTransactionRollback() throws Exception {
+        MessageConsumer consumer = session.createConsumer(queue);
+
         String frame =
             "CONNECT\n" +
             "login: brianm\n" +
@@ -409,7 +408,7 @@ public class StompTest extends CombinationTestSupport {
             "SEND\n" +
             "destination:/queue/" + getQueueName() + "\n" +
             "transaction: tx1\n" +
-            "\n\n" +
+            "\n" +
             "first message" +
             Stomp.NULL;
         sendFrame(frame);
@@ -422,11 +421,18 @@ public class StompTest extends CombinationTestSupport {
             Stomp.NULL;
         sendFrame(frame);
 
+        frame =
+            "BEGIN\n" +
+            "transaction: tx1\n" +
+            "\n\n" +
+            Stomp.NULL;
+        sendFrame(frame);
+
         frame =
             "SEND\n" +
             "destination:/queue/" + getQueueName() + "\n" +
             "transaction: tx1\n" +
-            "\n\n" +
+            "\n" +
             "second message" +
             Stomp.NULL;
         sendFrame(frame);
@@ -439,17 +445,18 @@ public class StompTest extends CombinationTestSupport {
         sendFrame(frame);
 
         // This test case is currently failing
+        waitForFrameToTakeEffect();
 
         //only second msg should be received since first msg was rolled back
-        MessageConsumer consumer = session.createConsumer(queue);
         TextMessage message = (TextMessage) consumer.receive(1000);
         assertNotNull(message);
-        assertEquals("second message", message.getText());
-
-
+        assertEquals("second message", message.getText().trim());
     }
 
-
-
-
+    protected void waitForFrameToTakeEffect() throws InterruptedException {
+        // bit of a dirty hack :)
+        // another option would be to force some kind of receipt to be returned
+        // from the frame
+        Thread.sleep(2000);
+    }
 }