This is an automated email from the ASF dual-hosted git repository. exceptionfactory pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push: new bd1ad8f9f4 NIFI-13389 Streamlined use of putAllAttributes() in ConsumeJMS bd1ad8f9f4 is described below commit bd1ad8f9f4c94ff9c93aa14b20ed60943662dc9a Author: Jim Steinebrey <jrsteineb...@gmail.com> AuthorDate: Tue Jun 11 14:28:13 2024 -0400 NIFI-13389 Streamlined use of putAllAttributes() in ConsumeJMS This closes #8955 Signed-off-by: David Handermann <exceptionfact...@apache.org> --- .../org/apache/nifi/jms/processors/ConsumeJMS.java | 27 ++++------------------ 1 file changed, 4 insertions(+), 23 deletions(-) diff --git a/nifi-extension-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java b/nifi-extension-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java index dfb18216c0..87444869bc 100644 --- a/nifi-extension-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java +++ b/nifi-extension-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java @@ -61,7 +61,6 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Map.Entry; import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.stream.Stream; @@ -364,12 +363,10 @@ public class ConsumeJMS extends AbstractJMSProcessor<JMSConsumer> { final Map<String, String> jmsHeaders = response.getMessageHeaders(); final Map<String, String> jmsProperties = response.getMessageProperties(); - - flowFile = updateFlowFileAttributesWithJMSAttributes(mergeJmsAttributes(jmsHeaders, jmsProperties), flowFile, processSession); - flowFile = processSession.putAttribute(flowFile, JMS_SOURCE_DESTINATION_NAME, destinationName); - flowFile = processSession.putAttribute(flowFile, JMS_MESSAGETYPE, response.getMessageType()); - - return flowFile; + Map<String, String> attributes = mergeJmsAttributes(jmsHeaders, jmsProperties); + attributes.put(JMS_SOURCE_DESTINATION_NAME, destinationName); + attributes.put(JMS_MESSAGETYPE, response.getMessageType()); + return processSession.putAllAttributes(flowFile, attributes); } private void processMessageSet(ProcessContext context, ProcessSession session, JMSConsumer consumer, String destinationName, String errorQueueName, @@ -496,22 +493,6 @@ public class ConsumeJMS extends AbstractJMSProcessor<JMSConsumer> { } } - /** - * Copies JMS attributes (i.e., headers and properties) as FF attributes. - * Given that FF attributes mandate that values are of type String, the - * copied values of JMS attributes will be "stringified" via - * String.valueOf(attribute). - */ - private FlowFile updateFlowFileAttributesWithJMSAttributes(Map<String, String> jmsAttributes, FlowFile flowFile, ProcessSession processSession) { - Map<String, String> attributes = new HashMap<>(); - for (Entry<String, String> entry : jmsAttributes.entrySet()) { - attributes.put(entry.getKey(), entry.getValue()); - } - - flowFile = processSession.putAllAttributes(flowFile, attributes); - return flowFile; - } - private Map<String, String> mergeJmsAttributes(Map<String, String> headers, Map<String, String> properties) { final Map<String, String> jmsAttributes = new HashMap<>(headers); properties.forEach((key, value) -> {