NIFI-1834 adding fixes to PutTCP
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/4a9d3c97 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/4a9d3c97 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/4a9d3c97 Branch: refs/heads/0.x Commit: 4a9d3c97ea14102e68be1709d868f0cc1f4418bf Parents: fd458d1 Author: jpercivall <[email protected]> Authored: Fri Jun 17 16:58:07 2016 -0400 Committer: jpercivall <[email protected]> Committed: Tue Jun 21 11:00:46 2016 -0400 ---------------------------------------------------------------------- nifi-commons/nifi-processor-utilities/pom.xml | 5 + .../util/put/AbstractPutEventProcessor.java | 37 ++- .../nifi/processors/splunk/PutSplunk.java | 8 - .../apache/nifi/processors/standard/PutTCP.java | 63 ++-- .../nifi/processors/standard/TCPTestServer.java | 167 ---------- .../nifi/processors/standard/TestPutTCP.java | 287 +---------------- .../nifi/processors/standard/TestPutTcpSSL.java | 68 ++++ .../processors/standard/util/TCPTestServer.java | 183 +++++++++++ .../standard/util/TestPutTCPCommon.java | 308 +++++++++++++++++++ pom.xml | 5 + 10 files changed, 643 insertions(+), 488 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/4a9d3c97/nifi-commons/nifi-processor-utilities/pom.xml ---------------------------------------------------------------------- diff --git a/nifi-commons/nifi-processor-utilities/pom.xml b/nifi-commons/nifi-processor-utilities/pom.xml index 8196818..0f3ee01 100644 --- a/nifi-commons/nifi-processor-utilities/pom.xml +++ b/nifi-commons/nifi-processor-utilities/pom.xml @@ -40,5 +40,10 @@ <groupId>commons-io</groupId> <artifactId>commons-io</artifactId> </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-ssl-context-service-api</artifactId> + <scope>provided</scope> + </dependency> </dependencies> </project> http://git-wip-us.apache.org/repos/asf/nifi/blob/4a9d3c97/nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/put/AbstractPutEventProcessor.java ---------------------------------------------------------------------- diff --git a/nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/put/AbstractPutEventProcessor.java b/nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/put/AbstractPutEventProcessor.java index be16b36..9fd0496 100644 --- a/nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/put/AbstractPutEventProcessor.java +++ b/nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/put/AbstractPutEventProcessor.java @@ -31,6 +31,7 @@ import org.apache.nifi.processor.util.put.sender.ChannelSender; import org.apache.nifi.processor.util.put.sender.DatagramChannelSender; import org.apache.nifi.processor.util.put.sender.SSLSocketChannelSender; import org.apache.nifi.processor.util.put.sender.SocketChannelSender; +import org.apache.nifi.ssl.SSLContextService; import javax.net.ssl.SSLContext; import java.io.IOException; @@ -123,11 +124,11 @@ public abstract class AbstractPutEventProcessor extends AbstractSessionFactoryPr .name("Outgoing Message Delimiter") .description("Specifies the delimiter to use when sending messages out over the same TCP stream. The delimiter is appended to each FlowFile message " + "that is transmitted over the stream so that the receiver can determine when one message ends and the next message begins. Users should " - + "ensure that the FlowFile content does not contain the delimiter character to avoid errors. If it is not possible to define a delimiter " - + "character that is not present in the FlowFile content then the user must use another processor to change the encoding of the data e.g. Base64 " - + "encoding.") - .required(false) + + "ensure that the FlowFile content does not contain the delimiter character to avoid errors. In order to use a new line character you can " + + "enter '\\n'. For a tab character use '\\t'. Finally for a carriage return use '\\r'.") + .required(true) .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .defaultValue("\\n") .expressionLanguageSupported(true) .build(); public static final PropertyDescriptor CONNECTION_PER_FLOWFILE = new PropertyDescriptor.Builder() @@ -135,8 +136,17 @@ public abstract class AbstractPutEventProcessor extends AbstractSessionFactoryPr .description("Specifies whether to send each FlowFile's content on an individual connection.") .required(true) .defaultValue("false") - .addValidator(StandardValidators.BOOLEAN_VALIDATOR) + .allowableValues("true", "false") + .build(); + + public static final PropertyDescriptor SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder() + .name("SSL Context Service") + .description("The Controller Service to use in order to obtain an SSL Context. If this property is set, " + + "messages will be sent over a secure connection.") + .required(false) + .identifiesControllerService(SSLContextService.class) .build(); + public static final Relationship REL_SUCCESS = new Relationship.Builder() .name("success") .description("FlowFiles that are sent successfully to the destination are sent out this relationship.") @@ -543,4 +553,21 @@ public abstract class AbstractPutEventProcessor extends AbstractSessionFactoryPr } } + /** + * Gets the current value of the "Outgoing Message Delimiter" property and parses the special characters. + * + * @param context + * - the current process context. + * @param flowFile + * - the FlowFile being processed. + * + * @return String containing the Delimiter value. + */ + protected String getOutgoingMessageDelimiter(final ProcessContext context, final FlowFile flowFile) { + String delimiter = context.getProperty(OUTGOING_MESSAGE_DELIMITER).evaluateAttributeExpressions(flowFile).getValue(); + if (delimiter != null) { + delimiter = delimiter.replace("\\n", "\n").replace("\\r", "\r").replace("\\t", "\t"); + } + return delimiter; + } } http://git-wip-us.apache.org/repos/asf/nifi/blob/4a9d3c97/nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java b/nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java index aece3b8..c461b0f 100644 --- a/nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java +++ b/nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java @@ -60,14 +60,6 @@ import java.util.concurrent.TimeUnit; "the FlowFile will be sent directly to Splunk as if it were a single message.") public class PutSplunk extends AbstractPutEventProcessor { - public static final PropertyDescriptor SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder() - .name("SSL Context Service") - .description("The Controller Service to use in order to obtain an SSL Context. If this property is set, " + - "messages will be sent over a secure connection.") - .required(false) - .identifiesControllerService(SSLContextService.class) - .build(); - public static final char NEW_LINE_CHAR = '\n'; @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/4a9d3c97/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutTCP.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutTCP.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutTCP.java index de22200..3354b09 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutTCP.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutTCP.java @@ -19,6 +19,7 @@ package org.apache.nifi.processors.standard; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; +import java.nio.charset.Charset; import java.util.Arrays; import java.util.List; import java.util.concurrent.TimeUnit; @@ -39,7 +40,11 @@ import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.processor.io.InputStreamCallback; import org.apache.nifi.processor.util.put.AbstractPutEventProcessor; import org.apache.nifi.processor.util.put.sender.ChannelSender; +import org.apache.nifi.ssl.SSLContextService; import org.apache.nifi.stream.io.StreamUtils; +import org.apache.nifi.util.StopWatch; + +import javax.net.ssl.SSLContext; /** * <p> @@ -57,7 +62,7 @@ import org.apache.nifi.stream.io.StreamUtils; * <li><b>Port</b> - The TCP port of the destination TCP server.</li> * </ul> * </p> - * + * * <p> * This processor has the following optional properties: * <ul> @@ -65,7 +70,7 @@ import org.apache.nifi.stream.io.StreamUtils; * <li><b>Idle Connection Expiration</b> - The time threshold after which a TCP sender is deemed eligible for pruning - the associated TCP connection will be closed after this timeout.</li> * <li><b>Max Size of Socket Send Buffer</b> - The maximum size of the socket send buffer that should be used. This is a suggestion to the Operating System to indicate how big the socket buffer should * be. If this value is set too low, the buffer may fill up before the data can be read, and incoming data will be dropped.</li> - * <li><b>Outgoing Message Delimiter/b> - A string to append to the end of each FlowFiles content to indicate the end of the message to the TCP server.</li> + * <li><b>Outgoing Message Delimiter</b> - A string to append to the end of each FlowFiles content to indicate the end of the message to the TCP server.</li> * <li><b>Timeout</b> - The timeout period for determining an error has occurred whilst connecting or sending data.</li> * </ul> * </p> @@ -106,12 +111,19 @@ public class PutTCP extends AbstractPutEventProcessor { final int port = context.getProperty(PORT).asInteger(); final int timeout = context.getProperty(TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue(); final int bufferSize = context.getProperty(MAX_SOCKET_SEND_BUFFER_SIZE).asDataSize(DataUnit.B).intValue(); + final SSLContextService sslContextService = (SSLContextService) context.getProperty(SSL_CONTEXT_SERVICE).asControllerService(); - return createSender(protocol, hostname, port, timeout, bufferSize, null); + SSLContext sslContext = null; + if (sslContextService != null) { + sslContext = sslContextService.createSSLContext(SSLContextService.ClientAuth.REQUIRED); + } + + return createSender(protocol, hostname, port, timeout, bufferSize, sslContext); } /** - * Creates a Universal Resource Identifier (URI) for this processor. Constructs a URI of the form TCP://<host>:<port> where the host and port values are taken from the configured property values. + * Creates a Universal Resource Identifier (URI) for this processor. Constructs a URI of the form TCP://< host >:< port > where the host and port + * values are taken from the configured property values. * * @param context * - the current process context. @@ -134,7 +146,11 @@ public class PutTCP extends AbstractPutEventProcessor { */ @Override protected List<PropertyDescriptor> getAdditionalProperties() { - return Arrays.asList(CONNECTION_PER_FLOWFILE, OUTGOING_MESSAGE_DELIMITER, TIMEOUT); + return Arrays.asList(CONNECTION_PER_FLOWFILE, + OUTGOING_MESSAGE_DELIMITER, + TIMEOUT, + SSL_CONTEXT_SERVICE, + CHARSET); } /** @@ -163,17 +179,20 @@ public class PutTCP extends AbstractPutEventProcessor { } try { - String delimiter = getDelimiter(context, flowFile); + String outgoingMessageDelimiter = getOutgoingMessageDelimiter(context, flowFile); ByteArrayOutputStream content = readContent(session, flowFile); - if (delimiter != null) { - content = appendDelimiter(content, delimiter); + if (outgoingMessageDelimiter != null) { + Charset charset = Charset.forName(context.getProperty(CHARSET).getValue()); + content = appendDelimiter(content, outgoingMessageDelimiter, charset); } + StopWatch stopWatch = new StopWatch(true); sender.send(content.toByteArray()); + session.getProvenanceReporter().send(flowFile, transitUri, stopWatch.getElapsed(TimeUnit.MILLISECONDS)); session.transfer(flowFile, REL_SUCCESS); session.commit(); } catch (Exception e) { - getLogger().error("Exception while handling a process session, transferring {} to failure.", new Object[] { flowFile }, e); onFailure(context, session, flowFile); + getLogger().error("Exception while handling a process session, transferring {} to failure.", new Object[] { flowFile }, e); } finally { // If we are going to use this sender again, then relinquish it back to the pool. if (!isConnectionPerFlowFile(context)) { @@ -185,7 +204,7 @@ public class PutTCP extends AbstractPutEventProcessor { } /** - * event handler method to perform the required actions when a failure has occurred. The FlowFile is penalized, forwarded to the failure relationship and the context is yielded. + * Event handler method to perform the required actions when a failure has occurred. The FlowFile is penalized, forwarded to the failure relationship and the context is yielded. * * @param context * - the current process context. @@ -225,7 +244,7 @@ public class PutTCP extends AbstractPutEventProcessor { /** * Helper method to append a delimiter to the message contents. - * + * * @param content * - the message contents. * @param delimiter @@ -233,30 +252,12 @@ public class PutTCP extends AbstractPutEventProcessor { * * @return ByteArrayOutputStream object containing the new message contents. */ - protected ByteArrayOutputStream appendDelimiter(final ByteArrayOutputStream content, final String delimiter) { - content.write(delimiter.getBytes(), 0, delimiter.length()); + protected ByteArrayOutputStream appendDelimiter(final ByteArrayOutputStream content, final String delimiter, Charset charSet) { + content.write(delimiter.getBytes(charSet), 0, delimiter.length()); return content; } /** - * Gets the current value of the "Outgoing Message Delimiter" property. - * - * @param context - * - the current process context. - * @param flowFile - * - the FlowFile being processed. - * - * @return String containing the Delimiter value. - */ - protected String getDelimiter(final ProcessContext context, final FlowFile flowFile) { - String delimiter = context.getProperty(OUTGOING_MESSAGE_DELIMITER).evaluateAttributeExpressions(flowFile).getValue(); - if (delimiter != null) { - delimiter = delimiter.replace("\\n", "\n").replace("\\r", "\r").replace("\\t", "\t"); - } - return delimiter; - } - - /** * Gets the current value of the "Connection Per FlowFile" property. * * @param context http://git-wip-us.apache.org/repos/asf/nifi/blob/4a9d3c97/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TCPTestServer.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TCPTestServer.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TCPTestServer.java deleted file mode 100644 index 5ec12e4..0000000 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TCPTestServer.java +++ /dev/null @@ -1,167 +0,0 @@ -/* - * 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 java.io.IOException; -import java.io.InputStream; -import java.net.InetAddress; -import java.net.ServerSocket; -import java.net.Socket; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.ArrayBlockingQueue; - -public class TCPTestServer implements Runnable { - - private final InetAddress ipAddress; - private final int port; - private final String messageDelimiter; - private volatile ServerSocket serverSocket; - private final ArrayBlockingQueue<List<Byte>> recvQueue; - private volatile Socket connectionSocket; - public final static String DEFAULT_MESSAGE_DELIMITER = "\n"; - private volatile int totalNumConnections = 0; - - public TCPTestServer(final InetAddress ipAddress, final int port, final ArrayBlockingQueue<List<Byte>> recvQueue) { - this(ipAddress, port, recvQueue, DEFAULT_MESSAGE_DELIMITER); - } - - public TCPTestServer(final InetAddress ipAddress, final int port, final ArrayBlockingQueue<List<Byte>> recvQueue, final String messageDelimiter) { - this.ipAddress = ipAddress; - this.port = port; - this.recvQueue = recvQueue; - this.messageDelimiter = messageDelimiter; - } - - public synchronized void startServer() throws IOException { - if (!isServerRunning()) { - serverSocket = new ServerSocket(port, 0, ipAddress); - Thread t = new Thread(this); - t.setName(this.getClass().getSimpleName()); - t.start(); - } - } - - public synchronized void shutdown() { - shutdownConnection(); - shutdownServer(); - } - - public synchronized void shutdownServer() { - if (isServerRunning()) { - try { - serverSocket.close(); - } catch (IOException ioe) { - // Do Nothing. - } - } - } - - public synchronized void shutdownConnection() { - if (isConnected()) { - try { - connectionSocket.close(); - } catch (IOException ioe) { - // Do Nothing. - } - } - } - - private void storeReceivedMessage(final List<Byte> message) { - recvQueue.add(message); - } - - private boolean isServerRunning() { - return serverSocket != null && !serverSocket.isClosed(); - } - - private boolean isConnected() { - return connectionSocket != null && !connectionSocket.isClosed(); - } - - public List<Byte> getReceivedMessage() { - return recvQueue.poll(); - } - - public int getTotalNumConnections() { - return totalNumConnections; - } - - protected boolean isDelimiterPresent(final List<Byte> message) { - if (messageDelimiter != null && message.size() >= messageDelimiter.length()) { - for (int i = 1; i <= messageDelimiter.length(); i++) { - if (message.get(message.size() - i).byteValue() == messageDelimiter.charAt(messageDelimiter.length() - i)) { - if (i == messageDelimiter.length()) { - return true; - } - } else { - break; - } - } - } - return false; - } - - protected boolean removeDelimiter(final List<Byte> message) { - if (isDelimiterPresent(message)) { - final int messageSize = message.size(); - for (int i = 1; i <= messageDelimiter.length(); i++) { - message.remove(messageSize - i); - } - return true; - } - - return false; - } - - @Override - public void run() { - try { - while (isServerRunning()) { - connectionSocket = serverSocket.accept(); - totalNumConnections++; - InputStream in = connectionSocket.getInputStream(); - while (isConnected()) { - final List<Byte> message = new ArrayList<Byte>(); - while (true) { - final int c = in.read(); - if (c < 0) { - if (!message.isEmpty()) { - storeReceivedMessage(message); - } - shutdownConnection(); - break; - } - - message.add((byte) c); - - if (removeDelimiter(message)) { - storeReceivedMessage(message); - break; - } - } - } - } - } catch (Exception e) { - // Do Nothing - } finally { - shutdownConnection(); - shutdownServer(); - } - - } -} http://git-wip-us.apache.org/repos/asf/nifi/blob/4a9d3c97/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTCP.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTCP.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTCP.java index 14f4d6c..2d5254c 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTCP.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTCP.java @@ -16,234 +16,25 @@ */ package org.apache.nifi.processors.standard; -import static org.junit.Assert.assertArrayEquals; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import java.net.InetAddress; -import java.util.List; -import java.util.concurrent.ArrayBlockingQueue; +import org.apache.nifi.processors.standard.util.TestPutTCPCommon; -import org.apache.commons.lang3.ArrayUtils; -import org.apache.nifi.util.TestRunner; -import org.apache.nifi.util.TestRunners; -import org.junit.After; -import org.junit.Before; -import org.junit.Test; +public class TestPutTCP extends TestPutTCPCommon { -public class TestPutTCP { - - private final static String TCP_SERVER_ADDRESS = "127.0.0.1"; - private final static String UNKNOWN_HOST = "fgdsfgsdffd"; - private final static String INVALID_IP_ADDRESS = "300.300.300.300"; - private final static int TCP_SERVER_PORT = 54674; - private final static int MIN_INVALID_PORT = 0; - private final static int MIN_VALID_PORT = 1; - private final static int MAX_VALID_PORT = 65535; - private final static int MAX_INVALID_PORT = 65536; - private final static int BUFFER_SIZE = 1024; - private final static int VALID_LARGE_FILE_SIZE = 32768; - private final static int VALID_SMALL_FILE_SIZE = 64; - private final static int LOAD_TEST_ITERATIONS = 500; - private final static int LOAD_TEST_THREAD_COUNT = 1; - private final static int DEFAULT_ITERATIONS = 1; - private final static int DEFAULT_THREAD_COUNT = 1; - private final static char CONTENT_CHAR = 'x'; - private final static int DATA_WAIT_PERIOD = 1000; - private final static int DEFAULT_TEST_TIMEOUT_PERIOD = 10000; - private final static int LONG_TEST_TIMEOUT_PERIOD = 100000; - private final static String OUTGOING_MESSAGE_DELIMITER = "\n"; - private final static String OUTGOING_MESSAGE_DELIMITER_MULTI_CHAR = "{delimiter}\r\n"; - - private TCPTestServer server; - private TestRunner runner; - private ArrayBlockingQueue<List<Byte>> recvQueue; - - // Test Data - private final static String[] EMPTY_FILE = { "" }; - private final static String[] VALID_FILES = { "abcdefghijklmnopqrstuvwxyz", "zyxwvutsrqponmlkjihgfedcba", "12345678", "343424222", "!@£$%^&*()_+:|{}[];\\" }; - - @Before - public void setup() throws Exception { - recvQueue = new ArrayBlockingQueue<List<Byte>>(BUFFER_SIZE); - runner = TestRunners.newTestRunner(PutTCP.class); - } - - private TCPTestServer createTestServer(final String address, final int port, final ArrayBlockingQueue<List<Byte>> recvQueue, final String delimiter) throws Exception { - TCPTestServer server = new TCPTestServer(InetAddress.getByName(address), port, recvQueue, delimiter); - server.startServer(); - return server; - } - - @After - public void cleanup() throws Exception { - runner.shutdown(); - removeTestServer(server); - } - - private void removeTestServer(TCPTestServer server) { - if (server != null) { - server.shutdown(); - server = null; - } - } - - @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) - public void testValidFiles() throws Exception { - server = createTestServer(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, recvQueue, OUTGOING_MESSAGE_DELIMITER); - configureProperties(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, OUTGOING_MESSAGE_DELIMITER, false, true); - sendTestData(VALID_FILES); - checkReceivedAllData(recvQueue, VALID_FILES); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, 1); - } - - @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) - public void testPruneSenders() throws Exception { - server = createTestServer(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, recvQueue, OUTGOING_MESSAGE_DELIMITER); - configureProperties(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, OUTGOING_MESSAGE_DELIMITER, false, true); - sendTestData(VALID_FILES); - Thread.sleep(10); - checkRelationships(VALID_FILES.length, 0); - checkReceivedAllData(recvQueue, VALID_FILES); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, 1); - runner.setProperty(PutTCP.IDLE_EXPIRATION, "1 second"); - Thread.sleep(2000); - runner.run(1, false, false); - runner.clearTransferState(); - sendTestData(VALID_FILES); - checkReceivedAllData(recvQueue, VALID_FILES); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, 2); - } - - @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) - public void testMultiCharDelimiter() throws Exception { - server = createTestServer(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, recvQueue, OUTGOING_MESSAGE_DELIMITER_MULTI_CHAR); - configureProperties(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, OUTGOING_MESSAGE_DELIMITER_MULTI_CHAR, false, true); - sendTestData(VALID_FILES); - checkReceivedAllData(recvQueue, VALID_FILES); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, 1); - } - - @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) - public void testConnectionPerFlowFile() throws Exception { - server = createTestServer(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, recvQueue, null); - configureProperties(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, null, true, true); - sendTestData(VALID_FILES); - checkReceivedAllData(recvQueue, VALID_FILES); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, VALID_FILES.length); - } - - @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) - public void testConnectionFailure() throws Exception { - server = createTestServer(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, recvQueue, OUTGOING_MESSAGE_DELIMITER); - configureProperties(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, OUTGOING_MESSAGE_DELIMITER, false, true); - sendTestData(VALID_FILES); - checkReceivedAllData(recvQueue, VALID_FILES); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, 1); - removeTestServer(server); - runner.clearTransferState(); - sendTestData(VALID_FILES); - Thread.sleep(10); - // checkRelationships(0, 5); - checkNoDataReceived(recvQueue); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, 1); - server = createTestServer(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, recvQueue, OUTGOING_MESSAGE_DELIMITER); - configureProperties(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, OUTGOING_MESSAGE_DELIMITER, false, true); - sendTestData(VALID_FILES); - checkReceivedAllData(recvQueue, VALID_FILES); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, 1); - } - - @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) - public void testEmptyFile() throws Exception { - server = createTestServer(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, recvQueue, OUTGOING_MESSAGE_DELIMITER); - configureProperties(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, null, false, true); - sendTestData(EMPTY_FILE); - Thread.sleep(10); - checkRelationships(EMPTY_FILE.length, 0); - checkNoDataReceived(recvQueue); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, 1); - } - - @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) - public void testlargeValidFile() throws Exception { - server = createTestServer(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, recvQueue, null); - configureProperties(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, null, true, true); - final String[] testData = createContent(VALID_LARGE_FILE_SIZE); - sendTestData(testData); - checkReceivedAllData(recvQueue, testData); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, testData.length); - } - - @Test(timeout = LONG_TEST_TIMEOUT_PERIOD) - public void testInvalidIPAddress() throws Exception { - server = createTestServer(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, recvQueue, OUTGOING_MESSAGE_DELIMITER); - configureProperties(INVALID_IP_ADDRESS, TCP_SERVER_PORT, OUTGOING_MESSAGE_DELIMITER, false, true); - sendTestData(VALID_FILES); - Thread.sleep(10); - checkRelationships(0, VALID_FILES.length); - checkNoDataReceived(recvQueue); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, 0); - } - - @Test(timeout = LONG_TEST_TIMEOUT_PERIOD) - public void testUnknownHostname() throws Exception { - server = createTestServer(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, recvQueue, OUTGOING_MESSAGE_DELIMITER); - configureProperties(UNKNOWN_HOST, TCP_SERVER_PORT, OUTGOING_MESSAGE_DELIMITER, false, true); - sendTestData(VALID_FILES); - Thread.sleep(10); - checkRelationships(0, VALID_FILES.length); - checkNoDataReceived(recvQueue); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, 0); - } - - @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) - public void testInvalidPort() throws Exception { - configureProperties(UNKNOWN_HOST, MIN_INVALID_PORT, OUTGOING_MESSAGE_DELIMITER, false, false); - configureProperties(UNKNOWN_HOST, MIN_VALID_PORT, OUTGOING_MESSAGE_DELIMITER, false, true); - configureProperties(UNKNOWN_HOST, MAX_VALID_PORT, OUTGOING_MESSAGE_DELIMITER, false, true); - configureProperties(UNKNOWN_HOST, MAX_INVALID_PORT, OUTGOING_MESSAGE_DELIMITER, false, false); - } - - @Test(timeout = LONG_TEST_TIMEOUT_PERIOD) - public void testLoadTest() throws Exception { - server = createTestServer(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, recvQueue, OUTGOING_MESSAGE_DELIMITER); - Thread.sleep(1000); - final String[] testData = createContent(VALID_SMALL_FILE_SIZE); - configureProperties(TCP_SERVER_ADDRESS, TCP_SERVER_PORT, OUTGOING_MESSAGE_DELIMITER, false, true); - sendTestData(testData, LOAD_TEST_ITERATIONS, LOAD_TEST_THREAD_COUNT); - checkReceivedAllData(recvQueue, testData, LOAD_TEST_ITERATIONS); - checkInputQueueIsEmpty(); - checkTotalNumConnections(server, 1); - } - - private void checkTotalNumConnections(final TCPTestServer server, final int expectedTotalNumConnections) { - assertEquals(expectedTotalNumConnections, server.getTotalNumConnections()); + public TestPutTCP() { + super(); + ssl = false; } - private void configureProperties(final String host, final int port, final String outgoingMessageDelimiter, final boolean connectionPerFlowFile, final boolean expectValid) { + @Override + public void configureProperties(String host, int port, String outgoingMessageDelimiter, boolean connectionPerFlowFile, boolean expectValid) { runner.setProperty(PutTCP.HOSTNAME, host); runner.setProperty(PutTCP.PORT, Integer.toString(port)); if (outgoingMessageDelimiter != null) { runner.setProperty(PutTCP.OUTGOING_MESSAGE_DELIMITER, outgoingMessageDelimiter); } - if (connectionPerFlowFile) { - runner.setProperty(PutTCP.CONNECTION_PER_FLOWFILE, "true"); - } else { - runner.setProperty(PutTCP.CONNECTION_PER_FLOWFILE, "false"); - } + + runner.setProperty(PutTCP.CONNECTION_PER_FLOWFILE, String.valueOf(connectionPerFlowFile)); + if (expectValid) { runner.assertValid(); } else { @@ -251,63 +42,5 @@ public class TestPutTCP { } } - private void sendTestData(final String[] testData) { - sendTestData(testData, DEFAULT_ITERATIONS, DEFAULT_THREAD_COUNT); - } - - private void sendTestData(final String[] testData, final int iterations, final int threadCount) { - runner.setThreadCount(threadCount); - for (int i = 0; i < iterations; i++) { - for (String item : testData) { - runner.enqueue(item.getBytes()); - } - runner.run(testData.length, false, i == 0 ? true : false); - } - } - - private void checkRelationships(final int successCount, final int failedCount) { - runner.assertTransferCount(PutTCP.REL_SUCCESS, successCount); - runner.assertTransferCount(PutTCP.REL_FAILURE, failedCount); - } - - private void checkNoDataReceived(final ArrayBlockingQueue<List<Byte>> recvQueue) throws Exception { - Thread.sleep(DATA_WAIT_PERIOD); - assertNull(recvQueue.poll()); - } - - private void checkInputQueueIsEmpty() { - runner.assertQueueEmpty(); - } - - private void checkReceivedAllData(final ArrayBlockingQueue<List<Byte>> recvQueue, final String[] sentData) throws Exception { - checkReceivedAllData(recvQueue, sentData, DEFAULT_ITERATIONS); - } - - private void checkReceivedAllData(final ArrayBlockingQueue<List<Byte>> recvQueue, final String[] sentData, final int iterations) throws Exception { - // check each sent FlowFile was successfully sent and received. - for (int i = 0; i < iterations; i++) { - for (String item : sentData) { - List<Byte> message = recvQueue.take(); - assertNotNull(message); - Byte[] messageBytes = new Byte[message.size()]; - assertArrayEquals(item.getBytes(), ArrayUtils.toPrimitive(message.toArray(messageBytes))); - } - } - - runner.assertTransferCount(PutTCP.REL_SUCCESS, sentData.length * iterations); - runner.clearTransferState(); - - // Check that we have no unexpected extra data. - assertNull(recvQueue.poll()); - } - - private String[] createContent(final int size) { - final char[] content = new char[size]; - for (int i = 0; i < size; i++) { - content[i] = CONTENT_CHAR; - } - - return new String[] { new String(content) }; - } } http://git-wip-us.apache.org/repos/asf/nifi/blob/4a9d3c97/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTcpSSL.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTcpSSL.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTcpSSL.java new file mode 100644 index 0000000..b758987 --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTcpSSL.java @@ -0,0 +1,68 @@ +/* + * 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.processors.standard.util.TestPutTCPCommon; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.ssl.StandardSSLContextService; + +import java.util.HashMap; +import java.util.Map; + +public class TestPutTcpSSL extends TestPutTCPCommon { + private static Map<String, String> sslProperties; + + public TestPutTcpSSL() { + super(); + ssl = true; + + sslProperties = createSslProperties(); + } + + @Override + public void configureProperties(String host, int port, String outgoingMessageDelimiter, boolean connectionPerFlowFile, boolean expectValid) throws InitializationException { + runner.setProperty(PutTCP.HOSTNAME, host); + runner.setProperty(PutTCP.PORT, Integer.toString(port)); + + final StandardSSLContextService sslService = new StandardSSLContextService(); + runner.addControllerService("ssl-context", sslService, sslProperties); + runner.enableControllerService(sslService); + runner.setProperty(PutTCP.SSL_CONTEXT_SERVICE, "ssl-context"); + + if (outgoingMessageDelimiter != null) { + runner.setProperty(PutTCP.OUTGOING_MESSAGE_DELIMITER, outgoingMessageDelimiter); + } + runner.setProperty(PutTCP.CONNECTION_PER_FLOWFILE, String.valueOf(connectionPerFlowFile)); + + if (expectValid) { + runner.assertValid(); + } else { + runner.assertNotValid(); + } + } + + private static Map<String, String> createSslProperties() { + final Map<String, String> map = new HashMap<>(); + map.put(StandardSSLContextService.KEYSTORE.getName(), "src/test/resources/localhost-ks.jks"); + map.put(StandardSSLContextService.KEYSTORE_PASSWORD.getName(), "localtest"); + map.put(StandardSSLContextService.KEYSTORE_TYPE.getName(), "JKS"); + map.put(StandardSSLContextService.TRUSTSTORE.getName(), "src/test/resources/localhost-ts.jks"); + map.put(StandardSSLContextService.TRUSTSTORE_PASSWORD.getName(), "localtest"); + map.put(StandardSSLContextService.TRUSTSTORE_TYPE.getName(), "JKS"); + return map; + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/4a9d3c97/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/TCPTestServer.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/TCPTestServer.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/TCPTestServer.java new file mode 100644 index 0000000..01d492e --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/TCPTestServer.java @@ -0,0 +1,183 @@ +/* + * 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.util; + +import org.apache.nifi.security.util.SslContextFactory; + +import javax.net.ServerSocketFactory; +import javax.net.ssl.SSLContext; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; + +public class TCPTestServer implements Runnable { + + private final InetAddress ipAddress; + private int port; + private final String messageDelimiter; + private volatile ServerSocket serverSocket; + private final ArrayBlockingQueue<List<Byte>> recvQueue; + private volatile Socket connectionSocket; + public final static String DEFAULT_MESSAGE_DELIMITER = "\n"; + private volatile int totalNumConnections = 0; + + public TCPTestServer(final InetAddress ipAddress, final ArrayBlockingQueue<List<Byte>> recvQueue) { + this(ipAddress, recvQueue, DEFAULT_MESSAGE_DELIMITER); + } + + public TCPTestServer(final InetAddress ipAddress, final ArrayBlockingQueue<List<Byte>> recvQueue, final String messageDelimiter) { + this.ipAddress = ipAddress; + this.recvQueue = recvQueue; + this.messageDelimiter = messageDelimiter; + } + + public synchronized void startServer(boolean ssl) throws Exception { + if (!isServerRunning()) { + if(ssl){ + final SSLContext sslCtx = SslContextFactory.createSslContext("src/test/resources/localhost-ks.jks","localtest".toCharArray(), "JKS", "src/test/resources/localhost-ts.jks", + "localtest".toCharArray(), "JKS", SslContextFactory.ClientAuth.REQUIRED, "TLS"); + + ServerSocketFactory sslSocketFactory = sslCtx.getServerSocketFactory(); + serverSocket = sslSocketFactory.createServerSocket(0, 0, ipAddress); + } else { + serverSocket = new ServerSocket(0, 0, ipAddress); + } + Thread t = new Thread(this); + t.setName(this.getClass().getSimpleName()); + t.start(); + port = serverSocket.getLocalPort(); + } + } + + public synchronized void shutdown() { + shutdownConnection(); + shutdownServer(); + } + + public synchronized void shutdownServer() { + if (isServerRunning()) { + try { + serverSocket.close(); + } catch (IOException ioe) { + // Do Nothing. + } + } + } + + public synchronized void shutdownConnection() { + if (isConnected()) { + try { + connectionSocket.close(); + } catch (IOException ioe) { + // Do Nothing. + } + } + } + + public int getPort(){ + return port; + } + + private void storeReceivedMessage(final List<Byte> message) { + recvQueue.add(message); + } + + private boolean isServerRunning() { + return serverSocket != null && !serverSocket.isClosed(); + } + + private boolean isConnected() { + return connectionSocket != null && !connectionSocket.isClosed(); + } + + public List<Byte> getReceivedMessage() { + return recvQueue.poll(); + } + + public int getTotalNumConnections() { + return totalNumConnections; + } + + protected boolean isDelimiterPresent(final List<Byte> message) { + if (messageDelimiter != null && message.size() >= messageDelimiter.length()) { + for (int i = 1; i <= messageDelimiter.length(); i++) { + if (message.get(message.size() - i).byteValue() == messageDelimiter.charAt(messageDelimiter.length() - i)) { + if (i == messageDelimiter.length()) { + return true; + } + } else { + break; + } + } + } + return false; + } + + protected boolean removeDelimiter(final List<Byte> message) { + if (isDelimiterPresent(message)) { + final int messageSize = message.size(); + for (int i = 1; i <= messageDelimiter.length(); i++) { + message.remove(messageSize - i); + } + return true; + } + + return false; + } + + @Override + public void run() { + try { + while (isServerRunning()) { + connectionSocket = serverSocket.accept(); + totalNumConnections++; + InputStream in = connectionSocket.getInputStream(); + while (isConnected()) { + final List<Byte> message = new ArrayList<Byte>(); + while (true) { + final int c = in.read(); + if (c < 0) { + if (!message.isEmpty()) { + storeReceivedMessage(message); + } + shutdownConnection(); + break; + } + + message.add((byte) c); + + if (removeDelimiter(message)) { + storeReceivedMessage(message); + break; + } + } + } + } + } catch (Exception e) { + // Do Nothing + } finally { + shutdownConnection(); + shutdownServer(); + } + + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/4a9d3c97/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/TestPutTCPCommon.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/TestPutTCPCommon.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/TestPutTCPCommon.java new file mode 100644 index 0000000..ed52c35 --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/TestPutTCPCommon.java @@ -0,0 +1,308 @@ +/* + * 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.util; + +import org.apache.commons.lang3.ArrayUtils; +import org.apache.nifi.processors.standard.PutTCP; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.util.TestRunner; +import org.apache.nifi.util.TestRunners; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.net.InetAddress; +import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +public abstract class TestPutTCPCommon { + private final static String TCP_SERVER_ADDRESS = "127.0.0.1"; + private final static String UNKNOWN_HOST = "fgdsfgsdffd"; + private final static String INVALID_IP_ADDRESS = "300.300.300.300"; + private final static int MIN_INVALID_PORT = 0; + private final static int MIN_VALID_PORT = 1; + private final static int MAX_VALID_PORT = 65535; + private final static int MAX_INVALID_PORT = 65536; + private final static int BUFFER_SIZE = 1024; + private final static int VALID_LARGE_FILE_SIZE = 32768; + private final static int VALID_SMALL_FILE_SIZE = 64; + private final static int LOAD_TEST_ITERATIONS = 500; + private final static int LOAD_TEST_THREAD_COUNT = 1; + private final static int DEFAULT_ITERATIONS = 1; + private final static int DEFAULT_THREAD_COUNT = 1; + private final static char CONTENT_CHAR = 'x'; + private final static int DATA_WAIT_PERIOD = 1000; + private final static int DEFAULT_TEST_TIMEOUT_PERIOD = 10000; + private final static int LONG_TEST_TIMEOUT_PERIOD = 100000; + private final static String OUTGOING_MESSAGE_DELIMITER = "\n"; + private final static String OUTGOING_MESSAGE_DELIMITER_MULTI_CHAR = "{delimiter}\r\n"; + + private TCPTestServer server; + private int tcp_server_port; + private ArrayBlockingQueue<List<Byte>> recvQueue; + + public boolean ssl; + public TestRunner runner; + + // Test Data + private final static String[] EMPTY_FILE = { "" }; + private final static String[] VALID_FILES = { "abcdefghijklmnopqrstuvwxyz", "zyxwvutsrqponmlkjihgfedcba", "12345678", "343424222", "!@£$%^&*()_+:|{}[];\\" }; + + @Before + public void setup() throws Exception { + recvQueue = new ArrayBlockingQueue<List<Byte>>(BUFFER_SIZE); + runner = TestRunners.newTestRunner(PutTCP.class); + } + + private synchronized TCPTestServer createTestServer(final String address, final ArrayBlockingQueue<List<Byte>> recvQueue, final String delimiter) throws Exception { + TCPTestServer server = new TCPTestServer(InetAddress.getByName(address), recvQueue, delimiter); + server.startServer(ssl); + tcp_server_port = server.getPort(); + return server; + } + + @After + public void cleanup() throws Exception { + runner.shutdown(); + removeTestServer(server); + } + + private void removeTestServer(TCPTestServer server) { + if (server != null) { + server.shutdown(); + server = null; + } + } + + @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) + public void testValidFiles() throws Exception { + server = createTestServer(TCP_SERVER_ADDRESS, recvQueue, OUTGOING_MESSAGE_DELIMITER); + configureProperties(TCP_SERVER_ADDRESS, tcp_server_port, OUTGOING_MESSAGE_DELIMITER, false, true); + sendTestData(VALID_FILES); + checkReceivedAllData(recvQueue, VALID_FILES); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, 1); + } + + @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) + public void testPruneSenders() throws Exception { + server = createTestServer(TCP_SERVER_ADDRESS, recvQueue, OUTGOING_MESSAGE_DELIMITER); + configureProperties(TCP_SERVER_ADDRESS, tcp_server_port, OUTGOING_MESSAGE_DELIMITER, false, true); + sendTestData(VALID_FILES); + Thread.sleep(10); + checkRelationships(VALID_FILES.length, 0); + checkReceivedAllData(recvQueue, VALID_FILES); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, 1); + runner.setProperty(PutTCP.IDLE_EXPIRATION, "1 second"); + Thread.sleep(2000); + runner.run(1, false, false); + runner.clearTransferState(); + sendTestData(VALID_FILES); + checkReceivedAllData(recvQueue, VALID_FILES); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, 2); + } + + @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) + public void testMultiCharDelimiter() throws Exception { + server = createTestServer(TCP_SERVER_ADDRESS, recvQueue, OUTGOING_MESSAGE_DELIMITER_MULTI_CHAR); + configureProperties(TCP_SERVER_ADDRESS, tcp_server_port, OUTGOING_MESSAGE_DELIMITER_MULTI_CHAR, false, true); + sendTestData(VALID_FILES); + checkReceivedAllData(recvQueue, VALID_FILES); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, 1); + } + + @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) + public void testConnectionPerFlowFile() throws Exception { + server = createTestServer(TCP_SERVER_ADDRESS, recvQueue, OUTGOING_MESSAGE_DELIMITER); + configureProperties(TCP_SERVER_ADDRESS, tcp_server_port, OUTGOING_MESSAGE_DELIMITER, true, true); + sendTestData(VALID_FILES); + checkReceivedAllData(recvQueue, VALID_FILES); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, VALID_FILES.length); + } + + @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) + public void testConnectionFailure() throws Exception { + server = createTestServer(TCP_SERVER_ADDRESS, recvQueue, OUTGOING_MESSAGE_DELIMITER); + configureProperties(TCP_SERVER_ADDRESS, tcp_server_port, OUTGOING_MESSAGE_DELIMITER, false, true); + sendTestData(VALID_FILES); + checkReceivedAllData(recvQueue, VALID_FILES); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, 1); + removeTestServer(server); + runner.clearTransferState(); + sendTestData(VALID_FILES); + Thread.sleep(10); + checkNoDataReceived(recvQueue); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, 1); + server = createTestServer(TCP_SERVER_ADDRESS, recvQueue, OUTGOING_MESSAGE_DELIMITER); + configureProperties(TCP_SERVER_ADDRESS, tcp_server_port, OUTGOING_MESSAGE_DELIMITER, false, true); + sendTestData(VALID_FILES); + checkReceivedAllData(recvQueue, VALID_FILES); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, 1); + } + + @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) + public void testEmptyFile() throws Exception { + server = createTestServer(TCP_SERVER_ADDRESS, recvQueue, OUTGOING_MESSAGE_DELIMITER); + configureProperties(TCP_SERVER_ADDRESS, tcp_server_port, OUTGOING_MESSAGE_DELIMITER, false, true); + sendTestData(EMPTY_FILE); + Thread.sleep(10); + checkRelationships(EMPTY_FILE.length, 0); + checkEmptyMessageReceived(recvQueue); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, 1); + } + + @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) + public void testlargeValidFile() throws Exception { + server = createTestServer(TCP_SERVER_ADDRESS, recvQueue, OUTGOING_MESSAGE_DELIMITER); + configureProperties(TCP_SERVER_ADDRESS, tcp_server_port, OUTGOING_MESSAGE_DELIMITER, true, true); + final String[] testData = createContent(VALID_LARGE_FILE_SIZE); + sendTestData(testData); + checkReceivedAllData(recvQueue, testData); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, testData.length); + } + + @Test(timeout = LONG_TEST_TIMEOUT_PERIOD) + public void testInvalidIPAddress() throws Exception { + server = createTestServer(TCP_SERVER_ADDRESS, recvQueue, OUTGOING_MESSAGE_DELIMITER); + configureProperties(INVALID_IP_ADDRESS, tcp_server_port, OUTGOING_MESSAGE_DELIMITER, false, true); + sendTestData(VALID_FILES); + Thread.sleep(10); + checkRelationships(0, VALID_FILES.length); + checkNoDataReceived(recvQueue); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, 0); + } + + @Test(timeout = LONG_TEST_TIMEOUT_PERIOD) + public void testUnknownHostname() throws Exception { + server = createTestServer(TCP_SERVER_ADDRESS, recvQueue, OUTGOING_MESSAGE_DELIMITER); + configureProperties(UNKNOWN_HOST, tcp_server_port, OUTGOING_MESSAGE_DELIMITER, false, true); + sendTestData(VALID_FILES); + Thread.sleep(10); + checkRelationships(0, VALID_FILES.length); + checkNoDataReceived(recvQueue); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, 0); + } + + @Test(timeout = DEFAULT_TEST_TIMEOUT_PERIOD) + public void testInvalidPort() throws Exception { + configureProperties(UNKNOWN_HOST, MIN_INVALID_PORT, OUTGOING_MESSAGE_DELIMITER, false, false); + configureProperties(UNKNOWN_HOST, MIN_VALID_PORT, OUTGOING_MESSAGE_DELIMITER, false, true); + configureProperties(UNKNOWN_HOST, MAX_VALID_PORT, OUTGOING_MESSAGE_DELIMITER, false, true); + configureProperties(UNKNOWN_HOST, MAX_INVALID_PORT, OUTGOING_MESSAGE_DELIMITER, false, false); + } + + @Test(timeout = LONG_TEST_TIMEOUT_PERIOD) + public void testLoadTest() throws Exception { + server = createTestServer(TCP_SERVER_ADDRESS, recvQueue, OUTGOING_MESSAGE_DELIMITER); + Thread.sleep(1000); + final String[] testData = createContent(VALID_SMALL_FILE_SIZE); + configureProperties(TCP_SERVER_ADDRESS, tcp_server_port, OUTGOING_MESSAGE_DELIMITER, false, true); + sendTestData(testData, LOAD_TEST_ITERATIONS, LOAD_TEST_THREAD_COUNT); + checkReceivedAllData(recvQueue, testData, LOAD_TEST_ITERATIONS); + checkInputQueueIsEmpty(); + checkTotalNumConnections(server, 1); + } + + private void checkTotalNumConnections(final TCPTestServer server, final int expectedTotalNumConnections) { + assertEquals(expectedTotalNumConnections, server.getTotalNumConnections()); + } + + public abstract void configureProperties(final String host, final int port, final String outgoingMessageDelimiter, final boolean connectionPerFlowFile, + final boolean expectValid) throws InitializationException; + + private void sendTestData(final String[] testData) { + sendTestData(testData, DEFAULT_ITERATIONS, DEFAULT_THREAD_COUNT); + } + + private void sendTestData(final String[] testData, final int iterations, final int threadCount) { + runner.setThreadCount(threadCount); + for (int i = 0; i < iterations; i++) { + for (String item : testData) { + runner.enqueue(item.getBytes()); + } + runner.run(testData.length, false, i == 0 ? true : false); + } + } + + private void checkRelationships(final int successCount, final int failedCount) { + runner.assertTransferCount(PutTCP.REL_SUCCESS, successCount); + runner.assertTransferCount(PutTCP.REL_FAILURE, failedCount); + } + + private void checkNoDataReceived(final ArrayBlockingQueue<List<Byte>> recvQueue) throws Exception { + Thread.sleep(DATA_WAIT_PERIOD); + assertNull(recvQueue.poll()); + } + + private void checkEmptyMessageReceived(final ArrayBlockingQueue<List<Byte>> recvQueue) throws Exception { + Thread.sleep(DATA_WAIT_PERIOD); + assertEquals(0, recvQueue.poll().size()); + } + + private void checkInputQueueIsEmpty() { + runner.assertQueueEmpty(); + } + + private void checkReceivedAllData(final ArrayBlockingQueue<List<Byte>> recvQueue, final String[] sentData) throws Exception { + checkReceivedAllData(recvQueue, sentData, DEFAULT_ITERATIONS); + } + + private void checkReceivedAllData(final ArrayBlockingQueue<List<Byte>> recvQueue, final String[] sentData, final int iterations) throws Exception { + // check each sent FlowFile was successfully sent and received. + for (int i = 0; i < iterations; i++) { + for (String item : sentData) { + List<Byte> message = recvQueue.take(); + assertNotNull(message); + Byte[] messageBytes = new Byte[message.size()]; + assertArrayEquals(item.getBytes(), ArrayUtils.toPrimitive(message.toArray(messageBytes))); + } + } + + runner.assertTransferCount(PutTCP.REL_SUCCESS, sentData.length * iterations); + runner.clearTransferState(); + + // Check that we have no unexpected extra data. + assertNull(recvQueue.poll()); + } + + private String[] createContent(final int size) { + final char[] content = new char[size]; + + for (int i = 0; i < size; i++) { + content[i] = CONTENT_CHAR; + } + + return new String[] { new String(content) }; + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/4a9d3c97/pom.xml ---------------------------------------------------------------------- diff --git a/pom.xml b/pom.xml index 039f12b..e99e0b1 100644 --- a/pom.xml +++ b/pom.xml @@ -887,6 +887,11 @@ language governing permissions and limitations under the License. --> </dependency> <dependency> <groupId>org.apache.nifi</groupId> + <artifactId>nifi-ssl-context-service-api</artifactId> + <version>0.7.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> <artifactId>nifi-distributed-cache-services-nar</artifactId> <version>0.7.0-SNAPSHOT</version> <type>nar</type>
