You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ro...@apache.org on 2019/04/26 12:07:41 UTC
[activemq] branch activemq-5.15.x updated: AMQ-7189: ensure
messages with defauled durable field are internally classed non-persistent
by the 'native' transformer
This is an automated email from the ASF dual-hosted git repository.
robbie pushed a commit to branch activemq-5.15.x
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/activemq-5.15.x by this push:
new e09610b AMQ-7189: ensure messages with defauled durable field are internally classed non-persistent by the 'native' transformer
e09610b is described below
commit e09610b16ceeeefb61f5662a570d04b51617b7c2
Author: Robbie Gemmell <ro...@apache.org>
AuthorDate: Fri Apr 26 12:55:14 2019 +0100
AMQ-7189: ensure messages with defauled durable field are internally classed non-persistent by the 'native' transformer
(cherry picked from commit aff9413cdbe88041552369787d1ba001c2b4ffd4)
---
.../transport/amqp/message/InboundTransformer.java | 2 +
.../amqp/interop/AmqpSendReceiveNativeTest.java | 25 +++++++++++
.../amqp/interop/AmqpSendReceiveTest.java | 52 +++++++++++++++++++---
3 files changed, 74 insertions(+), 5 deletions(-)
diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/message/InboundTransformer.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/message/InboundTransformer.java
index 5a68c77..9b74da1 100644
--- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/message/InboundTransformer.java
+++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/message/InboundTransformer.java
@@ -91,6 +91,8 @@ public abstract class InboundTransformer {
if (header.getDurable() != null) {
jms.setPersistent(header.getDurable().booleanValue());
+ } else {
+ jms.setPersistent(false);
}
if (header.getPriority() != null) {
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSendReceiveNativeTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSendReceiveNativeTest.java
new file mode 100644
index 0000000..db75d10
--- /dev/null
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSendReceiveNativeTest.java
@@ -0,0 +1,25 @@
+/*
+ * 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.activemq.transport.amqp.interop;
+
+public class AmqpSendReceiveNativeTest extends AmqpSendReceiveTest {
+
+ @Override
+ protected String getAmqpTransformer() {
+ return "native";
+ }
+}
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSendReceiveTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSendReceiveTest.java
index cb60e54..a86edbb 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSendReceiveTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSendReceiveTest.java
@@ -31,6 +31,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import javax.jms.DeliveryMode;
import javax.jms.Queue;
import javax.jms.Topic;
@@ -46,6 +47,10 @@ import org.apache.activemq.transport.amqp.client.AmqpReceiver;
import org.apache.activemq.transport.amqp.client.AmqpSender;
import org.apache.activemq.transport.amqp.client.AmqpSession;
import org.apache.activemq.util.Wait;
+import org.apache.qpid.proton.amqp.UnsignedByte;
+import org.apache.qpid.proton.amqp.messaging.AmqpValue;
+import org.apache.qpid.proton.amqp.messaging.Header;
+import org.apache.qpid.proton.message.Message;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
@@ -496,6 +501,20 @@ public class AmqpSendReceiveTest extends AmqpClientTestSupport {
@Test(timeout = 60000)
public void testMessageWithNoHeaderNotMarkedDurable() throws Exception {
+ doMessageNotMarkedDurableTestImpl(false, false);
+ }
+
+ @Test(timeout = 60000)
+ public void testMessageWithHeaderAndDefaultedNonDurableNotMarkedDurable() throws Exception {
+ doMessageNotMarkedDurableTestImpl(true, false);
+ }
+
+ @Test(timeout = 60000)
+ public void testMessageWithHeaderAndMarkedNonDurableNotMarkedDurable() throws Exception {
+ doMessageNotMarkedDurableTestImpl(true, true);
+ }
+
+ private void doMessageNotMarkedDurableTestImpl(boolean sendHeaderWithPriority, boolean explicitSetNonDurable) throws Exception {
AmqpClient client = createAmqpClient();
AmqpConnection connection = trackConnection(client.connect());
AmqpSession session = connection.createSession();
@@ -503,16 +522,39 @@ public class AmqpSendReceiveTest extends AmqpClientTestSupport {
AmqpSender sender = session.createSender("queue://" + getTestName());
AmqpReceiver receiver1 = session.createReceiver("queue://" + getTestName());
- // Create default message that should be sent as non-durable
- AmqpMessage message1 = new AmqpMessage();
- message1.setText("Test-Message -> non-durable");
- message1.setMessageId("ID:Message:1");
+ Message protonMessage = Message.Factory.create();
+ protonMessage.setMessageId("ID:Message:1");
+ protonMessage.setBody(new AmqpValue("Test-Message -> non-durable"));
+ if(sendHeaderWithPriority) {
+ Header header = new Header();
+ if(explicitSetNonDurable) {
+ header.setDurable(false);
+ }
+ header.setPriority(UnsignedByte.valueOf((byte) 5));
+
+ protonMessage.setHeader(header);
+ } else {
+ assertNull("Should not have a header", protonMessage.getHeader());
+ }
+
+ AmqpMessage message1 = new AmqpMessage(protonMessage);
+
sender.send(message1);
+ final QueueViewMBean queueView = getProxyToQueue(getTestName());
+ assertNotNull(queueView);
+
+ assertEquals(1, queueView.getQueueSize());
+
+ List<javax.jms.Message> messages = (List<javax.jms.Message>) queueView.browseMessages();
+ assertEquals(1, messages.size());
+ javax.jms.Message queueMessage = messages.get(0);
+ assertEquals("Queued message should not be persistent", DeliveryMode.NON_PERSISTENT, queueMessage.getJMSDeliveryMode());
+
receiver1.flow(1);
AmqpMessage message2 = receiver1.receive(50, TimeUnit.SECONDS);
assertNotNull("Should have read a message", message2);
- assertFalse("Second message sent should not be durable", message2.isDurable());
+ assertFalse("Received message should not be durable", message2.isDurable());
message2.accept();
sender.close();