You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by gf...@apache.org on 2026/04/17 12:39:31 UTC
(camel) branch main updated: CAMEL-23164: remove camel-ftp dependency to camel-mina-sftp
This is an automated email from the ASF dual-hosted git repository.
gfournier pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 627174305ac5 CAMEL-23164: remove camel-ftp dependency to camel-mina-sftp
627174305ac5 is described below
commit 627174305ac5168f5634462598d0ff170c5fcf7b
Author: Gaelle Fournier <ga...@gmail.com>
AuthorDate: Fri Apr 17 10:10:40 2026 +0200
CAMEL-23164: remove camel-ftp dependency to camel-mina-sftp
---
components/camel-mina-sftp/pom.xml | 14 +-
.../file/remote/mina/MinaSftpEndpoint.java | 4 +-
.../MinaSftpChangedExclusiveReadLockStrategy.java | 238 +++++++++++++++++++++
.../strategy/MinaSftpProcessStrategyFactory.java | 123 +++++++++++
.../remote/mina/strategy/MinaStrategyUtil.java | 57 +++++
5 files changed, 421 insertions(+), 15 deletions(-)
diff --git a/components/camel-mina-sftp/pom.xml b/components/camel-mina-sftp/pom.xml
index ea08212642c0..ddaf15e45ea0 100644
--- a/components/camel-mina-sftp/pom.xml
+++ b/components/camel-mina-sftp/pom.xml
@@ -39,19 +39,7 @@
<!-- Core Camel dependencies -->
<dependency>
<groupId>org.apache.camel</groupId>
- <artifactId>camel-ftp</artifactId>
- <exclusions>
- <!-- Exclude JSch - we use Apache MINA SSHD instead -->
- <exclusion>
- <groupId>com.github.mwiede</groupId>
- <artifactId>jsch</artifactId>
- </exclusion>
- <!-- Exclude commons-net - we only support SFTP, not FTP/FTPS -->
- <exclusion>
- <groupId>commons-net</groupId>
- <artifactId>commons-net</artifactId>
- </exclusion>
- </exclusions>
+ <artifactId>camel-ftp-common</artifactId>
</dependency>
<!-- Apache MINA SSHD for SFTP operations -->
<dependency>
diff --git a/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/MinaSftpEndpoint.java b/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/MinaSftpEndpoint.java
index 94a4ff765c9f..cee38caa334a 100644
--- a/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/MinaSftpEndpoint.java
+++ b/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/MinaSftpEndpoint.java
@@ -27,7 +27,7 @@ import org.apache.camel.component.file.remote.RemoteFileEndpoint;
import org.apache.camel.component.file.remote.RemoteFileOperations;
import org.apache.camel.component.file.remote.SftpDefaultMoveExistingFileStrategy;
import org.apache.camel.component.file.remote.SftpRemoteFile;
-import org.apache.camel.component.file.remote.strategy.SftpProcessStrategyFactory;
+import org.apache.camel.component.file.remote.mina.strategy.MinaSftpProcessStrategyFactory;
import org.apache.camel.component.file.strategy.FileMoveExistingStrategy;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriEndpoint;
@@ -147,7 +147,7 @@ public class MinaSftpEndpoint extends RemoteFileEndpoint<SftpRemoteFile> {
@Override
protected GenericFileProcessStrategy<SftpRemoteFile> createGenericFileStrategy() {
- return new SftpProcessStrategyFactory().createGenericFileProcessStrategy(getCamelContext(), getParamsAsMap());
+ return new MinaSftpProcessStrategyFactory().createGenericFileProcessStrategy(getCamelContext(), getParamsAsMap());
}
@Override
diff --git a/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/strategy/MinaSftpChangedExclusiveReadLockStrategy.java b/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/strategy/MinaSftpChangedExclusiveReadLockStrategy.java
new file mode 100644
index 000000000000..6f276377d1ec
--- /dev/null
+++ b/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/strategy/MinaSftpChangedExclusiveReadLockStrategy.java
@@ -0,0 +1,238 @@
+/*
+ * 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.camel.component.file.remote.mina.strategy;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.component.file.GenericFile;
+import org.apache.camel.component.file.GenericFileEndpoint;
+import org.apache.camel.component.file.GenericFileExclusiveReadLockStrategy;
+import org.apache.camel.component.file.GenericFileOperations;
+import org.apache.camel.component.file.remote.SftpRemoteFile;
+import org.apache.camel.spi.CamelLogger;
+import org.apache.camel.util.StopWatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MinaSftpChangedExclusiveReadLockStrategy implements GenericFileExclusiveReadLockStrategy<SftpRemoteFile> {
+ private static final Logger LOG = LoggerFactory.getLogger(MinaSftpChangedExclusiveReadLockStrategy.class);
+ private long timeout;
+ private long checkInterval = 5000;
+ private LoggingLevel readLockLoggingLevel = LoggingLevel.WARN;
+ private long minLength = 1;
+ private long minAge;
+ private boolean fastExistsCheck;
+
+ @Override
+ public void prepareOnStartup(
+ GenericFileOperations<SftpRemoteFile> tGenericFileOperations,
+ GenericFileEndpoint<SftpRemoteFile> tGenericFileEndpoint)
+ throws Exception {
+ // noop
+ }
+
+ @Override
+ public boolean acquireExclusiveReadLock(
+ GenericFileOperations<SftpRemoteFile> operations, GenericFile<SftpRemoteFile> file, Exchange exchange)
+ throws Exception {
+ boolean exclusive = false;
+
+ LOG.trace("Waiting for exclusive read lock to file: {}", file);
+
+ long lastModified = Long.MIN_VALUE;
+ long length = Long.MIN_VALUE;
+ StopWatch watch = new StopWatch();
+ long startTime = System.currentTimeMillis();
+
+ while (!exclusive) {
+ // timeout check
+ if (timeout > 0) {
+ long delta = watch.taken();
+ if (delta > timeout) {
+ CamelLogger.log(LOG, readLockLoggingLevel,
+ "Cannot acquire read lock within " + timeout + " millis. Will skip the file: " + file);
+ // we could not get the lock within the timeout period, so
+ // return false
+ return false;
+ }
+ }
+
+ long newLastModified = 0;
+ long newLength = 0;
+ // operations.listFiles returns SftpRemoteFile[] so
+ // do not use generic in the List files
+ Object[] files;
+ if (fastExistsCheck) {
+ // use the absolute file path to only pickup the file we want to
+ // check, this avoids expensive
+ // list operations if we have a lot of files in the directory
+ String path = file.getAbsoluteFilePath();
+ if (path.equals("/") || path.equals("\\")) {
+ // special for root (= home) directory
+ LOG.trace("Using fast exists to update file information in home directory");
+ files = operations.listFiles();
+ } else {
+ LOG.trace("Using fast exists to update file information for {}", path);
+ files = operations.listFiles(path);
+ }
+ } else {
+ String path = file.getParent();
+ if (path.equals("/") || path.equals("\\")) {
+ // special for root (= home) directory
+ LOG.trace(
+ "Using full directory listing in home directory to update file information. Consider enabling fastExistsCheck option.");
+ files = operations.listFiles();
+ } else {
+ LOG.trace(
+ "Using full directory listing to update file information for {}. Consider enabling fastExistsCheck option.",
+ path);
+ files = operations.listFiles(path);
+ }
+ }
+ LOG.trace("List files {} found {} files", file.getAbsoluteFilePath(), files.length);
+ for (Object f : files) {
+ SftpRemoteFile rf = (SftpRemoteFile) f;
+ boolean match;
+ if (fastExistsCheck) {
+ // uses the absolute file path as well
+ match = rf.getFilename().equals(file.getAbsoluteFilePath())
+ || rf.getFilename().equals(file.getFileNameOnly());
+ } else {
+ match = rf.getFilename().equals(file.getFileNameOnly());
+ }
+ if (match) {
+ newLastModified = rf.getLastModified();
+ newLength = rf.getFileLength();
+ }
+ }
+
+ LOG.trace("Previous last modified: {}, new last modified: {}", lastModified, newLastModified);
+ LOG.trace("Previous length: {}, new length: {}", length, newLength);
+ long newOlderThan = startTime + watch.taken() - minAge;
+ LOG.trace("New older than threshold: {}", newOlderThan);
+
+ if (newLength >= minLength && (minAge == 0 && newLastModified == lastModified && newLength == length
+ || minAge != 0 && newLastModified < newOlderThan)) {
+ LOG.trace("Read lock acquired.");
+ exclusive = true;
+ } else {
+ // set new base file change information
+ lastModified = newLastModified;
+ length = newLength;
+
+ boolean interrupted = sleep();
+ if (interrupted) {
+ // we were interrupted while sleeping, we are likely being
+ // shutdown so return false
+ return false;
+ }
+ }
+ }
+
+ return exclusive;
+ }
+
+ private boolean sleep() {
+ LOG.trace("Exclusive read lock not granted. Sleeping for {} millis.", checkInterval);
+ try {
+ Thread.sleep(checkInterval);
+ return false;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.debug("Sleep interrupted while waiting for exclusive read lock, so breaking out");
+ return true;
+ }
+ }
+
+ @Override
+ public void releaseExclusiveReadLockOnAbort(
+ GenericFileOperations<SftpRemoteFile> operations, GenericFile<SftpRemoteFile> file, Exchange exchange)
+ throws Exception {
+ // noop
+ }
+
+ @Override
+ public void releaseExclusiveReadLockOnRollback(
+ GenericFileOperations<SftpRemoteFile> operations, GenericFile<SftpRemoteFile> file, Exchange exchange)
+ throws Exception {
+ // noop
+ }
+
+ @Override
+ public void releaseExclusiveReadLockOnCommit(
+ GenericFileOperations<SftpRemoteFile> operations, GenericFile<SftpRemoteFile> file, Exchange exchange)
+ throws Exception {
+ // noop
+ }
+
+ public long getTimeout() {
+ return timeout;
+ }
+
+ @Override
+ public void setTimeout(long timeout) {
+ this.timeout = timeout;
+ }
+
+ public long getCheckInterval() {
+ return checkInterval;
+ }
+
+ @Override
+ public void setCheckInterval(long checkInterval) {
+ this.checkInterval = checkInterval;
+ }
+
+ @Override
+ public void setReadLockLoggingLevel(LoggingLevel readLockLoggingLevel) {
+ this.readLockLoggingLevel = readLockLoggingLevel;
+ }
+
+ public long getMinLength() {
+ return minLength;
+ }
+
+ public void setMinLength(long minLength) {
+ this.minLength = minLength;
+ }
+
+ public long getMinAge() {
+ return minAge;
+ }
+
+ public void setMinAge(long minAge) {
+ this.minAge = minAge;
+ }
+
+ public boolean isFastExistsCheck() {
+ return fastExistsCheck;
+ }
+
+ public void setFastExistsCheck(boolean fastExistsCheck) {
+ this.fastExistsCheck = fastExistsCheck;
+ }
+
+ @Override
+ public void setMarkerFiler(boolean markerFiler) {
+ // noop - not supported by ftp
+ }
+
+ @Override
+ public void setDeleteOrphanLockFiles(boolean deleteOrphanLockFiles) {
+ // noop - not supported by ftp
+ }
+}
diff --git a/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/strategy/MinaSftpProcessStrategyFactory.java b/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/strategy/MinaSftpProcessStrategyFactory.java
new file mode 100644
index 000000000000..156132676a55
--- /dev/null
+++ b/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/strategy/MinaSftpProcessStrategyFactory.java
@@ -0,0 +1,123 @@
+/*
+ * 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.camel.component.file.remote.mina.strategy;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Expression;
+import org.apache.camel.component.file.GenericFileExclusiveReadLockStrategy;
+import org.apache.camel.component.file.GenericFileProcessStrategy;
+import org.apache.camel.component.file.GenericFileProcessStrategyFactory;
+import org.apache.camel.component.file.remote.SftpRemoteFile;
+import org.apache.camel.component.file.strategy.GenericFileDeleteProcessStrategy;
+import org.apache.camel.component.file.strategy.GenericFileExpressionRenamer;
+import org.apache.camel.component.file.strategy.GenericFileNoOpProcessStrategy;
+import org.apache.camel.component.file.strategy.GenericFileRenameExclusiveReadLockStrategy;
+import org.apache.camel.component.file.strategy.GenericFileRenameProcessStrategy;
+import org.apache.camel.util.ObjectHelper;
+
+public final class MinaSftpProcessStrategyFactory implements GenericFileProcessStrategyFactory<SftpRemoteFile> {
+
+ @Override
+ public GenericFileProcessStrategy<SftpRemoteFile> createGenericFileProcessStrategy(
+ CamelContext context, Map<String, Object> params) {
+
+ // We assume a value is present only if its value not null for String
+ // and 'true' for boolean
+ Expression moveExpression = (Expression) params.get("move");
+ Expression moveFailedExpression = (Expression) params.get("moveFailed");
+ Expression preMoveExpression = (Expression) params.get("preMove");
+ boolean isNoop = params.get("noop") != null;
+ boolean isDelete = params.get("delete") != null;
+ boolean isMove = moveExpression != null || preMoveExpression != null || moveFailedExpression != null;
+
+ if (isDelete) {
+ GenericFileDeleteProcessStrategy<SftpRemoteFile> strategy = new GenericFileDeleteProcessStrategy<>();
+ strategy.setExclusiveReadLockStrategy(getExclusiveReadLockStrategy(params));
+ if (preMoveExpression != null) {
+ GenericFileExpressionRenamer<SftpRemoteFile> renamer = new GenericFileExpressionRenamer<>();
+ renamer.setExpression(preMoveExpression);
+ strategy.setBeginRenamer(renamer);
+ }
+ if (moveFailedExpression != null) {
+ GenericFileExpressionRenamer<SftpRemoteFile> renamer = new GenericFileExpressionRenamer<>();
+ renamer.setExpression(moveFailedExpression);
+ strategy.setFailureRenamer(renamer);
+ }
+ return strategy;
+ } else if (isMove || isNoop) {
+ GenericFileRenameProcessStrategy<SftpRemoteFile> strategy = new GenericFileRenameProcessStrategy<>();
+ strategy.setExclusiveReadLockStrategy(getExclusiveReadLockStrategy(params));
+ if (!isNoop && moveExpression != null) {
+ // move on commit is only possible if not noop
+ GenericFileExpressionRenamer<SftpRemoteFile> renamer = new GenericFileExpressionRenamer<>();
+ renamer.setExpression(moveExpression);
+ strategy.setCommitRenamer(renamer);
+ }
+ // both move and noop supports pre move
+ if (moveFailedExpression != null) {
+ GenericFileExpressionRenamer<SftpRemoteFile> renamer = new GenericFileExpressionRenamer<>();
+ renamer.setExpression(moveFailedExpression);
+ strategy.setFailureRenamer(renamer);
+ }
+ // both move and noop supports pre move
+ if (preMoveExpression != null) {
+ GenericFileExpressionRenamer<SftpRemoteFile> renamer = new GenericFileExpressionRenamer<>();
+ renamer.setExpression(preMoveExpression);
+ strategy.setBeginRenamer(renamer);
+ }
+ return strategy;
+ } else {
+ // default strategy will do nothing
+ GenericFileNoOpProcessStrategy<SftpRemoteFile> strategy = new GenericFileNoOpProcessStrategy<>();
+ strategy.setExclusiveReadLockStrategy(getExclusiveReadLockStrategy(params));
+ return strategy;
+ }
+ }
+
+ private static GenericFileExclusiveReadLockStrategy getExclusiveReadLockStrategy(Map<String, Object> params) {
+ GenericFileExclusiveReadLockStrategy strategy
+ = (GenericFileExclusiveReadLockStrategy) params.get("exclusiveReadLockStrategy");
+ if (strategy != null) {
+ return strategy;
+ }
+
+ // no explicit strategy set then fallback to readLock option
+ String readLock = (String) params.get("readLock");
+ if (ObjectHelper.isNotEmpty(readLock)) {
+ if ("none".equals(readLock) || "false".equals(readLock)) {
+ return null;
+ } else if ("rename".equals(readLock)) {
+ GenericFileRenameExclusiveReadLockStrategy<SftpRemoteFile> readLockStrategy
+ = new GenericFileRenameExclusiveReadLockStrategy<>();
+
+ MinaStrategyUtil.setup(readLockStrategy, params);
+
+ return readLockStrategy;
+ } else if ("changed".equals(readLock)) {
+ MinaSftpChangedExclusiveReadLockStrategy readLockStrategy = new MinaSftpChangedExclusiveReadLockStrategy();
+
+ MinaStrategyUtil.setup(readLockStrategy, params);
+
+ return readLockStrategy;
+ }
+ }
+
+ return null;
+ }
+}
diff --git a/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/strategy/MinaStrategyUtil.java b/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/strategy/MinaStrategyUtil.java
new file mode 100644
index 000000000000..7dcbb40ab711
--- /dev/null
+++ b/components/camel-mina-sftp/src/main/java/org/apache/camel/component/file/remote/mina/strategy/MinaStrategyUtil.java
@@ -0,0 +1,57 @@
+/*
+ * 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.camel.component.file.remote.mina.strategy;
+
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.component.file.strategy.GenericFileRenameExclusiveReadLockStrategy;
+
+final class MinaStrategyUtil {
+
+ private MinaStrategyUtil() {
+
+ }
+
+ static <T> void ifNotEmpty(Map<String, Object> params, String key, Class<T> clazz, Consumer<T> consumer) {
+ Object o = params.get(key);
+
+ if (o != null) {
+ consumer.accept(clazz.cast(o));
+ }
+ }
+
+ static void setup(GenericFileRenameExclusiveReadLockStrategy<?> readLockStrategy, Map<String, Object> params) {
+ ifNotEmpty(params, "readLockTimeout", Long.class, readLockStrategy::setTimeout);
+ ifNotEmpty(params, "readLockCheckInterval", Long.class, readLockStrategy::setCheckInterval);
+ ifNotEmpty(params, "readLockMarkerFile", Boolean.class, readLockStrategy::setMarkerFiler);
+ ifNotEmpty(params, "readLockLoggingLevel", LoggingLevel.class, readLockStrategy::setReadLockLoggingLevel);
+ }
+
+ static void setup(MinaSftpChangedExclusiveReadLockStrategy readLockStrategy, Map<String, Object> params) {
+ ifNotEmpty(params, "readLockTimeout", Long.class, readLockStrategy::setTimeout);
+ ifNotEmpty(params, "readLockCheckInterval", Long.class, readLockStrategy::setCheckInterval);
+ ifNotEmpty(params, "readLockMinLength", Long.class, readLockStrategy::setMinLength);
+ ifNotEmpty(params, "readLockMinAge", Long.class, readLockStrategy::setMinAge);
+ ifNotEmpty(params, "fastExistsCheck", Boolean.class, readLockStrategy::setFastExistsCheck);
+ ifNotEmpty(params, "readLockMarkerFile", Boolean.class, readLockStrategy::setMarkerFiler);
+ ifNotEmpty(params, "readLockLoggingLevel", LoggingLevel.class, readLockStrategy::setReadLockLoggingLevel);
+ }
+
+}