- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我正在尝试从多个 SFTP 服务器下载文件,然后处理这些文件。但我无法获取远程SFTP服务器的信息,例如:IpAddress、remoteDirectory,具体取决于哪个文件MessageHandler处理。相反,Payload 只包含本地下载的文件的信息。这是我在指南中使用的源代码: How to dynamically define file filter pattern for Spring Integration SFTP Inbound Adapter?
SFTIntegration.java
import com.jcraft.jsch.ChannelSftp.LsEntry;
import java.io.File;
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.file.remote.aop.RotatingServerAdvice;
import org.springframework.integration.file.remote.session.DelegatingSessionFactory;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.sftp.dsl.Sftp;
import org.springframework.integration.sftp.dsl.SftpInboundChannelAdapterSpec;
import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Component;
/**
* flow.
*/
@Configuration
@Component
public class SFTIntegration {
public static final String TIMEZONE_UTC = "UTC";
public static final String TIMESTAMP_FORMAT_OF_FILES = "yyyyMMddHHmmssSSS";
public static final String TEMPORARY_FILE_SUFFIX = ".part";
public static final int POLLER_FIXED_PERIOD_DELAY = 60000;
public static final int MAX_MESSAGES_PER_POLL = 100;
private static final Logger LOG = LoggerFactory.getLogger(SFTIntegration.class);
private static final String CHANNEL_INTERMEDIATE_STAGE = "intermediateChannel";
@Autowired
private ImportHandler importHandler;
/** database access repository */
private final SFTPServerConfigRepo SFTPServerConfigRepo;
@Value("${sftp.local.directory.download:${java.io.tmpdir}/localDownload}")
private String localTemporaryPath;
public SFTIntegration(final SFTPServerConfigRepo SFTPServerConfigRepo) {
this.SFTPServerConfigRepo = SFTPServerConfigRepo;
}
/**
* The default poller with 5s, 100 messages, RotatingServerAdvice and transaction.
*
* @return default poller.
*/
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata poller() {
return Pollers
.fixedDelay(POLLER_FIXED_PERIOD_DELAY)
.advice(advice())
.maxMessagesPerPoll(MAX_MESSAGES_PER_POLL)
.transactional()
.get();
}
/**
* The direct channel for the flow.
*
* @return MessageChannel
*/
@Bean
public MessageChannel stockIntermediateChannel() {
return new DirectChannel();
}
/**
* Get the files from a remote directory. Add a timestamp to the filename
* and write them to a local temporary folder.
*
* @return IntegrationFlow
*/
@Bean
public IntegrationFlow collectionInboundFlowFromSFTPServer() {
// Source definition
final SftpInboundChannelAdapterSpec sourceSpec = Sftp.inboundAdapter(delegatingSFtpSessionFactory())
.preserveTimestamp(true)
.patternFilter("*.*")
.deleteRemoteFiles(true)
.maxFetchSize(MAX_MESSAGES_PER_POLL)
.remoteDirectory("/")
.localDirectory(new File(localTemporaryPath))
.temporaryFileSuffix(TEMPORARY_FILE_SUFFIX)
.localFilenameExpression(new FunctionExpression<String>(s -> {
final int fileTypeSepPos = s.lastIndexOf('.');
return
DateTimeFormatter
.ofPattern(TIMESTAMP_FORMAT_OF_FILES)
.withZone(ZoneId.of(TIMEZONE_UTC))
.format(Instant.now())
+ "_"
+ s.substring(0, fileTypeSepPos)
+ s.substring(fileTypeSepPos);
}));
// Poller definition
final Consumer<SourcePollingChannelAdapterSpec> collectionInboundPoller = endpointConfigurer -> endpointConfigurer
.id("collectionInboundPoller")
.autoStartup(true)
.poller(poller());
return IntegrationFlows
.from(sourceSpec, collectionInboundPoller)
.transform(File.class, p -> {
// log step
LOG.info("flow=collectionInboundFlowFromSFTPServer, message=incoming file: " + p);
return p;
})
.channel(CHANNEL_INTERMEDIATE_STAGE)
.get();
}
@Bean
public IntegrationFlow collectionIntermediateStageChannel() {
return IntegrationFlows
.from(CHANNEL_INTERMEDIATE_STAGE)
.handle(importHandler)
.channel(new NullChannel())
.get();
}
public DefaultSftpSessionFactory createNewSftpSessionFactory(final SFTPServerConfig pc) {
final DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(
false);
factory.setHost(pc.getServerIp());
factory.setPort(pc.getPort());
factory.setUser(pc.getUsername());
factory.setPassword(pc.getPassword());
factory.setAllowUnknownKeys(true);
return factory;
}
@Bean
public DelegatingSessionFactory<LsEntry> delegatingSFtpSessionFactory() {
final List<SFTPServerConfig> partnerConnections = SFTPServerConfigRepo.findAll();
if (partnerConnections.isEmpty()) {
return null;
}
final Map<Object, SessionFactory<LsEntry>> factories = new LinkedHashMap<>(10);
for (SFTPServerConfig pc : partnerConnections) {
// create a factory for every key containing server type, url and port
if (factories.get(pc.getKey()) == null) {
factories.put(pc.getKey(), createNewSftpSessionFactory(pc));
}
}
// use the first SF as the default
return new DelegatingSessionFactory<>(factories, factories.values().iterator().next());
}
@Bean
public RotatingServerAdvice advice() {
final List<SFTPServerConfig> sftpConnections = SFTPServerConfigRepo.findAll();
final List<RotatingServerAdvice.KeyDirectory> keyDirectories = new ArrayList<>();
for (SFTPServerConfig pc : sftpConnections) {
keyDirectories
.add(new RotatingServerAdvice.KeyDirectory(pc.getKey(), pc.getServerPath()));
}
return new RotatingServerAdvice(delegatingSFtpSessionFactory(), keyDirectories, true);
}
}
ImportHandler.java
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;
@Service
public class ImportHandler {
public void handle(Message<?> message) {
System.out.println("Hello " + message);
System.out.println(message.getPayload());
System.out.println(message.getHeaders());
//How can I get the information of remote server Ip address, remoteDirectory here where the file comes from
}
}
如果您有任何想法,请告诉我。非常感谢!.
最佳答案
目前不支持;请open a new feature request .
关于java - Spring Integration Java DSL SFTP如何在处理程序中获取远程SFTP服务器信息,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/57606912/
谁能解释一下 Server.MapPath(".")、Server.MapPath("~")、Server.MapPath(@"之间的区别\") 和 Server.MapPath("/")? 最佳答案
我不知道,为什么我们要使用 Server.UrlEncode() & Server.UrlDecode()?!在 QueryString 中我们看到 URL 中的任何内容,那么为什么我们要对它们进行编
我已经通过 WHM 在我的一个域上安装了 ssl 证书。网站正在使用 https://xyz.com . 但是它不适用于 https://www.xyz.com .我已经检查了证书,它也适用于 www
我已经使用 WMI 检测操作系统上是否存在防病毒软件,itz 正常工作并通过使用命名空间向我显示防病毒信息,例如 win xp 和 window7 上的名称和实例 ID:\root\SecurityC
我们有 hive 0.10 版本,我们想知道是否应该使用 Hive Server 1 或 Hive Server2。另一个问题是连接到在端口 10000 上运行的 Hive 服务器,使用 3rd 方工
我想在 C++ 中使用 Windows Server API 设置一个 HTTPS 服务器,我使用了示例代码,它在 HTTP 上工作正常,但我就是不能让它在 HTTPS 上工作。 (我不想要客户端 S
我写了一个非常基本的类来发送电子邮件。我用 smtp 服务器对其进行了测试,它工作正常,但是当我尝试使用我公司的交换服务器时,它给出了这个异常: SMTP 服务器需要安全连接或客户端未通过身份验证。服
我的应用程序包含一个“网关”DataSnap REST 服务器,它是所有客户端的第一个访问点。根据客户端在请求中传递的用户名(基本身份验证),请求需要重定向到另一个 DataSnap 服务器。我的问题
我有一个 Tomcat 服务器和一个 Glassfish4 服务器。我的 Servlet 在 Tomcat 服务器上启动得很好,但在 Glassfish4 服务器上给我一个“HTTP Status 4
我在 vmware 上创建了一个 ubuntu 服务器。我用它作为文件服务器。如果我通过托管虚拟机的计算机进行连接,则可以访问它。我无法从同一网络上的其他计算机执行此操作。提前致谢! 最佳答案 首先确
如何重启 Rails 服务器?我从 开始 rails server -d 所以服务器是分离的 我知道的唯一方法就是去做ps 辅助 | grep rails 并 kill -9关于过程#但是像这样杀死进
我实际上正在尝试找到编写一个简单的 XMPP 服务器的最佳方法,或者找到一个占用空间非常小的服务器。我只关心XMPP的核心功能(状态、消息传递、群组消息传递)。目前还在学习 XMPP 协议(proto
我实际上正在尝试找到编写简单 XMPP 服务器的最佳方法,或者找到一个占用空间非常小的方法。我只关心 XMPP 的核心功能(统计、消息、组消息)。目前也在学习 XMPP 协议(protocol),所以
我们正在尝试从 Java JAX-RS 适配器访问 SOAP 1.1 Web 服务。 我们正在使用从 WSDL 生成的 SOAP 客户端。 但是当解码 SOAP 故障时,我们得到以下异常: ... C
目前,我和许多其他人正在多个平台(Windows、OS X 和可能的 Linux)上使用 Python HTTP 服务器。我们正在使用 Python HTTP 服务器来测试 JavaScript 游戏
我有一个连续运行的服务器程序(C#/.NET 2.0 on Linux with mono),我想从 PHP 脚本连接到它以在网站上显示状态信息。 目的是创建一个(某种)实时浏览器游戏(无 Flash
所以我有一个单页客户端应用程序。 正常流程: 应用程序 -> OAuth2 服务器 -> 应用程序 我们有自己的 OAuth2 服务器,因此人们可以登录应用程序并获取与用户实体关联的 access_t
我们刚刚将测试 Web 服务器从 Server 2008 升级到 Server 2012 R2。我们有一个部署我们网站的批处理脚本。当它将站点推送到服务器时,它现在失败了。奇怪的是,我可以使用相同的发
建议一些加载SpagoBI服务器的方法,我尝试了所有方法来解析spagobi服务器。在 Catalina 中,错误是 - * SEVERE: Unable to process Jar entry [
当我们点击应用程序服务器(apache tomcat)时,它会创建一个线程来处理我们的请求并与 tomcat 连接,建立连接,tomcat 创建另一个线程来处理请求并将其传递给连接,连接线程将其传递给
我是一名优秀的程序员,十分优秀!