nandorsoma commented on code in PR #6987:
URL: https://github.com/apache/nifi/pull/6987#discussion_r1150586980
##########
nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java:
##########
@@ -305,6 +326,97 @@ public void accept(final JMSResponse response) {
}
}
+ private void processSingleMessage(ProcessSession processSession,
JMSConsumer consumer, String destinationName, String errorQueueName,
+ boolean durable, boolean shared, String
subscriptionName, String messageSelector, String charset) {
+
+ consumer.consumeSingleMessage(destinationName, errorQueueName,
durable, shared, subscriptionName, messageSelector, charset, response -> {
+ if (response == null) {
+ return;
+ }
+
+ try {
+ final FlowFile flowFile =
createFlowFileFromMessage(processSession, destinationName, response);
+
+ processSession.getProvenanceReporter().receive(flowFile,
destinationName);
+ processSession.transfer(flowFile, REL_SUCCESS);
+ processSession.commitAsync(
+ () -> withLog(() -> acknowledge(response)),
+ __ -> withLog(() -> response.reject()));
+ } catch (final Throwable t) {
+ response.reject();
+ throw t;
+ }
+ });
+ }
+
+ private FlowFile createFlowFileFromMessage(ProcessSession processSession,
String destinationName, JMSResponse response) {
+ FlowFile flowFile = processSession.create();
+ flowFile = processSession.write(flowFile, out ->
out.write(response.getMessageBody()));
+
+ final Map<String, String> jmsHeaders = response.getMessageHeaders();
+ final Map<String, String> jmsProperties =
response.getMessageProperties();
+
+ flowFile = updateFlowFileAttributesWithJMSAttributes(jmsHeaders,
flowFile, processSession);
+ flowFile = updateFlowFileAttributesWithJMSAttributes(jmsProperties,
flowFile, processSession);
+ flowFile = processSession.putAttribute(flowFile,
JMS_SOURCE_DESTINATION_NAME, destinationName);
+
+ flowFile = processSession.putAttribute(flowFile, JMS_MESSAGETYPE,
response.getMessageType());
+
+ return flowFile;
+ }
+
+ private void processMessageSet(ProcessContext context, ProcessSession
session, JMSConsumer consumer, String destinationName,String errorQueueName,
+ boolean durable, boolean shared, String
subscriptionName, String messageSelector, String charset) {
+
+ final RecordReaderFactory readerFactory =
context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
+ final RecordSetWriterFactory writerFactory =
context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class);
+ final OutputStrategy outputStrategy =
OutputStrategy.valueOf(context.getProperty(OUTPUT_STRATEGY).getValue());
+
+ consumer.consumeMessageSet(destinationName, errorQueueName, durable,
shared, subscriptionName, messageSelector, charset, jmsResponses -> {
Review Comment:
Yes, my original idea was to move the fully parameterized MessageConsumer to
the onScheduled hook. With the reworked approach it is no longer possible. I
don't think it make sense to move it anymore.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
