You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by tk...@apache.org on 2016/01/04 06:26:12 UTC
nifi git commit: NIFI-1300 Corrects inconsistent error behavior in
PutJMS and added PutJMS unit testing
Repository: nifi
Updated Branches:
refs/heads/master 34bd2061f -> de2dd93f2
NIFI-1300 Corrects inconsistent error behavior in PutJMS and added PutJMS unit testing
Reviewed by Tony Kurc (tkurc@apache.org) This closes #145
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/de2dd93f
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/de2dd93f
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/de2dd93f
Branch: refs/heads/master
Commit: de2dd93f26dd34271632451621d5521fae11c979
Parents: 34bd206
Author: Joe Skora <js...@gmail.com>
Authored: Thu Dec 17 12:27:56 2015 -0500
Committer: Tony Kurc <tr...@gmail.com>
Committed: Mon Jan 4 00:26:21 2016 -0500
----------------------------------------------------------------------
.../apache/nifi/processors/standard/PutJMS.java | 5 +-
.../nifi/processors/standard/TestPutJMS.java | 565 +++++++++++++++++++
2 files changed, 568 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/de2dd93f/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutJMS.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutJMS.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutJMS.java
index b8902a9..7405ca5 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutJMS.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutJMS.java
@@ -251,8 +251,9 @@ public class PutJMS extends AbstractProcessor {
final String flowFileDescription = successfulFlowFiles.size() > 10 ? successfulFlowFiles.size() + " FlowFiles" : successfulFlowFiles.toString();
logger.info("Sent {} to JMS Server and transferred to 'success'", new Object[]{flowFileDescription});
} catch (JMSException e) {
- logger.error("Failed to commit JMS Session due to {}; rolling back session", new Object[]{e});
- session.rollback();
+ logger.error("Failed to commit JMS Session due to {} and transferred to 'failure'", new Object[]{e});
+ session.transfer(flowFiles, REL_FAILURE);
+ context.yield();
wrappedProducer.close(logger);
}
} finally {
http://git-wip-us.apache.org/repos/asf/nifi/blob/de2dd93f/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutJMS.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutJMS.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutJMS.java
new file mode 100644
index 0000000..acba96c
--- /dev/null
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutJMS.java
@@ -0,0 +1,565 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.standard;
+
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processors.standard.util.JmsFactory;
+import org.apache.nifi.processors.standard.util.JmsProperties;
+import org.apache.nifi.processors.standard.util.WrappedMessageProducer;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+
+public class TestPutJMS {
+
+ private final String TEST_PROVIDER = JmsProperties.ACTIVEMQ_PROVIDER;
+ private final String TEST_URL = "vm://localhost?broker.persistent=false";
+ private final String TEST_DEST_TYPE = JmsProperties.DESTINATION_TYPE_QUEUE;
+ private final String TEST_DEST_NAME = "queue.testing";
+ private final String TEST_ACK_MODE = JmsProperties.ACK_MODE_AUTO;
+
+ private String testQueueSuffix() {
+ final StackTraceElement[] trace = Thread.currentThread().getStackTrace();
+ return "." + trace[2].getMethodName();
+ }
+
+ private void injectFieldValue(Class klass, Object instance, String fieldName, Object fieldValue) throws NoSuchFieldException, IllegalAccessException {
+ Field field = klass.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(instance, fieldValue);
+ }
+
+ @Test
+ public void testGetRelationships() {
+ final PutJMS putJMS = new PutJMS();
+ final Set<Relationship> relationships = putJMS.getRelationships();
+ assertEquals(2, relationships.size());
+ assertTrue(relationships.contains(PutJMS.REL_FAILURE));
+ assertTrue(relationships.contains(PutJMS.REL_SUCCESS));
+ }
+
+ @Test
+ public void testCleanupResources() throws JMSException, NoSuchFieldException, IllegalAccessException {
+ final PutJMS putJMS = new PutJMS();
+ final TestRunner runnerPut = TestRunners.newTestRunner(putJMS);
+ runnerPut.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPut.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPut.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPut.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+
+ final Queue<WrappedMessageProducer> wrappedMessageProducerQueue = (Queue) spy(new LinkedBlockingQueue<>());
+ injectFieldValue(PutJMS.class, putJMS, "producerQueue", wrappedMessageProducerQueue);
+
+ final WrappedMessageProducer wrappedProducer = JmsFactory.createMessageProducer(runnerPut.getProcessContext(), true);
+ wrappedMessageProducerQueue.offer(wrappedProducer);
+
+ assertNotNull(wrappedMessageProducerQueue.peek());
+ putJMS.cleanupResources();
+ assertNull(wrappedMessageProducerQueue.peek());
+ }
+
+
+ @Test
+ public void testCreateMessageDirectly() throws JMSException {
+ final PutJMS putJMS = new PutJMS();
+ final TestRunner runnerPut = TestRunners.newTestRunner(putJMS);
+ runnerPut.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPut.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPut.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPut.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerPut.setProperty(JmsProperties.ACKNOWLEDGEMENT_MODE, TEST_ACK_MODE);
+
+ final WrappedMessageProducer wrappedProducer = JmsFactory.createMessageProducer(runnerPut.getProcessContext(), true);
+ final Session jmsSession = wrappedProducer.getSession();
+ final MessageProducer producer = wrappedProducer.getProducer();
+ final Message message = jmsSession.createTextMessage("createMessageDirectly");
+
+ producer.send(message);
+ jmsSession.commit();
+
+ final GetJMSQueue getJmsQueue = new GetJMSQueue();
+ final TestRunner runnerGet = TestRunners.newTestRunner(getJmsQueue);
+
+ runnerGet.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerGet.setProperty(JmsProperties.URL, TEST_URL);
+ runnerGet.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerGet.setProperty(JmsProperties.ACKNOWLEDGEMENT_MODE, TEST_ACK_MODE);
+
+ runnerGet.run();
+
+ final List<MockFlowFile> flowFiles = runnerGet.getFlowFilesForRelationship(
+ new Relationship.Builder().name("success").build());
+
+ assertEquals(1, flowFiles.size());
+ final MockFlowFile successFlowFile = flowFiles.get(0);
+ successFlowFile.assertContentEquals("createMessageDirectly");
+ successFlowFile.assertAttributeEquals("jms.JMSDestination", TEST_DEST_NAME + testQueueSuffix());
+ producer.close();
+ jmsSession.close();
+ }
+
+ @Test
+ public void testPutGetAttributesAndProps() throws JMSException {
+ final PutJMS putJMS = spy(new PutJMS());
+ final TestRunner runnerPut = TestRunners.newTestRunner(putJMS);
+ runnerPut.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPut.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPut.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPut.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerPut.setProperty(JmsProperties.REPLY_TO_QUEUE, TEST_DEST_NAME + testQueueSuffix() + ".reply");
+ runnerPut.setProperty(JmsProperties.ATTRIBUTES_TO_JMS_PROPS, "true");
+
+ runnerPut.run();
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("filename", "file1.txt");
+ attributes.put("jms.string", "banana");
+ attributes.put("jms.integer", "50");
+ attributes.put("jms.integer.type", "integer");
+ attributes.put("jms.float", "3.14159");
+ attributes.put("jms.float.type", "float");
+ attributes.put("jms.boolean", "true");
+ attributes.put("jms.boolean.type", "boolean");
+ attributes.put("jms.long", "123456789");
+ attributes.put("jms.long.type", "long");
+ attributes.put("jms.short", "16384");
+ attributes.put("jms.short.type", "short");
+ attributes.put("jms.byte", "127");
+ attributes.put("jms.byte.type", "byte");
+ attributes.put("jms.double", "3.1415626547");
+ attributes.put("jms.double.type", "double");
+ attributes.put("jms.object", "{\"id\":215, \"name\": \"john doe\"}");
+ attributes.put("jms.object.type", "object");
+ attributes.put("jms.eyes", "blue");
+ attributes.put("jms.eyes.type", "color");
+ attributes.put("jms.badinteger", "3.14");
+ attributes.put("jms.badinteger.type", "integer");
+ runnerPut.enqueue("putGetMessage".getBytes(), attributes);
+
+ runnerPut.run();
+
+ assertEquals(0, runnerPut.getFlowFilesForRelationship(PutJMS.REL_FAILURE).size());
+ assertEquals(1, runnerPut.getFlowFilesForRelationship(PutJMS.REL_SUCCESS).size());
+
+ final GetJMSQueue getJmsQueue = new GetJMSQueue();
+ final TestRunner runnerGet = TestRunners.newTestRunner(getJmsQueue);
+ runnerGet.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerGet.setProperty(JmsProperties.URL, TEST_URL);
+ runnerGet.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerGet.setProperty(JmsProperties.ACKNOWLEDGEMENT_MODE, TEST_ACK_MODE);
+ runnerGet.setProperty(JmsProperties.JMS_PROPS_TO_ATTRIBUTES, "true");
+
+ runnerGet.run();
+
+ assertEquals(1, runnerGet.getFlowFilesForRelationship(GetJMSQueue.REL_SUCCESS).size());
+
+ final List<MockFlowFile> flowFilesGet = runnerGet.getFlowFilesForRelationship(GetJMSQueue.REL_SUCCESS);
+
+ assertEquals(1, flowFilesGet.size());
+ final MockFlowFile successFlowFile = flowFilesGet.get(0);
+
+ successFlowFile.assertContentEquals("putGetMessage");
+ successFlowFile.assertAttributeEquals("jms.JMSDestination", TEST_DEST_NAME + testQueueSuffix());
+ successFlowFile.assertAttributeEquals("jms.JMSReplyTo", "queue://" + TEST_DEST_NAME + testQueueSuffix() + ".reply");
+ successFlowFile.assertAttributeEquals("jms.string", "banana");
+ successFlowFile.assertAttributeEquals("jms.integer", "50");
+ successFlowFile.assertAttributeEquals("jms.float", "3.14159");
+ successFlowFile.assertAttributeEquals("jms.boolean", "true");
+ successFlowFile.assertAttributeEquals("jms.long", "123456789");
+ successFlowFile.assertAttributeEquals("jms.short", "16384");
+ successFlowFile.assertAttributeEquals("jms.byte", "127");
+ successFlowFile.assertAttributeEquals("jms.double", "3.1415626547");
+ successFlowFile.assertAttributeEquals("jms.object", "{\"id\":215, \"name\": \"john doe\"}");
+ successFlowFile.assertAttributeEquals("jms.eyes", null);
+ successFlowFile.assertAttributeEquals("jms.badinteger", null);
+ }
+
+ @Test
+ public void testPutGetMessageTypes() throws JMSException {
+ final GetJMSQueue getJmsQueue = new GetJMSQueue();
+ final TestRunner runnerGet = TestRunners.newTestRunner(getJmsQueue);
+ runnerGet.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerGet.setProperty(JmsProperties.URL, TEST_URL);
+ runnerGet.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerGet.setProperty(JmsProperties.ACKNOWLEDGEMENT_MODE, TEST_ACK_MODE);
+ runnerGet.setProperty(JmsProperties.JMS_PROPS_TO_ATTRIBUTES, "true");
+
+ //------------------------------------------------------------
+ final PutJMS putJMStext = spy(new PutJMS());
+ final TestRunner runnerPutText = TestRunners.newTestRunner(putJMStext);
+ runnerPutText.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPutText.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPutText.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPutText.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerPutText.setProperty(JmsProperties.MESSAGE_TYPE, JmsProperties.MSG_TYPE_TEXT);
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("filename", "file1.txt");
+ runnerPutText.enqueue("putGetTextMessage", attributes);
+
+ runnerPutText.run();
+
+ assertEquals(0, runnerPutText.getFlowFilesForRelationship(PutJMS.REL_FAILURE).size());
+ assertEquals(1, runnerPutText.getFlowFilesForRelationship(PutJMS.REL_SUCCESS).size());
+
+ runnerGet.run();
+
+ final List<MockFlowFile> ffText = runnerGet.getFlowFilesForRelationship(GetJMSQueue.REL_SUCCESS);
+ assertEquals(1, ffText.size());
+ final MockFlowFile successText = ffText.get(0);
+ successText.assertContentEquals("putGetTextMessage");
+
+ //------------------------------------------------------------
+ final PutJMS putJMSempty = spy(new PutJMS());
+ final TestRunner runnerPutEmpty = TestRunners.newTestRunner(putJMSempty);
+ runnerPutEmpty.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPutEmpty.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPutEmpty.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPutEmpty.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerPutEmpty.setProperty(JmsProperties.MESSAGE_TYPE, JmsProperties.MSG_TYPE_EMPTY);
+
+ final Map<String, String> attributesEmpty = new HashMap<>();
+ attributesEmpty.put("filename", "file1.txt");
+ runnerPutEmpty.enqueue("putGetEmptyMessage", attributesEmpty);
+
+ runnerPutEmpty.run();
+
+ assertEquals(0, runnerPutEmpty.getFlowFilesForRelationship(PutJMS.REL_FAILURE).size());
+ assertEquals(1, runnerPutEmpty.getFlowFilesForRelationship(PutJMS.REL_SUCCESS).size());
+
+ runnerGet.clearTransferState();
+ runnerGet.run();
+
+ final List<MockFlowFile> ffEmpty = runnerGet.getFlowFilesForRelationship(GetJMSQueue.REL_SUCCESS);
+ assertEquals(1, ffEmpty.size());
+ final MockFlowFile successEmpty = ffEmpty.get(0);
+ successEmpty.assertContentEquals("");
+
+ //------------------------------------------------------------
+ final PutJMS putJMSstream = spy(new PutJMS());
+ final TestRunner runnerPutStream = TestRunners.newTestRunner(putJMSstream);
+ runnerPutStream.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPutStream.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPutStream.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPutStream.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerPutStream.setProperty(JmsProperties.MESSAGE_TYPE, JmsProperties.MSG_TYPE_STREAM);
+
+ final Map<String, String> attributesStream = new HashMap<>();
+ attributesStream.put("filename", "file1.txt");
+ runnerPutStream.enqueue("putGetStreamMessage", attributesStream);
+
+ runnerGet.clearTransferState();
+ runnerPutStream.run();
+
+ assertEquals(0, runnerPutStream.getFlowFilesForRelationship(PutJMS.REL_FAILURE).size());
+ assertEquals(1, runnerPutStream.getFlowFilesForRelationship(PutJMS.REL_SUCCESS).size());
+
+ runnerGet.run();
+
+ final List<MockFlowFile> ffStream = runnerGet.getFlowFilesForRelationship(GetJMSQueue.REL_SUCCESS);
+ assertEquals(1, ffStream.size());
+ final MockFlowFile successStream = ffStream.get(0);
+ successStream.assertContentEquals("putGetStreamMessage");
+
+ //------------------------------------------------------------
+ final PutJMS putJMSmap = spy(new PutJMS());
+ final TestRunner runnerPutMap = TestRunners.newTestRunner(putJMSmap);
+ runnerPutMap.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPutMap.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPutMap.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPutMap.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerPutMap.setProperty(JmsProperties.MESSAGE_TYPE, JmsProperties.MSG_TYPE_MAP);
+
+ final Map<String, String> attributesMap = new HashMap<>();
+ attributesMap.put("filename", "file1.txt");
+ runnerPutMap.enqueue("putGetMapMessage", attributesMap);
+
+ runnerGet.clearTransferState();
+ runnerPutMap.run();
+
+ assertEquals(0, runnerPutMap.getFlowFilesForRelationship(PutJMS.REL_FAILURE).size());
+ assertEquals(1, runnerPutMap.getFlowFilesForRelationship(PutJMS.REL_SUCCESS).size());
+
+ runnerGet.run();
+
+ final List<MockFlowFile> ffMap = runnerGet.getFlowFilesForRelationship(GetJMSQueue.REL_SUCCESS);
+ assertEquals(1, ffMap.size());
+ final MockFlowFile successMap = ffMap.get(0);
+ successMap.assertContentEquals("");
+
+ //------------------------------------------------------------
+ final PutJMS putJMSByte = spy(new PutJMS());
+ final TestRunner runnerPutByte = TestRunners.newTestRunner(putJMSByte);
+ runnerPutByte.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPutByte.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPutByte.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPutByte.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerPutByte.setProperty(JmsProperties.MESSAGE_TYPE, JmsProperties.MSG_TYPE_BYTE);
+
+ final Map<String, String> attributesByte = new HashMap<>();
+ attributesByte.put("filename", "file1.txt");
+ runnerPutByte.enqueue("putGetTextMessage", attributesByte);
+
+ runnerPutByte.run();
+
+ assertEquals(0, runnerPutByte.getFlowFilesForRelationship(PutJMS.REL_FAILURE).size());
+ assertEquals(1, runnerPutByte.getFlowFilesForRelationship(PutJMS.REL_SUCCESS).size());
+
+ runnerGet.clearTransferState();
+ runnerGet.run();
+
+ final List<MockFlowFile> ffByte = runnerGet.getFlowFilesForRelationship(GetJMSQueue.REL_SUCCESS);
+ assertEquals(1, ffByte.size());
+ final MockFlowFile successByte = ffByte.get(0);
+ successByte.assertContentEquals("putGetTextMessage");
+ }
+
+ @Test
+ public void testTTL() throws JMSException, InterruptedException {
+ final PutJMS putJMS = spy(new PutJMS());
+ final TestRunner runnerPut = TestRunners.newTestRunner(putJMS);
+ runnerPut.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPut.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPut.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPut.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerPut.setProperty(JmsProperties.MESSAGE_TTL, "3 s");
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("filename", "file1.txt");
+ runnerPut.enqueue("ttl10secNotExpired".getBytes(), attributes);
+
+ runnerPut.run();
+
+ assertEquals(0, runnerPut.getFlowFilesForRelationship(PutJMS.REL_FAILURE).size());
+ assertEquals(1, runnerPut.getFlowFilesForRelationship(PutJMS.REL_SUCCESS).size());
+
+ final GetJMSQueue getJmsQueue = new GetJMSQueue();
+ final TestRunner runnerGet = TestRunners.newTestRunner(getJmsQueue);
+ runnerGet.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerGet.setProperty(JmsProperties.URL, TEST_URL);
+ runnerGet.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerGet.setProperty(JmsProperties.ACKNOWLEDGEMENT_MODE, TEST_ACK_MODE);
+ runnerGet.setProperty(JmsProperties.TIMEOUT, "1 s");
+
+ runnerGet.run();
+
+ final List<MockFlowFile> flowFiles1 = runnerGet.getFlowFilesForRelationship(GetJMSQueue.REL_SUCCESS);
+ assertEquals(1, flowFiles1.size());
+ final MockFlowFile successFlowFile1 = flowFiles1.get(0);
+ successFlowFile1.assertContentEquals("ttl10secNotExpired");
+
+ runnerPut.clearTransferState();
+ runnerGet.clearTransferState();
+
+ runnerPut.setProperty(JmsProperties.MESSAGE_TTL, "1 s");
+ runnerPut.enqueue("ttl1secExpired".getBytes(), attributes);
+
+ runnerPut.run();
+
+ assertEquals(0, runnerPut.getFlowFilesForRelationship(PutJMS.REL_FAILURE).size());
+ assertEquals(1, runnerPut.getFlowFilesForRelationship(PutJMS.REL_SUCCESS).size());
+
+ Thread.sleep(2000L);
+ runnerGet.run();
+
+ final List<MockFlowFile> flowFiles2 = runnerGet.getFlowFilesForRelationship(GetJMSQueue.REL_SUCCESS);
+ assertEquals(0, flowFiles2.size());
+ }
+
+ @Test
+ public void testFailureOnFileExceedsBufferSize() throws JMSException {
+ final PutJMS putJMS = spy(new PutJMS());
+ final TestRunner runnerPut = TestRunners.newTestRunner(putJMS);
+ runnerPut.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPut.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPut.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPut.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerPut.setProperty(JmsProperties.MAX_BUFFER_SIZE, "10 B");
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("filename", "file1.txt");
+ runnerPut.enqueue("failureOnFileExceedsBufferSize".getBytes(), attributes);
+
+ runnerPut.run();
+
+ assertEquals(1, runnerPut.getFlowFilesForRelationship(PutJMS.REL_FAILURE).size());
+ assertEquals(0, runnerPut.getFlowFilesForRelationship(PutJMS.REL_SUCCESS).size());
+ }
+
+ @Test(expected = NumberFormatException.class)
+ public void testBadMessagePriorityValueFails() throws JMSException {
+ final PutJMS putJMS = spy(new PutJMS());
+ final TestRunner runnerPut = TestRunners.newTestRunner(putJMS);
+ runnerPut.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPut.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPut.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPut.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerPut.setProperty(JmsProperties.MESSAGE_PRIORITY, "negative one");
+ assertEquals(PutJMS.DEFAULT_MESSAGE_PRIORITY, runnerPut.getProcessContext().getProperty(JmsProperties.MESSAGE_PRIORITY).asInteger().intValue());
+ }
+
+ @Test
+ public void testBadMessagePriorityRunSucceeds() throws JMSException {
+ final PutJMS putJMS = spy(new PutJMS());
+ final TestRunner runnerPut = TestRunners.newTestRunner(putJMS);
+ runnerPut.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPut.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPut.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPut.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+ runnerPut.setProperty(JmsProperties.MESSAGE_PRIORITY, "negative one");
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("filename", "file1.txt");
+ runnerPut.enqueue("badMessagePriorityRunSucceeds".getBytes(), attributes);
+
+ runnerPut.run();
+
+ assertEquals(0, runnerPut.getFlowFilesForRelationship(PutJMS.REL_FAILURE).size());
+ assertEquals(1, runnerPut.getFlowFilesForRelationship(PutJMS.REL_SUCCESS).size());
+ }
+
+ @Test
+ public void testPutSendRoutesToFailure() throws JMSException, NoSuchFieldException, IllegalAccessException {
+
+ final PutJMS putJMS = spy(new PutJMS());
+
+ final TestRunner runnerPut = TestRunners.newTestRunner(putJMS);
+ runnerPut.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPut.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPut.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPut.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+
+ final ProcessContext context = runnerPut.getProcessContext();
+
+ final Queue<WrappedMessageProducer> wrappedMessageProducerQueue = (Queue) spy(new LinkedBlockingQueue<>());
+ injectFieldValue(PutJMS.class, putJMS, "producerQueue", wrappedMessageProducerQueue);
+
+ final WrappedMessageProducer wrappedMessageProducer = spy(JmsFactory.createMessageProducer(context, true));
+ final MessageProducer messageProducer = spy(wrappedMessageProducer.getProducer());
+
+ doAnswer(new Answer<WrappedMessageProducer>() {
+ @Override
+ public WrappedMessageProducer answer(InvocationOnMock invocationOnMock) {
+ return wrappedMessageProducer;
+ }
+ }).when(wrappedMessageProducerQueue).poll();
+ assertEquals(wrappedMessageProducer, wrappedMessageProducerQueue.poll());
+
+ doAnswer(new Answer<MessageProducer>() {
+ @Override
+ public MessageProducer answer(InvocationOnMock invocationOnMock) {
+ return messageProducer;
+ }
+ }).when(wrappedMessageProducer).getProducer();
+
+ doThrow(new JMSException("force send to fail")).when(messageProducer).send(any(Message.class));
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("filename", "file1.txt");
+ runnerPut.enqueue("putSendRoutesToFailure".getBytes(), attributes);
+
+ runnerPut.run();
+
+ assertEquals(0, runnerPut.getFlowFilesForRelationship(PutJMS.REL_SUCCESS).size());
+ assertEquals(1, runnerPut.getFlowFilesForRelationship(PutJMS.REL_FAILURE).size());
+
+ final List<MockFlowFile> flowFilesFail = runnerPut.getFlowFilesForRelationship(PutJMS.REL_FAILURE);
+ assertEquals(1, flowFilesFail.size());
+ }
+
+ @Test
+ public void testPutCommitRoutesToFailure() throws JMSException, NoSuchFieldException, IllegalAccessException {
+
+ final PutJMS putJMS = spy(new PutJMS());
+
+ final TestRunner runnerPut = TestRunners.newTestRunner(putJMS);
+ runnerPut.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER);
+ runnerPut.setProperty(JmsProperties.URL, TEST_URL);
+ runnerPut.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE);
+ runnerPut.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix());
+
+ final ProcessContext context = runnerPut.getProcessContext();
+ final Queue<WrappedMessageProducer> wrappedMessageProducerQueue = (Queue) spy(new LinkedBlockingQueue<>());
+ injectFieldValue(PutJMS.class, putJMS, "producerQueue", wrappedMessageProducerQueue);
+
+ final WrappedMessageProducer wrappedMessageProducer = spy(JmsFactory.createMessageProducer(context, true));
+ final MessageProducer messageProducer = spy(wrappedMessageProducer.getProducer());
+ final Connection connection = JmsFactory.createConnection(context);
+ final Session jmsSession = spy(JmsFactory.createSession(context, connection, true));
+
+ doAnswer(new Answer<WrappedMessageProducer>() {
+ @Override
+ public WrappedMessageProducer answer(InvocationOnMock invocationOnMock) {
+ return wrappedMessageProducer;
+ }
+ }).when(wrappedMessageProducerQueue).poll();
+
+ doAnswer(new Answer<MessageProducer>() {
+ @Override
+ public MessageProducer answer(InvocationOnMock invocationOnMock) {
+ return messageProducer;
+ }
+ }).when(wrappedMessageProducer).getProducer();
+
+ doAnswer(new Answer<Session>() {
+ @Override
+ public Session answer(InvocationOnMock invocationOnMock) {
+ return jmsSession;
+ }
+ }).when(wrappedMessageProducer).getSession();
+
+ doThrow(new JMSException("force commit to fail")).when(jmsSession).commit();
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("filename", "file1.txt");
+ runnerPut.enqueue("putCommitRoutesToFailure".getBytes(), attributes);
+
+ runnerPut.run();
+
+ assertEquals(0, runnerPut.getFlowFilesForRelationship(PutJMS.REL_SUCCESS).size());
+ assertEquals(1, runnerPut.getFlowFilesForRelationship(PutJMS.REL_FAILURE).size());
+
+ final List<MockFlowFile> flowFilesFail = runnerPut.getFlowFilesForRelationship(PutJMS.REL_FAILURE);
+ assertEquals(1, flowFilesFail.size());
+ }
+}
\ No newline at end of file