diff options
author | Robert Godfrey <rgodfrey@apache.org> | 2014-08-25 15:12:10 +0000 |
---|---|---|
committer | Robert Godfrey <rgodfrey@apache.org> | 2014-08-25 15:12:10 +0000 |
commit | 78a00e2a3a1bbc7486de0fad72603617958062c3 (patch) | |
tree | 75526f90e858d5cc3f8d228e4e1c3c42c7808a10 /qpid/java/systests/src/test/java/org/apache/qpid/test/unit/ack/ClientAcknowledgeTest.java | |
parent | 0ed130958f49a4d4cc5307c6d6f00d04da5f834d (diff) | |
download | qpid-python-78a00e2a3a1bbc7486de0fad72603617958062c3.tar.gz |
Merging from trunk r1617322:1617736 in the Java tree
git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/0.30@1620337 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/java/systests/src/test/java/org/apache/qpid/test/unit/ack/ClientAcknowledgeTest.java')
-rw-r--r-- | qpid/java/systests/src/test/java/org/apache/qpid/test/unit/ack/ClientAcknowledgeTest.java | 82 |
1 files changed, 82 insertions, 0 deletions
diff --git a/qpid/java/systests/src/test/java/org/apache/qpid/test/unit/ack/ClientAcknowledgeTest.java b/qpid/java/systests/src/test/java/org/apache/qpid/test/unit/ack/ClientAcknowledgeTest.java new file mode 100644 index 0000000000..291e1697ca --- /dev/null +++ b/qpid/java/systests/src/test/java/org/apache/qpid/test/unit/ack/ClientAcknowledgeTest.java @@ -0,0 +1,82 @@ +/* + * 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.ack; + +import org.apache.qpid.test.utils.QpidBrokerTestCase; + +import javax.jms.Connection; +import javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.MessageProducer; +import javax.jms.Queue; +import javax.jms.Session; + +public class ClientAcknowledgeTest extends QpidBrokerTestCase +{ + private static final long ONE_DAY_MS = 1000l * 60 * 60 * 24; + private Connection _connection; + private Queue _queue; + private Session _consumerSession; + private MessageConsumer _consumer; + private MessageProducer _producer; + + @Override + protected void setUp() throws Exception + { + super.setUp(); + _queue = getTestQueue(); + _connection = getConnection(); + } + + /** + * Test that message.acknowledge actually acknowledges, regardless of + * the flusher thread period, by restarting the broker after calling + * acknowledge, and then verifying after restart that the message acked + * is no longer present. This test requires a persistent store. + */ + public void testClientAckWithLargeFlusherPeriod() throws Exception + { + setTestClientSystemProperty("qpid.session.max_ack_delay", Long.toString(ONE_DAY_MS)); + _consumerSession = _connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); + _consumer = _consumerSession.createConsumer(_queue); + _connection.start(); + + _producer = _consumerSession.createProducer(_queue); + _producer.send(createNextMessage(_consumerSession, 1)); + _producer.send(createNextMessage(_consumerSession, 2)); + + Message message = _consumer.receive(1000l); + assertNotNull("Message has not been received", message); + assertEquals("Unexpected message is received", 1, message.getIntProperty(INDEX)); + message.acknowledge(); + + //restart broker to allow verification of the acks + //without explicitly closing connection (which acks) + restartBroker(); + + // try to receive the message again, which should fail (as it was ackd) + _connection = getConnection(); + _connection.start(); + _consumerSession = _connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); + _consumer = _consumerSession.createConsumer(_queue); + message = _consumer.receive(1000l); + assertNotNull("Message has not been received", message); + assertEquals("Unexpected message is received", 2, message.getIntProperty(INDEX)); + } +} |