nandorsoma commented on code in PR #6987:
URL: https://github.com/apache/nifi/pull/6987#discussion_r1151199672
##########
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);
+ session.commitAsync(
+ () -> withLog(() ->
acknowledge(processedMessages, failedMessages)),
+ __ -> withLog(() ->
reject(processedMessages, failedMessages))
+ );
+ session.adjustCounter(COUNTER_RECORDS_RECEIVED,
processedMessages.size() + failedMessages.size(), false);
+ session.adjustCounter(COUNTER_RECORDS_PROCESSED,
processedMessages.size(), false);
+ }
+
+ @Override
+ public void onParseFailure(FlowFile flowFile,
JMSResponse message, Exception e) {
+ final FlowFile failedMessage =
createFlowFileFromMessage(session, destinationName, message);
+ session.transfer(failedMessage, REL_PARSE_FAILURE);
+ session.adjustCounter(COUNTER_PARSE_FAILURES, 1,
false);
+ }
+
+ @Override
+ public void onFailure(FlowFile flowFile,
List<JMSResponse> processedMessages, List<JMSResponse> failedMessages,
Exception e) {
+ reject(processedMessages, failedMessages);
+ // It would be nicer to call rollback and yield
here, but we are rethrowing the exception to have the same error handling with
processSingleMessage.
+ throw new ProcessException(e);
+ }
+ }
+ );
Review Comment:
The topmost layer served multiple purposes. It has not just wrapped the
difference of how provenance events needed to be generated, but it was also
designed to work with demarcated and single FlowFiles. Since we agreed they are
not in scope for now, I removed the extra classes.
--
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]
