SFTP Dynamic Session Factory Message Handler ⇐ Java

[Anonymous]

Ich versuche, Dateien basierend auf der Anforderung an mehrere SFTP -Serverorte hochzuladen. Ich habe die SFTP -Verbindungsdetails aus der Datenbank gelesen und eine Sitzungsfabrik erstellt, die in einer Karte von Sitzungsfabriken auf der Grundlage eines Schlüssels gespeichert ist. Wenn ich nun versuche, MessageHandler für dasselbe zu erstellen, muss ich einen dynamischen Nachrichtenhandler mit der in der Karte erstellten und gespeicherten Sitzungsfabrik erstellen. Ich bin mir nicht sicher, wie man diese Karte und MessageHandler dynamisch verwendet. Konnte nicht genug Quelle finden. Ich habe meinen Code unten gegeben. Ich würde gerne wissen, was dafür der beste Ansatz ist. Ich habe die Federintegrationsströme gesehen, aber ich konnte nicht bekommen, wie man das dynamisch benutzt. Kann jemand bitte dabei helfen?

Code: Select all

public class SFTPService {

private final MessageChannel uploadEmployerSftpChannel;

@Retryable(
retryFor = { MessagingException.class, IOException.class },
backoff = @Backoff(delay = 5000)
)
public void uploadFromSftpConnection(@NonNull final SftpConnection sftpConnection,
@NonNull final Resource file,
@NonNull final String groupId) throws IOException {
if (UploadEnabled) {
log.info("Uploading file {} to server via InputStream with sftpRemoteDir header: {}", file.getFilename(), sftpConnection.getRemoteDirectory());

final Message message = MessageBuilder.withPayload(file.getInputStream())
.setHeader(SFTP_REMOTE_DIR, sftpConnection.getRemoteDirectory())
.setHeader(FILE_NAME, file.getFilename())
.setHeader(GROUP_ID, groupId)
.build();
uploadEmployerSftpChannel.send(message);
log.info("File successfully uploaded to client server path: {}", sftpConnection.getRemoteDirectory());
} else {
log.info("upload is disabled.  Skipping upload of file {} to client server.", file.getFilename());
}
}
}
< /code>
SFTP -Nachrichtenhandler < /p>
@RequiredArgsConstructor
@Configuration
public class SftpRuntimeSessionFactoryLocator {

private final Map sftpSessionFactoryMap = new HashMap();
private final SftpConnectionService sftpConnectionService;

public SessionFactory getSessionFactory(final Object groupId) {
if (!sftpSessionFactoryMap.containsKey(groupId)) {
sftpSessionFactoryMap.put(groupId, generateSessionFactory(groupId.toString()));
}
return sftpSessionFactoryMap.get(groupId);
}

public SessionFactory generateSessionFactory(final String key) {
//get sftp connection details from database
final SftpConnection sftpConnection = getSftpConnection(key);

final DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost(sftpConnection.getHost());
factory.setPort(sftpConnection.getPort());
factory.setPassword(sftpConnection.getPassword());
factory.setUser(sftpConnection.getUser());
factory.setAllowUnknownKeys(sftpConnection.isAllowUnknownKeys());
return new CachingSessionFactory(factory);
}

private SftpConnection getSftpConnection(final String groupId) {
return sftpConnectionService.getSFtpConnectionDetailsByGroupId(groupId);
}

@Bean
public MessageChannel uploadEmployerSftpChannel() {
return new DirectChannel();
}

@ServiceActivator(inputChannel = "uploadEmployerSftpChannel")
public MessageHandler dynamicSftpMessageHandler() {
return message -> {
final SessionFactory sftpSessionFactory = getSessionFactory(message.getHeaders().get("groupId"));

final SftpMessageHandler handler = new SftpMessageHandler(sftpSessionFactory);
handler.setUseTemporaryFileName(false);

handler.setRemoteDirectoryExpression(new FunctionExpression