Fixes to transaction handling.

- This line, and those below, will be ignored--

M    java/broker/test/src/org/apache/qpid/server/UnitTests.java
M    java/broker/test/src/org/apache/qpid/server/queue/DeliveryManagerTest.java
A    java/broker/test/src/org/apache/qpid/server/txn
AM   java/broker/test/src/org/apache/qpid/server/txn/UnitTests.java
AM   java/broker/test/src/org/apache/qpid/server/txn/TxnBufferTest.java
M    java/broker/test/src/org/apache/qpid/server/util/UnitTests.java
AM   java/broker/test/src/org/apache/qpid/server/util/OrderedMapHelperTest.java
M    java/broker/src/org/apache/qpid/server/queue/SubscriptionImpl.java
M    java/broker/src/org/apache/qpid/server/queue/DeliveryManager.java
AM   java/broker/src/org/apache/qpid/server/queue/FailedDequeueException.java
M    java/broker/src/org/apache/qpid/server/queue/AMQMessage.java
M    java/broker/src/org/apache/qpid/server/queue/AMQQueue.java
M    java/broker/src/org/apache/qpid/server/queue/Subscription.java
M    java/broker/src/org/apache/qpid/server/queue/NoConsumersException.java
M    java/broker/src/org/apache/qpid/server/AMQChannel.java
M    java/broker/src/org/apache/qpid/server/txn/TxnBuffer.java
M    java/broker/src/org/apache/qpid/server/txn/TxnOp.java
AM   java/broker/src/org/apache/qpid/server/util/OrderedMapHelper.java
M    java/broker/src/org/apache/qpid/server/handler/TxCommitHandler.java
AM   java/client/test/src/org/apache/qpid/transacted/UnitTests.java
M    java/client/test/src/org/apache/qpid/client/AllClientUnitTests.java


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

6 files changed, 442 insertions, 3 deletions

diff --git a/java/broker/test/src/org/apache/qpid/server/UnitTests.java b/java/broker/test/src/org/apache/qpid/server/UnitTests.java
index 377b1fb64e..3372584c07 100644
--- a/java/broker/test/src/org/apache/qpid/server/UnitTests.java
+++ b/java/broker/test/src/org/apache/qpid/server/UnitTests.java
@@ -28,6 +28,7 @@ import org.junit.runners.Suite;
         org.apache.qpid.server.protocol.UnitTests.class,
         org.apache.qpid.server.queue.UnitTests.class,
         org.apache.qpid.server.store.UnitTests.class,
+        org.apache.qpid.server.txn.UnitTests.class,
         org.apache.qpid.server.util.UnitTests.class
         })
 public class UnitTests
diff --git a/java/broker/test/src/org/apache/qpid/server/queue/DeliveryManagerTest.java b/java/broker/test/src/org/apache/qpid/server/queue/DeliveryManagerTest.java
index f84d426877..ef287e079b 100644
--- a/java/broker/test/src/org/apache/qpid/server/queue/DeliveryManagerTest.java
+++ b/java/broker/test/src/org/apache/qpid/server/queue/DeliveryManagerTest.java
@@ -138,7 +138,9 @@ public class DeliveryManagerTest extends MessageTestHelper
     @Test (expected=NoConsumersException.class)
     public void noConsumers() throws AMQException
     {
-        _mgr.deliver("Me", message(true));
+        AMQMessage msg = message(true);
+        _mgr.deliver("Me", msg);
+        msg.checkDeliveredToConsumer();        
     }
 
     @Test (expected=NoConsumersException.class)
@@ -147,7 +149,9 @@ public class DeliveryManagerTest extends MessageTestHelper
         TestSubscription s = new TestSubscription("A");
         _subscriptions.addSubscriber(s);
         s.setSuspended(true);
-        _mgr.deliver("Me", message(true));
+        AMQMessage msg = message(true);
+        _mgr.deliver("Me", msg);
+        msg.checkDeliveredToConsumer();
     }
 
     public static junit.framework.Test suite()
