You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jo...@apache.org on 2020/08/10 17:11:53 UTC
[nifi] branch main updated (58e324e -> 51091a5)
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git.
from 58e324e NIFI-7522 bumping from Apache Kafka 2.5 to Apache Kafka 2.6
new 71d155e NIFI-7338 fixed provenance, top level dir, chatty smbj log, pw validation update accessmask to be more restrictive
new 51091a5 NIFI-7338 This closes #4169. Added comment about com.hierynomus.smbj.SMBClient syncronicity
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
nifi-assembly/LICENSE | 38 +-
nifi-assembly/NOTICE | 11 +-
nifi-assembly/pom.xml | 6 +
.../src/main/resources/conf/logback.xml | 3 +
.../nifi-smb-nar}/pom.xml | 6 +-
.../src/main/resources/META-INF/LICENSE | 93 ++--
.../src/main/resources/META-INF}/NOTICE | 13 +-
.../nifi-smb-bundle/nifi-smb-processors}/pom.xml | 17 +-
.../org/apache/nifi/processors/smb/GetSmbFile.java | 544 +++++++++++++++++++++
.../org/apache/nifi/processors/smb/PutSmbFile.java | 334 +++++++++++++
.../services/org.apache.nifi.processor.Processor | 3 +-
.../apache/nifi/processors/smb/GetSmbFileTest.java | 296 +++++++++++
.../apache/nifi/processors/smb/PutSmbFileTest.java | 224 +++++++++
.../{nifi-poi-bundle => nifi-smb-bundle}/pom.xml | 6 +-
nifi-nar-bundles/pom.xml | 1 +
.../resources/conf/clustered/node1/logback.xml | 3 +
.../resources/conf/clustered/node2/logback.xml | 3 +
.../src/test/resources/conf/default/logback.xml | 3 +
18 files changed, 1533 insertions(+), 71 deletions(-)
copy nifi-nar-bundles/{nifi-tcp-bundle/nifi-tcp-nar => nifi-smb-bundle/nifi-smb-nar}/pom.xml (91%)
copy nifi-nar-bundles/{nifi-email-bundle/nifi-email-nar => nifi-smb-bundle/nifi-smb-nar}/src/main/resources/META-INF/LICENSE (84%)
copy nifi-nar-bundles/{nifi-rules-action-handler-bundle/nifi-rules-action-handler-nar/src/main/resources => nifi-smb-bundle/nifi-smb-nar/src/main/resources/META-INF}/NOTICE (55%)
copy {nifi-maven-archetypes/nifi-processor-bundle-archetype/src/main/resources/archetype-resources/nifi-__artifactBaseName__-processors => nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors}/pom.xml (81%)
create mode 100644 nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/GetSmbFile.java
create mode 100644 nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java
copy {nifi-maven-archetypes/nifi-processor-bundle-archetype/src/main/resources/archetype-resources/nifi-__artifactBaseName__-processors => nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors}/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor (90%)
create mode 100644 nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java
create mode 100644 nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/PutSmbFileTest.java
copy nifi-nar-bundles/{nifi-poi-bundle => nifi-smb-bundle}/pom.xml (91%)
[nifi] 01/02: NIFI-7338 fixed provenance, top level dir,
chatty smbj log, pw validation update accessmask to be more restrictive
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 71d155e874e98d5904c11ce1972b566fb77f523b
Author: si <si...@roche.com>
AuthorDate: Mon Mar 30 11:20:23 2020 +0200
NIFI-7338 fixed provenance, top level dir,
chatty smbj log, pw validation
update accessmask to be more restrictive
Author: si <si...@roche.com>
Signed-off-by: Joe Witt <jo...@apache.org>
---
nifi-assembly/LICENSE | 38 +-
nifi-assembly/NOTICE | 11 +-
nifi-assembly/pom.xml | 6 +
.../src/main/resources/conf/logback.xml | 3 +
.../nifi-smb-bundle/nifi-smb-nar/pom.xml | 41 ++
.../src/main/resources/META-INF/LICENSE | 280 +++++++++++
.../src/main/resources/META-INF/NOTICE | 19 +
.../nifi-smb-bundle/nifi-smb-processors/pom.xml | 60 +++
.../org/apache/nifi/processors/smb/GetSmbFile.java | 541 +++++++++++++++++++++
.../org/apache/nifi/processors/smb/PutSmbFile.java | 334 +++++++++++++
.../services/org.apache.nifi.processor.Processor | 16 +
.../apache/nifi/processors/smb/GetSmbFileTest.java | 290 +++++++++++
.../apache/nifi/processors/smb/PutSmbFileTest.java | 224 +++++++++
nifi-nar-bundles/nifi-smb-bundle/pom.xml | 35 ++
nifi-nar-bundles/pom.xml | 1 +
.../resources/conf/clustered/node1/logback.xml | 3 +
.../resources/conf/clustered/node2/logback.xml | 3 +
.../src/test/resources/conf/default/logback.xml | 3 +
18 files changed, 1902 insertions(+), 6 deletions(-)
diff --git a/nifi-assembly/LICENSE b/nifi-assembly/LICENSE
index 96872ae..279ab0d 100644
--- a/nifi-assembly/LICENSE
+++ b/nifi-assembly/LICENSE
@@ -951,7 +951,7 @@ The binary distribution of this product bundles 'ANTLR 2'
The binary distribution of this product bundles 'Bouncy Castle JDK 1.5'
under an MIT style license.
- Copyright (c) 2000 - 2015 The Legion of the Bouncy Castle Inc. (http://www.bouncycastle.org)
+ Copyright (c) 2000 - 2019 The Legion of the Bouncy Castle Inc. (http://www.bouncycastle.org)
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
@@ -1005,7 +1005,7 @@ The binary distribution of this product bundles 'ANTLR 2'
The binary distribution of this product bundles 'Slf4j' which is available under
an MIT license.
- Copyright (c) 2004-2013 QOS.ch
+ Copyright (c) 2004-2017 QOS.ch
All rights reserved.
Permission is hereby granted, free of charge, to any person obtaining
@@ -2985,6 +2985,29 @@ The binary distribution of this product bundles 'Project Lombok 1.18.6' which is
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
+The binary distribution of this product bundles 'mbassador' under an MIT style
+ license.
+
+ Copyright (c) 2012 Benjamin Diedrichsen
+
+ Permission is hereby granted, free of charge, to any person obtaining a
+ copy of this software and associated documentation files (the "Software"),
+ to deal in the Software without restriction, including without limitation
+ the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ and/or sell copies of the Software, and to permit persons to whom the
+ Software is furnished to do so, subject to the following conditions:
+
+ The above copyright notice and this permission notice shall be included in
+ all copies or substantial portions of the Software.
+
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ THE SOFTWARE.
+
The binary distribution of this product bundles 'Extensions on qpid-proton-j library 1.2.2' which is available under an MIT license.
Copyright (c) Microsoft Corporation. All rights reserved.
@@ -3016,8 +3039,10 @@ The binary distribution of this product bundles 'Jakarta Activation API 1.2.1' a
Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
- Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
- Neither the name of the Eclipse Foundation, Inc. nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.
+ Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer
+ in the documentation and/or other materials provided with the distribution.
+ Neither the name of the Eclipse Foundation, Inc. nor the names of its contributors may be used to endorse or promote products
+ derived from this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED
WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
@@ -3025,4 +3050,7 @@ The binary distribution of this product bundles 'Jakarta Activation API 1.2.1' a
INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
- ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
\ No newline at end of file
+ ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ DEALINGS IN THE SOFTWARE.
diff --git a/nifi-assembly/NOTICE b/nifi-assembly/NOTICE
index acc25b3..112a274 100644
--- a/nifi-assembly/NOTICE
+++ b/nifi-assembly/NOTICE
@@ -1823,7 +1823,7 @@ The following binary components are provided under the Apache Software License v
Copyright 2008-2018 The Apache Software Foundation
(ASLv2) sshj
- The following NOTICE information applie:
+ The following NOTICE information applies:
sshj - SSHv2 library for Java
Copyright 2010-2012 sshj contributors
@@ -1834,6 +1834,7 @@ The following binary components are provided under the Apache Software License v
- Apache MINA SSHD
- Apache Commons-Net
+<<<<<<< HEAD
(ASLv2) Nimbus Language Tags
The following NOTICE information applies:
Nimbus Language Tags
@@ -1924,6 +1925,14 @@ The following binary components are provided under the Apache Software License v
Reactive Streams Netty Driver
Copyright 2020, Project Reactor
+ (ALSv2) smbj
+ The following NOTICE information applies:
+ Copyright (C)2016 - SMBJ Contributors
+
+ (ASLv2) asn-one
+ The following NOTICE information applies:
+ Copyright 2016 Jeroen van Erp <je...@hierynomus.com>
+
************************
Common Development and Distribution License 1.1
************************
diff --git a/nifi-assembly/pom.xml b/nifi-assembly/pom.xml
index a13d9a4..a4b62b8 100644
--- a/nifi-assembly/pom.xml
+++ b/nifi-assembly/pom.xml
@@ -603,6 +603,12 @@ language governing permissions and limitations under the License. -->
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-smb-nar</artifactId>
+ <version>1.12.0-SNAPSHOT</version>
+ <type>nar</type>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
<artifactId>nifi-windows-event-log-nar</artifactId>
<version>1.12.0-SNAPSHOT</version>
<type>nar</type>
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml
index 9b7d45c..c6ddcbe 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml
@@ -124,6 +124,9 @@
<logger name="net.schmizz.sshj" level="WARN" />
<logger name="com.hierynomus.sshj" level="WARN" />
+ <!-- Suppress non-error messages from SMBJ which was emitting large amounts of INFO logs by default -->
+ <logger name="com.hierynomus.smbj" level="WARN" />
+
<!-- These log messages would normally go to the USER_FILE log, but they belong in the APP_FILE -->
<logger name="org.apache.nifi.web.security.requests" level="INFO" additivity="false">
<appender-ref ref="APP_FILE"/>
diff --git a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-nar/pom.xml b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-nar/pom.xml
new file mode 100644
index 0000000..3c48a55
--- /dev/null
+++ b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-nar/pom.xml
@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-smb-bundle</artifactId>
+ <version>1.12.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-smb-nar</artifactId>
+ <version>1.12.0-SNAPSHOT</version>
+ <packaging>nar</packaging>
+ <properties>
+ <maven.javadoc.skip>true</maven.javadoc.skip>
+ <source.skip>true</source.skip>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-smb-processors</artifactId>
+ <version>1.12.0-SNAPSHOT</version>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-nar/src/main/resources/META-INF/LICENSE b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-nar/src/main/resources/META-INF/LICENSE
new file mode 100644
index 0000000..823e44e
--- /dev/null
+++ b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-nar/src/main/resources/META-INF/LICENSE
@@ -0,0 +1,280 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed 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.
+
+APACHE NIFI SUBCOMPONENTS:
+
+The Apache NiFi project contains subcomponents with separate copyright
+notices and license terms. Your use of the source code for the these
+subcomponents is subject to the terms and conditions of the following
+licenses.
+
+ The binary distribution of this product bundles 'mbassador' under an MIT style
+ license.
+
+ Copyright (c) 2012 Benjamin Diedrichsen
+
+ Permission is hereby granted, free of charge, to any person obtaining a
+ copy of this software and associated documentation files (the "Software"),
+ to deal in the Software without restriction, including without limitation
+ the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ and/or sell copies of the Software, and to permit persons to whom the
+ Software is furnished to do so, subject to the following conditions:
+
+ The above copyright notice and this permission notice shall be included in
+ all copies or substantial portions of the Software.
+
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ DEALINGS IN THE SOFTWARE.
+
+ The binary distribution of this product bundles 'Bouncy Castle JDK 1.5'
+ under an MIT style license.
+
+ Copyright (c) 2000 - 2019 The Legion of the Bouncy Castle Inc. (http://www.bouncycastle.org)
+
+ Permission is hereby granted, free of charge, to any person obtaining a copy
+ of this software and associated documentation files (the "Software"), to deal
+ in the Software without restriction, including without limitation the rights
+ to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ copies of the Software, and to permit persons to whom the Software is
+ furnished to do so, subject to the following conditions:
+
+ The above copyright notice and this permission notice shall be included in
+ all copies or substantial portions of the Software.
+
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ THE SOFTWARE.
+
+ The binary distribution of this product bundles 'Slf4j' which is available under
+ an MIT license.
+
+ Copyright (c) 2004-2017 QOS.ch
+ All rights reserved.
+
+ Permission is hereby granted, free of charge, to any person obtaining
+ a copy of this software and associated documentation files (the
+ "Software"), to deal in the Software without restriction, including
+ without limitation the rights to use, copy, modify, merge, publish,
+ distribute, sublicense, and/or sell copies of the Software, and to
+ permit persons to whom the Software is furnished to do so, subject to
+ the following conditions:
+
+ The above copyright notice and this permission notice shall be
+ included in all copies or substantial portions of the Software.
+
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+ LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+ OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-nar/src/main/resources/META-INF/NOTICE b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-nar/src/main/resources/META-INF/NOTICE
new file mode 100644
index 0000000..83a81be
--- /dev/null
+++ b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-nar/src/main/resources/META-INF/NOTICE
@@ -0,0 +1,19 @@
+nifi-smb-nar
+Copyright 2019 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
+******************
+Apache Software License v2
+******************
+
+The following binary components are provided under the Apache Software License v2
+
+ (ALSv2) smbj
+ The following NOTICE information applies:
+ Copyright (C)2016 - SMBJ Contributors
+
+ (ASLv2) asn-one
+ The following NOTICE information applies:
+ Copyright 2016 Jeroen van Erp <je...@hierynomus.com>
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/pom.xml b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/pom.xml
new file mode 100644
index 0000000..30b9cef
--- /dev/null
+++ b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/pom.xml
@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-smb-bundle</artifactId>
+ <version>1.12.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-smb-processors</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ <version>1.12.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>com.hierynomus</groupId>
+ <artifactId>smbj</artifactId>
+ <version>0.10.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>1.12.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/GetSmbFile.java b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/GetSmbFile.java
new file mode 100644
index 0000000..0aac011
--- /dev/null
+++ b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/GetSmbFile.java
@@ -0,0 +1,541 @@
+/*
+ * 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.smb;
+
+import com.hierynomus.msdtyp.AccessMask;
+import com.hierynomus.mserref.NtStatus;
+import com.hierynomus.msfscc.FileAttributes;
+import com.hierynomus.msfscc.fileinformation.FileAllInformation;
+import com.hierynomus.msfscc.fileinformation.FileBasicInformation;
+import com.hierynomus.msfscc.fileinformation.FileIdBothDirectoryInformation;
+import com.hierynomus.mssmb2.SMB2CreateDisposition;
+import com.hierynomus.mssmb2.SMB2CreateOptions;
+import com.hierynomus.mssmb2.SMB2ShareAccess;
+import com.hierynomus.mssmb2.SMBApiException;
+import com.hierynomus.smbj.SMBClient;
+import com.hierynomus.smbj.auth.AuthenticationContext;
+import com.hierynomus.smbj.connection.Connection;
+import com.hierynomus.smbj.session.Session;
+import com.hierynomus.smbj.share.DiskShare;
+import com.hierynomus.smbj.share.File;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.TriggerWhenEmpty;
+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.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.InputStream;
+import java.net.URI;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.List;
+import java.util.Set;
+import java.util.HashSet;
+import java.util.Collections;
+import java.util.ListIterator;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.EnumSet;
+import java.util.ArrayList;
+import java.util.Locale;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.regex.Pattern;
+
+@TriggerWhenEmpty
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@Tags({"samba, smb, cifs, files, get"})
+@CapabilityDescription("Reads file from a samba network location to FlowFiles. " +
+ "Use this processor instead of a cifs mounts if share access control is important. " +
+ "Configure the Hostname, Share and Directory accordingly: \\\\[Hostname]\\[Share]\\[path\\to\\Directory]")
+@SeeAlso({PutSmbFile.class})
+@WritesAttributes({
+ @WritesAttribute(attribute = "filename", description = "The filename is set to the name of the file on the network share"),
+ @WritesAttribute(attribute = "path", description = "The path is set to the relative path of the file's network share name. For example, "
+ + "if the input is set to \\\\hostname\\share\\tmp, files picked up from \\tmp will have the path attribute set to tmp"),
+ @WritesAttribute(attribute = "file.creationTime", description = "The date and time that the file was created. May not work on all file systems"),
+ @WritesAttribute(attribute = "file.lastModifiedTime", description = "The date and time that the file was last modified. May not work on all "
+ + "file systems"),
+ @WritesAttribute(attribute = "file.lastAccessTime", description = "The date and time that the file was last accessed. May not work on all "
+ + "file systems"),
+ @WritesAttribute(attribute = "absolute.path", description = "The full path from where a file was picked up. This includes "
+ + "the hostname and the share name")})
+public class GetSmbFile extends AbstractProcessor {
+ public static final String SHARE_ACCESS_NONE = "none";
+ public static final String SHARE_ACCESS_READ = "read";
+ public static final String SHARE_ACCESS_READDELETE = "read, delete";
+ public static final String SHARE_ACCESS_READWRITEDELETE = "read, write, delete";
+
+ public static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder()
+ .name("Hostname")
+ .description("The network host to which files should be written.")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor SHARE = new PropertyDescriptor.Builder()
+ .name("Share")
+ .description("The network share to which files should be written. This is the \"first folder\"" +
+ "after the hostname: \\\\hostname\\[share]\\dir1\\dir2")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor DIRECTORY = new PropertyDescriptor.Builder()
+ .name("Directory")
+ .description("The network folder to which files should be written. This is the remaining relative " +
+ "path after the share: \\\\hostname\\share\\[dir1\\dir2].")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .build();
+ public static final PropertyDescriptor DOMAIN = new PropertyDescriptor.Builder()
+ .name("Domain")
+ .description("The domain used for authentication. Optional, in most cases username and password is sufficient.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder()
+ .name("Username")
+ .description("The username used for authentication. If no username is set then anonymous authentication is attempted.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder()
+ .name("Password")
+ .description("The password used for authentication. Required if Username is set.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .sensitive(true)
+ .build();
+ public static final PropertyDescriptor SHARE_ACCESS = new PropertyDescriptor.Builder()
+ .name("Share Access Strategy")
+ .description("Indicates which shared access are granted on the file during the read. " +
+ "None is the most restrictive, but the safest setting to prevent corruption.")
+ .required(true)
+ .defaultValue(SHARE_ACCESS_NONE)
+ .allowableValues(SHARE_ACCESS_NONE, SHARE_ACCESS_READ, SHARE_ACCESS_READDELETE, SHARE_ACCESS_READWRITEDELETE)
+ .build();
+ public static final PropertyDescriptor RECURSE = new PropertyDescriptor.Builder()
+ .name("Recurse Subdirectories")
+ .description("Indicates whether or not to pull files from subdirectories")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .build();
+ public static final PropertyDescriptor KEEP_SOURCE_FILE = new PropertyDescriptor.Builder()
+ .name("Keep Source File")
+ .description("If true, the file is not deleted after it has been copied to the Content Repository; "
+ + "this causes the file to be picked up continually and is useful for testing purposes. "
+ + "If not keeping original NiFi will need write permissions on the directory it is pulling "
+ + "from otherwise it will ignore the file.")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+ public static final PropertyDescriptor FILE_FILTER = new PropertyDescriptor.Builder()
+ .name("File Filter")
+ .description("Only files whose names match the given regular expression will be picked up")
+ .required(false)
+ .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor PATH_FILTER = new PropertyDescriptor.Builder()
+ .name("Path Filter")
+ .description("When " + RECURSE.getName() + " is true, then only subdirectories whose path matches the given regular expression will be scanned")
+ .required(false)
+ .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor IGNORE_HIDDEN_FILES = new PropertyDescriptor.Builder()
+ .name("Ignore Hidden Files")
+ .description("Indicates whether or not hidden files should be ignored")
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .required(true)
+ .build();
+ public static final PropertyDescriptor POLLING_INTERVAL = new PropertyDescriptor.Builder()
+ .name("Polling Interval")
+ .description("Indicates how long to wait before performing a directory listing")
+ .required(true)
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("0 sec")
+ .build();
+ public static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("Batch Size")
+ .description("The maximum number of files to pull in each iteration")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("10")
+ .build();
+
+ public static final String FILE_CREATION_TIME_ATTRIBUTE = "file.creationTime";
+ public static final String FILE_LAST_MODIFY_TIME_ATTRIBUTE = "file.lastModifiedTime";
+ public static final String FILE_LAST_ACCESS_TIME_ATTRIBUTE = "file.lastAccessTime";
+
+ public static final String FILE_MODIFY_DATE_ATTR_FORMAT = "yyyy-MM-dd'T'HH:mm:ssZ";
+ final static DateFormat dateFormatter = new SimpleDateFormat(FILE_MODIFY_DATE_ATTR_FORMAT, Locale.US);
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success").description("All files are routed to success").build();
+
+ private List<PropertyDescriptor> descriptors;
+ private Set<Relationship> relationships;
+
+
+ private final BlockingQueue<String> fileQueue = new LinkedBlockingQueue<>();
+ private final Set<String> inProcess = new HashSet<>(); // guarded by queueLock
+ private final Set<String> recentlyProcessed = new HashSet<>(); // guarded by queueLock
+ private final Lock queueLock = new ReentrantLock();
+
+ private final Lock listingLock = new ReentrantLock();
+
+ private final AtomicLong queueLastUpdated = new AtomicLong(0L);
+
+ private SMBClient smbClient = null;
+
+ private Pattern filePattern;
+ private Pattern pathPattern;
+ private boolean ignoreHidden;
+ private Set<SMB2ShareAccess> sharedAccess;
+
+ @Override
+ protected void init(final ProcessorInitializationContext context) {
+ final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
+ descriptors.add(HOSTNAME);
+ descriptors.add(SHARE);
+ descriptors.add(DIRECTORY);
+ descriptors.add(DOMAIN);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+ descriptors.add(SHARE_ACCESS);
+ descriptors.add(FILE_FILTER);
+ descriptors.add(PATH_FILTER);
+ descriptors.add(BATCH_SIZE);
+ descriptors.add(KEEP_SOURCE_FILE);
+ descriptors.add(RECURSE);
+ descriptors.add(POLLING_INTERVAL);
+ descriptors.add(IGNORE_HIDDEN_FILES);
+ this.descriptors = Collections.unmodifiableList(descriptors);
+
+ final Set<Relationship> relationships = new HashSet<Relationship>();
+ relationships.add(REL_SUCCESS);
+ this.relationships = Collections.unmodifiableSet(relationships);
+
+ if (this.smbClient == null) {
+ initSmbClient();
+ }
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return this.relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ initiateFilterFile(context);
+ fileQueue.clear();
+
+ switch (context.getProperty(SHARE_ACCESS).getValue()) {
+ case SHARE_ACCESS_NONE:
+ sharedAccess = Collections.<SMB2ShareAccess>emptySet();
+ break;
+ case SHARE_ACCESS_READ:
+ sharedAccess = EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ);
+ break;
+ case SHARE_ACCESS_READDELETE:
+ sharedAccess = EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ, SMB2ShareAccess.FILE_SHARE_DELETE);
+ break;
+ case SHARE_ACCESS_READWRITEDELETE:
+ sharedAccess = EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ, SMB2ShareAccess.FILE_SHARE_WRITE, SMB2ShareAccess.FILE_SHARE_DELETE);
+ break;
+ }
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+ Collection<ValidationResult> set = new ArrayList<>();
+ if (validationContext.getProperty(USERNAME).isSet() && !validationContext.getProperty(PASSWORD).isSet()) {
+ set.add(new ValidationResult.Builder().explanation("Password must be set if username is supplied.").build());
+ }
+ return set;
+ }
+
+ private void initSmbClient() {
+ initSmbClient(new SMBClient());
+ }
+
+ public void initSmbClient(SMBClient smbClient) {
+ this.smbClient = smbClient;
+ }
+
+ private void initiateFilterFile(final ProcessContext context) {
+ final String filePatternStr = context.getProperty(FILE_FILTER).getValue();
+ filePattern = filePatternStr == null ? null : Pattern.compile(filePatternStr);
+ final String pathPatternStr = context.getProperty(PATH_FILTER).getValue();
+ pathPattern = pathPatternStr == null ? null : Pattern.compile(pathPatternStr);
+ ignoreHidden = context.getProperty(IGNORE_HIDDEN_FILES).asBoolean().booleanValue();
+ }
+
+ private boolean filterFile(final String directory, final String filename, final long fileAttributes) {
+ if (pathPattern != null && !pathPattern.matcher(directory).matches()) {
+ return false;
+ }
+ if (filePattern != null && !filePattern.matcher(filename).matches()) {
+ return false;
+ }
+ if (ignoreHidden && (fileAttributes & FileAttributes.FILE_ATTRIBUTE_HIDDEN.getValue()) != 0) {
+ return false;
+ }
+ return true;
+ }
+
+ private Set<String> performListing(final DiskShare diskShare, final String directory, final String filter, final boolean recurseSubdirectories) {
+ final Set<String> queue = new HashSet<>();
+ if (!diskShare.folderExists(directory)) {
+ return queue;
+ }
+
+ final List<FileIdBothDirectoryInformation> children = diskShare.list(directory);
+ if (children == null) {
+ return queue;
+ }
+
+ for (final FileIdBothDirectoryInformation child : children) {
+ final String filename = child.getFileName();
+ if (filename.equals(".") || filename.equals("..")) {
+ continue;
+ }
+ String fullPath;
+ if (directory.isEmpty()) {
+ fullPath = filename;
+ } else {
+ fullPath = directory + "\\" + filename;
+ }
+ final long fileAttributes = child.getFileAttributes();
+ if ((fileAttributes & FileAttributes.FILE_ATTRIBUTE_DIRECTORY.getValue()) != 0) {
+ if (recurseSubdirectories) {
+ queue.addAll(performListing(diskShare, fullPath, filter, true));
+ }
+ } else if (filterFile(directory, filename, fileAttributes)) {
+ queue.add(fullPath);
+ }
+ }
+
+ return queue;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+
+ final ComponentLog logger = getLogger();
+
+ final String hostname = context.getProperty(HOSTNAME).getValue();
+ final String shareName = context.getProperty(SHARE).getValue();
+
+ final String domain = context.getProperty(DOMAIN).getValue();
+ final String username = context.getProperty(USERNAME).getValue();
+ final String password = context.getProperty(PASSWORD).getValue();
+
+ AuthenticationContext ac = null;
+ if (username != null && password != null) {
+ ac = new AuthenticationContext(
+ username,
+ password.toCharArray(),
+ domain);
+ } else {
+ ac = AuthenticationContext.anonymous();
+ }
+
+
+ try (Connection connection = smbClient.connect(hostname);
+ Session smbSession = connection.authenticate(ac);
+ DiskShare share = (DiskShare) smbSession.connectShare(shareName)) {
+ String directory = context.getProperty(DIRECTORY).evaluateAttributeExpressions().getValue();
+ if (directory == null) {
+ directory = "";
+ }
+ final boolean keepingSourceFile = context.getProperty(KEEP_SOURCE_FILE).asBoolean();
+ final String filter = context.getProperty(FILE_FILTER).getValue();
+
+ if (fileQueue.size() < 100) {
+ final long pollingMillis = context.getProperty(POLLING_INTERVAL).asTimePeriod(TimeUnit.MILLISECONDS);
+ if ((queueLastUpdated.get() < System.currentTimeMillis() - pollingMillis) && listingLock.tryLock()) {
+ try {
+ final Set<String> listing = performListing(share, directory, filter, context.getProperty(RECURSE).asBoolean().booleanValue());
+
+ queueLock.lock();
+ try {
+ listing.removeAll(inProcess);
+ if (!keepingSourceFile) {
+ listing.removeAll(recentlyProcessed);
+ }
+
+ fileQueue.clear();
+ fileQueue.addAll(listing);
+
+ queueLastUpdated.set(System.currentTimeMillis());
+ recentlyProcessed.clear();
+
+ if (listing.isEmpty()) {
+ context.yield();
+ }
+ } finally {
+ queueLock.unlock();
+ }
+ } finally {
+ listingLock.unlock();
+ }
+ }
+ }
+
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ final List<String> files = new ArrayList<>(batchSize);
+ queueLock.lock();
+ try {
+ fileQueue.drainTo(files, batchSize);
+ if (files.isEmpty()) {
+ return;
+ } else {
+ inProcess.addAll(files);
+ }
+ } finally {
+ queueLock.unlock();
+ }
+
+ final ListIterator<String> itr = files.listIterator();
+ FlowFile flowFile = null;
+
+ try {
+ while (itr.hasNext()) {
+ final String file = itr.next();
+ final String[] fileSplits = file.split("\\\\");
+ final String filename = fileSplits[fileSplits.length - 1];
+ final String filePath = String.join("\\", Arrays.copyOf(fileSplits, fileSplits.length-1));
+ final URI uri = new URI("smb", hostname, "/" + file.replace('\\', '/'), null);
+
+ flowFile = session.create();
+ final long importStart = System.nanoTime();
+
+ try (File f = share.openFile(
+ file,
+ EnumSet.of(AccessMask.GENERIC_READ),
+ EnumSet.of(FileAttributes.FILE_ATTRIBUTE_NORMAL),
+ sharedAccess,
+ SMB2CreateDisposition.FILE_OPEN,
+ EnumSet.of(SMB2CreateOptions.FILE_SEQUENTIAL_ONLY));
+ InputStream is = f.getInputStream()) {
+
+ flowFile = session.importFrom(is, flowFile);
+
+ final long importNanos = System.nanoTime() - importStart;
+ final long importMillis = TimeUnit.MILLISECONDS.convert(importNanos, TimeUnit.NANOSECONDS);
+ final FileAllInformation fileInfo = f.getFileInformation();
+ final FileBasicInformation fileBasicInfo = fileInfo.getBasicInformation();
+
+ flowFile = session.putAttribute(flowFile, CoreAttributes.FILENAME.key(), filename);
+ flowFile = session.putAttribute(flowFile, CoreAttributes.PATH.key(), filePath);
+ flowFile = session.putAttribute(flowFile, CoreAttributes.ABSOLUTE_PATH.key(), "\\\\" + hostname + "\\" + shareName + "\\" + file);
+ flowFile = session.putAttribute(flowFile, FILE_CREATION_TIME_ATTRIBUTE, dateFormatter.format(fileBasicInfo.getCreationTime().toDate()));
+ flowFile = session.putAttribute(flowFile, FILE_LAST_ACCESS_TIME_ATTRIBUTE, dateFormatter.format(fileBasicInfo.getLastAccessTime().toDate()));
+ flowFile = session.putAttribute(flowFile, FILE_LAST_MODIFY_TIME_ATTRIBUTE, dateFormatter.format(fileBasicInfo.getLastWriteTime().toDate()));
+ flowFile = session.putAttribute(flowFile, HOSTNAME.getName(), hostname);
+ flowFile = session.putAttribute(flowFile, SHARE.getName(), shareName);
+ session.getProvenanceReporter().receive(flowFile, uri.toString(), importMillis);
+
+ session.transfer(flowFile, REL_SUCCESS);
+ logger.info("added {} to flow", new Object[]{flowFile});
+
+ } catch (SMBApiException e) {
+ // do not fail whole batch if a single file cannot be accessed
+ if (e.getStatus() == NtStatus.STATUS_SHARING_VIOLATION) {
+ logger.info("Could not acquire sharing access for file {}", new Object[]{file});
+ if (flowFile != null) {
+ session.remove(flowFile);
+ }
+ continue;
+ } else {
+ throw e;
+ }
+ }
+
+ try {
+ if (!keepingSourceFile) {
+ share.rm(file);
+ }
+ } catch (SMBApiException e) {
+ logger.error("Could not remove file {}", new Object[]{file});
+ }
+
+ if (!isScheduled()) { // if processor stopped, put the rest of the files back on the queue.
+ queueLock.lock();
+ try {
+ while (itr.hasNext()) {
+ final String nextFile = itr.next();
+ fileQueue.add(nextFile);
+ inProcess.remove(nextFile);
+ }
+ } finally {
+ queueLock.unlock();
+ }
+ }
+ }
+ session.commit();
+ } catch (final Exception e) {
+ logger.error("Failed to retrieve files due to {}", e);
+
+ // anything that we've not already processed needs to be put back on the queue
+ if (flowFile != null) {
+ session.remove(flowFile);
+ }
+ } finally {
+ queueLock.lock();
+ try {
+ inProcess.removeAll(files);
+ recentlyProcessed.addAll(files);
+ } finally {
+ queueLock.unlock();
+ }
+ }
+ } catch (Exception e) {
+ logger.error("Could not establish smb connection because of error {}", new Object[]{e});
+ context.yield();
+ }
+ }
+}
diff --git a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java
new file mode 100644
index 0000000..64c528f
--- /dev/null
+++ b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java
@@ -0,0 +1,334 @@
+/*
+ * 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.smb;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+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.processor.exception.ProcessException;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.net.URI;
+
+import com.hierynomus.smbj.SMBClient;
+import com.hierynomus.smbj.connection.Connection;
+import com.hierynomus.smbj.auth.AuthenticationContext;
+import com.hierynomus.smbj.share.DiskShare;
+import com.hierynomus.smbj.session.Session;
+import com.hierynomus.msfscc.FileAttributes;
+import com.hierynomus.msdtyp.AccessMask;
+import com.hierynomus.mssmb2.SMB2ShareAccess;
+import com.hierynomus.mssmb2.SMB2CreateDisposition;
+import com.hierynomus.mssmb2.SMB2CreateOptions;
+import com.hierynomus.smbj.share.File;
+import java.io.OutputStream;
+import java.util.EnumSet;
+
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"samba, smb, cifs, files, put"})
+@CapabilityDescription("Writes the contents of a FlowFile to a samba network location. " +
+ "Use this processor instead of a cifs mounts if share access control is important." +
+ "Configure the Hostname, Share and Directory accordingly: \\\\[Hostname]\\[Share]\\[path\\to\\Directory]")
+@SeeAlso({GetSmbFile.class})
+@ReadsAttributes({@ReadsAttribute(attribute="filename", description="The filename to use when writing the FlowFile to the network folder.")})
+public class PutSmbFile extends AbstractProcessor {
+ public static final String SHARE_ACCESS_NONE = "none";
+ public static final String SHARE_ACCESS_READ = "read";
+ public static final String SHARE_ACCESS_READDELETE = "read, delete";
+ public static final String SHARE_ACCESS_READWRITEDELETE = "read, write, delete";
+
+ public static final String REPLACE_RESOLUTION = "replace";
+ public static final String IGNORE_RESOLUTION = "ignore";
+ public static final String FAIL_RESOLUTION = "fail";
+
+ public static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder()
+ .name("Hostname")
+ .description("The network host to which files should be written.")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor SHARE = new PropertyDescriptor.Builder()
+ .name("Share")
+ .description("The network share to which files should be written. This is the \"first folder\"" +
+ "after the hostname: \\\\hostname\\[share]\\dir1\\dir2")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor DIRECTORY = new PropertyDescriptor.Builder()
+ .name("Directory")
+ .description("The network folder to which files should be written. This is the remaining relative " +
+ "path after the share: \\\\hostname\\share\\[dir1\\dir2]. You may use expression language.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .build();
+ public static final PropertyDescriptor DOMAIN = new PropertyDescriptor.Builder()
+ .name("Domain")
+ .description("The domain used for authentication. Optional, in most cases username and password is sufficient.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder()
+ .name("Username")
+ .description("The username used for authentication. If no username is set then anonymous authentication is attempted.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder()
+ .name("Password")
+ .description("The password used for authentication. Required if Username is set.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .sensitive(true)
+ .build();
+ public static final PropertyDescriptor CREATE_DIRS = new PropertyDescriptor.Builder()
+ .name("Create Missing Directories")
+ .description("If true, then missing destination directories will be created. If false, flowfiles are penalized and sent to failure.")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+ public static final PropertyDescriptor SHARE_ACCESS = new PropertyDescriptor.Builder()
+ .name("Share Access Strategy")
+ .description("Indicates which shared access are granted on the file during the write. " +
+ "None is the most restrictive, but the safest setting to prevent corruption.")
+ .required(true)
+ .defaultValue(SHARE_ACCESS_NONE)
+ .allowableValues(SHARE_ACCESS_NONE, SHARE_ACCESS_READ, SHARE_ACCESS_READDELETE, SHARE_ACCESS_READWRITEDELETE)
+ .build();
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the output directory")
+ .required(true)
+ .defaultValue(REPLACE_RESOLUTION)
+ .allowableValues(REPLACE_RESOLUTION, IGNORE_RESOLUTION, FAIL_RESOLUTION)
+ .build();
+ public static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("Batch Size")
+ .description("The maximum number of files to put in each iteration")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("100")
+ .build();
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to the output network path 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 the output network path for some reason are transferred to this relationship")
+ .build();
+
+ private List<PropertyDescriptor> descriptors;
+
+ private Set<Relationship> relationships;
+
+ private SMBClient smbClient = null;
+ private Set<SMB2ShareAccess> sharedAccess;
+
+ @Override
+ protected void init(final ProcessorInitializationContext context) {
+ final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
+ descriptors.add(HOSTNAME);
+ descriptors.add(SHARE);
+ descriptors.add(DIRECTORY);
+ descriptors.add(DOMAIN);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+ descriptors.add(CREATE_DIRS);
+ descriptors.add(SHARE_ACCESS);
+ descriptors.add(CONFLICT_RESOLUTION);
+ descriptors.add(BATCH_SIZE);
+ this.descriptors = Collections.unmodifiableList(descriptors);
+
+ final Set<Relationship> relationships = new HashSet<Relationship>();
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+ this.relationships = Collections.unmodifiableSet(relationships);
+
+ if (this.smbClient == null) {
+ initSmbClient();
+ }
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return this.relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+
+ switch (context.getProperty(SHARE_ACCESS).getValue()) {
+ case SHARE_ACCESS_NONE:
+ sharedAccess = Collections.<SMB2ShareAccess>emptySet();
+ break;
+ case SHARE_ACCESS_READ:
+ sharedAccess = EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ);
+ break;
+ case SHARE_ACCESS_READDELETE:
+ sharedAccess = EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ, SMB2ShareAccess.FILE_SHARE_DELETE);
+ break;
+ case SHARE_ACCESS_READWRITEDELETE:
+ sharedAccess = EnumSet.of(SMB2ShareAccess.FILE_SHARE_READ, SMB2ShareAccess.FILE_SHARE_WRITE, SMB2ShareAccess.FILE_SHARE_DELETE);
+ break;
+ }
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+ Collection<ValidationResult> set = Collections.emptySet();
+ if (validationContext.getProperty(USERNAME).isSet() && !validationContext.getProperty(PASSWORD).isSet()) {
+ set.add(new ValidationResult.Builder().explanation("Password must be set if username is supplied.").build());
+ }
+ return set;
+ }
+
+ private void initSmbClient() {
+ initSmbClient(new SMBClient());
+ }
+
+ void initSmbClient(SMBClient smbClient) {
+ this.smbClient = smbClient;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+ List<FlowFile> flowFiles = session.get(batchSize);
+ if ( flowFiles.isEmpty() ) {
+ return;
+ }
+ final ComponentLog logger = getLogger();
+ logger.debug("Processing next {} flowfiles", new Object[]{flowFiles.size()});
+
+ final String hostname = context.getProperty(HOSTNAME).getValue();
+ final String shareName = context.getProperty(SHARE).getValue();
+
+ final String domain = context.getProperty(DOMAIN).getValue();
+ final String username = context.getProperty(USERNAME).getValue();
+ String password = context.getProperty(PASSWORD).getValue();
+
+ AuthenticationContext ac = null;
+ if (username != null && password != null) {
+ ac = new AuthenticationContext(
+ username,
+ password.toCharArray(),
+ domain);
+ } else {
+ ac = AuthenticationContext.anonymous();
+ }
+
+ try (Connection connection = smbClient.connect(hostname);
+ Session smbSession = connection.authenticate(ac);
+ DiskShare share = (DiskShare) smbSession.connectShare(shareName)) {
+
+ for (FlowFile flowFile : flowFiles) {
+ String directory = context.getProperty(DIRECTORY).evaluateAttributeExpressions(flowFile).getValue();
+ final String filename = flowFile.getAttribute(CoreAttributes.FILENAME.key());
+ final long sendStart = System.nanoTime();
+ String fullPath;
+
+ if (directory == null) {
+ directory = "";
+ fullPath = filename;
+ } else {
+ fullPath = directory + "\\" + filename;
+
+ // missing directory handling
+ if (context.getProperty(CREATE_DIRS).asBoolean() && !share.folderExists(directory)) {
+ logger.debug("Creating folder {}", new Object[]{directory});
+ share.mkdir(directory);
+ }
+ }
+
+ final URI uri = new URI("smb", hostname, "/" + fullPath.replace('\\', '/'), null);
+
+ // replace strategy handling
+ SMB2CreateDisposition createDisposition = SMB2CreateDisposition.FILE_OVERWRITE_IF;
+ final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
+ if (!conflictResolution.equals(REPLACE_RESOLUTION) && share.fileExists(fullPath)) {
+ if (conflictResolution.equals(IGNORE_RESOLUTION)) {
+ session.transfer(flowFile, REL_SUCCESS);
+ logger.info("Transferring {} to success because file with same name already exists", new Object[]{flowFile});
+ continue;
+ } else if (conflictResolution.equals(FAIL_RESOLUTION)) {
+ flowFile = session.penalize(flowFile);
+ logger.warn("Penalizing {} and routing to failure as configured because file with the same name already exists", new Object[]{flowFile});
+ session.transfer(flowFile, REL_FAILURE);
+ continue;
+ }
+ }
+
+
+ try (File f = share.openFile(
+ fullPath,
+ EnumSet.of(AccessMask.GENERIC_WRITE),
+ EnumSet.of(FileAttributes.FILE_ATTRIBUTE_NORMAL),
+ sharedAccess,
+ createDisposition,
+ EnumSet.of(SMB2CreateOptions.FILE_WRITE_THROUGH));
+ OutputStream os = f.getOutputStream()) {
+
+ session.exportTo(flowFile, os);
+
+ final long sendNanos = System.nanoTime() - sendStart;
+ final long sendMillis = TimeUnit.MILLISECONDS.convert(sendNanos, TimeUnit.NANOSECONDS);
+ session.getProvenanceReporter().send(flowFile, uri.toString(), sendMillis);
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (Exception e) {
+ flowFile = session.penalize(flowFile);
+ session.transfer(flowFile, REL_FAILURE);
+ logger.error("Penalizing {} and routing to 'failure' because of error {}", new Object[]{flowFile, e});
+ }
+ }
+ } catch (Exception e) {
+ session.transfer(flowFiles, REL_FAILURE);
+ logger.error("Could not establish smb connection because of error {}", new Object[]{e});
+ }
+ }
+}
diff --git a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
new file mode 100644
index 0000000..bc5320b
--- /dev/null
+++ b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -0,0 +1,16 @@
+# 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.
+org.apache.nifi.processors.smb.PutSmbFile
+org.apache.nifi.processors.smb.GetSmbFile
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java
new file mode 100644
index 0000000..e9ab958
--- /dev/null
+++ b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java
@@ -0,0 +1,290 @@
+/*
+ * 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.smb;
+
+import com.hierynomus.msdtyp.FileTime;
+import com.hierynomus.msfscc.FileAttributes;
+import com.hierynomus.msfscc.fileinformation.FileAllInformation;
+import com.hierynomus.msfscc.fileinformation.FileBasicInformation;
+import com.hierynomus.msfscc.fileinformation.FileIdBothDirectoryInformation;
+import com.hierynomus.mssmb2.SMB2CreateDisposition;
+import com.hierynomus.smbj.SMBClient;
+import com.hierynomus.smbj.auth.AuthenticationContext;
+import com.hierynomus.smbj.connection.Connection;
+import com.hierynomus.smbj.session.Session;
+import com.hierynomus.smbj.share.DiskShare;
+import com.hierynomus.smbj.share.File;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.MockitoAnnotations;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anySet;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.times;
+
+public class GetSmbFileTest {
+ private TestRunner testRunner;
+
+ private SMBClient smbClient;
+ private Connection connection;
+ private Session session;
+ private DiskShare diskShare;
+ private ByteArrayOutputStream baOutputStream;
+
+ private final static String HOSTNAME = "host";
+ private final static String SHARE = "share";
+ private final static String DIRECTORY = "nifi\\input";
+ private final static String DOMAIN = "";
+ private final static String USERNAME = "user";
+ private final static String PASSWORD = "pass";
+
+ private void setupSmbProcessor() throws IOException {
+ smbClient = mock(SMBClient.class);
+ connection = mock(Connection.class);
+ session = mock(Session.class);
+ diskShare = mock(DiskShare.class);
+
+ baOutputStream = new ByteArrayOutputStream();
+
+ when(smbClient.connect(any(String.class))).thenReturn(connection);
+ when(connection.authenticate(any(AuthenticationContext.class))).thenReturn(session);
+ when(session.connectShare(SHARE)).thenReturn(diskShare);
+
+
+ testRunner.setProperty(GetSmbFile.HOSTNAME, HOSTNAME);
+ testRunner.setProperty(GetSmbFile.SHARE, SHARE);
+ testRunner.setProperty(GetSmbFile.DIRECTORY, DIRECTORY);
+ if (!DOMAIN.isEmpty()) {
+ testRunner.setProperty(GetSmbFile.DOMAIN, DOMAIN);
+ }
+ testRunner.setProperty(GetSmbFile.USERNAME, USERNAME);
+ testRunner.setProperty(GetSmbFile.PASSWORD, PASSWORD);
+
+ GetSmbFile GetSmbFile = (GetSmbFile) testRunner.getProcessor();
+ GetSmbFile.initSmbClient(smbClient);
+
+ }
+
+ private FileIdBothDirectoryInformation mockFile(String path, String filename, String fileContent, long fileAttributes) {
+ File smbfile = mock(File.class);
+ final String fullpath = path + "\\" + filename;
+ when(diskShare.openFile(
+ eq(fullpath),
+ anySet(),
+ anySet(),
+ anySet(),
+ any(SMB2CreateDisposition.class),
+ anySet()
+ )).thenReturn(smbfile);
+ when(smbfile.getFileName()).thenReturn(filename);
+
+ if (fileContent != null) {
+ InputStream is = new ByteArrayInputStream(fileContent.getBytes(StandardCharsets.UTF_8));
+ when(smbfile.getInputStream()).thenReturn(is);
+ }
+
+ FileIdBothDirectoryInformation fdInfo = mock(FileIdBothDirectoryInformation.class);
+ when(fdInfo.getFileName()).thenReturn(filename);
+ when(fdInfo.getFileAttributes()).thenReturn(fileAttributes);
+
+ FileAllInformation fileAllInfo = mock(FileAllInformation.class);
+ FileTime fileTime = FileTime.ofEpochMillis(0);
+ FileBasicInformation fileBasicInfo = new FileBasicInformation(fileTime, fileTime, fileTime, fileTime, 0);
+ when(smbfile.getFileInformation()).thenReturn(fileAllInfo);
+ when(fileAllInfo.getBasicInformation()).thenReturn(fileBasicInfo);
+ return fdInfo;
+ }
+
+ private FileIdBothDirectoryInformation mockFile(String path, String filename, String fileContent) {
+ return mockFile(path, filename, fileContent, FileAttributes.FILE_ATTRIBUTE_NORMAL.getValue());
+ }
+
+ private void verifyOpenFile(String path, String filename, int times) {
+ final String fullpath = path + "\\" + filename;
+ verify(diskShare, times(times)).openFile(
+ eq(fullpath),
+ anySet(),
+ anySet(),
+ anySet(),
+ any(SMB2CreateDisposition.class),
+ anySet()
+ );
+ }
+
+ private FileIdBothDirectoryInformation mockDir(String path, List<FileIdBothDirectoryInformation> files) {
+ final String[] fileSplits = path.split("\\\\");
+ final String filename = fileSplits[fileSplits.length - 1];
+ when(diskShare.folderExists(path)).thenReturn(true);
+ when(diskShare.list(path)).thenReturn(files);
+
+ FileIdBothDirectoryInformation fdInfo = mock(FileIdBothDirectoryInformation.class);
+ when(fdInfo.getFileName()).thenReturn(filename);
+ when(fdInfo.getFileAttributes()).thenReturn(FileAttributes.FILE_ATTRIBUTE_DIRECTORY.getValue());
+ return fdInfo;
+ }
+
+ @Before
+ public void init() throws IOException {
+ testRunner = TestRunners.newTestRunner(GetSmbFile.class);
+ MockitoAnnotations.initMocks(this);
+ setupSmbProcessor();
+ }
+
+ @Test
+ public void testOpenFileCalled() throws IOException {
+ FileIdBothDirectoryInformation file1 = mockFile(DIRECTORY, "file1.txt", "abc");
+ mockDir(DIRECTORY, new ArrayList<FileIdBothDirectoryInformation>(){{
+ add(file1);
+ }});
+ testRunner.run();
+ verifyOpenFile(DIRECTORY, "file1.txt", 1);
+ verifyOpenFile(DIRECTORY, "file2.txt", 0);
+ }
+
+ @Test
+ public void testHiddenFile() throws IOException {
+ testRunner.setProperty(GetSmbFile.IGNORE_HIDDEN_FILES, "true");
+ FileIdBothDirectoryInformation file1 = mockFile(DIRECTORY, "file1.txt", "abc", FileAttributes.FILE_ATTRIBUTE_HIDDEN.getValue());
+ FileIdBothDirectoryInformation file2 = mockFile(DIRECTORY, "file2.txt", "abc", FileAttributes.FILE_ATTRIBUTE_NORMAL.getValue());
+ mockDir(DIRECTORY, new ArrayList<FileIdBothDirectoryInformation>(){{
+ add(file1);
+ add(file2);
+ }});
+ testRunner.run();
+ verifyOpenFile(DIRECTORY, "file1.txt", 0);
+ verifyOpenFile(DIRECTORY, "file2.txt", 1);
+ }
+
+ @Test
+ public void testFileFilter() throws IOException {
+ testRunner.setProperty(GetSmbFile.FILE_FILTER, "file[0-9]\\.txt");
+ mockDir(DIRECTORY, new ArrayList<FileIdBothDirectoryInformation>(){{
+ add(mockFile(DIRECTORY, "something_else.txt", "abc"));
+ add(mockFile(DIRECTORY, "file1.txt", "abc"));
+ add(mockFile(DIRECTORY, "file2.txt", "abc"));
+ }});
+ testRunner.run();
+ verifyOpenFile(DIRECTORY, "something_else.txt", 0);
+ verifyOpenFile(DIRECTORY, "file1.txt", 1);
+ verifyOpenFile(DIRECTORY, "file2.txt", 1);
+ testRunner.assertTransferCount(GetSmbFile.REL_SUCCESS, 2);
+ }
+
+ @Test
+ public void testNonRecurse() throws IOException {
+ testRunner.setProperty(GetSmbFile.RECURSE, "false");
+ String subdir = DIRECTORY + "\\subdir1";
+ mockDir(DIRECTORY, new ArrayList<FileIdBothDirectoryInformation>(){{
+ add(mockFile(DIRECTORY, "file1.txt", "abc"));
+ add(mockFile(DIRECTORY, "file2.txt", "abc"));
+ add(mockDir(subdir, new ArrayList<FileIdBothDirectoryInformation>(){{
+ add(mockFile(subdir, "file3.txt", "abc"));
+ }}));
+ }});
+
+ testRunner.run();
+ verifyOpenFile(DIRECTORY, "file1.txt", 1);
+ verifyOpenFile(DIRECTORY, "file2.txt", 1);
+ verifyOpenFile(subdir, "file3.txt", 0);
+ testRunner.assertTransferCount(GetSmbFile.REL_SUCCESS, 2);
+ }
+
+ @Test
+ public void testRecurse() throws IOException {
+ testRunner.setProperty(GetSmbFile.RECURSE, "true");
+ String subdir = DIRECTORY + "\\subdir1";
+ mockDir(DIRECTORY, new ArrayList<FileIdBothDirectoryInformation>(){{
+ add(mockFile(DIRECTORY, "file1.txt", "abc"));
+ add(mockFile(DIRECTORY, "file2.txt", "abc"));
+ add(mockDir(subdir, new ArrayList<FileIdBothDirectoryInformation>(){{
+ add(mockFile(subdir, "file3.txt", "abc"));
+ }}));
+ }});
+
+
+ testRunner.run();
+ verifyOpenFile(DIRECTORY, "file1.txt", 1);
+ verifyOpenFile(DIRECTORY, "file2.txt", 1);
+ verifyOpenFile(subdir, "file3.txt", 1);
+ testRunner.assertTransferCount(GetSmbFile.REL_SUCCESS, 3);
+ }
+
+ @Test
+ public void testPathFilter() throws IOException {
+ testRunner.setProperty(GetSmbFile.RECURSE, "true");
+ testRunner.setProperty(GetSmbFile.PATH_FILTER, ".*\\\\subdir[0-9]");
+ String subdir1 = DIRECTORY + "\\subdir1";
+ String subdir2 = DIRECTORY + "\\subdir2";
+ String subdir3 = DIRECTORY + "\\foo";
+ mockDir(DIRECTORY, new ArrayList<FileIdBothDirectoryInformation>(){{
+ add(mockDir(subdir1, new ArrayList<FileIdBothDirectoryInformation>(){{
+ add(mockFile(subdir1, "file1.txt", "abc"));
+ }}));
+ add(mockDir(subdir2, new ArrayList<FileIdBothDirectoryInformation>(){{
+ add(mockFile(subdir2, "file2.txt", "abc"));
+ }}));
+ add(mockDir(subdir3, new ArrayList<FileIdBothDirectoryInformation>(){{
+ add(mockFile(subdir3, "file3.txt", "abc"));
+ }}));
+ }});
+
+
+ testRunner.run();
+ verifyOpenFile(subdir1, "file1.txt", 1);
+ verifyOpenFile(subdir2, "file2.txt", 1);
+ verifyOpenFile(subdir3, "file3.txt", 0);
+ testRunner.assertTransferCount(GetSmbFile.REL_SUCCESS, 2);
+ }
+
+ @Test
+ public void testBatchSize() throws IOException {
+ int batchSize = 10;
+ testRunner.setProperty(GetSmbFile.BATCH_SIZE, Integer.toString(batchSize));
+ List<FileIdBothDirectoryInformation> fileList = new ArrayList<FileIdBothDirectoryInformation>();
+ for (int i=0; i<batchSize*2; i++) {
+ fileList.add(mockFile(DIRECTORY, "file" + i + ".txt", Integer.toString(i)));
+ }
+ mockDir(DIRECTORY, fileList);
+
+ testRunner.run();
+ testRunner.assertTransferCount(GetSmbFile.REL_SUCCESS, batchSize);
+ testRunner.run();
+ testRunner.assertTransferCount(GetSmbFile.REL_SUCCESS, batchSize*2);
+ List<String> queuedFileNames = testRunner.getFlowFilesForRelationship(GetSmbFile.REL_SUCCESS).stream()
+ .map(f -> f.getAttribute(CoreAttributes.FILENAME.key()))
+ .collect(Collectors.toList());
+ for (int i=0; i<batchSize*2; i++) {
+ queuedFileNames.contains("file" + i + ".txt");
+ }
+ }
+}
diff --git a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/PutSmbFileTest.java b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/PutSmbFileTest.java
new file mode 100644
index 0000000..3d6e4a6
--- /dev/null
+++ b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/PutSmbFileTest.java
@@ -0,0 +1,224 @@
+/*
+ * 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.smb;
+
+import com.hierynomus.smbj.share.File;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertArrayEquals;
+import static org.mockito.ArgumentMatchers.anySet;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.any;
+import org.mockito.Captor;
+import org.mockito.MockitoAnnotations;
+import org.mockito.ArgumentCaptor;
+
+import com.hierynomus.smbj.SMBClient;
+import com.hierynomus.smbj.connection.Connection;
+import com.hierynomus.smbj.auth.AuthenticationContext;
+import com.hierynomus.smbj.share.DiskShare;
+import com.hierynomus.smbj.session.Session;
+import com.hierynomus.mssmb2.SMB2CreateDisposition;
+import com.hierynomus.mssmb2.SMB2ShareAccess;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Set;
+
+
+public class PutSmbFileTest {
+
+ private TestRunner testRunner;
+
+ private SMBClient smbClient;
+ private Connection connection;
+ private Session session;
+ private DiskShare diskShare;
+ private File smbfile;
+ private ByteArrayOutputStream baOutputStream;
+
+ private final static String HOSTNAME = "smbhostname";
+ private final static String SHARE = "smbshare";
+ private final static String DIRECTORY = "smbdirectory\\subdir";
+ private final static String DOMAIN = "mydomain";
+ private final static String USERNAME = "myusername";
+ private final static String PASSWORD = "mypassword";
+
+ @Captor
+ private ArgumentCaptor<Set<SMB2ShareAccess>> shareAccessSet;
+ @Captor
+ private ArgumentCaptor<AuthenticationContext> authenticationContext;
+
+ private void setupSmbProcessor() throws IOException {
+ smbClient = mock(SMBClient.class);
+ connection = mock(Connection.class);
+ session = mock(Session.class);
+ diskShare = mock(DiskShare.class);
+ smbfile = mock(File.class);
+ baOutputStream = new ByteArrayOutputStream();
+
+ when(smbClient.connect(any(String.class))).thenReturn(connection);
+ when(connection.authenticate(any(AuthenticationContext.class))).thenReturn(session);
+ when(session.connectShare(SHARE)).thenReturn(diskShare);
+ when(diskShare.openFile(
+ any(String.class),
+ anySet(),
+ anySet(),
+ anySet(),
+ any(SMB2CreateDisposition.class),
+ anySet()
+ )).thenReturn(smbfile);
+ when(smbfile.getOutputStream()).thenReturn(baOutputStream);
+
+ testRunner.setProperty(PutSmbFile.HOSTNAME, HOSTNAME);
+ testRunner.setProperty(PutSmbFile.SHARE, SHARE);
+ testRunner.setProperty(PutSmbFile.DIRECTORY, DIRECTORY);
+ testRunner.setProperty(PutSmbFile.DOMAIN, DOMAIN);
+ testRunner.setProperty(PutSmbFile.USERNAME, USERNAME);
+ testRunner.setProperty(PutSmbFile.PASSWORD, PASSWORD);
+
+
+ PutSmbFile PutSmbFile = (PutSmbFile) testRunner.getProcessor();
+ PutSmbFile.initSmbClient(smbClient);
+ }
+
+ private void testDirectoryCreation(String dirFlag, int times) throws IOException {
+ when(diskShare.folderExists(DIRECTORY)).thenReturn(false);
+
+ testRunner.setProperty(PutSmbFile.CREATE_DIRS, dirFlag);
+ testRunner.enqueue("data");
+ testRunner.run();
+
+ verify(diskShare, times(times)).mkdir(DIRECTORY);
+ }
+
+ private Set<SMB2ShareAccess> testOpenFileShareAccess() throws IOException {
+ testRunner.enqueue("data");
+ testRunner.run();
+
+ verify(diskShare, times(1)).openFile(
+ any(String.class),
+ anySet(),
+ anySet(),
+ shareAccessSet.capture(),
+ any(SMB2CreateDisposition.class),
+ anySet()
+ );
+ return shareAccessSet.getValue();
+ }
+
+ @Before
+ public void init() throws IOException {
+ testRunner = TestRunners.newTestRunner(PutSmbFile.class);
+ MockitoAnnotations.initMocks(this);
+ setupSmbProcessor();
+ }
+
+ @Test
+ public void testNormalAuth() throws IOException {
+ testRunner.enqueue("data");
+ testRunner.run();
+
+ verify(connection).authenticate(authenticationContext.capture());
+ AuthenticationContext acObj = authenticationContext.getValue();
+ assertEquals(acObj.getUsername(), USERNAME);
+ assertEquals(acObj.getDomain(), DOMAIN);
+ assertArrayEquals(acObj.getPassword(), PASSWORD.toCharArray());
+ }
+
+ @Test
+ public void testAnonymousAuth() throws IOException {
+ testRunner.removeProperty(PutSmbFile.USERNAME);
+ testRunner.enqueue("data");
+ testRunner.run();
+
+ verify(connection).authenticate(authenticationContext.capture());
+ AuthenticationContext acObj = authenticationContext.getValue();
+ AuthenticationContext compAc = AuthenticationContext.anonymous();
+ assertEquals(acObj.getUsername(), compAc.getUsername());
+ assertEquals(acObj.getDomain(), compAc.getDomain());
+ assertArrayEquals(acObj.getPassword(), compAc.getPassword());
+ }
+
+ @Test
+ public void testDirExistsWithoutCreate() throws IOException {
+ testDirectoryCreation("false", 0);
+ }
+
+ @Test
+ public void testDirExistsWithCreate() throws IOException {
+ testDirectoryCreation("true", 1);
+ }
+
+ @Test
+ public void testFileShareNone() throws IOException {
+ testRunner.setProperty(PutSmbFile.SHARE_ACCESS, PutSmbFile.SHARE_ACCESS_NONE);
+ Set<SMB2ShareAccess> shareAccessSet = testOpenFileShareAccess();
+ assertTrue(shareAccessSet.isEmpty());
+ }
+
+ @Test
+ public void testFileShareRead() throws IOException {
+ testRunner.setProperty(PutSmbFile.SHARE_ACCESS, PutSmbFile.SHARE_ACCESS_READ);
+ Set<SMB2ShareAccess> shareAccessSet = testOpenFileShareAccess();
+ assertTrue(shareAccessSet.contains(SMB2ShareAccess.FILE_SHARE_READ));
+ }
+
+ @Test
+ public void testFileShareReadWriteDelete() throws IOException {
+ testRunner.setProperty(PutSmbFile.SHARE_ACCESS, PutSmbFile.SHARE_ACCESS_READWRITEDELETE);
+ Set<SMB2ShareAccess> shareAccessSet = testOpenFileShareAccess();
+ assertTrue(shareAccessSet.contains(SMB2ShareAccess.FILE_SHARE_READ));
+ assertTrue(shareAccessSet.contains(SMB2ShareAccess.FILE_SHARE_WRITE));
+ assertTrue(shareAccessSet.contains(SMB2ShareAccess.FILE_SHARE_DELETE));
+ }
+
+ @Test
+ public void testFileExistsFail() throws IOException {
+ testRunner.setProperty(PutSmbFile.CONFLICT_RESOLUTION, PutSmbFile.FAIL_RESOLUTION);
+ when(diskShare.fileExists(any(String.class))).thenReturn(true);
+ testRunner.assertAllFlowFilesTransferred(PutSmbFile.REL_FAILURE);
+ }
+
+ @Test
+ public void testFileExistsIgnore() throws IOException {
+ testRunner.setProperty(PutSmbFile.CONFLICT_RESOLUTION, PutSmbFile.IGNORE_RESOLUTION);
+ when(diskShare.fileExists(any(String.class))).thenReturn(true);
+ testRunner.assertAllFlowFilesTransferred(PutSmbFile.REL_SUCCESS);
+ }
+
+ @Test
+ public void testConnectionError() throws IOException {
+ String emsg = "mock connection exception";
+ when(smbClient.connect(any(String.class))).thenThrow(new IOException(emsg));
+
+ testRunner.enqueue("1");
+ testRunner.enqueue("2");
+ testRunner.enqueue("3");
+ testRunner.run();
+
+ testRunner.assertAllFlowFilesTransferred(PutSmbFile.REL_FAILURE, 3);
+ }
+}
diff --git a/nifi-nar-bundles/nifi-smb-bundle/pom.xml b/nifi-nar-bundles/nifi-smb-bundle/pom.xml
new file mode 100644
index 0000000..1364fac
--- /dev/null
+++ b/nifi-nar-bundles/nifi-smb-bundle/pom.xml
@@ -0,0 +1,35 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-nar-bundles</artifactId>
+ <version>1.12.0-SNAPSHOT</version>
+ </parent>
+
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-smb-bundle</artifactId>
+ <version>1.12.0-SNAPSHOT</version>
+ <packaging>pom</packaging>
+
+ <modules>
+ <module>nifi-smb-processors</module>
+ <module>nifi-smb-nar</module>
+ </modules>
+
+</project>
diff --git a/nifi-nar-bundles/pom.xml b/nifi-nar-bundles/pom.xml
index 53e61a0..afda746 100755
--- a/nifi-nar-bundles/pom.xml
+++ b/nifi-nar-bundles/pom.xml
@@ -69,6 +69,7 @@
<module>nifi-mqtt-bundle</module>
<module>nifi-evtx-bundle</module>
<module>nifi-slack-bundle</module>
+ <module>nifi-smb-bundle</module>
<module>nifi-snmp-bundle</module>
<module>nifi-datadog-bundle</module>
<module>nifi-windows-event-log-bundle</module>
diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/logback.xml b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/logback.xml
index 4b707ec..b0c4571 100644
--- a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/logback.xml
+++ b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/logback.xml
@@ -124,6 +124,9 @@
<logger name="net.schmizz.sshj" level="WARN" />
<logger name="com.hierynomus.sshj" level="WARN" />
+ <!-- Suppress non-error messages from SMBJ which was emitting large amounts of INFO logs by default -->
+ <logger name="com.hierynomus.smbj" level="WARN" />
+
<!--
Logger for capturing user events. We do not want to propagate these
log events to the root logger. These messages are only sent to the
diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/logback.xml b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/logback.xml
index d22e83f..530cc19 100644
--- a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/logback.xml
+++ b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/logback.xml
@@ -126,6 +126,9 @@
<logger name="net.schmizz.sshj" level="WARN" />
<logger name="com.hierynomus.sshj" level="WARN" />
+ <!-- Suppress non-error messages from SMBJ which was emitting large amounts of INFO logs by default -->
+ <logger name="com.hierynomus.smbj" level="WARN" />
+
<!--
Logger for capturing user events. We do not want to propagate these
log events to the root logger. These messages are only sent to the
diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/logback.xml b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/logback.xml
index 01b936f..5e0b22e 100644
--- a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/logback.xml
+++ b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/logback.xml
@@ -124,6 +124,9 @@
<logger name="net.schmizz.sshj" level="WARN" />
<logger name="com.hierynomus.sshj" level="WARN" />
+ <!-- Suppress non-error messages from SMBJ which was emitting large amounts of INFO logs by default -->
+ <logger name="com.hierynomus.smbj" level="WARN" />
+
<!--
Logger for capturing user events. We do not want to propagate these
log events to the root logger. These messages are only sent to the
[nifi] 02/02: NIFI-7338 This closes #4169. Added comment about
com.hierynomus.smbj.SMBClient syncronicity
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 51091a5fca8dde3116c46e3135de14c658140c1e
Author: bibistroc <ma...@gbarbu.eu>
AuthorDate: Wed Aug 5 12:17:21 2020 +0300
NIFI-7338 This closes #4169. Added comment about com.hierynomus.smbj.SMBClient syncronicity
- added file.size attribute on GetSmbFile
Signed-off-by: Joe Witt <jo...@apache.org>
---
.../src/main/java/org/apache/nifi/processors/smb/GetSmbFile.java | 5 ++++-
.../src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java | 2 +-
.../test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java | 6 ++++++
3 files changed, 11 insertions(+), 2 deletions(-)
diff --git a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/GetSmbFile.java b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/GetSmbFile.java
index 0aac011..f7caa21 100644
--- a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/GetSmbFile.java
+++ b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/GetSmbFile.java
@@ -203,6 +203,7 @@ public class GetSmbFile extends AbstractProcessor {
public static final String FILE_CREATION_TIME_ATTRIBUTE = "file.creationTime";
public static final String FILE_LAST_MODIFY_TIME_ATTRIBUTE = "file.lastModifiedTime";
public static final String FILE_LAST_ACCESS_TIME_ATTRIBUTE = "file.lastAccessTime";
+ public static final String FILE_SIZE_ATTRIBUTE = "file.size";
public static final String FILE_MODIFY_DATE_ATTR_FORMAT = "yyyy-MM-dd'T'HH:mm:ssZ";
final static DateFormat dateFormatter = new SimpleDateFormat(FILE_MODIFY_DATE_ATTR_FORMAT, Locale.US);
@@ -222,7 +223,7 @@ public class GetSmbFile extends AbstractProcessor {
private final AtomicLong queueLastUpdated = new AtomicLong(0L);
- private SMBClient smbClient = null;
+ private SMBClient smbClient = null; // this gets synchronized when the `connect` method is called
private Pattern filePattern;
private Pattern pathPattern;
@@ -468,6 +469,7 @@ public class GetSmbFile extends AbstractProcessor {
final long importMillis = TimeUnit.MILLISECONDS.convert(importNanos, TimeUnit.NANOSECONDS);
final FileAllInformation fileInfo = f.getFileInformation();
final FileBasicInformation fileBasicInfo = fileInfo.getBasicInformation();
+ final long fileSize = fileInfo.getStandardInformation().getEndOfFile();
flowFile = session.putAttribute(flowFile, CoreAttributes.FILENAME.key(), filename);
flowFile = session.putAttribute(flowFile, CoreAttributes.PATH.key(), filePath);
@@ -475,6 +477,7 @@ public class GetSmbFile extends AbstractProcessor {
flowFile = session.putAttribute(flowFile, FILE_CREATION_TIME_ATTRIBUTE, dateFormatter.format(fileBasicInfo.getCreationTime().toDate()));
flowFile = session.putAttribute(flowFile, FILE_LAST_ACCESS_TIME_ATTRIBUTE, dateFormatter.format(fileBasicInfo.getLastAccessTime().toDate()));
flowFile = session.putAttribute(flowFile, FILE_LAST_MODIFY_TIME_ATTRIBUTE, dateFormatter.format(fileBasicInfo.getLastWriteTime().toDate()));
+ flowFile = session.putAttribute(flowFile, FILE_SIZE_ATTRIBUTE, String.valueOf(fileSize));
flowFile = session.putAttribute(flowFile, HOSTNAME.getName(), hostname);
flowFile = session.putAttribute(flowFile, SHARE.getName(), shareName);
session.getProvenanceReporter().receive(flowFile, uri.toString(), importMillis);
diff --git a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java
index 64c528f..1d8a9e0 100644
--- a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java
+++ b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java
@@ -162,7 +162,7 @@ public class PutSmbFile extends AbstractProcessor {
private Set<Relationship> relationships;
- private SMBClient smbClient = null;
+ private SMBClient smbClient = null; // this gets synchronized when the `connect` method is called
private Set<SMB2ShareAccess> sharedAccess;
@Override
diff --git a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java
index e9ab958..83b6ad3 100644
--- a/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java
+++ b/nifi-nar-bundles/nifi-smb-bundle/nifi-smb-processors/src/test/java/org/apache/nifi/processors/smb/GetSmbFileTest.java
@@ -21,6 +21,7 @@ import com.hierynomus.msfscc.FileAttributes;
import com.hierynomus.msfscc.fileinformation.FileAllInformation;
import com.hierynomus.msfscc.fileinformation.FileBasicInformation;
import com.hierynomus.msfscc.fileinformation.FileIdBothDirectoryInformation;
+import com.hierynomus.msfscc.fileinformation.FileStandardInformation;
import com.hierynomus.mssmb2.SMB2CreateDisposition;
import com.hierynomus.smbj.SMBClient;
import com.hierynomus.smbj.auth.AuthenticationContext;
@@ -120,8 +121,13 @@ public class GetSmbFileTest {
FileAllInformation fileAllInfo = mock(FileAllInformation.class);
FileTime fileTime = FileTime.ofEpochMillis(0);
FileBasicInformation fileBasicInfo = new FileBasicInformation(fileTime, fileTime, fileTime, fileTime, 0);
+ FileStandardInformation fileStandardInformation = mock(FileStandardInformation.class);
+
when(smbfile.getFileInformation()).thenReturn(fileAllInfo);
when(fileAllInfo.getBasicInformation()).thenReturn(fileBasicInfo);
+ when(fileAllInfo.getStandardInformation()).thenReturn(fileStandardInformation);
+ when(fileStandardInformation.getEndOfFile()).thenReturn((long) 0);
+
return fdInfo;
}