This commit is contained in:
Clebert Suconic 2020-06-09 15:18:45 -04:00
commit 0a2c4399bc
2 changed files with 57 additions and 1 deletions

View File

@ -958,6 +958,7 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
@Override @Override
public final long getExpiration() { public final long getExpiration() {
ensureMessageDataScanned();
return expiration; return expiration;
} }

View File

@ -16,8 +16,15 @@
*/ */
package org.apache.activemq.artemis.tests.integration.client; package org.apache.activemq.artemis.tests.integration.client;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import org.apache.activemq.artemis.api.core.Message; import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.QueueConfiguration; import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString; import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.ClientConsumer; import org.apache.activemq.artemis.api.core.client.ClientConsumer;
import org.apache.activemq.artemis.api.core.client.ClientMessage; import org.apache.activemq.artemis.api.core.client.ClientMessage;
@ -29,7 +36,9 @@ import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.Queue; import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings; import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.RandomUtil; import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.qpid.jms.JmsConnectionFactory;
import org.junit.Assert; import org.junit.Assert;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
@ -70,6 +79,50 @@ public class MessageExpirationTest extends ActiveMQTestBase {
session.deleteQueue(queue); session.deleteQueue(queue);
} }
@Test
public void testAmqpJmsReloaded() throws Exception {
SimpleString queue = RandomUtil.randomSimpleString();
SimpleString expiry = RandomUtil.randomSimpleString();
server.createQueue(new QueueConfiguration(queue).setRoutingType(RoutingType.ANYCAST));
server.createQueue(new QueueConfiguration(expiry));
server.getAddressSettingsRepository().addMatch(queue.toString(), new AddressSettings().setExpiryAddress(expiry));
ConnectionFactory cf = new JmsConnectionFactory("amqp://localhost:61616");
Connection connection = cf.createConnection();
Session session = connection.createSession();
MessageProducer producer = session.createProducer(session.createQueue(queue.toString()));
producer.setTimeToLive(EXPIRATION);
for (int i = 0; i < 20; i++) {
javax.jms.Message message = session.createMessage();
producer.send(message);
}
connection.close();
Wait.assertEquals(20L, () -> server.locateQueue(queue).getMessageCount(), 2000, 100);
Wait.assertEquals(0L, () -> server.locateQueue(expiry).getMessageCount(), 2000, 100);
server.stop();
server.start();
Thread.sleep(EXPIRATION * 2);
Wait.assertEquals(0L, () -> server.locateQueue(queue).getMessageCount(), 2000, 100);
Wait.assertEquals(20L, () -> server.locateQueue(expiry).getMessageCount(), 2000, 100);
connection = cf.createConnection();
session = connection.createSession();
MessageConsumer consumer = session.createConsumer(session.createQueue(queue.toString()));
connection.start();
for (int i = 0; i < 20; i++) {
javax.jms.Message message2 = consumer.receiveNoWait();
Assert.assertNull(message2);
}
consumer.close();
}
@Test @Test
public void testMessageExpiredWithoutExpiryAddressWithExpiryDelayOverride() throws Exception { public void testMessageExpiredWithoutExpiryAddressWithExpiryDelayOverride() throws Exception {
SimpleString address = RandomUtil.randomSimpleString(); SimpleString address = RandomUtil.randomSimpleString();
@ -235,7 +288,9 @@ public class MessageExpirationTest extends ActiveMQTestBase {
public void setUp() throws Exception { public void setUp() throws Exception {
super.setUp(); super.setUp();
server = createServer(false); server = createServer(true);
server.getConfiguration().addAcceptorConfiguration("amqp", "tcp://127.0.0.1:61616");
server.getConfiguration().setMessageExpiryScanPeriod(200);
server.start(); server.start();
locator = createInVMNonHALocator(); locator = createInVMNonHALocator();
sf = createSessionFactory(locator); sf = createSessionFactory(locator);