Added: incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/testcases/RollbackTest.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/testcases/RollbackTest.java?rev=651133&view=auto ============================================================================== --- incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/testcases/RollbackTest.java (added) +++ incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/testcases/RollbackTest.java Wed Apr 23 18:54:20 2008 @@ -0,0 +1,132 @@ +/* + * + * 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.testcases; + +import org.apache.qpid.test.framework.Circuit; +import org.apache.qpid.test.framework.FrameworkBaseCase; +import org.apache.qpid.test.framework.MessagingTestConfigProperties; +import static org.apache.qpid.test.framework.MessagingTestConfigProperties.*; +import org.apache.qpid.test.framework.sequencers.CircuitFactory; + +import org.apache.qpid.junit.extensions.util.ParsedProperties; +import org.apache.qpid.junit.extensions.util.TestContextProperties; + +/** + * RollbackTest tests the rollback ability of transactional messaging. + * + * <p/><table id="crc"><caption>CRC Card</caption> + * <tr><th> Responsibilities <th> Collaborations + * <tr><td> Check messages sent but rolled back are never received. + * <tr><td> Check messages received but rolled back are redelivered on subsequent receives. + * <tr><td> Attempting to rollback outside of a transaction results in an IllegalStateException. + * </table> + */ +public class RollbackTest extends FrameworkBaseCase +{ + /** Used to read the tests configurable properties through. */ + ParsedProperties testProps; + + /** + * Creates a new test case with the specified name. + * + * @param name The test case name. + */ + public RollbackTest(String name) + { + super(name); + } + + /** Check messages sent but rolled back are never received. */ + public void testRolledbackMessageNotDelivered() + { + // Ensure transactional sessions are on. + testProps.setProperty(TRANSACTED_PUBLISHER_PROPNAME, true); + testProps.setProperty(ROLLBACK_PUBLISHER_PROPNAME, true); + + // Run the default test sequence over the test circuit checking for no errors. + CircuitFactory circuitFactory = getCircuitFactory(); + Circuit testCircuit = circuitFactory.createCircuit(testProps); + + assertNoFailures(testCircuit.test(1, + assertionList(testCircuit.getPublisher().noExceptionsAssertion(testProps), + testCircuit.getReceiver().noMessagesReceivedAssertion(testProps)))); + } + + /** Check messages received but rolled back are redelivered on subsequent receives. */ + public void testRolledbackMessagesSubsequentlyReceived() + { + // Ensure transactional sessions are on. + testProps.setProperty(TRANSACTED_RECEIVER_PROPNAME, true); + testProps.setProperty(ROLLBACK_RECEIVER_PROPNAME, true); + + // Run the default test sequence over the test circuit checking for no errors. + CircuitFactory circuitFactory = getCircuitFactory(); + Circuit testCircuit = circuitFactory.createCircuit(testProps); + + assertNoFailures(testCircuit.test(1, + assertionList(testCircuit.getPublisher().noExceptionsAssertion(testProps), + testCircuit.getReceiver().allMessagesReceivedAssertion(testProps)))); + } + + /** Attempting to rollback outside of a transaction results in an IllegalStateException. */ + public void testRollbackUnavailableOutsideTransactionPublisher() + { + // Ensure transactional sessions are on. + testProps.setProperty(TRANSACTED_PUBLISHER_PROPNAME, false); + testProps.setProperty(ROLLBACK_PUBLISHER_PROPNAME, true); + + // Run the default test sequence over the test circuit checking for no errors. + CircuitFactory circuitFactory = getCircuitFactory(); + Circuit testCircuit = circuitFactory.createCircuit(testProps); + + assertNoFailures(testCircuit.test(1, assertionList(testCircuit.getPublisher().channelClosedAssertion(testProps)))); + } + + /** Attempting to rollback outside of a transaction results in an IllegalStateException. */ + public void testRollbackUnavailableOutsideTransactionReceiver() + { + // Ensure transactional sessions are on. + testProps.setProperty(TRANSACTED_RECEIVER_PROPNAME, false); + testProps.setProperty(ROLLBACK_RECEIVER_PROPNAME, true); + + // Run the default test sequence over the test circuit checking for no errors. + CircuitFactory circuitFactory = getCircuitFactory(); + Circuit testCircuit = circuitFactory.createCircuit(testProps); + + assertNoFailures(testCircuit.test(1, assertionList(testCircuit.getReceiver().channelClosedAssertion(testProps)))); + } + + /** + * Sets up all tests to have an active outward route and consumer by default. + * + * @throws Exception Any exceptions are allowed to fall through and fail the test. + */ + protected void setUp() throws Exception + { + super.setUp(); + + testProps = TestContextProperties.getInstance(MessagingTestConfigProperties.defaults); + + /** Bind the receivers consumer by default. */ + testProps.setProperty(RECEIVER_CONSUMER_BIND_PROPNAME, true); + testProps.setProperty(RECEIVER_CONSUMER_ACTIVE_PROPNAME, true); + } +}
Added: incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/testcases/TTLTest.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/testcases/TTLTest.java?rev=651133&view=auto ============================================================================== --- incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/testcases/TTLTest.java (added) +++ incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/testcases/TTLTest.java Wed Apr 23 18:54:20 2008 @@ -0,0 +1,151 @@ +/* + * + * 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.testcases; + +import org.apache.qpid.test.framework.Circuit; +import org.apache.qpid.test.framework.FrameworkBaseCase; +import static org.apache.qpid.test.framework.MessagingTestConfigProperties.ACK_MODE_PROPNAME; +import static org.apache.qpid.test.framework.MessagingTestConfigProperties.PUBSUB_PROPNAME; +import org.apache.qpid.test.framework.TestUtils; +import org.apache.qpid.test.framework.localcircuit.LocalCircuitImpl; +import org.apache.qpid.test.framework.sequencers.CircuitFactory; + +import javax.jms.*; + +import java.util.LinkedList; +import java.util.List; +import java.util.Random; + +/** + * TTLTest checks that time-to-live is applied to messages. The test sends messages with a variety of TTL stamps on them + * then after a pause attempts to receive those messages. Only messages with a large enough TTL to have survived the pause + * should be receiveable. This test case also applies an additional assertion against the broker, that the message store + * is empty at the end of the test. + * + * <p/>This test is designed to run over local circuits only, as it must control a timed pause between sending and receiving + * messages to that TTL can be applied to purge some of the messages. + * + * <p/><table id="crc"><caption>CRC Card</caption> + * <tr><th> Responsibilities <th> Collaborations + * <tr><td> + * </table> + * + * @todo Use an interface or other method to mark this test as local only. + * + * @todo Implement the message store assertion for in-vm broker. Could also be done for external broker, for example + * by using diagnostic exchange. + * + * @todo Implement and add a queue depth assertion too. This might already be in another test to copy from. + * + * @todo Create variations on test theme, for different ack mode and tx and message sizes etc. + * + * @todo Add an allowable margin of error to the test, as ttl may not be precise. + */ +public class TTLTest extends FrameworkBaseCase +{ + /** + * Creates a new test case with the specified name. + * + * @param name The test case name. + */ + public TTLTest(String name) + { + super(name); + } + + /** + * Checks that all messages sent with a TTL shorter than a pause between sending them and attempting to receive them + * will fail to arrive. Once all messages have been purged by TTL or received, check that they no longer exist on + * the broker. + * + * @throws javax.jms.JMSException Allowed to fall through and fail test. + */ + public void testTTLP2P() throws JMSException + { + String errorMessages = ""; + Random r = new Random(); + + // Used to accumulate correctly received messages in. + List<Message> receivedMessages = new LinkedList<Message>(); + + // Set up the test properties to match the test case requirements. + testProps.setProperty(ACK_MODE_PROPNAME, Session.AUTO_ACKNOWLEDGE); + testProps.setProperty(PUBSUB_PROPNAME, false); + + // Create the test circuit from the test configuration parameters. + CircuitFactory circuitFactory = getCircuitFactory(); + Circuit testCircuit = circuitFactory.createCircuit(testProps); + + // This test case assumes it is using a local circuit. + LocalCircuitImpl localCircuit = (LocalCircuitImpl) testCircuit; + + Session producerSession = localCircuit.getLocalPublisherCircuitEnd().getSession(); + MessageProducer producer = localCircuit.getLocalPublisherCircuitEnd().getProducer(); + MessageConsumer consumer = localCircuit.getLocalReceiverCircuitEnd().getConsumer(); + + // Send some tests messages, with random TTLs, some shorter and some longer than the pause time. + for (int i = 0; i < 100; i++) + { + Message testMessage = TestUtils.createTestMessageOfSize(producerSession, 10); + + // Set the TTL on the message and record its value in the message headers. + long ttl = 500 + r.nextInt(1500); + producer.setTimeToLive(ttl); + testMessage.setLongProperty("testTTL", ttl); + + producer.send(testMessage); + // producerSession.commit(); + } + + // Inject a pause to allow some messages to be purged by TTL. + TestUtils.pause(1000); + + // Attempt to receive back all of the messages, confirming by the message time stamps and TTLs that only + // those received should have avoided being purged by the TTL. + boolean timedOut = false; + + while (!timedOut) + { + Message testMessage = consumer.receive(1000); + + long ttl = testMessage.getLongProperty("testTTL"); + long timeStamp = testMessage.getJMSTimestamp(); + long now = System.currentTimeMillis(); + + if ((timeStamp + ttl) < now) + { + errorMessages += + "Received message [sent: " + timeStamp + ", ttl: " + ttl + ", received: " + now + + "] which should have been purged by its TTL.\n"; + } + /*else + { + receivedMessages.add(testMessage); + }*/ + } + + // Check that the queue and message store on the broker are empty. + // assertTrue("Message store is not empty.", messageStoreEmpty.apply()); + // assertTrue("Queue is not empty.", queueEmpty.apply()); + + assertTrue(errorMessages, "".equals(errorMessages)); + } +} Added: incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ack/AcknowledgeTest.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ack/AcknowledgeTest.java?rev=651133&view=auto ============================================================================== --- incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ack/AcknowledgeTest.java (added) +++ incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ack/AcknowledgeTest.java Wed Apr 23 18:54:20 2008 @@ -0,0 +1,174 @@ +package org.apache.qpid.test.unit.ack; + +/* + * + * 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. + * + */ + +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.MessageProducer; +import javax.jms.Queue; +import javax.jms.Session; +import javax.jms.TextMessage; + +import org.apache.log4j.Logger; +import org.apache.qpid.client.AMQConnectionFactory; +import org.apache.qpid.client.AMQDestination; +import org.apache.qpid.client.AMQQueue; +import org.apache.qpid.client.AMQSession; +import org.apache.qpid.client.transport.TransportConnection; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.test.VMTestCase; + +public class AcknowledgeTest extends VMTestCase +{ + private static final int NUM_MESSAGES = 50; + private Connection _con; + private Queue _queue; + private MessageProducer _producer; + private Session _producerSession; + private Session _consumerSession; + private MessageConsumer _consumerA; + private MessageConsumer _consumerB; + + @Override + protected void setUp() throws Exception + { + super.setUp(); + _queue = (Queue) _context.lookup("queue"); + + //CreateQueue + ((ConnectionFactory) _context.lookup("connection")).createConnection().createSession(false, Session.AUTO_ACKNOWLEDGE).createConsumer(_queue).close(); + + //Create Producer put some messages on the queue + _con = ((ConnectionFactory) _context.lookup("connection")).createConnection(); + _con.start(); + } + + private void init(boolean transacted, int mode) throws JMSException { + _producerSession = _con.createSession(false, Session.AUTO_ACKNOWLEDGE); + _consumerSession = _con.createSession(transacted, mode); + _producer = _producerSession.createProducer(_queue); + _consumerA = _consumerSession.createConsumer(_queue); + } + + @Override + protected void tearDown() throws Exception + { + super.tearDown(); + try + { + TransportConnection.killAllVMBrokers(); + ApplicationRegistry.removeAll(); + } + catch (Exception e) + { + fail("Unable to clean up"); + } + + } + + private void consumeMessages(int toConsume, MessageConsumer consumer) throws JMSException + { + Message msg; + for (int i = 0; i < toConsume; i++) + { + msg = consumer.receive(1000); + assertNotNull("Message " + i + " was null!", msg); + assertEquals("message " + i, ((TextMessage) msg).getText()); + } + } + + private void sendMessages(int totalMessages) throws JMSException + { + for (int i = 0; i < totalMessages; i++) + { + _producer.send(_producerSession.createTextMessage("message " + i)); + } + } + + private void testMessageAck(boolean transacted, int mode) throws Exception + { + init(transacted, mode); + sendMessages(NUM_MESSAGES/2); + Thread.sleep(1500); + _consumerB = _consumerSession.createConsumer(_queue); + sendMessages(NUM_MESSAGES/2); + int count = 0; + Message msg = _consumerB.receive(1500); + while (msg != null) + { + if (mode == Session.CLIENT_ACKNOWLEDGE) + { + msg.acknowledge(); + } + count++; + msg = _consumerB.receive(1500); + } + if (transacted) + { + _consumerSession.commit(); + } + _consumerA.close(); + _consumerB.close(); + _consumerSession.close(); + assertEquals("Wrong number of messages on queue", NUM_MESSAGES - count, + ((AMQSession) _producerSession).getQueueDepth((AMQDestination) _queue)); + + // Clean up messages that may be left on the queue + _consumerSession = _con.createSession(transacted, mode); + _consumerA = _consumerSession.createConsumer(_queue); + msg = _consumerA.receive(1500); + while (msg != null) + { + if (mode == Session.CLIENT_ACKNOWLEDGE) + { + msg.acknowledge(); + } + msg = _consumerA.receive(1500); + } + _consumerA.close(); + if (transacted) + { + _consumerSession.commit(); + } + _consumerSession.close(); + super.tearDown(); + } + + public void test2ConsumersAutoAck() throws Exception + { + testMessageAck(false, Session.AUTO_ACKNOWLEDGE); + } + + public void test2ConsumersClientAck() throws Exception + { + testMessageAck(true, Session.CLIENT_ACKNOWLEDGE); + } + + public void test2ConsumersTx() throws Exception + { + testMessageAck(true, Session.AUTO_ACKNOWLEDGE); + } + +} Modified: incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/util/ConversationFactory.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/util/ConversationFactory.java?rev=651133&r1=651132&r2=651133&view=diff ============================================================================== --- incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/util/ConversationFactory.java (original) +++ incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/util/ConversationFactory.java Wed Apr 23 18:54:20 2008 @@ -92,7 +92,7 @@ * * <p/><table id="crc"><caption>CRC Card</caption> * <tr><th> Responsibilities <th> Collaborations - * <tr><th> Associate messages to an ongoing conversation using correlation ids. + * <tr><td> Associate messages to an ongoing conversation using correlation ids. * <tr><td> Auto manage sessions for conversations. * <tr><td> Store messages not in a conversation in dead letter box. * </table> Modified: incubator/qpid/trunk/qpid/specs/amqp.0-8.xml URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/specs/amqp.0-8.xml?rev=651133&r1=651132&r2=651133&view=diff ============================================================================== --- incubator/qpid/trunk/qpid/specs/amqp.0-8.xml (original) +++ incubator/qpid/trunk/qpid/specs/amqp.0-8.xml Wed Apr 23 18:54:20 2008 @@ -790,7 +790,7 @@ class of the method. </doc> </field> - <field name="method id" domain="class id"> + <field name="method id" domain="method id"> failing method ID <doc> When the close is provoked by a method exception, this is the @@ -1354,6 +1354,7 @@ </method> <method name="bound" synchronous="1" index="22"> + <chassis name="server" implement="SHOULD"/> <field name="exchange" domain="exchange name"/> <field name = "routing key" type = "shortstr"> Message routing key @@ -1368,6 +1369,7 @@ <method name="bound-ok" synchronous="1" index="23"> <field name="reply code" domain="reply code"/> <field name="reply text" domain="reply text"/> + <chassis name="client" implement="SHOULD"/> </method> </class> Modified: incubator/qpid/trunk/qpid/specs/amqp.0-9.xml URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/specs/amqp.0-9.xml?rev=651133&r1=651132&r2=651133&view=diff ============================================================================== --- incubator/qpid/trunk/qpid/specs/amqp.0-9.xml (original) +++ incubator/qpid/trunk/qpid/specs/amqp.0-9.xml Wed Apr 23 18:54:20 2008 @@ -1659,6 +1659,29 @@ <doc>This method confirms the deletion of an exchange.</doc> <chassis name = "client" implement = "MUST" /> </method> + + <!-- RG : Added Exchange.bound and Exchange.bound-ok --> + <method name="bound" synchronous="1" index="22"> + <chassis name="server" implement="SHOULD"/> + <field name="exchange" domain="exchange-name"/> + <field name = "routing-key" type = "shortstr"> + Message routing key + <doc> + Specifies the routing key for the message. The routing key is + used for routing messages depending on the exchange configuration. + </doc> + </field> + <field name = "queue" domain = "queue name"/> + </method> + + <method name="bound-ok" synchronous="1" index="23"> + <field name="reply-code" domain="reply-code"/> + <field name="reply-text" domain="reply-text"/> + <chassis name="client" implement="SHOULD"/> + </method> + + + </class> <!-- == QUEUE ============================================================ --> @@ -2704,7 +2727,9 @@ </doc> </field> - <field name = "filter" domain = "table" label = "arguments for consuming"> + <!-- RG changed name from filter to arguments on basic.consume: this is inline with qpid0-8 and 0-10 and has no effect on the wire level encoding + <field name = "arguments" domain = "table" label = "arguments for consuming"> --> + <field name = "arguments" domain = "table" label = "arguments for consuming"> <doc> A set of filters for the consume. The syntax and semantics of these filters depends on the providers implementation. @@ -3183,6 +3208,45 @@ </doc> </field> </method> + + + <!-- RG : Added recover-sync and recover-sync-ok to give a synchronous recover without interfering with the correct 0-9 recover method --> + <method name = "recover-sync" index = "102"> + redeliver unacknowledged messages + <doc> + This method asks the broker to redeliver all unacknowledged messages on a + specified channel. Zero or more messages may be redelivered. This method + is only allowed on non-transacted channels. + </doc> + <chassis name = "server" implement = "MUST" /> + + <field name = "requeue" type = "bit"> + requeue the message + <doc> + If this field is zero, the message will be redelivered to the original + recipient. If this bit is 1, the server will attempt to requeue the + message, potentially then delivering it to an alternative subscriber. + </doc> + </field> + <doc name="rule"> + The server MUST set the redelivered flag on all messages that are resent. + </doc> + <doc name="rule"> + The server MUST raise a channel exception if this is called on a + transacted channel. + </doc> + <response name="recover-sync-ok"/> + </method> + <method name="recover-sync-ok" synchronous="1" index="101"> + confirm a successful recover + <doc> + This method confirms to the client that the recover succeeded. + Note that if an recover fails, the server raises a channel exception. + </doc> + <chassis name="client" implement="MUST"/> + </method> + + </class> <!-- == FILE ============================================================= -->