diff --git a/java/broker/test/src/org/apache/qpid/server/txn/TxnBufferTest.java b/java/broker/test/src/org/apache/qpid/server/txn/TxnBufferTest.java
new file mode 100644
index 0000000000..2b49752023
--- /dev/null
+++ b/java/broker/test/src/org/apache/qpid/server/txn/TxnBufferTest.java
@@ -0,0 +1,305 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.server.txn;
+
+import junit.framework.JUnit4TestAdapter;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.Ignore;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.TestableMemoryMessageStore;
+
+import java.util.LinkedList;
+
+public class TxnBufferTest
+{
+    private final LinkedList<MockOp> ops = new LinkedList<MockOp>();  
+
+    @Before
+    public void setup() throws Exception
+    {
+    }
+
+    @Test
+    public void commit() throws AMQException
+    {
+        MockStore store = new MockStore();
+
+        TxnBuffer buffer = new TxnBuffer(store);
+        buffer.enlist(new MockOp().expectPrepare().expectCommit());
+        //check relative ordering
+        MockOp op = new MockOp().expectPrepare().expectPrepare().expectCommit().expectCommit();
+        buffer.enlist(op);
+        buffer.enlist(op);
+        buffer.enlist(new MockOp().expectPrepare().expectCommit());
+
+        buffer.commit();
+
+        validateOps();
+        store.validate();
+    }
+
+    @Test
+    public void rollback() throws AMQException
+    {
+        MockStore store = new MockStore();
+
+        TxnBuffer buffer = new TxnBuffer(store);
+        buffer.enlist(new MockOp().expectRollback());
+        buffer.enlist(new MockOp().expectRollback());
+        buffer.enlist(new MockOp().expectRollback());
+
+        buffer.rollback();
+
+        validateOps();
+        store.validate();
+    }
+
+    @Test
+    public void commitWithFailureDuringPrepare() throws AMQException
+    {
+        MockStore store = new MockStore();
+        store.expectBegin().expectAbort();
+
+        TxnBuffer buffer = new TxnBuffer(store);
+        buffer.containsPersistentChanges();
+        buffer.enlist(new MockOp().expectPrepare().expectUndoPrepare());
+        buffer.enlist(new TxnTester(store));
+        buffer.enlist(new MockOp().expectPrepare().expectUndoPrepare());
+        buffer.enlist(new FailedPrepare());
+        buffer.enlist(new MockOp());
+
+        buffer.commit();        
+        validateOps();
+        store.validate();
+    }
+
+    @Test
+    public void commitWithPersistance() throws AMQException
+    {
+        MockStore store = new MockStore();
+        store.expectBegin().expectCommit();
+
+        TxnBuffer buffer = new TxnBuffer(store);
+        buffer.enlist(new MockOp().expectPrepare().expectCommit());
+        buffer.enlist(new MockOp().expectPrepare().expectCommit());
+        buffer.enlist(new MockOp().expectPrepare().expectCommit());
+        buffer.enlist(new TxnTester(store));
+        buffer.containsPersistentChanges();
+
+        buffer.commit();
+        validateOps();
+        store.validate();
+    }
+
+    private void validateOps()
+    {
+        for(MockOp op : ops)
+        {
+            op.validate();
+        }
+    }
+
+    public static junit.framework.Test suite()
+    {
+        return new JUnit4TestAdapter(TxnBufferTest.class);
+    }
+
+    class MockOp implements TxnOp
+    {        
+        final Object PREPARE = "PREPARE";
+        final Object COMMIT = "COMMIT";
+        final Object UNDO_PREPARE = "UNDO_PREPARE";
+        final Object ROLLBACK = "ROLLBACK";
+
+        private final LinkedList expected = new LinkedList();
+
+        MockOp()
+        {
+            ops.add(this);
+        }
+
+        public void prepare()
+        {
+            assertEquals(expected.removeLast(), PREPARE);
+        }
+
+        public void commit()
+        {
+            assertEquals(expected.removeLast(), COMMIT);
+        }
+
+        public void undoPrepare()
+        {
+            assertEquals(expected.removeLast(), UNDO_PREPARE);
+        }
+
+        public void rollback()
+        {
+            assertEquals(expected.removeLast(), ROLLBACK);
+        }
+
+        private MockOp expect(Object optype)
+        {
+            expected.addFirst(optype);
+            return this;
+        }
+
+        MockOp expectPrepare()
+        {
+            return expect(PREPARE);
+        }
+
+        MockOp expectCommit()
+        {
+            return expect(COMMIT);
+        }
+
+        MockOp expectUndoPrepare()
+        {
+            return expect(UNDO_PREPARE);
+        }
+
+        MockOp expectRollback()
+        {
+            return expect(ROLLBACK);
+        }
+
+        void validate()
+        {
+            assertEquals("Expected ops were not all invoked", new LinkedList(), expected);
+        }
+
+        void clear()
+        {
+            expected.clear();
+        }
+    }
+
+    class MockStore extends TestableMemoryMessageStore
+    {
+        final Object BEGIN = "BEGIN";
+        final Object ABORT = "ABORT";
+        final Object COMMIT = "COMMIT";
+
+        private final LinkedList expected = new LinkedList();
+        private boolean inTran;
+
+        public void beginTran() throws AMQException
+        {
+            assertEquals(expected.removeLast(), BEGIN);
+            inTran = true;
+        }
+        
+        public void commitTran() throws AMQException
+        {
+            assertEquals(expected.removeLast(), COMMIT);
+            inTran = false;
+        }
+        
+        public void abortTran() throws AMQException
+        {
+            assertEquals(expected.removeLast(), ABORT);
+            inTran = false;
+        }
+
+        public boolean inTran()
+        {
+            return inTran;
+        }
+
+        private MockStore expect(Object optype)
+        {
+            expected.addFirst(optype);
+            return this;
+        }
+
+        MockStore expectBegin()
+        {
+            return expect(BEGIN);
+        }
+
+        MockStore expectCommit()
+        {
+            return expect(COMMIT);
+        }
+
+        MockStore expectAbort()
+        {
+            return expect(ABORT);
+        }
+
+        void clear()
+        {
+            expected.clear();
+        }
+
+        void validate()
+        {
+            assertEquals("Expected ops were not all invoked", new LinkedList(), expected);
+        }
+    }
+
+    class NullOp implements TxnOp
+    {        
+        public void prepare() throws AMQException
+        {
+        }
+        public void commit()
+        {
+        }
+        public void undoPrepare()
+        {
+        }
+        public void rollback()
+        {
+        }
+    }
+
+    class FailedPrepare extends NullOp
+    {        
+        public void prepare() throws AMQException
+        {
+            throw new AMQException("Fail!");
+        }
+    }
+
+    class TxnTester extends NullOp
+    {        
+        private final MessageStore store;
+
+        TxnTester(MessageStore store)
+        {
+            this.store = store;
+        }
+
+        public void prepare() throws AMQException
+        {
+            assertTrue("Expected prepare to be performed under txn", store.inTran());
+        }
+
+        public void commit()
+        {
+            assertTrue("Expected commit not to be performed under txn", !store.inTran());
+        }
+    }
+
+}
diff --git a/java/broker/test/src/org/apache/qpid/server/txn/UnitTests.java b/java/broker/test/src/org/apache/qpid/server/txn/UnitTests.java
new file mode 100644
index 0000000000..dd63b260fd
--- /dev/null
+++ b/java/broker/test/src/org/apache/qpid/server/txn/UnitTests.java
@@ -0,0 +1,34 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.server.txn;
+
+import junit.framework.JUnit4TestAdapter;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+@RunWith(Suite.class)
+@Suite.SuiteClasses({
+        TxnBufferTest.class
+})
+public class UnitTests
+{
+    public static junit.framework.Test suite()
+    {
+        return new JUnit4TestAdapter(UnitTests.class);
+    }
+}
diff --git a/java/broker/test/src/org/apache/qpid/server/util/OrderedMapHelperTest.java b/java/broker/test/src/org/apache/qpid/server/util/OrderedMapHelperTest.java
new file mode 100644
index 0000000000..74340e98c0
--- /dev/null
+++ b/java/broker/test/src/org/apache/qpid/server/util/OrderedMapHelperTest.java
@@ -0,0 +1,95 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.server.util;
+
+import junit.framework.JUnit4TestAdapter;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.Ignore;
+import org.apache.qpid.AMQException;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+public class OrderedMapHelperTest
+{
+    private final Object lock = new Object();
+    private final Map<Integer, String> map = new LinkedHashMap<Integer, String>();
+    private final OrderedMapHelper<Integer, String> helper = new OrderedMapHelper<Integer, String>(map, lock, 0);
+
+    @Before
+    public void setup() throws Exception
+    {
+        map.put(1, "One");
+        map.put(2, "Two");
+        map.put(5, "Five");
+        map.put(8, "Eight");
+        map.put(10, "Ten");
+    }
+
+    @Test
+    public void specific()
+    {
+        Map<Integer, String> slice = helper.getValues(5, false);
+        assertEquals(1, slice.size());
+        assertTrue(slice.containsKey(5));
+        assertTrue(slice.containsValue("Five"));
+        assertEquals("Five", slice.get(5));
+    }
+
+    @Test
+    public void multiple()
+    {
+        Map<Integer, String> slice = helper.getValues(5, true);
+        assertEquals(3, slice.size());
+
+        assertTrue(slice.containsKey(1));
+        assertTrue(slice.containsKey(2));
+        assertTrue(slice.containsKey(5));
+
+        assertEquals("One", slice.get(1));
+        assertEquals("Two", slice.get(2));
+        assertEquals("Five", slice.get(5));
+    }
+
+    @Test
+    public void all()
+    {
+        Map<Integer, String> slice = helper.getValues(0/*the 'wildcard'*/, true);
+        assertEquals(5, slice.size());
+
+        assertTrue(slice.containsKey(1));
+        assertTrue(slice.containsKey(2));
+        assertTrue(slice.containsKey(5));
+        assertTrue(slice.containsKey(8));
+        assertTrue(slice.containsKey(10));
+
+        assertEquals("One", slice.get(1));
+        assertEquals("Two", slice.get(2));
+        assertEquals("Five", slice.get(5));
+        assertEquals("Eight", slice.get(8));
+        assertEquals("Ten", slice.get(10));
+    }
+
+    public static junit.framework.Test suite()
+    {
+        return new JUnit4TestAdapter(OrderedMapHelperTest.class);
+    }
+}
diff --git a/java/broker/test/src/org/apache/qpid/server/util/UnitTests.java b/java/broker/test/src/org/apache/qpid/server/util/UnitTests.java
index d6cc471413..5c2242fb0d 100644
--- a/java/broker/test/src/org/apache/qpid/server/util/UnitTests.java
+++ b/java/broker/test/src/org/apache/qpid/server/util/UnitTests.java
@@ -22,7 +22,7 @@ import org.junit.runner.RunWith;
 import org.junit.runners.Suite;
 
 @RunWith(Suite.class)
-@Suite.SuiteClasses({LoggingProxyTest.class})
+@Suite.SuiteClasses({LoggingProxyTest.class, OrderedMapHelperTest.class})
 public class UnitTests
 {
     public static junit.framework.Test suite()