You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2024/06/17 18:04:16 UTC
(nifi) branch main updated: NIFI-13389 Streamlined use of putAllAttributes() in ConsumeJMS
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 <jr...@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 <ex...@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) -> {