nandorsoma commented on code in PR #6987:
URL: https://github.com/apache/nifi/pull/6987#discussion_r1150595064
##########
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 -> {
+ final MessageConsumer<JMSResponse> messageConsumer = new
MessageConsumer.Builder<JMSResponse>()
+ .withFlowFileWriter((new
RecordWriter.Builder<JMSResponse>()
+ .withReaderFactory(readerFactory)
+ .withWriterFactory(writerFactory)
+ .withSerializer(message ->
message.getMessageBody() == null ? new byte[0] : message.getMessageBody())
+ .withOutputStrategy(outputStrategy)
+ .withLogger(getLogger()).build()))
+ .withAttributeSupplier(message ->
mergeJmsAttributes(message.getMessageHeaders(), message.getMessageProperties()))
+
.withEventReporter(StandardRecordReceivedEventReporter.of(destinationName))
+ .build();
+
+ messageConsumer.consumeMessages(
+ session,
+ jmsResponses,
+ new MessageConsumerCallback<>() {
+ @Override
+ public void onSuccess(FlowFile flowFile,
List<JMSResponse> processedMessages, List<JMSResponse> failedMessages) {
+ session.transfer(flowFile, REL_SUCCESS);
Review Comment:
Before these methods are called, RecordWriter does a more detailed logging
that I can do here. Any logging here would be a duplication.
--
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]
