You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "krisztina-zsihovszki (via GitHub)" <gi...@apache.org> on 2023/01/26 08:47:55 UTC
[GitHub] [nifi] krisztina-zsihovszki opened a new pull request, #6892: NIFI-11059 PutBoxFile processor
krisztina-zsihovszki opened a new pull request, #6892:
URL: https://github.com/apache/nifi/pull/6892
<!-- 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. -->
# Summary
[NIFI-11059](https://issues.apache.org/jira/browse/NIFI-11059)
# Tracking
Please complete the following tracking steps prior to pull request creation.
### Issue Tracking
- [x] [Apache NiFi Jira](https://issues.apache.org/jira/browse/NIFI) issue created
### Pull Request Tracking
- [x] Pull Request title starts with Apache NiFi Jira issue number, such as `NIFI-11059`
- [x] Pull Request commit message starts with Apache NiFi Jira issue number, as such `NIFI-11059`
### Pull Request Formatting
- [x] Pull Request based on current revision of the `main` branch
- [x] Pull Request refers to a feature branch with one commit containing changes
# Verification
Please indicate the verification steps performed prior to pull request creation.
### Build
- [ ] Build completed using `mvn clean install -P contrib-check`
- [x] JDK 8
- [x] JDK 11
- [x] JDK 17
### Licensing
- [ ] New dependencies are compatible with the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0) according to the [License Policy](https://www.apache.org/legal/resolved.html)
- [ ] New dependencies are documented in applicable `LICENSE` and `NOTICE` files
### Documentation
- [x] Documentation formatting appears as expected in rendered files
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] turcsanyip commented on a diff in pull request #6892: NIFI-11059 PutBoxFile processor
Posted by "turcsanyip (via GitHub)" <gi...@apache.org>.
turcsanyip commented on code in PR #6892:
URL: https://github.com/apache/nifi/pull/6892#discussion_r1088117506
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor
+ .Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
Review Comment:
`Create Subfolder` property should depend on `Subfolder Name`.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/resources/docs/org.apache.nifi.processors.box.FetchBoxFile/additionalDetails.html:
##########
@@ -0,0 +1,49 @@
+<!DOCTYPE html>
+<html lang="en" xmlns="http://www.w3.org/1999/html">
+<!--
+ 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.
+ -->
+
+<head>
+ <meta charset="utf-8"/>
+ <title>FetchBoxFile</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
+</head>
+<body>
+
+<h1>Fetch Box files in NiFi</h1>
+
+<p>
+ This processor uses Box credentials for authentication to access Box.
+</p>
+<ol>
+ <li><b>Find File ID</b>
+ </br>
+ Usually FetchBoxFile is used with ListBoxFile and 'box.id' is set.</br></br>
+ In case 'box.id' is not available, you can find the ID of the file in the following way:
+ </br>
+ <ul>
+ <li>Click on the file.</li>
+ <li>The URL in the browser will include the file ID.
Review Comment:
Minor (it is referred as `File ID` below):
```suggestion
<li>The URL in the browser will include the File ID.
```
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/FetchBoxFile.java:
##########
@@ -114,40 +128,50 @@ public void onTrigger(ProcessContext context, ProcessSession session) throws Pro
}
String fileId = context.getProperty(FILE_ID).evaluateAttributeExpressions(flowFile).getValue();
- FlowFile outFlowFile = flowFile;
try {
- outFlowFile = fetchFile(fileId, session, outFlowFile);
-
- session.transfer(outFlowFile, REL_SUCCESS);
+ final long startNanos = System.nanoTime();
+ flowFile = fetchFile(fileId, session, flowFile);
+ final String boxUrlOfFile = BoxFileUtils.BOX_URL + fileId;
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().fetch(flowFile, boxUrlOfFile, transferMillis);
+ session.transfer(flowFile, REL_SUCCESS);
} catch (BoxAPIResponseException e) {
handleErrorResponse(session, fileId, flowFile, e);
} catch (Exception e) {
handleUnexpectedError(session, flowFile, fileId, e);
}
}
- FlowFile fetchFile(String fileId, ProcessSession session, FlowFile outFlowFile) {
- BoxFile boxFile = new BoxFile(boxAPIConnection, fileId);
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
- outFlowFile = session.write(outFlowFile, outputStream -> boxFile.download(outputStream));
+ BoxFile getBoxFile(String fileId) {
+ return new BoxFile(boxAPIConnection, fileId);
+ }
- return outFlowFile;
+ FlowFile fetchFile(String fileId, ProcessSession session, FlowFile flowFile) {
Review Comment:
I think it can be `private`.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor
+ .Builder()
+ .name("file-name")
Review Comment:
Minor formatting:
```suggestion
public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor.Builder()
.name("file-name")
```
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/FetchBoxFile.java:
##########
@@ -114,40 +128,50 @@ public void onTrigger(ProcessContext context, ProcessSession session) throws Pro
}
String fileId = context.getProperty(FILE_ID).evaluateAttributeExpressions(flowFile).getValue();
- FlowFile outFlowFile = flowFile;
try {
- outFlowFile = fetchFile(fileId, session, outFlowFile);
-
- session.transfer(outFlowFile, REL_SUCCESS);
+ final long startNanos = System.nanoTime();
+ flowFile = fetchFile(fileId, session, flowFile);
+ final String boxUrlOfFile = BoxFileUtils.BOX_URL + fileId;
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().fetch(flowFile, boxUrlOfFile, transferMillis);
+ session.transfer(flowFile, REL_SUCCESS);
} catch (BoxAPIResponseException e) {
handleErrorResponse(session, fileId, flowFile, e);
} catch (Exception e) {
handleUnexpectedError(session, flowFile, fileId, e);
}
}
- FlowFile fetchFile(String fileId, ProcessSession session, FlowFile outFlowFile) {
- BoxFile boxFile = new BoxFile(boxAPIConnection, fileId);
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
Review Comment:
Please move it back to its original place.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/resources/docs/org.apache.nifi.processors.box.FetchBoxFile/additionalDetails.html:
##########
@@ -0,0 +1,49 @@
+<!DOCTYPE html>
+<html lang="en" xmlns="http://www.w3.org/1999/html">
+<!--
+ 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.
+ -->
+
+<head>
+ <meta charset="utf-8"/>
+ <title>FetchBoxFile</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
+</head>
+<body>
+
+<h1>Fetch Box files in NiFi</h1>
+
+<p>
+ This processor uses Box credentials for authentication to access Box.
Review Comment:
This sentence is without context a bit. The points under it are related to File/Folder ID, not to the credentials.
It could be mentioned that the credentials are documented on the controller service or it could be removed at all.
For the File/Folder ID section, a separate subtitle (or intro note) should be added.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/resources/docs/org.apache.nifi.processors.box.PutBoxFile/additionalDetails.html:
##########
@@ -0,0 +1,48 @@
+<!DOCTYPE html>
+<html lang="en" xmlns="http://www.w3.org/1999/html">
+<!--
+ 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.
+ -->
+
+<head>
+ <meta charset="utf-8"/>
+ <title>PutBoxFile</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
+</head>
+<body>
+
+<h1>Upload files to Box from NiFi</h1>
+
+<p>
+ This processor uses Box credentials for authentication to access Box.
+</p>
+<ol>
+ <li><b>Find Folder ID</b>
+ <ul>
+ <li>Navigate to the folder to be listed in Box and enter it. The URL in your browser will include the ID at the end of
Review Comment:
The `folder to be listed` is a copy-paste error from the List processor.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/BoxFileAttributes.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.box;
+
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+
+public class BoxFileAttributes {
+
+ public static final String ID = "box.id";
+ public static final String ID_DESC = "The id of the file";
+
+ public static final String FILENAME = CoreAttributes.FILENAME.key();
+ public static final String FILENAME_DESC = "The name of the file";
+
+ public static final String PATH = CoreAttributes.PATH.key();
+ public static final String PATH_DESC = "The folder path where the file is located";
Review Comment:
I would rather use literals here than in `@WritesAttribute` in all the processors.
A reference to `CoreAttributes` can be added in a comment.
```suggestion
public static final String FILENAME = "filename"; // CoreAttributes.FILENAME.key()
public static final String FILENAME_DESC = "The name of the file";
public static final String PATH = "path"; //CoreAttributes.PATH.key()
public static final String PATH_DESC = "The folder path where the file is located";
```
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor
+ .Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION)
+ .allowableValues(FAIL_RESOLUTION, IGNORE_RESOLUTION, REPLACE_RESOLUTION)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ public static final int CONFLICT_RESPONSE_CODE = 409;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ final String subfolderName = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ BoxFile.Info uploadedFile = null;
+
+ final long startNanos = System.nanoTime();
+
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+
+ folderId = subfolderName != null ? getOrCreateParentSubfolder(subfolderName, folderId, createFolder).getID() : folderId;
+ final BoxFolder parentFolder = getFolder(folderId);
+ fullPath = BoxFileUtils.getPath(parentFolder.getInfo());
+
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+
+ final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ boolean isNewVersionUpload = false;
+
+ if (REPLACE_RESOLUTION.equals(conflictResolution)) {
+ final Optional<BoxFile.Info> alreadyUploadedFile = getFileByName(filename, parentFolder);
+
+ if (alreadyUploadedFile.isPresent()) {
+ BoxFile existingBoxFile = new BoxFile(boxAPIConnection, alreadyUploadedFile.get().getID());
+ existingBoxFile.uploadNewVersion(rawIn);
Review Comment:
`uploadedFile` should be assigned here too. Otherwise the FF attributes will not be populated and no provenance event will be sent for replaced files.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor
+ .Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION)
+ .allowableValues(FAIL_RESOLUTION, IGNORE_RESOLUTION, REPLACE_RESOLUTION)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ public static final int CONFLICT_RESPONSE_CODE = 409;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ final String subfolderName = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ BoxFile.Info uploadedFile = null;
+
+ final long startNanos = System.nanoTime();
+
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+
+ folderId = subfolderName != null ? getOrCreateParentSubfolder(subfolderName, folderId, createFolder).getID() : folderId;
+ final BoxFolder parentFolder = getFolder(folderId);
+ fullPath = BoxFileUtils.getPath(parentFolder.getInfo());
+
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+
+ final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ boolean isNewVersionUpload = false;
+
+ if (REPLACE_RESOLUTION.equals(conflictResolution)) {
+ final Optional<BoxFile.Info> alreadyUploadedFile = getFileByName(filename, parentFolder);
+
+ if (alreadyUploadedFile.isPresent()) {
+ BoxFile existingBoxFile = new BoxFile(boxAPIConnection, alreadyUploadedFile.get().getID());
+ existingBoxFile.uploadNewVersion(rawIn);
+ isNewVersionUpload = true;
+ }
+ }
+
+ if (!isNewVersionUpload) {
+ if (size > chunkUploadThreshold) {
+ uploadedFile = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFile = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+
+ } catch (BoxAPIResponseException e) {
+ handleUploadError(conflictResolution, filename, parentFolder, e);
+ }
+
+ if (uploadedFile != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFile);
+ final String url = BOX_URL + uploadedFile.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Unexpected exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ final BoxClientService boxClientService = context.getProperty(BoxClientService.BOX_CLIENT_SERVICE).asControllerService(BoxClientService.class);
+
+ boxAPIConnection = boxClientService.getBoxApiConnection();
+ }
+
+ BoxFolder getFolder(String folderId) {
+ return new BoxFolder(boxAPIConnection, folderId);
+ }
+
+ private BoxFolder.Info getOrCreateParentSubfolder(String folderName, String parentFolderId, boolean createFolder) {
+ final int indexOfPathSeparator = folderName.indexOf("/");
+
+ if (isMultiLevelFolder(indexOfPathSeparator, folderName)) {
+ final String mainFolderName = folderName.substring(0, indexOfPathSeparator);
+ final String subFolders = folderName.substring(indexOfPathSeparator + 1);
+ final BoxFolder.Info mainFolder = getOrCreateFolder(mainFolderName, parentFolderId, createFolder);
+ return getOrCreateParentSubfolder(subFolders, mainFolder.getID(), createFolder);
+ } else {
+ return getOrCreateFolder(folderName, parentFolderId, createFolder);
+ }
+ }
+
+ private BoxFolder.Info getOrCreateFolder(String folderName, String parentFolderId, boolean createFolder) {
+ final Optional<BoxFolder.Info> existingFolder = checkFolderExistence(folderName, parentFolderId);
+
+ if (existingFolder.isPresent()) {
+ return existingFolder.get();
+ }
+
+ if (createFolder) {
+ getLogger().debug("Create folder " + folderName + " parent id: " + parentFolderId);
+
+ final BoxFolder parentFolder = getFolder(parentFolderId);
+ return parentFolder.createFolder(folderName);
+ } else {
+ throw new ProcessException(format("The specified subfolder '%s' does not exist and '%s' is false.", folderName, CREATE_SUBFOLDER.getDisplayName()));
+ }
+ }
+
+ private Optional<BoxFolder.Info> checkFolderExistence(final String folderName, final String parentFolderId) {
+ final BoxFolder parentFolder = getFolder(parentFolderId);
+ return StreamSupport.stream(parentFolder.getChildren("name").spliterator(), false)
+ .filter(BoxFolder.Info.class::isInstance)
+ .map(BoxFolder.Info.class::cast)
+ .filter(info -> info.getName().equals(folderName))
+ .findAny();
+ }
+
+ private Optional<BoxFile.Info> getFileByName(final String filename, final BoxFolder parentFolder) {
+ return StreamSupport.stream(parentFolder.getChildren("name").spliterator(), false)
+ .filter(BoxFile.Info.class::isInstance)
+ .map(BoxFile.Info.class::cast)
+ .filter(info -> info.getName().equals(filename))
+ .findAny();
+ }
+
+ private boolean isMultiLevelFolder(int indexOfPathSeparator, String folderName) {
+ return indexOfPathSeparator > 0 && indexOfPathSeparator < folderName.length() - 1;
+ }
+
+ private void handleUploadError(final String conflictResolution, final String filename, BoxFolder folder, final BoxAPIException e) {
+ if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
+ handleConflict(conflictResolution, filename, folder, e);
+ } else {
+ throw new ProcessException(e);
+ }
+ }
+
+ private void handleConflict(final String conflictResolution, final String filename, BoxFolder folder, final BoxAPIException e) {
+ final String path = BoxFileUtils.getPath(folder.getInfo());
+
+ if (IGNORE_RESOLUTION.equals(conflictResolution)) {
+ getLogger().info("File with the same name '{}' already exists in '%s'. Remote file is not modified due to {} being set to '{}'.",
+ filename, path, CONFLICT_RESOLUTION.getDisplayName(), conflictResolution);
+ } else if (FAIL_RESOLUTION.equals(conflictResolution)) {
+ throw new ProcessException(format("File with the same name '%s' already exists in '%s'.", filename, path), e);
Review Comment:
In a very rare case we can end up here even if the conflict resolution is REPLACE (because a file with the same name can be created by a 3rd party after we checked its existence but before trying to upload our file in the normal / non-replacing way).
For this reason, it would be safer to fail in that case too. So I would use a simple `else` case here without a condition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] turcsanyip commented on a diff in pull request #6892: NIFI-11059 PutBoxFile processor
Posted by "turcsanyip (via GitHub)" <gi...@apache.org>.
turcsanyip commented on code in PR #6892:
URL: https://github.com/apache/nifi/pull/6892#discussion_r1093774027
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/test/java/org/apache/nifi/processors/box/PutBoxFileTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Arrays.asList;
+import static java.util.Collections.emptyList;
+import static java.util.Collections.singletonList;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.provenance.ProvenanceEventType;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+public class PutBoxFileTest extends AbstractBoxFileTest {
+
+ public static final String SUBFOLDER1_ID = "aaaa";
+ public static final String SUBFOLDER2_ID = "bbbb";
+ @Mock
+ protected BoxFolder.Info mockSubfolder1Info;
+
+ @Mock
+ protected BoxFolder.Info mockSubfolder2Info;
+
+ @Mock
+ protected BoxFolder mockSubfolder1;
+
+ @Mock
+ protected BoxFolder mockSubfolder2;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ final PutBoxFile testSubject = new PutBoxFile() {
+
+ @Override
+ BoxFolder getFolder(String folderId) {
+ if (folderId.equals(SUBFOLDER1_ID)) {
+ return mockSubfolder1;
+ } else if (folderId.equals(SUBFOLDER2_ID)) {
+ return mockSubfolder2;
+ }
+ return mockBoxFolder;
Review Comment:
Minor:
It would be nicer to create a lookup map with these items and simply get the folder by `folderId` in the method.
I guess `mockBoxFolder` also has an id too.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,431 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.REPLACE;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import com.box.sdk.BoxItem;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final int NUMBER_OF_RETRIES = 5;
+ public static final int WAIT_TIME_MS = 5000;
Review Comment:
It is hard to say good numbers for the waiting periods, but I would check the folder more frequently but for less time in total.
My suggestion:
```suggestion
public static final int NUMBER_OF_RETRIES = 10;
public static final int WAIT_TIME_MS = 1000;
```
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,431 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.REPLACE;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import com.box.sdk.BoxItem;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final int NUMBER_OF_RETRIES = 5;
+ public static final int WAIT_TIME_MS = 5000;
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded." +
+ " Please see Additional Details to obtain Folder ID.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor.Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .dependsOn(SUBFOLDER_NAME)
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(ConflictResolutionStrategy.FAIL.getValue())
+ .allowableValues(ConflictResolutionStrategy.class)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ private static final int CONFLICT_RESPONSE_CODE = 409;
+ private static final int NOT_FOUND_RESPONSE_CODE = 404;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ final BoxClientService boxClientService = context.getProperty(BoxClientService.BOX_CLIENT_SERVICE).asControllerService(BoxClientService.class);
+
+ boxAPIConnection = boxClientService.getBoxApiConnection();
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+ final ConflictResolutionStrategy conflictResolution = ConflictResolutionStrategy.forValue(context.getProperty(CONFLICT_RESOLUTION).getValue());
+
+ final long startNanos = System.nanoTime();
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+ final BoxFolder parentFolder = getOrCreateDirectParentFolder(context, flowFile);
+ fullPath = BoxFileUtils.getFolderPath(parentFolder.getInfo());
+ BoxFile.Info uploadedFileInfo = null;
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ if (REPLACE.equals(conflictResolution)) {
+ uploadedFileInfo = replaceBoxFileIfExists(parentFolder, filename, rawIn, size, chunkUploadThreshold);
+ }
+
+ if (uploadedFileInfo == null) {
+ uploadedFileInfo = createBoxFile(parentFolder, filename, rawIn, size, chunkUploadThreshold);
+ }
+ } catch (BoxAPIResponseException e) {
+ if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
+ handleConflict(conflictResolution, filename, parentFolder, e);
+ } else {
+ throw e;
+ }
+ }
+
+ if (uploadedFileInfo != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFileInfo);
+ final String url = BOX_URL + uploadedFileInfo.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Upload failed: File [{}] Folder [{}] Response Code [{}]", filename, fullPath, e.getResponseCode(), e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Upload failed: File [{}], Folder [{}]", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ BoxFolder getFolder(String folderId) {
+ return new BoxFolder(boxAPIConnection, folderId);
+ }
+
+ private BoxFolder getOrCreateDirectParentFolder(ProcessContext context, FlowFile flowFile ) {
+ final String subfolderPath = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ final String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ BoxFolder parentFolder = getFolderById(folderId);
+
+ if (subfolderPath != null) {
+ final Queue<String> subFolderNames = getSubFolderNames(subfolderPath);
+ parentFolder = getOrCreateSubfolders(subFolderNames, parentFolder, createFolder);
+ }
+
+ return parentFolder;
+ }
+
+ private BoxFile.Info replaceBoxFileIfExists(BoxFolder parentFolder, String filename, final InputStream inputStream, final long size, final long chunkUploadThreshold)
+ throws IOException, InterruptedException {
+ final Optional<BoxFile> existingBoxFileInfo = getFileByName(filename, parentFolder);
+ if (existingBoxFileInfo.isPresent()) {
+ final BoxFile existingBoxFile = existingBoxFileInfo.get();
+
+ if (size > chunkUploadThreshold) {
+ return existingBoxFile.uploadLargeFile(inputStream, size);
+ } else {
+ return existingBoxFile.uploadNewVersion(inputStream);
+ }
+ }
+ return null;
+ }
+
+ private BoxFile.Info createBoxFile(BoxFolder parentFolder, String filename, InputStream inputStream, long size, final long chunkUploadThreshold)
+ throws IOException, InterruptedException {
+ if (size > chunkUploadThreshold) {
+ return parentFolder.uploadLargeFile(inputStream, filename, size);
+ } else {
+ return parentFolder.uploadFile(inputStream, filename);
+ }
+ }
+
+ private Queue<String> getSubFolderNames(String subfolderPath) {
+ final Queue<String> subfolderNames = new LinkedList<>();
+ Collections.addAll(subfolderNames, subfolderPath.split("/"));
+ return subfolderNames;
+ }
+
+ private BoxFolder getOrCreateSubfolders(Queue<String> subFolderNames, BoxFolder parentFolder, boolean createFolder) {
+ final BoxFolder newParentFolder = getOrCreateFolder(subFolderNames.poll(), parentFolder, createFolder);
+
+ if (!subFolderNames.isEmpty()) {
+ return getOrCreateSubfolders(subFolderNames, newParentFolder, createFolder);
+ } else {
+ return newParentFolder;
+ }
+ }
+
+ private BoxFolder getOrCreateFolder(String folderName, BoxFolder parentFolder, boolean createFolder) {
+ final Optional<BoxFolder> existingFolder = getFolderByName(folderName, parentFolder);
+
+ if (existingFolder.isPresent()) {
+ return existingFolder.get();
+ }
+
+ if (!createFolder) {
+ throw new ProcessException(format("The specified subfolder [%s] does not exist and [%s] is false.",
+ folderName, CREATE_SUBFOLDER.getDisplayName()));
+ }
+
+ return createFolder(folderName, parentFolder);
+ }
+
+ private BoxFolder createFolder(final String folderName, final BoxFolder parentFolder) {
+ getLogger().info("Creating Folder [{}], Parent [{}]", folderName, parentFolder.getInfo().getID());
Review Comment:
Please avoid using `getInfo()` because it always involves a server turnaround.
```suggestion
getLogger().info("Creating Folder [{}], Parent [{}]", folderName, parentFolder.getID());
```
Also below in this method and in next method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] turcsanyip commented on a diff in pull request #6892: NIFI-11059 PutBoxFile processor
Posted by "turcsanyip (via GitHub)" <gi...@apache.org>.
turcsanyip commented on code in PR #6892:
URL: https://github.com/apache/nifi/pull/6892#discussion_r1088387924
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor
+ .Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION)
+ .allowableValues(FAIL_RESOLUTION, IGNORE_RESOLUTION, REPLACE_RESOLUTION)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ public static final int CONFLICT_RESPONSE_CODE = 409;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ final String subfolderName = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ BoxFile.Info uploadedFile = null;
+
+ final long startNanos = System.nanoTime();
+
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+
+ folderId = subfolderName != null ? getOrCreateParentSubfolder(subfolderName, folderId, createFolder).getID() : folderId;
+ final BoxFolder parentFolder = getFolder(folderId);
+ fullPath = BoxFileUtils.getPath(parentFolder.getInfo());
+
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+
+ final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ boolean isNewVersionUpload = false;
+
+ if (REPLACE_RESOLUTION.equals(conflictResolution)) {
+ final Optional<BoxFile.Info> alreadyUploadedFile = getFileByName(filename, parentFolder);
+
+ if (alreadyUploadedFile.isPresent()) {
+ BoxFile existingBoxFile = new BoxFile(boxAPIConnection, alreadyUploadedFile.get().getID());
+ existingBoxFile.uploadNewVersion(rawIn);
+ isNewVersionUpload = true;
+ }
+ }
+
+ if (!isNewVersionUpload) {
+ if (size > chunkUploadThreshold) {
+ uploadedFile = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFile = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+
+ } catch (BoxAPIResponseException e) {
+ handleUploadError(conflictResolution, filename, parentFolder, e);
+ }
+
+ if (uploadedFile != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFile);
+ final String url = BOX_URL + uploadedFile.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Unexpected exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ final BoxClientService boxClientService = context.getProperty(BoxClientService.BOX_CLIENT_SERVICE).asControllerService(BoxClientService.class);
+
+ boxAPIConnection = boxClientService.getBoxApiConnection();
+ }
+
+ BoxFolder getFolder(String folderId) {
+ return new BoxFolder(boxAPIConnection, folderId);
+ }
+
+ private BoxFolder.Info getOrCreateParentSubfolder(String folderName, String parentFolderId, boolean createFolder) {
+ final int indexOfPathSeparator = folderName.indexOf("/");
+
+ if (isMultiLevelFolder(indexOfPathSeparator, folderName)) {
+ final String mainFolderName = folderName.substring(0, indexOfPathSeparator);
+ final String subFolders = folderName.substring(indexOfPathSeparator + 1);
+ final BoxFolder.Info mainFolder = getOrCreateFolder(mainFolderName, parentFolderId, createFolder);
+ return getOrCreateParentSubfolder(subFolders, mainFolder.getID(), createFolder);
+ } else {
+ return getOrCreateFolder(folderName, parentFolderId, createFolder);
+ }
+ }
+
+ private BoxFolder.Info getOrCreateFolder(String folderName, String parentFolderId, boolean createFolder) {
+ final Optional<BoxFolder.Info> existingFolder = checkFolderExistence(folderName, parentFolderId);
+
+ if (existingFolder.isPresent()) {
+ return existingFolder.get();
+ }
+
+ if (createFolder) {
+ getLogger().debug("Create folder " + folderName + " parent id: " + parentFolderId);
+
+ final BoxFolder parentFolder = getFolder(parentFolderId);
+ return parentFolder.createFolder(folderName);
+ } else {
+ throw new ProcessException(format("The specified subfolder '%s' does not exist and '%s' is false.", folderName, CREATE_SUBFOLDER.getDisplayName()));
+ }
+ }
+
+ private Optional<BoxFolder.Info> checkFolderExistence(final String folderName, final String parentFolderId) {
+ final BoxFolder parentFolder = getFolder(parentFolderId);
+ return StreamSupport.stream(parentFolder.getChildren("name").spliterator(), false)
+ .filter(BoxFolder.Info.class::isInstance)
+ .map(BoxFolder.Info.class::cast)
+ .filter(info -> info.getName().equals(folderName))
+ .findAny();
+ }
+
+ private Optional<BoxFile.Info> getFileByName(final String filename, final BoxFolder parentFolder) {
+ return StreamSupport.stream(parentFolder.getChildren("name").spliterator(), false)
+ .filter(BoxFile.Info.class::isInstance)
+ .map(BoxFile.Info.class::cast)
+ .filter(info -> info.getName().equals(filename))
+ .findAny();
+ }
+
+ private boolean isMultiLevelFolder(int indexOfPathSeparator, String folderName) {
+ return indexOfPathSeparator > 0 && indexOfPathSeparator < folderName.length() - 1;
+ }
+
+ private void handleUploadError(final String conflictResolution, final String filename, BoxFolder folder, final BoxAPIException e) {
+ if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
+ handleConflict(conflictResolution, filename, folder, e);
+ } else {
+ throw new ProcessException(e);
Review Comment:
I would not wrap the exception here at all, just throw the original one. This way it would land in the right `catch` block of the outer `try-catch` where `BoxAPIResponseException`-s are handled.
It may be weird to rethrow an exception that was passed in as a parameter though. For this reason, it may be better to move this `if-else` back to the inner catch block instead of the separate method.
```
} catch (BoxAPIResponseException e) {
if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
handleConflict(conflictResolution, filename, parentFolder, e);
} else {
throw e;
}
}
session.transfer(flowFile, REL_SUCCESS);
} catch (BoxAPIResponseException e) {
getLogger().error("Exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
handleExpectedError(session, flowFile, e);
}
```
I think this way it is more straightforward which `BoxAPIResponseException` is handled specially and which go to the common handler block.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] asfgit closed pull request #6892: NIFI-11059 PutBoxFile processor
Posted by "asfgit (via GitHub)" <gi...@apache.org>.
asfgit closed pull request #6892: NIFI-11059 PutBoxFile processor
URL: https://github.com/apache/nifi/pull/6892
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #6892: NIFI-11059 PutBoxFile processor
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #6892:
URL: https://github.com/apache/nifi/pull/6892#discussion_r1088329051
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/test/java/org/apache/nifi/processors/box/FetchBoxFileTest.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.box;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.verify;
+
+import com.box.sdk.BoxFile;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.nifi.provenance.ProvenanceEventType;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+public class FetchBoxFileTest extends AbstractBoxFileTest{
+ @Mock
+ BoxFile mockBoxFile;
+
+ @BeforeEach
+ void setUp() throws Exception {
+
+ final FetchBoxFile testSubject = new FetchBoxFile() {
+ @Override
+ protected BoxFile getBoxFile(String fileId) {
+ return mockBoxFile;
+ }
+ };
+
+ testRunner = TestRunners.newTestRunner(testSubject);
+ super.setUp();
+ }
+
+ @Test
+ void testBoxIdFromFlowFileAttribute() {
+ testRunner.setProperty(FetchBoxFile.FILE_ID, "${box.id}");
+ final MockFlowFile inputFlowFile = new MockFlowFile(0);
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(BoxFileAttributes.ID, TEST_FILE_ID);
+ inputFlowFile.putAttributes(attributes);
+
+ final BoxFile.Info fetchedFileInfo = createFileInfo(TEST_PATH, MODIFIED_TIME);
+ doReturn(fetchedFileInfo).when(mockBoxFile).getInfo();
+
+ // WHEN
Review Comment:
The use of `WHEN` and `THEN` comments should be avoided.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/BoxFileAttributes.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.box;
+
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+
+public class BoxFileAttributes {
+
+ public static final String ID = "box.id";
+ public static final String ID_DESC = "The id of the file";
+
+ public static final String FILENAME = CoreAttributes.FILENAME.key();
Review Comment:
I recommend avoiding reproducing the `FILENAME` and `PATH` attributes and using the `CoreAttributes` enum directly.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/BoxFileUtils.java:
##########
@@ -0,0 +1,48 @@
+/*
+ * 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.box;
+
+import static java.lang.String.valueOf;
+
+import com.box.sdk.BoxFile.Info;
+import com.box.sdk.BoxItem;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public final class BoxFileUtils {
+
+ public static final String BOX_URL = "https://app.box.com/file/";
+
+ public static String getPath(BoxItem.Info info) {
+ return "/" + info.getPathCollection().stream()
+ .filter(pathItemInfo -> !pathItemInfo.getID().equals("0"))
+ .map(BoxItem.Info::getName)
+ .collect(Collectors.joining("/"));
+ }
+
+ public static Map<String, String> createAttributeMap(Info fileInfo) {
+ final Map<String, String> attributes = new HashMap<>();
Review Comment:
Recommend using `LinkedHashMap` instead of `HashMap` for deterministic ordering of attributes added.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/FetchBoxFile.java:
##########
@@ -114,40 +128,50 @@ public void onTrigger(ProcessContext context, ProcessSession session) throws Pro
}
String fileId = context.getProperty(FILE_ID).evaluateAttributeExpressions(flowFile).getValue();
- FlowFile outFlowFile = flowFile;
try {
- outFlowFile = fetchFile(fileId, session, outFlowFile);
-
- session.transfer(outFlowFile, REL_SUCCESS);
+ final long startNanos = System.nanoTime();
+ flowFile = fetchFile(fileId, session, flowFile);
+ final String boxUrlOfFile = BoxFileUtils.BOX_URL + fileId;
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().fetch(flowFile, boxUrlOfFile, transferMillis);
+ session.transfer(flowFile, REL_SUCCESS);
} catch (BoxAPIResponseException e) {
handleErrorResponse(session, fileId, flowFile, e);
} catch (Exception e) {
handleUnexpectedError(session, flowFile, fileId, e);
}
}
- FlowFile fetchFile(String fileId, ProcessSession session, FlowFile outFlowFile) {
- BoxFile boxFile = new BoxFile(boxAPIConnection, fileId);
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
- outFlowFile = session.write(outFlowFile, outputStream -> boxFile.download(outputStream));
+ BoxFile getBoxFile(String fileId) {
+ return new BoxFile(boxAPIConnection, fileId);
+ }
- return outFlowFile;
+ FlowFile fetchFile(String fileId, ProcessSession session, FlowFile flowFile) {
+ final BoxFile boxFile = getBoxFile(fileId);
+ flowFile = session.write(flowFile, outputStream -> boxFile.download(outputStream));
+ flowFile = session.putAllAttributes(flowFile, BoxFileUtils.createAttributeMap(boxFile.getInfo()));
+ return flowFile;
}
- private void handleErrorResponse(ProcessSession session, String fileId, FlowFile outFlowFile, BoxAPIResponseException e) {
+ private void handleErrorResponse(ProcessSession session, String fileId, FlowFile flowFile, BoxAPIResponseException e) {
getLogger().error("Couldn't fetch file with id '{}'", fileId, e);
- outFlowFile = session.putAttribute(outFlowFile, ERROR_CODE_ATTRIBUTE, "" + e.getResponseCode());
- outFlowFile = session.putAttribute(outFlowFile, ERROR_MESSAGE_ATTRIBUTE, e.getMessage());
-
- session.transfer(outFlowFile, REL_FAILURE);
+ flowFile = session.putAttribute(flowFile, ERROR_CODE, "" + e.getResponseCode());
Review Comment:
This is a good opportunity to adjust this approach. `String.valueOf(e.getResponseCode())` should be used instead of concatenating an empty string to the response code.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor
+ .Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION)
+ .allowableValues(FAIL_RESOLUTION, IGNORE_RESOLUTION, REPLACE_RESOLUTION)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ public static final int CONFLICT_RESPONSE_CODE = 409;
Review Comment:
Can this variable be `protected` or `private`?
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor
+ .Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION)
+ .allowableValues(FAIL_RESOLUTION, IGNORE_RESOLUTION, REPLACE_RESOLUTION)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ public static final int CONFLICT_RESPONSE_CODE = 409;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ final String subfolderName = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ BoxFile.Info uploadedFile = null;
+
+ final long startNanos = System.nanoTime();
+
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+
+ folderId = subfolderName != null ? getOrCreateParentSubfolder(subfolderName, folderId, createFolder).getID() : folderId;
+ final BoxFolder parentFolder = getFolder(folderId);
+ fullPath = BoxFileUtils.getPath(parentFolder.getInfo());
+
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+
+ final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ boolean isNewVersionUpload = false;
+
+ if (REPLACE_RESOLUTION.equals(conflictResolution)) {
+ final Optional<BoxFile.Info> alreadyUploadedFile = getFileByName(filename, parentFolder);
+
+ if (alreadyUploadedFile.isPresent()) {
+ BoxFile existingBoxFile = new BoxFile(boxAPIConnection, alreadyUploadedFile.get().getID());
+ existingBoxFile.uploadNewVersion(rawIn);
+ isNewVersionUpload = true;
+ }
+ }
+
+ if (!isNewVersionUpload) {
+ if (size > chunkUploadThreshold) {
+ uploadedFile = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFile = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+
+ } catch (BoxAPIResponseException e) {
+ handleUploadError(conflictResolution, filename, parentFolder, e);
+ }
+
+ if (uploadedFile != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFile);
+ final String url = BOX_URL + uploadedFile.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Unexpected exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
Review Comment:
```suggestion
getLogger().error("Upload failed: File [{}] folder [{}]", filename, fullPath, e);
```
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor
+ .Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION)
+ .allowableValues(FAIL_RESOLUTION, IGNORE_RESOLUTION, REPLACE_RESOLUTION)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ public static final int CONFLICT_RESPONSE_CODE = 409;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ final String subfolderName = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ BoxFile.Info uploadedFile = null;
+
+ final long startNanos = System.nanoTime();
+
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+
+ folderId = subfolderName != null ? getOrCreateParentSubfolder(subfolderName, folderId, createFolder).getID() : folderId;
+ final BoxFolder parentFolder = getFolder(folderId);
+ fullPath = BoxFileUtils.getPath(parentFolder.getInfo());
+
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+
+ final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ boolean isNewVersionUpload = false;
+
+ if (REPLACE_RESOLUTION.equals(conflictResolution)) {
+ final Optional<BoxFile.Info> alreadyUploadedFile = getFileByName(filename, parentFolder);
+
+ if (alreadyUploadedFile.isPresent()) {
+ BoxFile existingBoxFile = new BoxFile(boxAPIConnection, alreadyUploadedFile.get().getID());
+ existingBoxFile.uploadNewVersion(rawIn);
+ isNewVersionUpload = true;
+ }
+ }
+
+ if (!isNewVersionUpload) {
+ if (size > chunkUploadThreshold) {
+ uploadedFile = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFile = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+
+ } catch (BoxAPIResponseException e) {
+ handleUploadError(conflictResolution, filename, parentFolder, e);
+ }
+
+ if (uploadedFile != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFile);
+ final String url = BOX_URL + uploadedFile.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
Review Comment:
Recommend avoiding the wording `Exception occurred` and instead using something like the following. Also recommend including the Response Code for easier troubleshooting.
```suggestion
getLogger().error("Upload failed: File [{}] Folder [{}] Response Code [{}]", filename, fullPath, e.getResponseCode(), e);
```
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor
+ .Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION)
+ .allowableValues(FAIL_RESOLUTION, IGNORE_RESOLUTION, REPLACE_RESOLUTION)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ public static final int CONFLICT_RESPONSE_CODE = 409;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ final String subfolderName = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ BoxFile.Info uploadedFile = null;
+
+ final long startNanos = System.nanoTime();
+
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+
+ folderId = subfolderName != null ? getOrCreateParentSubfolder(subfolderName, folderId, createFolder).getID() : folderId;
+ final BoxFolder parentFolder = getFolder(folderId);
+ fullPath = BoxFileUtils.getPath(parentFolder.getInfo());
+
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+
+ final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ boolean isNewVersionUpload = false;
+
+ if (REPLACE_RESOLUTION.equals(conflictResolution)) {
+ final Optional<BoxFile.Info> alreadyUploadedFile = getFileByName(filename, parentFolder);
+
+ if (alreadyUploadedFile.isPresent()) {
+ BoxFile existingBoxFile = new BoxFile(boxAPIConnection, alreadyUploadedFile.get().getID());
+ existingBoxFile.uploadNewVersion(rawIn);
+ isNewVersionUpload = true;
+ }
+ }
+
+ if (!isNewVersionUpload) {
+ if (size > chunkUploadThreshold) {
+ uploadedFile = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFile = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+
+ } catch (BoxAPIResponseException e) {
+ handleUploadError(conflictResolution, filename, parentFolder, e);
+ }
+
+ if (uploadedFile != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFile);
+ final String url = BOX_URL + uploadedFile.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Unexpected exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ final BoxClientService boxClientService = context.getProperty(BoxClientService.BOX_CLIENT_SERVICE).asControllerService(BoxClientService.class);
+
+ boxAPIConnection = boxClientService.getBoxApiConnection();
+ }
+
+ BoxFolder getFolder(String folderId) {
+ return new BoxFolder(boxAPIConnection, folderId);
+ }
+
+ private BoxFolder.Info getOrCreateParentSubfolder(String folderName, String parentFolderId, boolean createFolder) {
+ final int indexOfPathSeparator = folderName.indexOf("/");
+
+ if (isMultiLevelFolder(indexOfPathSeparator, folderName)) {
+ final String mainFolderName = folderName.substring(0, indexOfPathSeparator);
+ final String subFolders = folderName.substring(indexOfPathSeparator + 1);
+ final BoxFolder.Info mainFolder = getOrCreateFolder(mainFolderName, parentFolderId, createFolder);
+ return getOrCreateParentSubfolder(subFolders, mainFolder.getID(), createFolder);
+ } else {
+ return getOrCreateFolder(folderName, parentFolderId, createFolder);
+ }
+ }
+
+ private BoxFolder.Info getOrCreateFolder(String folderName, String parentFolderId, boolean createFolder) {
+ final Optional<BoxFolder.Info> existingFolder = checkFolderExistence(folderName, parentFolderId);
+
+ if (existingFolder.isPresent()) {
+ return existingFolder.get();
+ }
+
+ if (createFolder) {
+ getLogger().debug("Create folder " + folderName + " parent id: " + parentFolderId);
Review Comment:
Log statements should use placeholders:
```suggestion
getLogger().debug("Creating Folder [{}] Parent [{}]", folderName, parentFolderId);
```
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor
+ .Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION)
+ .allowableValues(FAIL_RESOLUTION, IGNORE_RESOLUTION, REPLACE_RESOLUTION)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ public static final int CONFLICT_RESPONSE_CODE = 409;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ final String subfolderName = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ BoxFile.Info uploadedFile = null;
+
+ final long startNanos = System.nanoTime();
+
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+
+ folderId = subfolderName != null ? getOrCreateParentSubfolder(subfolderName, folderId, createFolder).getID() : folderId;
+ final BoxFolder parentFolder = getFolder(folderId);
+ fullPath = BoxFileUtils.getPath(parentFolder.getInfo());
+
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+
+ final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ boolean isNewVersionUpload = false;
+
+ if (REPLACE_RESOLUTION.equals(conflictResolution)) {
+ final Optional<BoxFile.Info> alreadyUploadedFile = getFileByName(filename, parentFolder);
+
+ if (alreadyUploadedFile.isPresent()) {
+ BoxFile existingBoxFile = new BoxFile(boxAPIConnection, alreadyUploadedFile.get().getID());
+ existingBoxFile.uploadNewVersion(rawIn);
+ isNewVersionUpload = true;
+ }
+ }
+
+ if (!isNewVersionUpload) {
+ if (size > chunkUploadThreshold) {
+ uploadedFile = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFile = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+
+ } catch (BoxAPIResponseException e) {
+ handleUploadError(conflictResolution, filename, parentFolder, e);
+ }
+
+ if (uploadedFile != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFile);
+ final String url = BOX_URL + uploadedFile.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Unexpected exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ final BoxClientService boxClientService = context.getProperty(BoxClientService.BOX_CLIENT_SERVICE).asControllerService(BoxClientService.class);
+
+ boxAPIConnection = boxClientService.getBoxApiConnection();
+ }
+
+ BoxFolder getFolder(String folderId) {
+ return new BoxFolder(boxAPIConnection, folderId);
+ }
+
+ private BoxFolder.Info getOrCreateParentSubfolder(String folderName, String parentFolderId, boolean createFolder) {
+ final int indexOfPathSeparator = folderName.indexOf("/");
+
+ if (isMultiLevelFolder(indexOfPathSeparator, folderName)) {
+ final String mainFolderName = folderName.substring(0, indexOfPathSeparator);
+ final String subFolders = folderName.substring(indexOfPathSeparator + 1);
+ final BoxFolder.Info mainFolder = getOrCreateFolder(mainFolderName, parentFolderId, createFolder);
+ return getOrCreateParentSubfolder(subFolders, mainFolder.getID(), createFolder);
+ } else {
+ return getOrCreateFolder(folderName, parentFolderId, createFolder);
+ }
+ }
+
+ private BoxFolder.Info getOrCreateFolder(String folderName, String parentFolderId, boolean createFolder) {
+ final Optional<BoxFolder.Info> existingFolder = checkFolderExistence(folderName, parentFolderId);
+
+ if (existingFolder.isPresent()) {
+ return existingFolder.get();
+ }
+
+ if (createFolder) {
+ getLogger().debug("Create folder " + folderName + " parent id: " + parentFolderId);
+
+ final BoxFolder parentFolder = getFolder(parentFolderId);
+ return parentFolder.createFolder(folderName);
+ } else {
+ throw new ProcessException(format("The specified subfolder '%s' does not exist and '%s' is false.", folderName, CREATE_SUBFOLDER.getDisplayName()));
+ }
+ }
+
+ private Optional<BoxFolder.Info> checkFolderExistence(final String folderName, final String parentFolderId) {
+ final BoxFolder parentFolder = getFolder(parentFolderId);
+ return StreamSupport.stream(parentFolder.getChildren("name").spliterator(), false)
+ .filter(BoxFolder.Info.class::isInstance)
+ .map(BoxFolder.Info.class::cast)
+ .filter(info -> info.getName().equals(folderName))
+ .findAny();
+ }
+
+ private Optional<BoxFile.Info> getFileByName(final String filename, final BoxFolder parentFolder) {
+ return StreamSupport.stream(parentFolder.getChildren("name").spliterator(), false)
+ .filter(BoxFile.Info.class::isInstance)
+ .map(BoxFile.Info.class::cast)
+ .filter(info -> info.getName().equals(filename))
+ .findAny();
+ }
+
+ private boolean isMultiLevelFolder(int indexOfPathSeparator, String folderName) {
+ return indexOfPathSeparator > 0 && indexOfPathSeparator < folderName.length() - 1;
+ }
+
+ private void handleUploadError(final String conflictResolution, final String filename, BoxFolder folder, final BoxAPIException e) {
+ if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
+ handleConflict(conflictResolution, filename, folder, e);
+ } else {
+ throw new ProcessException(e);
+ }
+ }
+
+ private void handleConflict(final String conflictResolution, final String filename, BoxFolder folder, final BoxAPIException e) {
+ final String path = BoxFileUtils.getPath(folder.getInfo());
+
+ if (IGNORE_RESOLUTION.equals(conflictResolution)) {
+ getLogger().info("File with the same name '{}' already exists in '%s'. Remote file is not modified due to {} being set to '{}'.",
+ filename, path, CONFLICT_RESOLUTION.getDisplayName(), conflictResolution);
+ } else if (FAIL_RESOLUTION.equals(conflictResolution)) {
+ throw new ProcessException(format("File with the same name '%s' already exists in '%s'.", filename, path), e);
Review Comment:
Recommend avoiding periods at the end of messages:
```suggestion
throw new ProcessException(format("File with the same name [%s] already exists in [%s]", filename, path), e);
```
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor
+ .Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION)
+ .allowableValues(FAIL_RESOLUTION, IGNORE_RESOLUTION, REPLACE_RESOLUTION)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ public static final int CONFLICT_RESPONSE_CODE = 409;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ final String subfolderName = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ BoxFile.Info uploadedFile = null;
+
+ final long startNanos = System.nanoTime();
+
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+
+ folderId = subfolderName != null ? getOrCreateParentSubfolder(subfolderName, folderId, createFolder).getID() : folderId;
+ final BoxFolder parentFolder = getFolder(folderId);
+ fullPath = BoxFileUtils.getPath(parentFolder.getInfo());
+
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+
+ final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ boolean isNewVersionUpload = false;
+
+ if (REPLACE_RESOLUTION.equals(conflictResolution)) {
+ final Optional<BoxFile.Info> alreadyUploadedFile = getFileByName(filename, parentFolder);
+
+ if (alreadyUploadedFile.isPresent()) {
+ BoxFile existingBoxFile = new BoxFile(boxAPIConnection, alreadyUploadedFile.get().getID());
+ existingBoxFile.uploadNewVersion(rawIn);
+ isNewVersionUpload = true;
+ }
+ }
+
+ if (!isNewVersionUpload) {
+ if (size > chunkUploadThreshold) {
+ uploadedFile = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFile = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+
+ } catch (BoxAPIResponseException e) {
+ handleUploadError(conflictResolution, filename, parentFolder, e);
+ }
+
+ if (uploadedFile != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFile);
+ final String url = BOX_URL + uploadedFile.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Unexpected exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ final BoxClientService boxClientService = context.getProperty(BoxClientService.BOX_CLIENT_SERVICE).asControllerService(BoxClientService.class);
+
+ boxAPIConnection = boxClientService.getBoxApiConnection();
+ }
+
+ BoxFolder getFolder(String folderId) {
+ return new BoxFolder(boxAPIConnection, folderId);
+ }
+
+ private BoxFolder.Info getOrCreateParentSubfolder(String folderName, String parentFolderId, boolean createFolder) {
+ final int indexOfPathSeparator = folderName.indexOf("/");
+
+ if (isMultiLevelFolder(indexOfPathSeparator, folderName)) {
+ final String mainFolderName = folderName.substring(0, indexOfPathSeparator);
+ final String subFolders = folderName.substring(indexOfPathSeparator + 1);
+ final BoxFolder.Info mainFolder = getOrCreateFolder(mainFolderName, parentFolderId, createFolder);
+ return getOrCreateParentSubfolder(subFolders, mainFolder.getID(), createFolder);
+ } else {
+ return getOrCreateFolder(folderName, parentFolderId, createFolder);
+ }
+ }
+
+ private BoxFolder.Info getOrCreateFolder(String folderName, String parentFolderId, boolean createFolder) {
+ final Optional<BoxFolder.Info> existingFolder = checkFolderExistence(folderName, parentFolderId);
+
+ if (existingFolder.isPresent()) {
+ return existingFolder.get();
+ }
+
+ if (createFolder) {
+ getLogger().debug("Create folder " + folderName + " parent id: " + parentFolderId);
+
+ final BoxFolder parentFolder = getFolder(parentFolderId);
+ return parentFolder.createFolder(folderName);
+ } else {
+ throw new ProcessException(format("The specified subfolder '%s' does not exist and '%s' is false.", folderName, CREATE_SUBFOLDER.getDisplayName()));
+ }
+ }
+
+ private Optional<BoxFolder.Info> checkFolderExistence(final String folderName, final String parentFolderId) {
+ final BoxFolder parentFolder = getFolder(parentFolderId);
+ return StreamSupport.stream(parentFolder.getChildren("name").spliterator(), false)
+ .filter(BoxFolder.Info.class::isInstance)
+ .map(BoxFolder.Info.class::cast)
+ .filter(info -> info.getName().equals(folderName))
+ .findAny();
+ }
+
+ private Optional<BoxFile.Info> getFileByName(final String filename, final BoxFolder parentFolder) {
+ return StreamSupport.stream(parentFolder.getChildren("name").spliterator(), false)
+ .filter(BoxFile.Info.class::isInstance)
+ .map(BoxFile.Info.class::cast)
+ .filter(info -> info.getName().equals(filename))
+ .findAny();
+ }
+
+ private boolean isMultiLevelFolder(int indexOfPathSeparator, String folderName) {
+ return indexOfPathSeparator > 0 && indexOfPathSeparator < folderName.length() - 1;
+ }
+
+ private void handleUploadError(final String conflictResolution, final String filename, BoxFolder folder, final BoxAPIException e) {
+ if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
+ handleConflict(conflictResolution, filename, folder, e);
+ } else {
+ throw new ProcessException(e);
Review Comment:
Recommend including a specific message along with the wrapped exception.
```suggestion
throw new ProcessException(String.format("File [%s] upload failed", filename), e);
```
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor
+ .Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION)
+ .allowableValues(FAIL_RESOLUTION, IGNORE_RESOLUTION, REPLACE_RESOLUTION)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ public static final int CONFLICT_RESPONSE_CODE = 409;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ final String subfolderName = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ BoxFile.Info uploadedFile = null;
+
+ final long startNanos = System.nanoTime();
+
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+
+ folderId = subfolderName != null ? getOrCreateParentSubfolder(subfolderName, folderId, createFolder).getID() : folderId;
+ final BoxFolder parentFolder = getFolder(folderId);
+ fullPath = BoxFileUtils.getPath(parentFolder.getInfo());
+
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+
+ final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ boolean isNewVersionUpload = false;
+
+ if (REPLACE_RESOLUTION.equals(conflictResolution)) {
+ final Optional<BoxFile.Info> alreadyUploadedFile = getFileByName(filename, parentFolder);
+
+ if (alreadyUploadedFile.isPresent()) {
+ BoxFile existingBoxFile = new BoxFile(boxAPIConnection, alreadyUploadedFile.get().getID());
+ existingBoxFile.uploadNewVersion(rawIn);
+ isNewVersionUpload = true;
+ }
+ }
+
+ if (!isNewVersionUpload) {
+ if (size > chunkUploadThreshold) {
+ uploadedFile = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFile = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+
+ } catch (BoxAPIResponseException e) {
+ handleUploadError(conflictResolution, filename, parentFolder, e);
+ }
+
+ if (uploadedFile != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFile);
+ final String url = BOX_URL + uploadedFile.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Unexpected exception occurred while uploading file '{}' to Box folder '{}'", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ final BoxClientService boxClientService = context.getProperty(BoxClientService.BOX_CLIENT_SERVICE).asControllerService(BoxClientService.class);
+
+ boxAPIConnection = boxClientService.getBoxApiConnection();
+ }
+
+ BoxFolder getFolder(String folderId) {
+ return new BoxFolder(boxAPIConnection, folderId);
+ }
+
+ private BoxFolder.Info getOrCreateParentSubfolder(String folderName, String parentFolderId, boolean createFolder) {
+ final int indexOfPathSeparator = folderName.indexOf("/");
+
+ if (isMultiLevelFolder(indexOfPathSeparator, folderName)) {
+ final String mainFolderName = folderName.substring(0, indexOfPathSeparator);
+ final String subFolders = folderName.substring(indexOfPathSeparator + 1);
+ final BoxFolder.Info mainFolder = getOrCreateFolder(mainFolderName, parentFolderId, createFolder);
+ return getOrCreateParentSubfolder(subFolders, mainFolder.getID(), createFolder);
+ } else {
+ return getOrCreateFolder(folderName, parentFolderId, createFolder);
+ }
+ }
+
+ private BoxFolder.Info getOrCreateFolder(String folderName, String parentFolderId, boolean createFolder) {
+ final Optional<BoxFolder.Info> existingFolder = checkFolderExistence(folderName, parentFolderId);
+
+ if (existingFolder.isPresent()) {
+ return existingFolder.get();
+ }
+
+ if (createFolder) {
+ getLogger().debug("Create folder " + folderName + " parent id: " + parentFolderId);
+
+ final BoxFolder parentFolder = getFolder(parentFolderId);
+ return parentFolder.createFolder(folderName);
+ } else {
+ throw new ProcessException(format("The specified subfolder '%s' does not exist and '%s' is false.", folderName, CREATE_SUBFOLDER.getDisplayName()));
+ }
+ }
+
+ private Optional<BoxFolder.Info> checkFolderExistence(final String folderName, final String parentFolderId) {
+ final BoxFolder parentFolder = getFolder(parentFolderId);
+ return StreamSupport.stream(parentFolder.getChildren("name").spliterator(), false)
+ .filter(BoxFolder.Info.class::isInstance)
+ .map(BoxFolder.Info.class::cast)
+ .filter(info -> info.getName().equals(folderName))
+ .findAny();
+ }
+
+ private Optional<BoxFile.Info> getFileByName(final String filename, final BoxFolder parentFolder) {
+ return StreamSupport.stream(parentFolder.getChildren("name").spliterator(), false)
+ .filter(BoxFile.Info.class::isInstance)
+ .map(BoxFile.Info.class::cast)
+ .filter(info -> info.getName().equals(filename))
+ .findAny();
+ }
+
+ private boolean isMultiLevelFolder(int indexOfPathSeparator, String folderName) {
+ return indexOfPathSeparator > 0 && indexOfPathSeparator < folderName.length() - 1;
+ }
+
+ private void handleUploadError(final String conflictResolution, final String filename, BoxFolder folder, final BoxAPIException e) {
+ if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
+ handleConflict(conflictResolution, filename, folder, e);
+ } else {
+ throw new ProcessException(e);
+ }
+ }
+
+ private void handleConflict(final String conflictResolution, final String filename, BoxFolder folder, final BoxAPIException e) {
+ final String path = BoxFileUtils.getPath(folder.getInfo());
+
+ if (IGNORE_RESOLUTION.equals(conflictResolution)) {
+ getLogger().info("File with the same name '{}' already exists in '%s'. Remote file is not modified due to {} being set to '{}'.",
Review Comment:
```suggestion
getLogger().info("File with the same name [{}] already exists in [%s] Remote file is not modified due to [{}] being set to [{}]",
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #6892: NIFI-11059 PutBoxFile processor
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #6892:
URL: https://github.com/apache/nifi/pull/6892#discussion_r1088363732
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,366 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String FAIL_RESOLUTION = "fail";
Review Comment:
Based on your work for NIFI-11068 in PR #6861, it looks like these variables can be replaced with the new `ConflictResolutionStrategy`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] turcsanyip commented on a diff in pull request #6892: NIFI-11059 PutBoxFile processor
Posted by "turcsanyip (via GitHub)" <gi...@apache.org>.
turcsanyip commented on code in PR #6892:
URL: https://github.com/apache/nifi/pull/6892#discussion_r1094353463
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/BoxFileAttributes.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.box;
+
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+
+public class BoxFileAttributes {
+
+ public static final String ID = "box.id";
+ public static final String ID_DESC = "The id of the file";
+
+ public static final String FILENAME = CoreAttributes.FILENAME.key();
+ public static final String FILENAME_DESC = "The name of the file";
+
+ public static final String PATH = CoreAttributes.PATH.key();
+ public static final String PATH_DESC = "The folder path where the file is located";
Review Comment:
It seems a better option to introduce special annotations for core attributes in a separate jira.
Closing this thread.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] turcsanyip commented on a diff in pull request #6892: NIFI-11059 PutBoxFile processor
Posted by "turcsanyip (via GitHub)" <gi...@apache.org>.
turcsanyip commented on code in PR #6892:
URL: https://github.com/apache/nifi/pull/6892#discussion_r1090685097
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/resources/docs/org.apache.nifi.processors.box.PutBoxFile/additionalDetails.html:
##########
@@ -0,0 +1,45 @@
+<!DOCTYPE html>
+<html lang="en" xmlns="http://www.w3.org/1999/html">
+<!--
+ 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.
+ -->
+
+<head>
+ <meta charset="utf-8"/>
+ <title>PutBoxFile</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
+</head>
+<body>
+
+<h1>Upload files to Box from NiFi</h1>
+
+<ol>
+ <li><b>Find Folder ID</b>
+ <ul>
+ <li>Navigate to the folder in Box and enter it. The URL in your browser will include the ID at the end of
+ the URL.
+ For example, if the URL were <code>https://app.box.com/folder/191632099757</code>, the
+ Folder ID would be <code>191632099757</code>
+ </li>
+ </ul>
+ </li>
+ <li><b>Set Folder ID in 'Folder ID' property</b>
+ </li>
+</ol>
+
+</body>
+</html>
+
+</body>
+</html>
Review Comment:
These tags are duplicated at the end of the HTML file.
Could you please remove them? Also on the List processor's page.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.REPLACE;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import com.box.sdk.BoxFolder.Info;
+import com.box.sdk.BoxItem;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded." +
+ " Please see Additional Details to obtain Folder ID.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor.Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .dependsOn(SUBFOLDER_NAME)
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(ConflictResolutionStrategy.FAIL.getValue())
+ .allowableValues(ConflictResolutionStrategy.class)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ private static final int CONFLICT_RESPONSE_CODE = 409;
+ private static final int NOT_FOUND_RESPONSE_CODE = 404;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+ final ConflictResolutionStrategy conflictResolution = ConflictResolutionStrategy.forValue(context.getProperty(CONFLICT_RESOLUTION).getValue());
+
+ final long startNanos = System.nanoTime();
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+ final BoxFolder parentFolder = getOrCreateDirectParentFolder(context, flowFile);
+ fullPath = BoxFileUtils.getFolderPath(parentFolder.getInfo());
+ BoxFile.Info uploadedFileInfo = null;
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ if (REPLACE.equals(conflictResolution)) {
+ uploadedFileInfo = replaceExistingBoxFile(parentFolder, filename, rawIn, size, chunkUploadThreshold);
+ }
+
+ if (uploadedFileInfo == null) {
+ if (size > chunkUploadThreshold) {
+ uploadedFileInfo = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFileInfo = parentFolder.uploadFile(rawIn, filename);
+ }
Review Comment:
It might make sense extracting it into `createBoxFile()` (similar to the replace case).
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.REPLACE;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import com.box.sdk.BoxFolder.Info;
+import com.box.sdk.BoxItem;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded." +
+ " Please see Additional Details to obtain Folder ID.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor.Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .dependsOn(SUBFOLDER_NAME)
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(ConflictResolutionStrategy.FAIL.getValue())
+ .allowableValues(ConflictResolutionStrategy.class)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ private static final int CONFLICT_RESPONSE_CODE = 409;
+ private static final int NOT_FOUND_RESPONSE_CODE = 404;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+ final ConflictResolutionStrategy conflictResolution = ConflictResolutionStrategy.forValue(context.getProperty(CONFLICT_RESOLUTION).getValue());
+
+ final long startNanos = System.nanoTime();
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+ final BoxFolder parentFolder = getOrCreateDirectParentFolder(context, flowFile);
+ fullPath = BoxFileUtils.getFolderPath(parentFolder.getInfo());
+ BoxFile.Info uploadedFileInfo = null;
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ if (REPLACE.equals(conflictResolution)) {
+ uploadedFileInfo = replaceExistingBoxFile(parentFolder, filename, rawIn, size, chunkUploadThreshold);
+ }
+
+ if (uploadedFileInfo == null) {
+ if (size > chunkUploadThreshold) {
+ uploadedFileInfo = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFileInfo = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+ } catch (BoxAPIResponseException e) {
+ if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
+ handleConflict(conflictResolution, filename, parentFolder, e);
+ } else {
+ throw e;
+ }
+ }
+
+ if (uploadedFileInfo != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFileInfo);
+ final String url = BOX_URL + uploadedFileInfo.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Upload failed: File [{}] Folder [{}] Response Code [{}]", filename, fullPath, e.getResponseCode(), e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Upload failed: File [{}], Folder [{}]", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ private BoxFolder getOrCreateDirectParentFolder(ProcessContext context, FlowFile flowFile ) {
+ final String subfolderPath = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ final String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ BoxFolder parentFolder = getFolderById(folderId);
+
+ if (subfolderPath != null) {
+ final Queue<String> subFolderNames = getSubFolderNames(subfolderPath);
+ parentFolder = getFolder(getOrCreateSubfolders(subFolderNames, parentFolder, createFolder).getID());
+ }
+
+ return parentFolder;
+ }
+
+ private BoxFile.Info replaceExistingBoxFile(BoxFolder parentFolder, String filename, final InputStream rawIn, final long size, final long chunkUploadThreshold)
+ throws IOException, InterruptedException {
+ final Optional<BoxFile.Info> existingBoxFileInfo = getFileByName(filename, parentFolder);
+ if (existingBoxFileInfo.isPresent()) {
+ final BoxFile existingBoxFile = new BoxFile(boxAPIConnection, existingBoxFileInfo.get().getID());
+
+ if (size > chunkUploadThreshold) {
+ return existingBoxFile.uploadLargeFile(rawIn, size);
+ } else {
+ return existingBoxFile.uploadNewVersion(rawIn);
+ }
+ }
+ return null;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ final BoxClientService boxClientService = context.getProperty(BoxClientService.BOX_CLIENT_SERVICE).asControllerService(BoxClientService.class);
+
+ boxAPIConnection = boxClientService.getBoxApiConnection();
+ }
+
+ BoxFolder getFolder(String folderId) {
+ return new BoxFolder(boxAPIConnection, folderId);
+ }
+
+ private Queue<String> getSubFolderNames(String subfolderPath) {
+ final Queue<String> subfolderNames = new LinkedList<>();
+ Collections.addAll(subfolderNames, subfolderPath.split("/"));
+ return subfolderNames;
+ }
+
+ private BoxFolder.Info getOrCreateSubfolders(Queue<String> subFolderNames, BoxFolder parentFolder, boolean createFolder) {
+ final BoxFolder.Info folderInfo = getOrCreateFolder(subFolderNames.poll(), parentFolder, createFolder);
+
+ if (!subFolderNames.isEmpty()) {
+ final BoxFolder newParentFolder = getFolder(folderInfo.getID());
+ return getOrCreateSubfolders(subFolderNames, newParentFolder, createFolder);
+ } else {
+ return folderInfo;
+ }
+ }
+
+ private BoxFolder.Info getOrCreateFolder(String folderName, BoxFolder parentFolder, boolean createFolder) {
+ final Optional<BoxFolder.Info> existingFolder = getFolderByName(folderName, parentFolder);
+
+ if (existingFolder.isPresent()) {
+ return existingFolder.get();
+ }
+
+ if (!createFolder) {
+ throw new ProcessException(format("The specified subfolder [%s] does not exist and [%s] is false.",
+ folderName, CREATE_SUBFOLDER.getDisplayName()));
+ }
+
+ return createFolder(folderName, parentFolder);
+ }
+
+ private Info createFolder(final String folderName, final BoxFolder parentFolder) {
+ getLogger().info("Creating Folder [{}], Parent [{}]", folderName, parentFolder.getInfo().getID());
+
+ try {
+ return parentFolder.createFolder(folderName);
+ } catch (BoxAPIResponseException e) {
+ getLogger().info("Folder [{}], Parent [{}] already existed", folderName, parentFolder.getInfo().getID());
+
+ if (e.getResponseCode() != CONFLICT_RESPONSE_CODE) {
+ throw e;
+ } else {
+ return getFolderByName(folderName, parentFolder).orElseThrow(() ->
+ new ProcessException(format("Created subfolder [%s] can not be found under [%s]", folderName, parentFolder.getInfo().getID()), e));
+ }
Review Comment:
I think the log message is only relevant for the `else` case.
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/FetchBoxFile.java:
##########
@@ -47,46 +62,51 @@
@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
@Tags({"box", "storage", "fetch"})
@CapabilityDescription("Fetches files from a Box Folder. Designed to be used in tandem with ListBoxFile.")
-@SeeAlso({ListBoxFile.class})
+@SeeAlso({ListBoxFile.class, PutBoxFile.class})
+@ReadsAttribute(attribute = ID, description = ID_DESC)
@WritesAttributes({
- @WritesAttribute(attribute = FetchBoxFile.ERROR_CODE_ATTRIBUTE, description = "The error code returned by Box when the fetch of a file fails"),
- @WritesAttribute(attribute = FetchBoxFile.ERROR_MESSAGE_ATTRIBUTE, description = "The error message returned by Box when the fetch of a file fails")
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)
})
public class FetchBoxFile extends AbstractProcessor {
- public static final String ERROR_CODE_ATTRIBUTE = "error.code";
- public static final String ERROR_MESSAGE_ATTRIBUTE = "error.message";
public static final PropertyDescriptor FILE_ID = new PropertyDescriptor
- .Builder().name("box-file-id")
- .displayName("File ID")
- .description("The ID of the File to fetch")
- .required(true)
- .defaultValue("${box.id}")
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
+ .Builder().name("box-file-id")
+ .displayName("File ID")
+ .description("The ID of the File to fetch")
+ .required(true)
+ .defaultValue("${box.id}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
public static final Relationship REL_SUCCESS =
- new Relationship.Builder()
- .name("success")
- .description("A FlowFile will be routed here for each successfully fetched File.")
- .build();
+ new Relationship.Builder()
+ .name("success")
+ .description("A FlowFile will be routed here for each successfully fetched File.")
+ .build();
public static final Relationship REL_FAILURE =
- new Relationship.Builder().name("failure")
- .description("A FlowFile will be routed here for each File for which fetch was attempted but failed.")
- .build();
-
- private static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(Arrays.asList(
- BoxClientService.BOX_CLIENT_SERVICE,
- FILE_ID
- ));
+ new Relationship.Builder()
+ .name("failure")
+ .description("A FlowFile will be routed here for each File for which fetch was attempted but failed.")
+ .build();
public static final Set<Relationship> relationships = Collections.unmodifiableSet(new HashSet<>(Arrays.asList(
Review Comment:
`RELATIONSHIPS`
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.REPLACE;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import com.box.sdk.BoxFolder.Info;
+import com.box.sdk.BoxItem;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded." +
+ " Please see Additional Details to obtain Folder ID.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor.Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .dependsOn(SUBFOLDER_NAME)
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(ConflictResolutionStrategy.FAIL.getValue())
+ .allowableValues(ConflictResolutionStrategy.class)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ private static final int CONFLICT_RESPONSE_CODE = 409;
+ private static final int NOT_FOUND_RESPONSE_CODE = 404;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+ final ConflictResolutionStrategy conflictResolution = ConflictResolutionStrategy.forValue(context.getProperty(CONFLICT_RESOLUTION).getValue());
+
+ final long startNanos = System.nanoTime();
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+ final BoxFolder parentFolder = getOrCreateDirectParentFolder(context, flowFile);
+ fullPath = BoxFileUtils.getFolderPath(parentFolder.getInfo());
+ BoxFile.Info uploadedFileInfo = null;
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ if (REPLACE.equals(conflictResolution)) {
+ uploadedFileInfo = replaceExistingBoxFile(parentFolder, filename, rawIn, size, chunkUploadThreshold);
Review Comment:
I'd suggest: `replaceBoxFileIfExists()`
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.REPLACE;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import com.box.sdk.BoxFolder.Info;
+import com.box.sdk.BoxItem;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded." +
+ " Please see Additional Details to obtain Folder ID.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor.Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .dependsOn(SUBFOLDER_NAME)
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(ConflictResolutionStrategy.FAIL.getValue())
+ .allowableValues(ConflictResolutionStrategy.class)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ private static final int CONFLICT_RESPONSE_CODE = 409;
+ private static final int NOT_FOUND_RESPONSE_CODE = 404;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+ final ConflictResolutionStrategy conflictResolution = ConflictResolutionStrategy.forValue(context.getProperty(CONFLICT_RESOLUTION).getValue());
+
+ final long startNanos = System.nanoTime();
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+ final BoxFolder parentFolder = getOrCreateDirectParentFolder(context, flowFile);
+ fullPath = BoxFileUtils.getFolderPath(parentFolder.getInfo());
+ BoxFile.Info uploadedFileInfo = null;
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ if (REPLACE.equals(conflictResolution)) {
+ uploadedFileInfo = replaceExistingBoxFile(parentFolder, filename, rawIn, size, chunkUploadThreshold);
+ }
+
+ if (uploadedFileInfo == null) {
+ if (size > chunkUploadThreshold) {
+ uploadedFileInfo = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFileInfo = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+ } catch (BoxAPIResponseException e) {
+ if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
+ handleConflict(conflictResolution, filename, parentFolder, e);
+ } else {
+ throw e;
+ }
+ }
+
+ if (uploadedFileInfo != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFileInfo);
+ final String url = BOX_URL + uploadedFileInfo.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Upload failed: File [{}] Folder [{}] Response Code [{}]", filename, fullPath, e.getResponseCode(), e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Upload failed: File [{}], Folder [{}]", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ private BoxFolder getOrCreateDirectParentFolder(ProcessContext context, FlowFile flowFile ) {
+ final String subfolderPath = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ final String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ BoxFolder parentFolder = getFolderById(folderId);
+
+ if (subfolderPath != null) {
+ final Queue<String> subFolderNames = getSubFolderNames(subfolderPath);
+ parentFolder = getFolder(getOrCreateSubfolders(subFolderNames, parentFolder, createFolder).getID());
+ }
+
+ return parentFolder;
+ }
+
+ private BoxFile.Info replaceExistingBoxFile(BoxFolder parentFolder, String filename, final InputStream rawIn, final long size, final long chunkUploadThreshold)
+ throws IOException, InterruptedException {
+ final Optional<BoxFile.Info> existingBoxFileInfo = getFileByName(filename, parentFolder);
+ if (existingBoxFileInfo.isPresent()) {
+ final BoxFile existingBoxFile = new BoxFile(boxAPIConnection, existingBoxFileInfo.get().getID());
+
+ if (size > chunkUploadThreshold) {
+ return existingBoxFile.uploadLargeFile(rawIn, size);
+ } else {
+ return existingBoxFile.uploadNewVersion(rawIn);
+ }
+ }
+ return null;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ final BoxClientService boxClientService = context.getProperty(BoxClientService.BOX_CLIENT_SERVICE).asControllerService(BoxClientService.class);
+
+ boxAPIConnection = boxClientService.getBoxApiConnection();
+ }
+
+ BoxFolder getFolder(String folderId) {
+ return new BoxFolder(boxAPIConnection, folderId);
+ }
+
+ private Queue<String> getSubFolderNames(String subfolderPath) {
+ final Queue<String> subfolderNames = new LinkedList<>();
+ Collections.addAll(subfolderNames, subfolderPath.split("/"));
+ return subfolderNames;
+ }
+
+ private BoxFolder.Info getOrCreateSubfolders(Queue<String> subFolderNames, BoxFolder parentFolder, boolean createFolder) {
+ final BoxFolder.Info folderInfo = getOrCreateFolder(subFolderNames.poll(), parentFolder, createFolder);
+
+ if (!subFolderNames.isEmpty()) {
+ final BoxFolder newParentFolder = getFolder(folderInfo.getID());
+ return getOrCreateSubfolders(subFolderNames, newParentFolder, createFolder);
+ } else {
+ return folderInfo;
+ }
+ }
+
+ private BoxFolder.Info getOrCreateFolder(String folderName, BoxFolder parentFolder, boolean createFolder) {
+ final Optional<BoxFolder.Info> existingFolder = getFolderByName(folderName, parentFolder);
+
+ if (existingFolder.isPresent()) {
+ return existingFolder.get();
+ }
+
+ if (!createFolder) {
+ throw new ProcessException(format("The specified subfolder [%s] does not exist and [%s] is false.",
+ folderName, CREATE_SUBFOLDER.getDisplayName()));
+ }
+
+ return createFolder(folderName, parentFolder);
+ }
+
+ private Info createFolder(final String folderName, final BoxFolder parentFolder) {
Review Comment:
```suggestion
private BoxFolder.Info createFolder(final String folderName, final BoxFolder parentFolder) {
```
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.REPLACE;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import com.box.sdk.BoxFolder.Info;
+import com.box.sdk.BoxItem;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded." +
+ " Please see Additional Details to obtain Folder ID.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor.Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .dependsOn(SUBFOLDER_NAME)
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(ConflictResolutionStrategy.FAIL.getValue())
+ .allowableValues(ConflictResolutionStrategy.class)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ private static final int CONFLICT_RESPONSE_CODE = 409;
+ private static final int NOT_FOUND_RESPONSE_CODE = 404;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+ final ConflictResolutionStrategy conflictResolution = ConflictResolutionStrategy.forValue(context.getProperty(CONFLICT_RESOLUTION).getValue());
+
+ final long startNanos = System.nanoTime();
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+ final BoxFolder parentFolder = getOrCreateDirectParentFolder(context, flowFile);
+ fullPath = BoxFileUtils.getFolderPath(parentFolder.getInfo());
+ BoxFile.Info uploadedFileInfo = null;
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ if (REPLACE.equals(conflictResolution)) {
+ uploadedFileInfo = replaceExistingBoxFile(parentFolder, filename, rawIn, size, chunkUploadThreshold);
+ }
+
+ if (uploadedFileInfo == null) {
+ if (size > chunkUploadThreshold) {
+ uploadedFileInfo = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFileInfo = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+ } catch (BoxAPIResponseException e) {
+ if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
+ handleConflict(conflictResolution, filename, parentFolder, e);
+ } else {
+ throw e;
+ }
+ }
+
+ if (uploadedFileInfo != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFileInfo);
+ final String url = BOX_URL + uploadedFileInfo.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Upload failed: File [{}] Folder [{}] Response Code [{}]", filename, fullPath, e.getResponseCode(), e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Upload failed: File [{}], Folder [{}]", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ private BoxFolder getOrCreateDirectParentFolder(ProcessContext context, FlowFile flowFile ) {
+ final String subfolderPath = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ final String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ BoxFolder parentFolder = getFolderById(folderId);
+
+ if (subfolderPath != null) {
+ final Queue<String> subFolderNames = getSubFolderNames(subfolderPath);
+ parentFolder = getFolder(getOrCreateSubfolders(subFolderNames, parentFolder, createFolder).getID());
+ }
+
+ return parentFolder;
+ }
+
+ private BoxFile.Info replaceExistingBoxFile(BoxFolder parentFolder, String filename, final InputStream rawIn, final long size, final long chunkUploadThreshold)
+ throws IOException, InterruptedException {
+ final Optional<BoxFile.Info> existingBoxFileInfo = getFileByName(filename, parentFolder);
+ if (existingBoxFileInfo.isPresent()) {
+ final BoxFile existingBoxFile = new BoxFile(boxAPIConnection, existingBoxFileInfo.get().getID());
+
+ if (size > chunkUploadThreshold) {
+ return existingBoxFile.uploadLargeFile(rawIn, size);
+ } else {
+ return existingBoxFile.uploadNewVersion(rawIn);
+ }
+ }
+ return null;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ final BoxClientService boxClientService = context.getProperty(BoxClientService.BOX_CLIENT_SERVICE).asControllerService(BoxClientService.class);
+
+ boxAPIConnection = boxClientService.getBoxApiConnection();
+ }
+
+ BoxFolder getFolder(String folderId) {
+ return new BoxFolder(boxAPIConnection, folderId);
+ }
+
+ private Queue<String> getSubFolderNames(String subfolderPath) {
+ final Queue<String> subfolderNames = new LinkedList<>();
+ Collections.addAll(subfolderNames, subfolderPath.split("/"));
+ return subfolderNames;
+ }
+
+ private BoxFolder.Info getOrCreateSubfolders(Queue<String> subFolderNames, BoxFolder parentFolder, boolean createFolder) {
+ final BoxFolder.Info folderInfo = getOrCreateFolder(subFolderNames.poll(), parentFolder, createFolder);
+
+ if (!subFolderNames.isEmpty()) {
+ final BoxFolder newParentFolder = getFolder(folderInfo.getID());
Review Comment:
```suggestion
final BoxFolder newParentFolder = folderInfo.getResource();
```
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.REPLACE;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import com.box.sdk.BoxFolder.Info;
+import com.box.sdk.BoxItem;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded." +
+ " Please see Additional Details to obtain Folder ID.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor.Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .dependsOn(SUBFOLDER_NAME)
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(ConflictResolutionStrategy.FAIL.getValue())
+ .allowableValues(ConflictResolutionStrategy.class)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ private static final int CONFLICT_RESPONSE_CODE = 409;
+ private static final int NOT_FOUND_RESPONSE_CODE = 404;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+ final ConflictResolutionStrategy conflictResolution = ConflictResolutionStrategy.forValue(context.getProperty(CONFLICT_RESOLUTION).getValue());
+
+ final long startNanos = System.nanoTime();
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+ final BoxFolder parentFolder = getOrCreateDirectParentFolder(context, flowFile);
+ fullPath = BoxFileUtils.getFolderPath(parentFolder.getInfo());
+ BoxFile.Info uploadedFileInfo = null;
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ if (REPLACE.equals(conflictResolution)) {
+ uploadedFileInfo = replaceExistingBoxFile(parentFolder, filename, rawIn, size, chunkUploadThreshold);
+ }
+
+ if (uploadedFileInfo == null) {
+ if (size > chunkUploadThreshold) {
+ uploadedFileInfo = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFileInfo = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+ } catch (BoxAPIResponseException e) {
+ if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
+ handleConflict(conflictResolution, filename, parentFolder, e);
+ } else {
+ throw e;
+ }
+ }
+
+ if (uploadedFileInfo != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFileInfo);
+ final String url = BOX_URL + uploadedFileInfo.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Upload failed: File [{}] Folder [{}] Response Code [{}]", filename, fullPath, e.getResponseCode(), e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Upload failed: File [{}], Folder [{}]", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ private BoxFolder getOrCreateDirectParentFolder(ProcessContext context, FlowFile flowFile ) {
+ final String subfolderPath = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ final String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ BoxFolder parentFolder = getFolderById(folderId);
+
+ if (subfolderPath != null) {
+ final Queue<String> subFolderNames = getSubFolderNames(subfolderPath);
+ parentFolder = getFolder(getOrCreateSubfolders(subFolderNames, parentFolder, createFolder).getID());
+ }
+
+ return parentFolder;
+ }
+
+ private BoxFile.Info replaceExistingBoxFile(BoxFolder parentFolder, String filename, final InputStream rawIn, final long size, final long chunkUploadThreshold)
+ throws IOException, InterruptedException {
+ final Optional<BoxFile.Info> existingBoxFileInfo = getFileByName(filename, parentFolder);
+ if (existingBoxFileInfo.isPresent()) {
+ final BoxFile existingBoxFile = new BoxFile(boxAPIConnection, existingBoxFileInfo.get().getID());
+
+ if (size > chunkUploadThreshold) {
+ return existingBoxFile.uploadLargeFile(rawIn, size);
+ } else {
+ return existingBoxFile.uploadNewVersion(rawIn);
+ }
+ }
+ return null;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
Review Comment:
Could you please move `onScheduled()` up, before `onTrigger()`?
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.REPLACE;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import com.box.sdk.BoxFolder.Info;
+import com.box.sdk.BoxItem;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded." +
+ " Please see Additional Details to obtain Folder ID.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor.Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .dependsOn(SUBFOLDER_NAME)
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(ConflictResolutionStrategy.FAIL.getValue())
+ .allowableValues(ConflictResolutionStrategy.class)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ private static final int CONFLICT_RESPONSE_CODE = 409;
+ private static final int NOT_FOUND_RESPONSE_CODE = 404;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+ final ConflictResolutionStrategy conflictResolution = ConflictResolutionStrategy.forValue(context.getProperty(CONFLICT_RESOLUTION).getValue());
+
+ final long startNanos = System.nanoTime();
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+ final BoxFolder parentFolder = getOrCreateDirectParentFolder(context, flowFile);
+ fullPath = BoxFileUtils.getFolderPath(parentFolder.getInfo());
+ BoxFile.Info uploadedFileInfo = null;
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ if (REPLACE.equals(conflictResolution)) {
+ uploadedFileInfo = replaceExistingBoxFile(parentFolder, filename, rawIn, size, chunkUploadThreshold);
+ }
+
+ if (uploadedFileInfo == null) {
+ if (size > chunkUploadThreshold) {
+ uploadedFileInfo = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFileInfo = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+ } catch (BoxAPIResponseException e) {
+ if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
+ handleConflict(conflictResolution, filename, parentFolder, e);
+ } else {
+ throw e;
+ }
+ }
+
+ if (uploadedFileInfo != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFileInfo);
+ final String url = BOX_URL + uploadedFileInfo.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Upload failed: File [{}] Folder [{}] Response Code [{}]", filename, fullPath, e.getResponseCode(), e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Upload failed: File [{}], Folder [{}]", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ private BoxFolder getOrCreateDirectParentFolder(ProcessContext context, FlowFile flowFile ) {
+ final String subfolderPath = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ final String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ BoxFolder parentFolder = getFolderById(folderId);
+
+ if (subfolderPath != null) {
+ final Queue<String> subFolderNames = getSubFolderNames(subfolderPath);
+ parentFolder = getFolder(getOrCreateSubfolders(subFolderNames, parentFolder, createFolder).getID());
+ }
+
+ return parentFolder;
+ }
+
+ private BoxFile.Info replaceExistingBoxFile(BoxFolder parentFolder, String filename, final InputStream rawIn, final long size, final long chunkUploadThreshold)
+ throws IOException, InterruptedException {
+ final Optional<BoxFile.Info> existingBoxFileInfo = getFileByName(filename, parentFolder);
+ if (existingBoxFileInfo.isPresent()) {
+ final BoxFile existingBoxFile = new BoxFile(boxAPIConnection, existingBoxFileInfo.get().getID());
Review Comment:
Using `BoxFile.Info.getResource()` would be more API compliant I think:
```suggestion
final BoxFile existingBoxFile = existingBoxFileInfo.get().getResource();
```
Though it requires more mocking in some tests:
```
when(mockSubfolder1Info.getResource()).thenReturn(mockSubfolder1);
when(mockSubfolder2Info.getResource()).thenReturn(mockSubfolder2);
```
##########
nifi-nar-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/PutBoxFile.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * 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.box;
+
+import static java.lang.String.format;
+import static java.lang.String.valueOf;
+import static java.util.Arrays.asList;
+import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_CODE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ERROR_MESSAGE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.FILENAME_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID;
+import static org.apache.nifi.processors.box.BoxFileAttributes.ID_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.PATH_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE;
+import static org.apache.nifi.processors.box.BoxFileAttributes.SIZE_DESC;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP;
+import static org.apache.nifi.processors.box.BoxFileAttributes.TIMESTAMP_DESC;
+import static org.apache.nifi.processors.box.BoxFileUtils.BOX_URL;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
+import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.REPLACE;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIException;
+import com.box.sdk.BoxAPIResponseException;
+import com.box.sdk.BoxFile;
+import com.box.sdk.BoxFolder;
+import com.box.sdk.BoxFolder.Info;
+import com.box.sdk.BoxItem;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.box.controllerservices.BoxClientService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy;
+
+
+@SeeAlso({ListBoxFile.class, FetchBoxFile.class})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"box", "storage", "put"})
+@CapabilityDescription("Puts content to a Box folder.")
+@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the Box object.")
+@WritesAttributes({
+ @WritesAttribute(attribute = ID, description = ID_DESC),
+ @WritesAttribute(attribute = "filename", description = FILENAME_DESC),
+ @WritesAttribute(attribute = "path", description = PATH_DESC),
+ @WritesAttribute(attribute = SIZE, description = SIZE_DESC),
+ @WritesAttribute(attribute = TIMESTAMP, description = TIMESTAMP_DESC),
+ @WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
+ @WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
+public class PutBoxFile extends AbstractProcessor {
+ public static final int CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES = 20 * 1024 * 1024;
+ public static final int CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES = 50 * 1024 * 1024;
+
+ public static final PropertyDescriptor FOLDER_ID = new PropertyDescriptor.Builder()
+ .name("box-folder-id")
+ .displayName("Folder ID")
+ .description("The ID of the folder where the file is uploaded." +
+ " Please see Additional Details to obtain Folder ID.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor.Builder()
+ .name("file-name")
+ .displayName("Filename")
+ .description("The name of the file to upload to the specified Box folder.")
+ .required(true)
+ .defaultValue("${filename}")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
+ .name("subfolder-name")
+ .displayName("Subfolder Name")
+ .description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the folder specified by 'Folder ID'."
+ + " Example: subFolder, subFolder1/subfolder2")
+ .addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
+ "Subfolder Name should not contain leading or trailing slash ('/') character."))
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
+ .name("create-folder")
+ .displayName("Create Subfolder")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .dependsOn(SUBFOLDER_NAME)
+ .description("Specifies whether to check if the subfolder exists and to automatically create it if it does not. " +
+ "Permission to list folders is required. ")
+ .build();
+
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the specified Box folder.")
+ .required(true)
+ .defaultValue(ConflictResolutionStrategy.FAIL.getValue())
+ .allowableValues(ConflictResolutionStrategy.class)
+ .build();
+
+ public static final PropertyDescriptor CHUNKED_UPLOAD_THRESHOLD = new PropertyDescriptor.Builder()
+ .name("chunked-upload-threshold")
+ .displayName("Chunked Upload Threshold")
+ .description("The maximum size of the content which is uploaded at once. FlowFiles larger than this threshold are uploaded in chunks."
+ + " Chunked upload is allowed for files larger than 20 MB. It is recommended to use chunked upload for files exceeding 50 MB.")
+ .defaultValue("20 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(CHUNKED_UPLOAD_LOWER_LIMIT_IN_BYTES, CHUNKED_UPLOAD_UPPER_LIMIT_IN_BYTES))
+ .required(false)
+ .build();
+
+ public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
+ BoxClientService.BOX_CLIENT_SERVICE,
+ FOLDER_ID,
+ SUBFOLDER_NAME,
+ CREATE_SUBFOLDER,
+ FILE_NAME,
+ CONFLICT_RESOLUTION,
+ CHUNKED_UPLOAD_THRESHOLD
+ ));
+
+ public static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to Box are transferred to this relationship.")
+ .build();
+
+ public static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to Box for some reason are transferred to this relationship.")
+ .build();
+
+ public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(asList(
+ REL_SUCCESS,
+ REL_FAILURE
+ )));
+
+ private static final int CONFLICT_RESPONSE_CODE = 409;
+ private static final int NOT_FOUND_RESPONSE_CODE = 404;
+
+ private volatile BoxAPIConnection boxAPIConnection;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final long chunkUploadThreshold = context.getProperty(CHUNKED_UPLOAD_THRESHOLD)
+ .asDataSize(DataUnit.B)
+ .longValue();
+ final ConflictResolutionStrategy conflictResolution = ConflictResolutionStrategy.forValue(context.getProperty(CONFLICT_RESOLUTION).getValue());
+
+ final long startNanos = System.nanoTime();
+ String fullPath = null;
+
+ try {
+ final long size = flowFile.getSize();
+ final BoxFolder parentFolder = getOrCreateDirectParentFolder(context, flowFile);
+ fullPath = BoxFileUtils.getFolderPath(parentFolder.getInfo());
+ BoxFile.Info uploadedFileInfo = null;
+
+ try (InputStream rawIn = session.read(flowFile)){
+
+ if (REPLACE.equals(conflictResolution)) {
+ uploadedFileInfo = replaceExistingBoxFile(parentFolder, filename, rawIn, size, chunkUploadThreshold);
+ }
+
+ if (uploadedFileInfo == null) {
+ if (size > chunkUploadThreshold) {
+ uploadedFileInfo = parentFolder.uploadLargeFile(rawIn, filename, size);
+ } else {
+ uploadedFileInfo = parentFolder.uploadFile(rawIn, filename);
+ }
+ }
+ } catch (BoxAPIResponseException e) {
+ if (e.getResponseCode() == CONFLICT_RESPONSE_CODE) {
+ handleConflict(conflictResolution, filename, parentFolder, e);
+ } else {
+ throw e;
+ }
+ }
+
+ if (uploadedFileInfo != null) {
+ final Map<String, String> attributes = BoxFileUtils.createAttributeMap(uploadedFileInfo);
+ final String url = BOX_URL + uploadedFileInfo.getID();
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, url, transferMillis);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (BoxAPIResponseException e) {
+ getLogger().error("Upload failed: File [{}] Folder [{}] Response Code [{}]", filename, fullPath, e.getResponseCode(), e);
+ handleExpectedError(session, flowFile, e);
+ } catch (Exception e) {
+ getLogger().error("Upload failed: File [{}], Folder [{}]", filename, fullPath, e);
+ handleUnexpectedError(session, flowFile, e);
+ }
+ }
+
+ private BoxFolder getOrCreateDirectParentFolder(ProcessContext context, FlowFile flowFile ) {
+ final String subfolderPath = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
+ final String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
+ BoxFolder parentFolder = getFolderById(folderId);
+
+ if (subfolderPath != null) {
+ final Queue<String> subFolderNames = getSubFolderNames(subfolderPath);
+ parentFolder = getFolder(getOrCreateSubfolders(subFolderNames, parentFolder, createFolder).getID());
+ }
+
+ return parentFolder;
+ }
+
+ private BoxFile.Info replaceExistingBoxFile(BoxFolder parentFolder, String filename, final InputStream rawIn, final long size, final long chunkUploadThreshold)
+ throws IOException, InterruptedException {
+ final Optional<BoxFile.Info> existingBoxFileInfo = getFileByName(filename, parentFolder);
+ if (existingBoxFileInfo.isPresent()) {
+ final BoxFile existingBoxFile = new BoxFile(boxAPIConnection, existingBoxFileInfo.get().getID());
+
+ if (size > chunkUploadThreshold) {
+ return existingBoxFile.uploadLargeFile(rawIn, size);
+ } else {
+ return existingBoxFile.uploadNewVersion(rawIn);
+ }
+ }
+ return null;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ final BoxClientService boxClientService = context.getProperty(BoxClientService.BOX_CLIENT_SERVICE).asControllerService(BoxClientService.class);
+
+ boxAPIConnection = boxClientService.getBoxApiConnection();
+ }
+
+ BoxFolder getFolder(String folderId) {
+ return new BoxFolder(boxAPIConnection, folderId);
+ }
+
+ private Queue<String> getSubFolderNames(String subfolderPath) {
+ final Queue<String> subfolderNames = new LinkedList<>();
+ Collections.addAll(subfolderNames, subfolderPath.split("/"));
+ return subfolderNames;
+ }
+
+ private BoxFolder.Info getOrCreateSubfolders(Queue<String> subFolderNames, BoxFolder parentFolder, boolean createFolder) {
+ final BoxFolder.Info folderInfo = getOrCreateFolder(subFolderNames.poll(), parentFolder, createFolder);
+
+ if (!subFolderNames.isEmpty()) {
+ final BoxFolder newParentFolder = getFolder(folderInfo.getID());
+ return getOrCreateSubfolders(subFolderNames, newParentFolder, createFolder);
+ } else {
+ return folderInfo;
+ }
+ }
+
+ private BoxFolder.Info getOrCreateFolder(String folderName, BoxFolder parentFolder, boolean createFolder) {
+ final Optional<BoxFolder.Info> existingFolder = getFolderByName(folderName, parentFolder);
+
+ if (existingFolder.isPresent()) {
+ return existingFolder.get();
+ }
+
+ if (!createFolder) {
+ throw new ProcessException(format("The specified subfolder [%s] does not exist and [%s] is false.",
+ folderName, CREATE_SUBFOLDER.getDisplayName()));
+ }
+
+ return createFolder(folderName, parentFolder);
+ }
+
+ private Info createFolder(final String folderName, final BoxFolder parentFolder) {
+ getLogger().info("Creating Folder [{}], Parent [{}]", folderName, parentFolder.getInfo().getID());
+
+ try {
+ return parentFolder.createFolder(folderName);
+ } catch (BoxAPIResponseException e) {
+ getLogger().info("Folder [{}], Parent [{}] already existed", folderName, parentFolder.getInfo().getID());
+
+ if (e.getResponseCode() != CONFLICT_RESPONSE_CODE) {
+ throw e;
+ } else {
+ return getFolderByName(folderName, parentFolder).orElseThrow(() ->
+ new ProcessException(format("Created subfolder [%s] can not be found under [%s]", folderName, parentFolder.getInfo().getID()), e));
+ }
+ }
+ }
+
+ private BoxFolder getFolderById(final String folderId) {
+ final BoxFolder folder = getFolder(folderId);
+ try {
+ //Error is returned for nonexistent folder only when a method is called on BoxFolder.
+ folder.getInfo();
+ } catch (BoxAPIResponseException e) {
+ if (e.getResponseCode() == NOT_FOUND_RESPONSE_CODE) {
+ throw new ProcessException(format("The Folder [%s] specified by [%s] does not exist", folderId, FOLDER_ID.getDisplayName()));
+ }
+ }
+ return folder;
+ }
+
+ private Optional<BoxFolder.Info> getFolderByName(final String folderName, final BoxFolder parentFolder) {
+ return getItemByName(folderName, parentFolder, BoxFolder.Info.class);
+ }
+
+ private Optional<BoxFile.Info> getFileByName(final String filename, final BoxFolder parentFolder) {
+ return getItemByName(filename, parentFolder, BoxFile.Info.class);
+ }
Review Comment:
Converting `BoxFolder.Info` to `BoxFolder` and then back appear in the code in several places.
I might consider converting the Info objects to BoxFolder/BoxFile right here, and use those objects in the rest of the code.
```
private Optional<BoxFile> getFileByName(final String filename, final BoxFolder parentFolder) {
return getItemByName(filename, parentFolder, BoxFile.Info.class).map(BoxFile.Info::getResource);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] turcsanyip commented on pull request #6892: NIFI-11059 PutBoxFile processor
Posted by "turcsanyip (via GitHub)" <gi...@apache.org>.
turcsanyip commented on PR #6892:
URL: https://github.com/apache/nifi/pull/6892#issuecomment-1413838085
@exceptionfactory Thanks for your review!
Merging to main...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org