pvillard31 closed pull request #3240: NIFI-5921: Added property to allow a user
to define a timeout on the ConsumeJMS …
URL: https://github.com/apache/nifi/pull/3240
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java
b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java
index ff05f6a922..1f8358e23e 100644
---
a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java
+++
b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java
@@ -48,6 +48,7 @@
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
/**
* Consuming JMS processor which upon each invocation of
@@ -128,6 +129,14 @@
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
.build();
+ static final PropertyDescriptor TIMEOUT = new PropertyDescriptor.Builder()
+ .name("Timeout")
+ .description("How long to wait to consume a message from the
remote broker before giving up.")
+ .required(true)
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("0 sec")
+
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .build();
public static final Relationship REL_SUCCESS = new Relationship.Builder()
.name("success")
@@ -154,6 +163,7 @@
_propertyDescriptors.add(DURABLE_SUBSCRIBER);
_propertyDescriptors.add(SHARED_SUBSCRIBER);
_propertyDescriptors.add(SUBSCRIPTION_NAME);
+ _propertyDescriptors.add(TIMEOUT);
thisPropertyDescriptors =
Collections.unmodifiableList(_propertyDescriptors);
Set<Relationship> _relationships = new HashSet<>();
@@ -177,8 +187,9 @@ protected void rendezvousWithJms(final ProcessContext
context, final ProcessSess
final boolean shared = sharedBoolean == null ? false : sharedBoolean;
final String subscriptionName =
context.getProperty(SUBSCRIPTION_NAME).evaluateAttributeExpressions().getValue();
final String charset =
context.getProperty(CHARSET).evaluateAttributeExpressions().getValue();
+ final long timeout =
context.getProperty(TIMEOUT).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS);
- consumer.consume(destinationName, durable, shared, subscriptionName,
charset, new ConsumerCallback() {
+ consumer.consume(destinationName, durable, shared, subscriptionName,
charset, timeout, new ConsumerCallback() {
@Override
public void accept(final JMSResponse response) {
if (response == null) {
diff --git
a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java
b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java
index a2c73b4e3b..e6e5913569 100644
---
a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java
+++
b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java
@@ -80,7 +80,7 @@ private MessageConsumer createMessageConsumer(final Session
session, final Strin
}
- public void consume(final String destinationName, final boolean durable,
final boolean shared, final String subscriberName, final String charset,
+ public void consume(final String destinationName, final boolean durable,
final boolean shared, final String subscriberName, final String charset, final
long timeout,
final ConsumerCallback consumerCallback) {
this.jmsTemplate.execute(new SessionCallback<Void>() {
@Override
@@ -88,7 +88,7 @@ public Void doInJms(final Session session) throws
JMSException {
final MessageConsumer msgConsumer =
createMessageConsumer(session, destinationName, durable, shared,
subscriberName);
try {
- final Message message =
msgConsumer.receive(JMSConsumer.this.jmsTemplate.getReceiveTimeout());
+ final Message message = msgConsumer.receive(timeout);
JMSResponse response = null;
if (message != null) {
diff --git
a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java
b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java
index 7812e7185a..eed9276d09 100644
---
a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java
+++
b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java
@@ -113,7 +113,7 @@ public Message createMessage(Session session) throws
JMSException {
});
JMSConsumer consumer = new JMSConsumer((CachingConnectionFactory)
jmsTemplate.getConnectionFactory(), jmsTemplate, mock(ComponentLog.class));
- consumer.consume(destinationName, false, false, null, "UTF-8", new
ConsumerCallback() {
+ consumer.consume(destinationName, false, false, null, "UTF-8",
1000, new ConsumerCallback() {
@Override
public void accept(JMSResponse response) {
// noop
@@ -143,7 +143,7 @@ public Message createMessage(Session session) throws
JMSException {
JMSConsumer consumer = new JMSConsumer((CachingConnectionFactory)
jmsTemplate.getConnectionFactory(), jmsTemplate, mock(ComponentLog.class));
final AtomicBoolean callbackInvoked = new AtomicBoolean();
- consumer.consume(destinationName, false, false, null, "UTF-8", new
ConsumerCallback() {
+ consumer.consume(destinationName, false, false, null, "UTF-8",
1000, new ConsumerCallback() {
@Override
public void accept(JMSResponse response) {
callbackInvoked.set(true);
@@ -190,7 +190,7 @@ public void accept(JMSResponse response) {
JMSConsumer consumer = new
JMSConsumer((CachingConnectionFactory) consumeTemplate.getConnectionFactory(),
consumeTemplate, mock(ComponentLog.class));
for (int j = 0; j < 1000 && msgCount.get() < 4000;
j++) {
- consumer.consume(destinationName, false, false,
null, "UTF-8", callback);
+ consumer.consume(destinationName, false, false,
null, "UTF-8", 1000, callback);
}
} finally {
((CachingConnectionFactory)
consumeTemplate.getConnectionFactory()).destroy();
@@ -229,7 +229,7 @@ public void validateMessageRedeliveryWhenNotAcked() throws
Exception {
JMSConsumer consumer = new JMSConsumer((CachingConnectionFactory)
jmsTemplate.getConnectionFactory(), jmsTemplate, mock(ComponentLog.class));
final AtomicBoolean callbackInvoked = new AtomicBoolean();
try {
- consumer.consume(destinationName, false, false, null, "UTF-8",
new ConsumerCallback() {
+ consumer.consume(destinationName, false, false, null, "UTF-8",
1000, new ConsumerCallback() {
@Override
public void accept(JMSResponse response) {
callbackInvoked.set(true);
@@ -246,7 +246,7 @@ public void accept(JMSResponse response) {
// should receive the same message, but will process it
successfully
while (!callbackInvoked.get()) {
- consumer.consume(destinationName, false, false, null, "UTF-8",
new ConsumerCallback() {
+ consumer.consume(destinationName, false, false, null, "UTF-8",
1000, new ConsumerCallback() {
@Override
public void accept(JMSResponse response) {
if (response == null) {
@@ -265,7 +265,7 @@ public void accept(JMSResponse response) {
// receiving next message and fail again
try {
while (!callbackInvoked.get()) {
- consumer.consume(destinationName, false, false, null,
"UTF-8", new ConsumerCallback() {
+ consumer.consume(destinationName, false, false, null,
"UTF-8", 1000, new ConsumerCallback() {
@Override
public void accept(JMSResponse response) {
if (response == null) {
@@ -287,7 +287,7 @@ public void accept(JMSResponse response) {
// should receive the same message, but will process it
successfully
try {
while (!callbackInvoked.get()) {
- consumer.consume(destinationName, false, false, null,
"UTF-8", new ConsumerCallback() {
+ consumer.consume(destinationName, false, false, null,
"UTF-8", 1000, new ConsumerCallback() {
@Override
public void accept(JMSResponse response) {
if (response == null) {
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
