http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/d29a2d68/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java index 0a076ca..21e6b4c 100644 --- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java +++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java @@ -56,54 +56,54 @@ public class FTPTransfer implements FileTransfer { public static final String PROXY_TYPE_HTTP = Proxy.Type.HTTP.name(); public static final String PROXY_TYPE_SOCKS = Proxy.Type.SOCKS.name(); - public static final PropertyDescriptor CONNECTION_MODE = new PropertyDescriptor.Builder(). - name("Connection Mode"). - description("The FTP Connection Mode"). - allowableValues(CONNECTION_MODE_ACTIVE, CONNECTION_MODE_PASSIVE). - defaultValue(CONNECTION_MODE_PASSIVE). - build(); - public static final PropertyDescriptor TRANSFER_MODE = new PropertyDescriptor.Builder(). - name("Transfer Mode"). - description("The FTP Transfer Mode"). - allowableValues(TRANSFER_MODE_BINARY, TRANSFER_MODE_ASCII). - defaultValue(TRANSFER_MODE_BINARY). - build(); - public static final PropertyDescriptor PORT = new PropertyDescriptor.Builder(). - name("Port"). - description("The port that the remote system is listening on for file transfers"). - addValidator(StandardValidators.PORT_VALIDATOR). - required(true). - defaultValue("21"). - build(); - public static final PropertyDescriptor PROXY_TYPE = new PropertyDescriptor.Builder(). - name("Proxy Type"). - description("Proxy type used for file transfers"). - allowableValues(PROXY_TYPE_DIRECT, PROXY_TYPE_HTTP, PROXY_TYPE_SOCKS). - defaultValue(PROXY_TYPE_DIRECT). - build(); - public static final PropertyDescriptor PROXY_HOST = new PropertyDescriptor.Builder(). - name("Proxy Host"). - description("The fully qualified hostname or IP address of the proxy server"). - addValidator(StandardValidators.NON_EMPTY_VALIDATOR). - build(); - public static final PropertyDescriptor PROXY_PORT = new PropertyDescriptor.Builder(). - name("Proxy Port"). - description("The port of the proxy server"). - addValidator(StandardValidators.PORT_VALIDATOR). - build(); - public static final PropertyDescriptor HTTP_PROXY_USERNAME = new PropertyDescriptor.Builder(). - name("Http Proxy Username"). - description("Http Proxy Username"). - addValidator(StandardValidators.NON_EMPTY_VALIDATOR). - required(false). - build(); - public static final PropertyDescriptor HTTP_PROXY_PASSWORD = new PropertyDescriptor.Builder(). - name("Http Proxy Password"). - description("Http Proxy Password"). - addValidator(StandardValidators.NON_EMPTY_VALIDATOR). - required(false). - sensitive(true). - build(); + public static final PropertyDescriptor CONNECTION_MODE = new PropertyDescriptor.Builder() + .name("Connection Mode") + .description("The FTP Connection Mode") + .allowableValues(CONNECTION_MODE_ACTIVE, CONNECTION_MODE_PASSIVE) + .defaultValue(CONNECTION_MODE_PASSIVE) + .build(); + public static final PropertyDescriptor TRANSFER_MODE = new PropertyDescriptor.Builder() + .name("Transfer Mode") + .description("The FTP Transfer Mode") + .allowableValues(TRANSFER_MODE_BINARY, TRANSFER_MODE_ASCII) + .defaultValue(TRANSFER_MODE_BINARY) + .build(); + public static final PropertyDescriptor PORT = new PropertyDescriptor.Builder() + .name("Port") + .description("The port that the remote system is listening on for file transfers") + .addValidator(StandardValidators.PORT_VALIDATOR) + .required(true) + .defaultValue("21") + .build(); + public static final PropertyDescriptor PROXY_TYPE = new PropertyDescriptor.Builder() + .name("Proxy Type") + .description("Proxy type used for file transfers") + .allowableValues(PROXY_TYPE_DIRECT, PROXY_TYPE_HTTP, PROXY_TYPE_SOCKS) + .defaultValue(PROXY_TYPE_DIRECT) + .build(); + public static final PropertyDescriptor PROXY_HOST = new PropertyDescriptor.Builder() + .name("Proxy Host") + .description("The fully qualified hostname or IP address of the proxy server") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + public static final PropertyDescriptor PROXY_PORT = new PropertyDescriptor.Builder() + .name("Proxy Port") + .description("The port of the proxy server") + .addValidator(StandardValidators.PORT_VALIDATOR) + .build(); + public static final PropertyDescriptor HTTP_PROXY_USERNAME = new PropertyDescriptor.Builder() + .name("Http Proxy Username") + .description("Http Proxy Username") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .required(false) + .build(); + public static final PropertyDescriptor HTTP_PROXY_PASSWORD = new PropertyDescriptor.Builder() + .name("Http Proxy Password") + .description("Http Proxy Password") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .required(false) + .sensitive(true) + .build(); private final ProcessorLog logger; @@ -135,8 +135,7 @@ public class FTPTransfer implements FileTransfer { client.disconnect(); } } catch (final Exception ex) { - logger.warn("Failed to close FTPClient due to {}", new Object[]{ex. - toString()}, ex); + logger.warn("Failed to close FTPClient due to {}", new Object[]{ex.toString()}, ex); } client = null; } @@ -149,13 +148,9 @@ public class FTPTransfer implements FileTransfer { @Override public List<FileInfo> getListing() throws IOException { - final String path = ctx.getProperty(FileTransfer.REMOTE_PATH). - evaluateAttributeExpressions(). - getValue(); + final String path = ctx.getProperty(FileTransfer.REMOTE_PATH).evaluateAttributeExpressions().getValue(); final int depth = 0; - final int maxResults = ctx. - getProperty(FileTransfer.REMOTE_POLL_BATCH_SIZE). - asInteger(); + final int maxResults = ctx.getProperty(FileTransfer.REMOTE_POLL_BATCH_SIZE).asInteger(); return getListing(path, depth, maxResults); } @@ -166,43 +161,27 @@ public class FTPTransfer implements FileTransfer { } if (depth >= 100) { - logger. - warn(this + " had to stop recursively searching directories at a recursive depth of " + depth + " to avoid memory issues"); + logger.warn(this + " had to stop recursively searching directories at a recursive depth of " + depth + " to avoid memory issues"); return listing; } - final boolean ignoreDottedFiles = ctx. - getProperty(FileTransfer.IGNORE_DOTTED_FILES). - asBoolean(); - final boolean recurse = ctx.getProperty(FileTransfer.RECURSIVE_SEARCH). - asBoolean(); - final String fileFilterRegex = ctx. - getProperty(FileTransfer.FILE_FILTER_REGEX). - getValue(); - final Pattern pattern = (fileFilterRegex == null) ? null : Pattern. - compile(fileFilterRegex); - final String pathFilterRegex = ctx. - getProperty(FileTransfer.PATH_FILTER_REGEX). - getValue(); - final Pattern pathPattern = (!recurse || pathFilterRegex == null) ? null : Pattern. - compile(pathFilterRegex); - final String remotePath = ctx.getProperty(FileTransfer.REMOTE_PATH). - evaluateAttributeExpressions(). - getValue(); + final boolean ignoreDottedFiles = ctx.getProperty(FileTransfer.IGNORE_DOTTED_FILES).asBoolean(); + final boolean recurse = ctx.getProperty(FileTransfer.RECURSIVE_SEARCH).asBoolean(); + final String fileFilterRegex = ctx.getProperty(FileTransfer.FILE_FILTER_REGEX).getValue(); + final Pattern pattern = (fileFilterRegex == null) ? null : Pattern.compile(fileFilterRegex); + final String pathFilterRegex = ctx.getProperty(FileTransfer.PATH_FILTER_REGEX).getValue(); + final Pattern pathPattern = (!recurse || pathFilterRegex == null) ? null : Pattern.compile(pathFilterRegex); + final String remotePath = ctx.getProperty(FileTransfer.REMOTE_PATH).evaluateAttributeExpressions().getValue(); // check if this directory path matches the PATH_FILTER_REGEX boolean pathFilterMatches = true; if (pathPattern != null) { Path reldir = path == null ? Paths.get(".") : Paths.get(path); if (remotePath != null) { - reldir = Paths.get(remotePath). - relativize(reldir); + reldir = Paths.get(remotePath).relativize(reldir); } - if (reldir != null && !reldir.toString(). - isEmpty()) { - if (!pathPattern.matcher(reldir.toString(). - replace("\\", "/")). - matches()) { + if (reldir != null && !reldir.toString().isEmpty()) { + if (!pathPattern.matcher(reldir.toString().replace("\\", "/")).matches()) { pathFilterMatches = false; } } @@ -213,14 +192,12 @@ public class FTPTransfer implements FileTransfer { int count = 0; final FTPFile[] files; - if (path == null || path.trim(). - isEmpty()) { + if (path == null || path.trim().isEmpty()) { files = client.listFiles("."); } else { files = client.listFiles(path); } - if (files.length == 0 && path != null && !path.trim(). - isEmpty()) { + if (files.length == 0 && path != null && !path.trim().isEmpty()) { // throw exception if directory doesn't exist final boolean cdSuccessful = setWorkingDirectory(path); if (!cdSuccessful) { @@ -239,24 +216,20 @@ public class FTPTransfer implements FileTransfer { } final File newFullPath = new File(path, filename); - final String newFullForwardPath = newFullPath.getPath(). - replace("\\", "/"); + final String newFullForwardPath = newFullPath.getPath().replace("\\", "/"); if (recurse && file.isDirectory()) { try { - listing. - addAll(getListing(newFullForwardPath, depth + 1, maxResults - count)); + listing.addAll(getListing(newFullForwardPath, depth + 1, maxResults - count)); } catch (final IOException e) { - logger. - error("Unable to get listing from " + newFullForwardPath + "; skipping this subdirectory"); + logger.error("Unable to get listing from " + newFullForwardPath + "; skipping this subdirectory"); } } // if is not a directory and is not a link and it matches // FILE_FILTER_REGEX - then let's add it if (!file.isDirectory() && !file.isSymbolicLink() && pathFilterMatches) { - if (pattern == null || pattern.matcher(filename). - matches()) { + if (pattern == null || pattern.matcher(filename).matches()) { listing.add(newFileInfo(file, path)); count++; } @@ -275,38 +248,27 @@ public class FTPTransfer implements FileTransfer { return null; } final File newFullPath = new File(path, file.getName()); - final String newFullForwardPath = newFullPath.getPath(). - replace("\\", "/"); + final String newFullForwardPath = newFullPath.getPath().replace("\\", "/"); StringBuilder perms = new StringBuilder(); - perms.append(file. - hasPermission(FTPFile.USER_ACCESS, FTPFile.READ_PERMISSION) ? "r" : "-"); - perms.append(file. - hasPermission(FTPFile.USER_ACCESS, FTPFile.WRITE_PERMISSION) ? "w" : "-"); - perms.append(file. - hasPermission(FTPFile.USER_ACCESS, FTPFile.EXECUTE_PERMISSION) ? "x" : "-"); - perms.append(file. - hasPermission(FTPFile.GROUP_ACCESS, FTPFile.READ_PERMISSION) ? "r" : "-"); - perms.append(file. - hasPermission(FTPFile.GROUP_ACCESS, FTPFile.WRITE_PERMISSION) ? "w" : "-"); - perms.append(file. - hasPermission(FTPFile.GROUP_ACCESS, FTPFile.EXECUTE_PERMISSION) ? "x" : "-"); - perms.append(file. - hasPermission(FTPFile.WORLD_ACCESS, FTPFile.READ_PERMISSION) ? "r" : "-"); - perms.append(file. - hasPermission(FTPFile.WORLD_ACCESS, FTPFile.WRITE_PERMISSION) ? "w" : "-"); - perms.append(file. - hasPermission(FTPFile.WORLD_ACCESS, FTPFile.EXECUTE_PERMISSION) ? "x" : "-"); + perms.append(file.hasPermission(FTPFile.USER_ACCESS, FTPFile.READ_PERMISSION) ? "r" : "-"); + perms.append(file.hasPermission(FTPFile.USER_ACCESS, FTPFile.WRITE_PERMISSION) ? "w" : "-"); + perms.append(file.hasPermission(FTPFile.USER_ACCESS, FTPFile.EXECUTE_PERMISSION) ? "x" : "-"); + perms.append(file.hasPermission(FTPFile.GROUP_ACCESS, FTPFile.READ_PERMISSION) ? "r" : "-"); + perms.append(file.hasPermission(FTPFile.GROUP_ACCESS, FTPFile.WRITE_PERMISSION) ? "w" : "-"); + perms.append(file.hasPermission(FTPFile.GROUP_ACCESS, FTPFile.EXECUTE_PERMISSION) ? "x" : "-"); + perms.append(file.hasPermission(FTPFile.WORLD_ACCESS, FTPFile.READ_PERMISSION) ? "r" : "-"); + perms.append(file.hasPermission(FTPFile.WORLD_ACCESS, FTPFile.WRITE_PERMISSION) ? "w" : "-"); + perms.append(file.hasPermission(FTPFile.WORLD_ACCESS, FTPFile.EXECUTE_PERMISSION) ? "x" : "-"); FileInfo.Builder builder = new FileInfo.Builder() - .filename(file.getName()). - fullPathFileName(newFullForwardPath). - directory(file.isDirectory()). - size(file.getSize()). - lastModifiedTime(file.getTimestamp(). - getTimeInMillis()). - permissions(perms.toString()). - owner(file.getUser()). - group(file.getGroup()); + .filename(file.getName()) + .fullPathFileName(newFullForwardPath) + .directory(file.isDirectory()) + .size(file.getSize()) + .lastModifiedTime(file.getTimestamp().getTimeInMillis()) + .permissions(perms.toString()) + .owner(file.getUser()) + .group(file.getGroup()); return builder.build(); } @@ -359,20 +321,16 @@ public class FTPTransfer implements FileTransfer { @Override public void ensureDirectoryExists(final FlowFile flowFile, final File directoryName) throws IOException { - if (directoryName.getParent() != null && !directoryName.getParentFile(). - equals(new File(File.separator))) { + if (directoryName.getParent() != null && !directoryName.getParentFile().equals(new File(File.separator))) { ensureDirectoryExists(flowFile, directoryName.getParentFile()); } - final String remoteDirectory = directoryName.getAbsolutePath(). - replace("\\", "/"). - replaceAll("^.\\:", ""); + final String remoteDirectory = directoryName.getAbsolutePath().replace("\\", "/").replaceAll("^.\\:", ""); final FTPClient client = getClient(flowFile); final boolean cdSuccessful = setWorkingDirectory(remoteDirectory); if (!cdSuccessful) { - logger. - debug("Remote Directory {} does not exist; creating it", new Object[]{remoteDirectory}); + logger.debug("Remote Directory {} does not exist; creating it", new Object[]{remoteDirectory}); if (client.makeDirectory(remoteDirectory)) { logger.debug("Created {}", new Object[]{remoteDirectory}); } else { @@ -410,26 +368,19 @@ public class FTPTransfer implements FileTransfer { fullPath = workingDir.endsWith("/") ? workingDir + filename : workingDir + "/" + filename; } - String tempFilename = ctx.getProperty(TEMP_FILENAME). - evaluateAttributeExpressions(flowFile). - getValue(); + String tempFilename = ctx.getProperty(TEMP_FILENAME).evaluateAttributeExpressions(flowFile).getValue(); if (tempFilename == null) { - final boolean dotRename = ctx.getProperty(DOT_RENAME). - asBoolean(); + final boolean dotRename = ctx.getProperty(DOT_RENAME).asBoolean(); tempFilename = dotRename ? "." + filename : filename; } final boolean storeSuccessful = client.storeFile(tempFilename, content); if (!storeSuccessful) { - throw new IOException("Failed to store file " + tempFilename + " to " + fullPath + " due to: " + client. - getReplyString()); + throw new IOException("Failed to store file " + tempFilename + " to " + fullPath + " due to: " + client.getReplyString()); } - final String lastModifiedTime = ctx.getProperty(LAST_MODIFIED_TIME). - evaluateAttributeExpressions(flowFile). - getValue(); - if (lastModifiedTime != null && !lastModifiedTime.trim(). - isEmpty()) { + final String lastModifiedTime = ctx.getProperty(LAST_MODIFIED_TIME).evaluateAttributeExpressions(flowFile).getValue(); + if (lastModifiedTime != null && !lastModifiedTime.trim().isEmpty()) { try { final DateFormat informat = new SimpleDateFormat(FILE_MODIFY_DATE_ATTR_FORMAT, Locale.US); final Date fileModifyTime = informat.parse(lastModifiedTime); @@ -437,43 +388,32 @@ public class FTPTransfer implements FileTransfer { final String time = outformat.format(fileModifyTime); if (!client.setModificationTime(tempFilename, time)) { // FTP server probably doesn't support MFMT command - logger. - warn("Could not set lastModifiedTime on {} to {}", new Object[]{flowFile, lastModifiedTime}); + logger.warn("Could not set lastModifiedTime on {} to {}", new Object[]{flowFile, lastModifiedTime}); } } catch (final Exception e) { - logger. - error("Failed to set lastModifiedTime on {} to {} due to {}", new Object[]{flowFile, lastModifiedTime, e}); + logger.error("Failed to set lastModifiedTime on {} to {} due to {}", new Object[]{flowFile, lastModifiedTime, e}); } } - final String permissions = ctx.getProperty(PERMISSIONS). - evaluateAttributeExpressions(flowFile). - getValue(); - if (permissions != null && !permissions.trim(). - isEmpty()) { + final String permissions = ctx.getProperty(PERMISSIONS).evaluateAttributeExpressions(flowFile).getValue(); + if (permissions != null && !permissions.trim().isEmpty()) { try { int perms = numberPermissions(permissions); if (perms >= 0) { - if (!client.sendSiteCommand("chmod " + Integer. - toOctalString(perms) + " " + tempFilename)) { - logger. - warn("Could not set permission on {} to {}", new Object[]{flowFile, permissions}); + if (!client.sendSiteCommand("chmod " + Integer.toOctalString(perms) + " " + tempFilename)) { + logger.warn("Could not set permission on {} to {}", new Object[]{flowFile, permissions}); } } } catch (final Exception e) { - logger. - error("Failed to set permission on {} to {} due to {}", new Object[]{flowFile, permissions, e}); + logger.error("Failed to set permission on {} to {} due to {}", new Object[]{flowFile, permissions, e}); } } if (!filename.equals(tempFilename)) { try { - logger. - debug("Renaming remote path from {} to {} for {}", new Object[]{tempFilename, filename, flowFile}); - final boolean renameSuccessful = client. - rename(tempFilename, filename); + logger.debug("Renaming remote path from {} to {} for {}", new Object[]{tempFilename, filename, flowFile}); + final boolean renameSuccessful = client.rename(tempFilename, filename); if (!renameSuccessful) { - throw new IOException("Failed to rename temporary file " + tempFilename + " to " + fullPath + " due to: " + client. - getReplyString()); + throw new IOException("Failed to rename temporary file " + tempFilename + " to " + fullPath + " due to: " + client.getReplyString()); } } catch (final IOException e) { try { @@ -495,8 +435,7 @@ public class FTPTransfer implements FileTransfer { setWorkingDirectory(path); } if (!client.deleteFile(remoteFileName)) { - throw new IOException("Failed to remove file " + remoteFileName + " due to " + client. - getReplyString()); + throw new IOException("Failed to remove file " + remoteFileName + " due to " + client.getReplyString()); } } @@ -505,8 +444,7 @@ public class FTPTransfer implements FileTransfer { final FTPClient client = getClient(null); final boolean success = client.removeDirectory(remoteDirectoryName); if (!success) { - throw new IOException("Failed to remove directory " + remoteDirectoryName + " due to " + client. - getReplyString()); + throw new IOException("Failed to remove directory " + remoteDirectoryName + " due to " + client.getReplyString()); } } @@ -525,14 +463,10 @@ public class FTPTransfer implements FileTransfer { if (!cmd.isEmpty()) { int result; result = client.sendCommand(cmd); - logger. - debug(this + " sent command to the FTP server: " + cmd + " for " + flowFile); - - if (FTPReply.isNegativePermanent(result) || FTPReply. - isNegativeTransient(result)) { - throw new IOException(this + " negative reply back from FTP server cmd: " - + cmd + " reply:" + result + ": " + client. - getReplyString() + " for " + flowFile); + logger.debug(this + " sent command to the FTP server: " + cmd + " for " + flowFile); + + if (FTPReply.isNegativePermanent(result) || FTPReply.isNegativeTransient(result)) { + throw new IOException(this + " negative reply back from FTP server cmd: " + cmd + " reply:" + result + ": " + client.getReplyString() + " for " + flowFile); } } } @@ -540,9 +474,7 @@ public class FTPTransfer implements FileTransfer { private FTPClient getClient(final FlowFile flowFile) throws IOException { if (client != null) { - String desthost = ctx.getProperty(HOSTNAME). - evaluateAttributeExpressions(flowFile). - getValue(); + String desthost = ctx.getProperty(HOSTNAME).evaluateAttributeExpressions(flowFile).getValue(); if (remoteHostName.equals(desthost)) { // destination matches so we can keep our current session resetWorkingDirectory(); @@ -553,38 +485,24 @@ public class FTPTransfer implements FileTransfer { } } - final Proxy.Type proxyType = Proxy.Type.valueOf(ctx. - getProperty(PROXY_TYPE). - getValue()); - final String proxyHost = ctx.getProperty(PROXY_HOST). - getValue(); - final Integer proxyPort = ctx.getProperty(PROXY_PORT). - asInteger(); + final Proxy.Type proxyType = Proxy.Type.valueOf(ctx.getProperty(PROXY_TYPE).getValue()); + final String proxyHost = ctx.getProperty(PROXY_HOST).getValue(); + final Integer proxyPort = ctx.getProperty(PROXY_PORT).asInteger(); FTPClient client; if (proxyType == Proxy.Type.HTTP) { - client = new FTPHTTPClient(proxyHost, proxyPort, ctx. - getProperty(HTTP_PROXY_USERNAME). - getValue(), ctx.getProperty(HTTP_PROXY_PASSWORD). - getValue()); + client = new FTPHTTPClient(proxyHost, proxyPort, ctx.getProperty(HTTP_PROXY_USERNAME).getValue(), ctx.getProperty(HTTP_PROXY_PASSWORD).getValue()); } else { client = new FTPClient(); if (proxyType == Proxy.Type.SOCKS) { - client. - setSocketFactory(new SocksProxySocketFactory(new Proxy(proxyType, new InetSocketAddress(proxyHost, proxyPort)))); + client.setSocketFactory(new SocksProxySocketFactory(new Proxy(proxyType, new InetSocketAddress(proxyHost, proxyPort)))); } } this.client = client; - client.setDataTimeout(ctx.getProperty(DATA_TIMEOUT). - asTimePeriod(TimeUnit.MILLISECONDS). - intValue()); - client.setDefaultTimeout(ctx.getProperty(CONNECTION_TIMEOUT). - asTimePeriod(TimeUnit.MILLISECONDS). - intValue()); + client.setDataTimeout(ctx.getProperty(DATA_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue()); + client.setDefaultTimeout(ctx.getProperty(CONNECTION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue()); client.setRemoteVerificationEnabled(false); - final String remoteHostname = ctx.getProperty(HOSTNAME). - evaluateAttributeExpressions(flowFile). - getValue(); + final String remoteHostname = ctx.getProperty(HOSTNAME).evaluateAttributeExpressions(flowFile).getValue(); this.remoteHostName = remoteHostname; InetAddress inetAddress = null; try { @@ -596,35 +514,26 @@ public class FTPTransfer implements FileTransfer { inetAddress = InetAddress.getByName(remoteHostname); } - client.connect(inetAddress, ctx.getProperty(PORT). - asInteger()); + client.connect(inetAddress, ctx.getProperty(PORT).asInteger()); this.closed = false; - client.setDataTimeout(ctx.getProperty(DATA_TIMEOUT). - asTimePeriod(TimeUnit.MILLISECONDS). - intValue()); - client.setSoTimeout(ctx.getProperty(CONNECTION_TIMEOUT). - asTimePeriod(TimeUnit.MILLISECONDS). - intValue()); - - final String username = ctx.getProperty(USERNAME). - getValue(); - final String password = ctx.getProperty(PASSWORD). - getValue(); + client.setDataTimeout(ctx.getProperty(DATA_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue()); + client.setSoTimeout(ctx.getProperty(CONNECTION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue()); + + final String username = ctx.getProperty(USERNAME).getValue(); + final String password = ctx.getProperty(PASSWORD).getValue(); final boolean loggedIn = client.login(username, password); if (!loggedIn) { throw new IOException("Could not login for user '" + username + "'"); } - final String connectionMode = ctx.getProperty(CONNECTION_MODE). - getValue(); + final String connectionMode = ctx.getProperty(CONNECTION_MODE).getValue(); if (connectionMode.equalsIgnoreCase(CONNECTION_MODE_ACTIVE)) { client.enterLocalActiveMode(); } else { client.enterLocalPassiveMode(); } - final String transferMode = ctx.getProperty(TRANSFER_MODE). - getValue(); + final String transferMode = ctx.getProperty(TRANSFER_MODE).getValue(); final int fileType = (transferMode.equalsIgnoreCase(TRANSFER_MODE_ASCII)) ? FTPClient.ASCII_FILE_TYPE : FTPClient.BINARY_FILE_TYPE; if (!client.setFileType(fileType)) { throw new IOException("Unable to set transfer mode to type " + transferMode); @@ -638,8 +547,7 @@ public class FTPTransfer implements FileTransfer { int number = -1; final Pattern rwxPattern = Pattern.compile("^[rwx-]{9}$"); final Pattern numPattern = Pattern.compile("\\d+"); - if (rwxPattern.matcher(perms). - matches()) { + if (rwxPattern.matcher(perms).matches()) { number = 0; if (perms.charAt(0) == 'r') { number |= 0x100; @@ -668,8 +576,7 @@ public class FTPTransfer implements FileTransfer { if (perms.charAt(8) == 'x') { number |= 0x1; } - } else if (numPattern.matcher(perms). - matches()) { + } else if (numPattern.matcher(perms).matches()) { try { number = Integer.parseInt(perms, 8); } catch (NumberFormatException ignore) {
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/d29a2d68/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPUtils.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPUtils.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPUtils.java index fa7722b..0e6a26f 100644 --- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPUtils.java +++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPUtils.java @@ -57,47 +57,30 @@ public class FTPUtils { public static final String NETWORK_SOCKET_TIMEOUT_KEY = "network.socket.timeout"; /** - * Creates a new FTPClient connected to an FTP server. The following - * properties must exist: + * Creates a new FTPClient connected to an FTP server. The following properties must exist: * <ul>Required Properties: - * <li>remote.host - The hostname or IP address of the FTP server to connect - * to</li> + * <li>remote.host - The hostname or IP address of the FTP server to connect to</li> * <li>remote.user - The username of the account to authenticate with</li> - * <li>remote.password = The password for the username to authenticate - * with</li> + * <li>remote.password = The password for the username to authenticate with</li> * </ul> * <ul>Optional Properties: - * <li>remote.port - The port on the FTP server to connect to. Defaults to - * FTP default.</li> - * <li>transfer.mode - The type of transfer for this connection ('ascii', - * 'binary'). Defaults to 'binary'</li> - * <li>connection.mode - The type of FTP connection to make ('active_local', - * 'passive_local'). Defaults to 'active_local'. In active_local the server - * initiates 'data connections' to the client where in passive_local the - * client initiates 'data connections' to the server.</li> - * <li>network.data.timeout - Default is 0. Sets the timeout in milliseconds - * for waiting to establish a new 'data connection' (not a control - * connection) when in ACTIVE_LOCAL mode. Also, this establishes the amount - * of time to wait on read calls on the data connection in either mode. A - * value of zero means do not timeout. Users should probably set a value - * here unless using very reliable communications links or else risk - * indefinite hangs that require a restart.</li> - * <li>network.socket.timeout - Default is 0. Sets the timeout in - * milliseconds to use when creating a new control channel socket and also a - * timeout to set when reading from a control socket. A value of zero means - * do not timeout. Users should probably set a value here unless using very - * reliable communications links or else risk indefinite hangs that require - * a restart.</li> + * <li>remote.port - The port on the FTP server to connect to. Defaults to FTP default.</li> + * <li>transfer.mode - The type of transfer for this connection ('ascii', 'binary'). Defaults to 'binary'</li> + * <li>connection.mode - The type of FTP connection to make ('active_local', 'passive_local'). Defaults to 'active_local'. In active_local the server initiates 'data connections' to the client + * where in passive_local the client initiates 'data connections' to the server.</li> + * <li>network.data.timeout - Default is 0. Sets the timeout in milliseconds for waiting to establish a new 'data connection' (not a control connection) when in ACTIVE_LOCAL mode. Also, this + * establishes the amount of time to wait on read calls on the data connection in either mode. A value of zero means do not timeout. Users should probably set a value here unless using very + * reliable communications links or else risk indefinite hangs that require a restart.</li> + * <li>network.socket.timeout - Default is 0. Sets the timeout in milliseconds to use when creating a new control channel socket and also a timeout to set when reading from a control socket. A + * value of zero means do not timeout. Users should probably set a value here unless using very reliable communications links or else risk indefinite hangs that require a restart.</li> * </ul> * * @param conf - * @param monitor if provided will be used to monitor FTP commands processed - * but may be null + * @param monitor if provided will be used to monitor FTP commands processed but may be null * @return FTPClient connected to FTP server as configured * @throws NullPointerException if either argument is null * @throws IllegalArgumentException if a required property is missing - * @throws NumberFormatException if any argument that must be an int cannot - * be converted to int + * @throws NumberFormatException if any argument that must be an int cannot be converted to int * @throws IOException if some problem occurs connecting to FTP server */ public static FTPClient connect(final FTPConfiguration conf, final ProtocolCommandListener monitor) throws IOException { @@ -112,11 +95,9 @@ public class FTPUtils { final String transferModeVal = conf.transferMode; final String transferMode = (null == transferModeVal) ? BINARY_TRANSFER_MODE : transferModeVal; final String networkDataTimeoutVal = conf.dataTimeout; - final int networkDataTimeout = (null == networkDataTimeoutVal) ? 0 : Integer. - parseInt(networkDataTimeoutVal); + final int networkDataTimeout = (null == networkDataTimeoutVal) ? 0 : Integer.parseInt(networkDataTimeoutVal); final String networkSocketTimeoutVal = conf.connectionTimeout; - final int networkSocketTimeout = (null == networkSocketTimeoutVal) ? 0 : Integer. - parseInt(networkSocketTimeoutVal); + final int networkSocketTimeout = (null == networkSocketTimeoutVal) ? 0 : Integer.parseInt(networkSocketTimeoutVal); final FTPClient client = new FTPClient(); if (networkDataTimeout > 0) { @@ -182,60 +163,47 @@ public class FTPUtils { @Override public void protocolCommandSent(final ProtocolCommandEvent event) { if (logger.isDebugEnabled()) { - logger.debug(processor + " : " + event.getMessage(). - trim()); + logger.debug(processor + " : " + event.getMessage().trim()); } } @Override public void protocolReplyReceived(final ProtocolCommandEvent event) { if (logger.isDebugEnabled()) { - logger.debug(processor + " : " + event.getMessage(). - trim()); + logger.debug(processor + " : " + event.getMessage().trim()); } } } /** - * Handles the logic required to change to the given directory RELATIVE TO - * THE CURRENT DIRECTORY which can include creating new directories needed. + * Handles the logic required to change to the given directory RELATIVE TO THE CURRENT DIRECTORY which can include creating new directories needed. * - * This will first attempt to change to the full path of the given directory - * outright. If that fails, then it will attempt to change from the top of - * the tree of the given directory all the way down to the final leaf node - * of the given directory. + * This will first attempt to change to the full path of the given directory outright. If that fails, then it will attempt to change from the top of the tree of the given directory all the way + * down to the final leaf node of the given directory. * * @param client - the ftp client with an already active connection * @param dirPath - the path to change or create directories to - * @param createDirs - if true will attempt to create any missing - * directories + * @param createDirs - if true will attempt to create any missing directories * @param processor - used solely for targetting logging output. * @throws IOException if any access problem occurs */ public static void changeWorkingDirectory(final FTPClient client, final String dirPath, final boolean createDirs, final Processor processor) throws IOException { final String currentWorkingDirectory = client.printWorkingDirectory(); final File dir = new File(dirPath); - logger. - debug(processor + " attempting to change directory from " + currentWorkingDirectory + " to " + dir. - getPath()); + logger.debug(processor + " attempting to change directory from " + currentWorkingDirectory + " to " + dir.getPath()); boolean dirExists = false; - final String forwardPaths = dir.getPath(). - replaceAll(Matcher.quoteReplacement("\\"), Matcher. - quoteReplacement("/")); + final String forwardPaths = dir.getPath().replaceAll(Matcher.quoteReplacement("\\"), Matcher.quoteReplacement("/")); //always use forward paths for long string attempt try { dirExists = client.changeWorkingDirectory(forwardPaths); if (dirExists) { - logger. - debug(processor + " changed working directory to '" + forwardPaths + "' from '" + currentWorkingDirectory + "'"); + logger.debug(processor + " changed working directory to '" + forwardPaths + "' from '" + currentWorkingDirectory + "'"); } else { - logger. - debug(processor + " could not change directory to '" + forwardPaths + "' from '" + currentWorkingDirectory + "' so trying the hard way."); + logger.debug(processor + " could not change directory to '" + forwardPaths + "' from '" + currentWorkingDirectory + "' so trying the hard way."); } } catch (final IOException ioe) { - logger. - debug(processor + " could not change directory to '" + forwardPaths + "' from '" + currentWorkingDirectory + "' so trying the hard way."); + logger.debug(processor + " could not change directory to '" + forwardPaths + "' from '" + currentWorkingDirectory + "' so trying the hard way."); } if (!dirExists) { //coulnd't navigate directly...begin hard work final Deque<String> stack = new LinkedList<>(); @@ -256,15 +224,12 @@ public class FTPUtils { exists = false; } if (!exists && createDirs) { - logger. - debug(processor + " creating new directory and changing to it " + dirName); + logger.debug(processor + " creating new directory and changing to it " + dirName); client.makeDirectory(dirName); - if (!(client.makeDirectory(dirName) || client. - changeWorkingDirectory(dirName))) { + if (!(client.makeDirectory(dirName) || client.changeWorkingDirectory(dirName))) { throw new IOException(processor + " could not create and change to newly created directory " + dirName); } else { - logger. - debug(processor + " successfully changed working directory to " + dirName); + logger.debug(processor + " successfully changed working directory to " + dirName); } } else if (!exists) { throw new IOException(processor + " could not change directory to '" + dirName + "' from '" + currentWorkingDirectory + "'"); http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/d29a2d68/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FileInfo.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FileInfo.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FileInfo.java index 6605ff6..c57b4e0 100644 --- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FileInfo.java +++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FileInfo.java @@ -67,8 +67,7 @@ public class FileInfo implements Comparable<FileInfo>, Serializable { public int hashCode() { final int prime = 31; int result = 1; - result = prime * result + ((fullPathFileName == null) ? 0 : fullPathFileName. - hashCode()); + result = prime * result + ((fullPathFileName == null) ? 0 : fullPathFileName.hashCode()); return result; } http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/d29a2d68/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FileTransfer.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FileTransfer.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FileTransfer.java index 3af00fa..ece0e59 100644 --- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FileTransfer.java +++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FileTransfer.java @@ -50,125 +50,125 @@ public interface FileTransfer extends Closeable { void ensureDirectoryExists(FlowFile flowFile, File remoteDirectory) throws IOException; - public static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder(). - name("Hostname"). - description("The fully qualified hostname or IP address of the remote system"). - addValidator(StandardValidators.NON_EMPTY_VALIDATOR). - required(true). - expressionLanguageSupported(true). - build(); - public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder(). - name("Username"). - description("Username"). - addValidator(StandardValidators.NON_EMPTY_VALIDATOR). - required(true). - build(); - public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder(). - name("Password"). - description("Password for the user account"). - addValidator(StandardValidators.NON_EMPTY_VALIDATOR). - required(false). - sensitive(true). - build(); - public static final PropertyDescriptor DATA_TIMEOUT = new PropertyDescriptor.Builder(). - name("Data Timeout"). - description("Amount of time to wait before timing out while transferring data"). - required(true). - addValidator(StandardValidators.TIME_PERIOD_VALIDATOR). - defaultValue("30 sec"). - build(); - public static final PropertyDescriptor CONNECTION_TIMEOUT = new PropertyDescriptor.Builder(). - name("Connection Timeout"). - description("Amount of time to wait before timing out while creating a connection"). - required(true). - addValidator(StandardValidators.TIME_PERIOD_VALIDATOR). - defaultValue("30 sec"). - build(); - public static final PropertyDescriptor REMOTE_PATH = new PropertyDescriptor.Builder(). - name("Remote Path"). - description("The path on the remote system from which to pull or push files"). - required(false). - addValidator(StandardValidators.NON_EMPTY_VALIDATOR). - expressionLanguageSupported(true). - build(); - public static final PropertyDescriptor CREATE_DIRECTORY = new PropertyDescriptor.Builder(). - name("Create Directory"). - description("Specifies whether or not the remote directory should be created if it does not exist."). - required(true). - allowableValues("true", "false"). - defaultValue("false"). - build(); - - public static final PropertyDescriptor USE_COMPRESSION = new PropertyDescriptor.Builder(). - name("Use Compression"). - description("Indicates whether or not ZLIB compression should be used when transferring files"). - allowableValues("true", "false"). - defaultValue("false"). - required(true). - build(); + public static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder() + .name("Hostname") + .description("The fully qualified hostname or IP address of the remote system") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .required(true) + .expressionLanguageSupported(true) + .build(); + public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() + .name("Username") + .description("Username") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .required(true) + .build(); + public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() + .name("Password") + .description("Password for the user account") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .required(false) + .sensitive(true) + .build(); + public static final PropertyDescriptor DATA_TIMEOUT = new PropertyDescriptor.Builder() + .name("Data Timeout") + .description("Amount of time to wait before timing out while transferring data") + .required(true) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .defaultValue("30 sec") + .build(); + public static final PropertyDescriptor CONNECTION_TIMEOUT = new PropertyDescriptor.Builder() + .name("Connection Timeout") + .description("Amount of time to wait before timing out while creating a connection") + .required(true) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .defaultValue("30 sec") + .build(); + public static final PropertyDescriptor REMOTE_PATH = new PropertyDescriptor.Builder() + .name("Remote Path") + .description("The path on the remote system from which to pull or push files") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(true) + .build(); + public static final PropertyDescriptor CREATE_DIRECTORY = new PropertyDescriptor.Builder() + .name("Create Directory") + .description("Specifies whether or not the remote directory should be created if it does not exist.") + .required(true) + .allowableValues("true", "false") + .defaultValue("false") + .build(); + + public static final PropertyDescriptor USE_COMPRESSION = new PropertyDescriptor.Builder() + .name("Use Compression") + .description("Indicates whether or not ZLIB compression should be used when transferring files") + .allowableValues("true", "false") + .defaultValue("false") + .required(true) + .build(); // GET-specific properties - public static final PropertyDescriptor RECURSIVE_SEARCH = new PropertyDescriptor.Builder(). - name("Search Recursively"). - description("If true, will pull files from arbitrarily nested subdirectories; otherwise, will not traverse subdirectories"). - required(true). - defaultValue("false"). - allowableValues("true", "false"). - build(); - public static final PropertyDescriptor FILE_FILTER_REGEX = new PropertyDescriptor.Builder(). - name("File Filter Regex"). - description("Provides a Java Regular Expression for filtering Filenames; if a filter is supplied, only files whose names match that Regular Expression will be fetched"). - required(false). - addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR). - build(); - public static final PropertyDescriptor PATH_FILTER_REGEX = new PropertyDescriptor.Builder(). - name("Path Filter Regex"). - description("When " + RECURSIVE_SEARCH.getName() + " is true, then only subdirectories whose path matches the given Regular Expression will be scanned"). - required(false). - addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR). - build(); - public static final PropertyDescriptor MAX_SELECTS = new PropertyDescriptor.Builder(). - name("Max Selects"). - description("The maximum number of files to pull in a single connection"). - defaultValue("100"). - required(true). - addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR). - build(); - public static final PropertyDescriptor REMOTE_POLL_BATCH_SIZE = new PropertyDescriptor.Builder(). - name("Remote Poll Batch Size"). - description("The value specifies how many file paths to find in a given directory on the remote system when doing a file listing. This value in general should not need to be modified but when polling against a remote system with a tremendous number of files this value can be critical. Setting this value too high can result very poor performance and setting it too low can cause the flow to be slower than normal."). - defaultValue("5000"). - addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR). - required(true). - build(); - public static final PropertyDescriptor DELETE_ORIGINAL = new PropertyDescriptor.Builder(). - name("Delete Original"). - description("Determines whether or not the file is deleted from the remote system after it has been successfully transferred"). - defaultValue("true"). - allowableValues("true", "false"). - required(true). - build(); - public static final PropertyDescriptor POLLING_INTERVAL = new PropertyDescriptor.Builder(). - name("Polling Interval"). - description("Determines how long to wait between fetching the listing for new files"). - addValidator(StandardValidators.TIME_PERIOD_VALIDATOR). - required(true). - defaultValue("60 sec"). - build(); - public static final PropertyDescriptor IGNORE_DOTTED_FILES = new PropertyDescriptor.Builder(). - name("Ignore Dotted Files"). - description("If true, files whose names begin with a dot (\".\") will be ignored"). - allowableValues("true", "false"). - defaultValue("true"). - required(true). - build(); - public static final PropertyDescriptor USE_NATURAL_ORDERING = new PropertyDescriptor.Builder(). - name("Use Natural Ordering"). - description("If true, will pull files in the order in which they are naturally listed; otherwise, the order in which the files will be pulled is not defined"). - allowableValues("true", "false"). - defaultValue("false"). - required(true). - build(); + public static final PropertyDescriptor RECURSIVE_SEARCH = new PropertyDescriptor.Builder() + .name("Search Recursively") + .description("If true, will pull files from arbitrarily nested subdirectories; otherwise, will not traverse subdirectories") + .required(true) + .defaultValue("false") + .allowableValues("true", "false") + .build(); + public static final PropertyDescriptor FILE_FILTER_REGEX = new PropertyDescriptor.Builder() + .name("File Filter Regex") + .description("Provides a Java Regular Expression for filtering Filenames; if a filter is supplied, only files whose names match that Regular Expression will be fetched") + .required(false) + .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR) + .build(); + public static final PropertyDescriptor PATH_FILTER_REGEX = new PropertyDescriptor.Builder() + .name("Path Filter Regex") + .description("When " + RECURSIVE_SEARCH.getName() + " is true, then only subdirectories whose path matches the given Regular Expression will be scanned") + .required(false) + .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR) + .build(); + public static final PropertyDescriptor MAX_SELECTS = new PropertyDescriptor.Builder() + .name("Max Selects") + .description("The maximum number of files to pull in a single connection") + .defaultValue("100") + .required(true) + .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR) + .build(); + public static final PropertyDescriptor REMOTE_POLL_BATCH_SIZE = new PropertyDescriptor.Builder() + .name("Remote Poll Batch Size") + .description("The value specifies how many file paths to find in a given directory on the remote system when doing a file listing. This value in general should not need to be modified but when polling against a remote system with a tremendous number of files this value can be critical. Setting this value too high can result very poor performance and setting it too low can cause the flow to be slower than normal.") + .defaultValue("5000") + .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR) + .required(true) + .build(); + public static final PropertyDescriptor DELETE_ORIGINAL = new PropertyDescriptor.Builder() + .name("Delete Original") + .description("Determines whether or not the file is deleted from the remote system after it has been successfully transferred") + .defaultValue("true") + .allowableValues("true", "false") + .required(true) + .build(); + public static final PropertyDescriptor POLLING_INTERVAL = new PropertyDescriptor.Builder() + .name("Polling Interval") + .description("Determines how long to wait between fetching the listing for new files") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .required(true) + .defaultValue("60 sec") + .build(); + public static final PropertyDescriptor IGNORE_DOTTED_FILES = new PropertyDescriptor.Builder() + .name("Ignore Dotted Files") + .description("If true, files whose names begin with a dot (\".\") will be ignored") + .allowableValues("true", "false") + .defaultValue("true") + .required(true) + .build(); + public static final PropertyDescriptor USE_NATURAL_ORDERING = new PropertyDescriptor.Builder() + .name("Use Natural Ordering") + .description("If true, will pull files in the order in which they are naturally listed; otherwise, the order in which the files will be pulled is not defined") + .allowableValues("true", "false") + .defaultValue("false") + .required(true) + .build(); // PUT-specific properties public static final String FILE_MODIFY_DATE_ATTR_FORMAT = "yyyy-MM-dd'T'HH:mm:ssZ"; @@ -179,66 +179,66 @@ public interface FileTransfer extends Closeable { public static final String CONFLICT_RESOLUTION_REJECT = "REJECT"; public static final String CONFLICT_RESOLUTION_FAIL = "FAIL"; public static final String CONFLICT_RESOLUTION_NONE = "NONE"; - public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder(). - name("Conflict Resolution"). - description("Determines how to handle the problem of filename collisions"). - required(true). - allowableValues(CONFLICT_RESOLUTION_REPLACE, CONFLICT_RESOLUTION_IGNORE, CONFLICT_RESOLUTION_RENAME, CONFLICT_RESOLUTION_REJECT, CONFLICT_RESOLUTION_FAIL, CONFLICT_RESOLUTION_NONE). - defaultValue(CONFLICT_RESOLUTION_NONE). - build(); - public static final PropertyDescriptor REJECT_ZERO_BYTE = new PropertyDescriptor.Builder(). - name("Reject Zero-Byte Files"). - description("Determines whether or not Zero-byte files should be rejected without attempting to transfer"). - allowableValues("true", "false"). - defaultValue("true"). - build(); - public static final PropertyDescriptor DOT_RENAME = new PropertyDescriptor.Builder(). - name("Dot Rename"). - description("If true, then the filename of the sent file is prepended with a \".\" and then renamed back to the original once the file is completely sent. Otherwise, there is no rename. This property is ignored if the Temporary Filename property is set."). - allowableValues("true", "false"). - defaultValue("true"). - build(); - public static final PropertyDescriptor TEMP_FILENAME = new PropertyDescriptor.Builder(). - name("Temporary Filename"). - description("If set, the filename of the sent file will be equal to the value specified during the transfer and after successful completion will be renamed to the original filename. If this value is set, the Dot Rename property is ignored."). - addValidator(StandardValidators.NON_EMPTY_VALIDATOR). - expressionLanguageSupported(true). - required(false). - build(); - public static final PropertyDescriptor LAST_MODIFIED_TIME = new PropertyDescriptor.Builder(). - name("Last Modified Time"). - description("The lastModifiedTime to assign to the file after transferring it. If not set, the lastModifiedTime will not be changed. Format must be yyyy-MM-dd'T'HH:mm:ssZ. You may also use expression language such as ${file.lastModifiedTime}. If the value is invalid, the processor will not be invalid but will fail to change lastModifiedTime of the file."). - required(false). - addValidator(StandardValidators.NON_EMPTY_VALIDATOR). - expressionLanguageSupported(true). - build(); - public static final PropertyDescriptor PERMISSIONS = new PropertyDescriptor.Builder(). - name("Permissions"). - description("The permissions to assign to the file after transferring it. Format must be either UNIX rwxrwxrwx with a - in place of denied permissions (e.g. rw-r--r--) or an octal number (e.g. 644). If not set, the permissions will not be changed. You may also use expression language such as ${file.permissions}. If the value is invalid, the processor will not be invalid but will fail to change permissions of the file."). - required(false). - addValidator(StandardValidators.NON_EMPTY_VALIDATOR). - expressionLanguageSupported(true). - build(); - public static final PropertyDescriptor REMOTE_OWNER = new PropertyDescriptor.Builder(). - name("Remote Owner"). - description("Integer value representing the User ID to set on the file after transferring it. If not set, the owner will not be set. You may also use expression language such as ${file.owner}. If the value is invalid, the processor will not be invalid but will fail to change the owner of the file."). - required(false). - addValidator(StandardValidators.NON_EMPTY_VALIDATOR). - expressionLanguageSupported(true). - build(); - public static final PropertyDescriptor REMOTE_GROUP = new PropertyDescriptor.Builder(). - name("Remote Group"). - description("Integer value representing the Group ID to set on the file after transferring it. If not set, the group will not be set. You may also use expression language such as ${file.group}. If the value is invalid, the processor will not be invalid but will fail to change the group of the file."). - required(false). - addValidator(StandardValidators.NON_EMPTY_VALIDATOR). - expressionLanguageSupported(true). - build(); - public static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder(). - name("Batch Size"). - description("The maximum number of FlowFiles to send in a single connection"). - required(true). - addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR). - defaultValue("500"). - build(); + public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder() + .name("Conflict Resolution") + .description("Determines how to handle the problem of filename collisions") + .required(true) + .allowableValues(CONFLICT_RESOLUTION_REPLACE, CONFLICT_RESOLUTION_IGNORE, CONFLICT_RESOLUTION_RENAME, CONFLICT_RESOLUTION_REJECT, CONFLICT_RESOLUTION_FAIL, CONFLICT_RESOLUTION_NONE) + .defaultValue(CONFLICT_RESOLUTION_NONE) + .build(); + public static final PropertyDescriptor REJECT_ZERO_BYTE = new PropertyDescriptor.Builder() + .name("Reject Zero-Byte Files") + .description("Determines whether or not Zero-byte files should be rejected without attempting to transfer") + .allowableValues("true", "false") + .defaultValue("true") + .build(); + public static final PropertyDescriptor DOT_RENAME = new PropertyDescriptor.Builder() + .name("Dot Rename") + .description("If true, then the filename of the sent file is prepended with a \".\" and then renamed back to the original once the file is completely sent. Otherwise, there is no rename. This property is ignored if the Temporary Filename property is set.") + .allowableValues("true", "false") + .defaultValue("true") + .build(); + public static final PropertyDescriptor TEMP_FILENAME = new PropertyDescriptor.Builder() + .name("Temporary Filename") + .description("If set, the filename of the sent file will be equal to the value specified during the transfer and after successful completion will be renamed to the original filename. If this value is set, the Dot Rename property is ignored.") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(true) + .required(false) + .build(); + public static final PropertyDescriptor LAST_MODIFIED_TIME = new PropertyDescriptor.Builder() + .name("Last Modified Time") + .description("The lastModifiedTime to assign to the file after transferring it. If not set, the lastModifiedTime will not be changed. Format must be yyyy-MM-dd'T'HH:mm:ssZ. You may also use expression language such as ${file.lastModifiedTime}. If the value is invalid, the processor will not be invalid but will fail to change lastModifiedTime of the file.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(true) + .build(); + public static final PropertyDescriptor PERMISSIONS = new PropertyDescriptor.Builder() + .name("Permissions") + .description("The permissions to assign to the file after transferring it. Format must be either UNIX rwxrwxrwx with a - in place of denied permissions (e.g. rw-r--r--) or an octal number (e.g. 644). If not set, the permissions will not be changed. You may also use expression language such as ${file.permissions}. If the value is invalid, the processor will not be invalid but will fail to change permissions of the file.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(true) + .build(); + public static final PropertyDescriptor REMOTE_OWNER = new PropertyDescriptor.Builder() + .name("Remote Owner") + .description("Integer value representing the User ID to set on the file after transferring it. If not set, the owner will not be set. You may also use expression language such as ${file.owner}. If the value is invalid, the processor will not be invalid but will fail to change the owner of the file.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(true) + .build(); + public static final PropertyDescriptor REMOTE_GROUP = new PropertyDescriptor.Builder() + .name("Remote Group") + .description("Integer value representing the Group ID to set on the file after transferring it. If not set, the group will not be set. You may also use expression language such as ${file.group}. If the value is invalid, the processor will not be invalid but will fail to change the group of the file.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(true) + .build(); + public static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder() + .name("Batch Size") + .description("The maximum number of FlowFiles to send in a single connection") + .required(true) + .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR) + .defaultValue("500") + .build(); } http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/d29a2d68/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/JmsFactory.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/JmsFactory.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/JmsFactory.java index 0f50cdf..35a65dc 100644 --- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/JmsFactory.java +++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/JmsFactory.java @@ -104,13 +104,9 @@ public class JmsFactory { final ConnectionFactory connectionFactory = createConnectionFactory(context); - final String username = context.getProperty(USERNAME). - getValue(); - final String password = context.getProperty(PASSWORD). - getValue(); - final Connection connection = (username == null && password == null) ? connectionFactory. - createConnection() - : connectionFactory.createConnection(username, password); + final String username = context.getProperty(USERNAME).getValue(); + final String password = context.getProperty(PASSWORD).getValue(); + final Connection connection = (username == null && password == null) ? connectionFactory.createConnection() : connectionFactory.createConnection(username, password); connection.setClientID(clientId); connection.start(); @@ -119,17 +115,12 @@ public class JmsFactory { public static Connection createConnection(final String url, final String jmsProvider, final String username, final String password, final int timeoutMillis) throws JMSException { final ConnectionFactory connectionFactory = createConnectionFactory(url, timeoutMillis, jmsProvider); - return (username == null && password == null) ? connectionFactory. - createConnection() : connectionFactory. - createConnection(username, password); + return (username == null && password == null) ? connectionFactory.createConnection() : connectionFactory.createConnection(username, password); } public static String createClientId(final ProcessContext context) { - final String clientIdPrefix = context.getProperty(CLIENT_ID_PREFIX). - getValue(); - return CLIENT_ID_FIXED_PREFIX + (clientIdPrefix == null ? "" : clientIdPrefix) + "-" + UUID. - randomUUID(). - toString(); + final String clientIdPrefix = context.getProperty(CLIENT_ID_PREFIX).getValue(); + return CLIENT_ID_FIXED_PREFIX + (clientIdPrefix == null ? "" : clientIdPrefix) + "-" + UUID.randomUUID().toString(); } public static boolean clientIdPrefixEquals(final String one, final String two) { @@ -138,14 +129,11 @@ public class JmsFactory { } else if (two == null) { return false; } - int uuidLen = UUID.randomUUID(). - toString(). - length(); + int uuidLen = UUID.randomUUID().toString().length(); if (one.length() <= uuidLen || two.length() <= uuidLen) { return false; } - return one.substring(0, one.length() - uuidLen). - equals(two.substring(0, two.length() - uuidLen)); + return one.substring(0, one.length() - uuidLen).equals(two.substring(0, two.length() - uuidLen)); } public static byte[] createByteArray(final Message message) throws JMSException { @@ -164,8 +152,7 @@ public class JmsFactory { } private static byte[] getMessageBytes(TextMessage message) throws JMSException { - return (message.getText() == null) ? new byte[0] : message.getText(). - getBytes(); + return (message.getText() == null) ? new byte[0] : message.getText().getBytes(); } private static byte[] getMessageBytes(BytesMessage message) throws JMSException { @@ -204,8 +191,7 @@ public class JmsFactory { String key = (String) elements.nextElement(); map.put(key, message.getString(key)); } - return map.toString(). - getBytes(); + return map.toString().getBytes(); } private static byte[] getMessageBytes(ObjectMessage message) throws JMSException { @@ -224,9 +210,7 @@ public class JmsFactory { } public static Session createSession(final ProcessContext context, final Connection connection, final boolean transacted) throws JMSException { - final String configuredAckMode = context. - getProperty(ACKNOWLEDGEMENT_MODE). - getValue(); + final String configuredAckMode = context.getProperty(ACKNOWLEDGEMENT_MODE).getValue(); return createSession(connection, configuredAckMode, transacted); } @@ -247,14 +231,11 @@ public class JmsFactory { Session jmsSession = null; try { connection = JmsFactory.createConnection(context); - jmsSession = JmsFactory. - createSession(context, connection, DEFAULT_IS_TRANSACTED); + jmsSession = JmsFactory.createSession(context, connection, DEFAULT_IS_TRANSACTED); - final String messageSelector = context.getProperty(MESSAGE_SELECTOR). - getValue(); + final String messageSelector = context.getProperty(MESSAGE_SELECTOR).getValue(); final Destination destination = createQueue(context); - final MessageConsumer messageConsumer = jmsSession. - createConsumer(destination, messageSelector, false); + final MessageConsumer messageConsumer = jmsSession.createConsumer(destination, messageSelector, false); return new WrappedMessageConsumer(connection, jmsSession, messageConsumer); } catch (JMSException e) { @@ -280,20 +261,15 @@ public class JmsFactory { Session jmsSession = null; try { connection = JmsFactory.createConnection(context, clientId); - jmsSession = JmsFactory. - createSession(context, connection, DEFAULT_IS_TRANSACTED); + jmsSession = JmsFactory.createSession(context, connection, DEFAULT_IS_TRANSACTED); - final String messageSelector = context.getProperty(MESSAGE_SELECTOR). - getValue(); + final String messageSelector = context.getProperty(MESSAGE_SELECTOR).getValue(); final Topic topic = createTopic(context); final MessageConsumer messageConsumer; - if (context.getProperty(DURABLE_SUBSCRIPTION). - asBoolean()) { - messageConsumer = jmsSession. - createDurableSubscriber(topic, clientId, messageSelector, false); + if (context.getProperty(DURABLE_SUBSCRIPTION).asBoolean()) { + messageConsumer = jmsSession.createDurableSubscriber(topic, clientId, messageSelector, false); } else { - messageConsumer = jmsSession. - createConsumer(topic, messageSelector, false); + messageConsumer = jmsSession.createConsumer(topic, messageSelector, false); } return new WrappedMessageConsumer(connection, jmsSession, messageConsumer); @@ -309,8 +285,7 @@ public class JmsFactory { } private static Destination getDestination(final ProcessContext context) throws JMSException { - final String destinationType = context.getProperty(DESTINATION_TYPE). - getValue(); + final String destinationType = context.getProperty(DESTINATION_TYPE).getValue(); switch (destinationType) { case DESTINATION_TYPE_TOPIC: return createTopic(context); @@ -330,12 +305,10 @@ public class JmsFactory { try { connection = JmsFactory.createConnection(context); - jmsSession = JmsFactory. - createSession(context, connection, transacted); + jmsSession = JmsFactory.createSession(context, connection, transacted); final Destination destination = getDestination(context); - final MessageProducer messageProducer = jmsSession. - createProducer(destination); + final MessageProducer messageProducer = jmsSession.createProducer(destination); return new WrappedMessageProducer(connection, jmsSession, messageProducer); } catch (JMSException e) { @@ -350,13 +323,11 @@ public class JmsFactory { } public static Destination createQueue(final ProcessContext context) { - return createQueue(context, context.getProperty(DESTINATION_NAME). - getValue()); + return createQueue(context, context.getProperty(DESTINATION_NAME).getValue()); } public static Queue createQueue(final ProcessContext context, final String queueName) { - return createQueue(context.getProperty(JMS_PROVIDER). - getValue(), queueName); + return createQueue(context.getProperty(JMS_PROVIDER).getValue(), queueName); } public static Queue createQueue(final String jmsProvider, final String queueName) { @@ -368,10 +339,8 @@ public class JmsFactory { } private static Topic createTopic(final ProcessContext context) { - final String topicName = context.getProperty(DESTINATION_NAME). - getValue(); - switch (context.getProperty(JMS_PROVIDER). - getValue()) { + final String topicName = context.getProperty(DESTINATION_NAME).getValue(); + switch (context.getProperty(JMS_PROVIDER).getValue()) { case ACTIVEMQ_PROVIDER: default: return new ActiveMQTopic(topicName); @@ -379,13 +348,9 @@ public class JmsFactory { } private static ConnectionFactory createConnectionFactory(final ProcessContext context) throws JMSException { - final String url = context.getProperty(URL). - getValue(); - final int timeoutMillis = context.getProperty(TIMEOUT). - asTimePeriod(TimeUnit.MILLISECONDS). - intValue(); - final String provider = context.getProperty(JMS_PROVIDER). - getValue(); + final String url = context.getProperty(URL).getValue(); + final int timeoutMillis = context.getProperty(TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue(); + final String provider = context.getProperty(JMS_PROVIDER).getValue(); return createConnectionFactory(url, timeoutMillis, provider); } @@ -412,8 +377,7 @@ public class JmsFactory { if (value == null) { attributes.put(ATTRIBUTE_PREFIX + propName, ""); - attributes. - put(ATTRIBUTE_PREFIX + propName + ATTRIBUTE_TYPE_SUFFIX, "Unknown"); + attributes.put(ATTRIBUTE_PREFIX + propName + ATTRIBUTE_TYPE_SUFFIX, "Unknown"); continue; } @@ -441,42 +405,30 @@ public class JmsFactory { propType = PROP_TYPE_OBJECT; } - attributes. - put(ATTRIBUTE_PREFIX + propName + ATTRIBUTE_TYPE_SUFFIX, propType); + attributes.put(ATTRIBUTE_PREFIX + propName + ATTRIBUTE_TYPE_SUFFIX, propType); } if (message.getJMSCorrelationID() != null) { - attributes.put(ATTRIBUTE_PREFIX + JMS_CORRELATION_ID, message. - getJMSCorrelationID()); + attributes.put(ATTRIBUTE_PREFIX + JMS_CORRELATION_ID, message.getJMSCorrelationID()); } if (message.getJMSDestination() != null) { - attributes.put(ATTRIBUTE_PREFIX + JMS_DESTINATION, message. - getJMSDestination(). - toString()); + attributes.put(ATTRIBUTE_PREFIX + JMS_DESTINATION, message.getJMSDestination().toString()); } if (message.getJMSMessageID() != null) { - attributes.put(ATTRIBUTE_PREFIX + JMS_MESSAGE_ID, message. - getJMSMessageID()); + attributes.put(ATTRIBUTE_PREFIX + JMS_MESSAGE_ID, message.getJMSMessageID()); } if (message.getJMSReplyTo() != null) { - attributes.put(ATTRIBUTE_PREFIX + JMS_REPLY_TO, message. - getJMSReplyTo(). - toString()); + attributes.put(ATTRIBUTE_PREFIX + JMS_REPLY_TO, message.getJMSReplyTo().toString()); } if (message.getJMSType() != null) { attributes.put(ATTRIBUTE_PREFIX + JMS_TYPE, message.getJMSType()); } - attributes.put(ATTRIBUTE_PREFIX + JMS_DELIVERY_MODE, String. - valueOf(message.getJMSDeliveryMode())); - attributes.put(ATTRIBUTE_PREFIX + JMS_EXPIRATION, String. - valueOf(message.getJMSExpiration())); - attributes.put(ATTRIBUTE_PREFIX + JMS_PRIORITY, String.valueOf(message. - getJMSPriority())); - attributes.put(ATTRIBUTE_PREFIX + JMS_REDELIVERED, String. - valueOf(message.getJMSRedelivered())); - attributes.put(ATTRIBUTE_PREFIX + JMS_TIMESTAMP, String.valueOf(message. - getJMSTimestamp())); + attributes.put(ATTRIBUTE_PREFIX + JMS_DELIVERY_MODE, String.valueOf(message.getJMSDeliveryMode())); + attributes.put(ATTRIBUTE_PREFIX + JMS_EXPIRATION, String.valueOf(message.getJMSExpiration())); + attributes.put(ATTRIBUTE_PREFIX + JMS_PRIORITY, String.valueOf(message.getJMSPriority())); + attributes.put(ATTRIBUTE_PREFIX + JMS_REDELIVERED, String.valueOf(message.getJMSRedelivered())); + attributes.put(ATTRIBUTE_PREFIX + JMS_TIMESTAMP, String.valueOf(message.getJMSTimestamp())); return attributes; } }
