This is an automated email from the ASF dual-hosted git repository. joewitt pushed a commit to branch support/nifi-1.9.x in repository https://gitbox.apache.org/repos/asf/nifi.git
commit e873d68eeb2b30a944a2b5959aeb3baa2adf260d Author: Mark Bean <[email protected]> AuthorDate: Fri Oct 5 14:08:28 2018 +0000 NIFI-5660: JMSPublisher should set some header properties in JmsTemplate instead of directly in the message NIFI-5660: Added lines to integration test to verify these header properties (added by Mike Moser) Signed-off-by: Mike Moser <[email protected]> This closes #3053 --- .../org/apache/nifi/jms/processors/AbstractJMSProcessor.java | 5 +++++ .../main/java/org/apache/nifi/jms/processors/JMSPublisher.java | 9 ++++++--- .../org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java | 4 ++++ 3 files changed, 15 insertions(+), 3 deletions(-) diff --git a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/AbstractJMSProcessor.java b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/AbstractJMSProcessor.java index f47cf78..e51238d 100644 --- a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/AbstractJMSProcessor.java +++ b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/AbstractJMSProcessor.java @@ -33,6 +33,7 @@ import org.springframework.jms.connection.UserCredentialsConnectionFactoryAdapte import org.springframework.jms.core.JmsTemplate; import javax.jms.ConnectionFactory; +import javax.jms.Message; import java.nio.charset.Charset; import java.util.ArrayList; import java.util.List; @@ -177,6 +178,10 @@ abstract class AbstractJMSProcessor<T extends JMSWorker> extends AbstractProcess } } if (worker != null) { + worker.jmsTemplate.setExplicitQosEnabled(false); + worker.jmsTemplate.setDeliveryMode(Message.DEFAULT_DELIVERY_MODE); + worker.jmsTemplate.setTimeToLive(Message.DEFAULT_TIME_TO_LIVE); + worker.jmsTemplate.setPriority(Message.DEFAULT_PRIORITY); workerPool.offer(worker); } } diff --git a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSPublisher.java b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSPublisher.java index c13f4b7..506de49 100644 --- a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSPublisher.java +++ b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSPublisher.java @@ -84,11 +84,14 @@ final class JMSPublisher extends JMSWorker { for (Entry<String, String> entry : flowFileAttributesToSend.entrySet()) { try { if (entry.getKey().equals(JmsHeaders.DELIVERY_MODE)) { - message.setJMSDeliveryMode(Integer.parseInt(entry.getValue())); + this.jmsTemplate.setDeliveryMode(Integer.parseInt(entry.getValue())); + this.jmsTemplate.setExplicitQosEnabled(true); } else if (entry.getKey().equals(JmsHeaders.EXPIRATION)) { - message.setJMSExpiration(Integer.parseInt(entry.getValue())); + this.jmsTemplate.setTimeToLive(Integer.parseInt(entry.getValue())); + this.jmsTemplate.setExplicitQosEnabled(true); } else if (entry.getKey().equals(JmsHeaders.PRIORITY)) { - message.setJMSPriority(Integer.parseInt(entry.getValue())); + this.jmsTemplate.setPriority(Integer.parseInt(entry.getValue())); + this.jmsTemplate.setExplicitQosEnabled(true); } else if (entry.getKey().equals(JmsHeaders.REDELIVERED)) { message.setJMSRedelivered(Boolean.parseBoolean(entry.getValue())); } else if (entry.getKey().equals(JmsHeaders.TIMESTAMP)) { diff --git a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java index 7812e71..12474ec 100644 --- a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java +++ b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java @@ -77,6 +77,8 @@ public class JMSPublisherConsumerIT { flowFileAttributes.put("illegal-property", "value"); flowFileAttributes.put("another.illegal", "value"); flowFileAttributes.put(JmsHeaders.REPLY_TO, "myTopic"); + flowFileAttributes.put(JmsHeaders.DELIVERY_MODE, "1"); + flowFileAttributes.put(JmsHeaders.PRIORITY, "1"); flowFileAttributes.put(JmsHeaders.EXPIRATION, "never"); // value expected to be integer, make sure non-integer doesn't cause problems publisher.publish(destinationName, "hellomq".getBytes(), flowFileAttributes); @@ -86,6 +88,8 @@ public class JMSPublisherConsumerIT { assertFalse(receivedMessage.propertyExists("illegal-property")); assertFalse(receivedMessage.propertyExists("another.illegal")); assertTrue(receivedMessage.getJMSReplyTo() instanceof Topic); + assertEquals(1, receivedMessage.getJMSDeliveryMode()); + assertEquals(1, receivedMessage.getJMSPriority()); assertEquals("myTopic", ((Topic) receivedMessage.getJMSReplyTo()).getTopicName()); } finally {
