インバウンドアダプタメッセージハンドラ内のアウトバウンドゲートウェイを使用したSpringSftpフェッチファイル
JavaDSLを使用するインバウンドアダプターを使用して、SFTPサーバーからPDFファイルをポーリングしています。PDFファイルをフェッチした後、アプリケーションがSFTPサーバー上に同じ名前のCSV形式で存在する構成ファイルをフェッチするという使用例があります。構成ファイルをフェッチした後、アプリケーションは構成ファイルで定義されたプロパティを使用して元のpdfファイルを処理し、アウトバウンドアダプターを使用してSFTPサーバーにアップロードします。
アウトバウンドゲートウェイを使用して、同じスレッドのハンドラー内で構成ファイルをフェッチする際に問題が発生します。
これが私のコードです：
統合フローの登録：
for (String client : clientsArr) {
this.flowContext.registration(getInboundIntegrationFlow(client)).register();
}
this.flowContext.registration(getOutboundIntegrationFlow()).register();
this.flowContext.registration(sftpGatewayGetIntegrationFlow()).register();
インバウンドアダプタ統合フロー：
@Autowired
private SftpPdfMessageHandler messageHandler;
private IntegrationFlow getInboundIntegrationFlow(String client) {
String remoteDirectory = getRemoteDirectory(client);
String localDirectory = getLocalDirectory(client);
String inboundAdapterId = getInboundAdapterId(client);
return IntegrationFlows
.from(Sftp.inboundAdapter(sftpSessionFactory())
.preserveTimestamp(true)
.remoteDirectory(remoteDirectory)
.autoCreateLocalDirectory(true)
.localDirectory(new File(localDirectory))
.maxFetchSize(Integer.parseInt(sftpProperties.getMaxFetchSize()))
.filter(new SftpSimplePatternFileListFilter(sftpProperties.getRemoteFileFilter()))
.deleteRemoteFiles(true),
e -> e.id(inboundAdapterId)
.autoStartup(true)
.poller(Pollers
.fixedDelay(Long.parseLong(sftpProperties.getPollPeriodInSeconds()), TimeUnit.SECONDS)
.receiveTimeout(Long.parseLong(sftpProperties.getPollerTimeout()))
.maxMessagesPerPoll(Long.parseLong(sftpProperties.getMaxMessagesPerPoll()))
))
.handle(inBoundHandler())
.get();
}
public MessageHandler inBoundHandler() {
return message -> {
File file = (File) message.getPayload();
messageHandler.handleMessage(file);
};
}
アウトバウンドアダプタ統合フロー：
private IntegrationFlow getOutboundIntegrationFlow() {
return IntegrationFlows.from("sftpOutboundChannel")
.handle(Sftp.outboundAdapter(sftpSessionFactory(), FileExistsMode.FAIL)
.remoteDirectoryExpression(String.format("headers['%s']", FileHeaders.REMOTE_DIRECTORY))).get();
}
@Bean("sftpOutboundChannel")
public MessageChannel sftpOutboundChannel() {
return new DirectChannel();
}
SFTPメッセージハンドラー：
@Async("sftpHandlerAsyncExecutor")
public void handleMessage(File originalFile) {
File configFile = fetchConfigFile();
/*
process original file and store processed file in output file path on local directory
*/
boolean success = uploadFileToSftpServer(outputFilePath, client, entity);
if (success) {
deleteFileFromLocal(originalFile);
}
}
アウトバウンドゲートウェイのGET統合フロー：
private IntegrationFlow sftpGatewayGetIntegrationFlow() {
return IntegrationFlows.from("sftpGetInputChannel")
.handle(Sftp.outboundGateway(sftpSessionFactory(),
AbstractRemoteFileOutboundGateway.Command.GET, "payload")
.options(AbstractRemoteFileOutboundGateway.Option.DELETE,
AbstractRemoteFileOutboundGateway.Option.PRESERVE_TIMESTAMP)
.localDirectoryExpression(String.format("headers['%s']", Constants.HEADER_LOCAL_DIRECTORY_NAME))
.autoCreateLocalDirectory(true))
.channel("nullChannel")
.get();
}
@Bean("sftpGetInputChannel")
public MessageChannel sftpGetInputChannel() {
return new DirectChannel();
}
messageHandler.handleMessage()メソッドは非同期で（ThreadPoolTaskExecutorを使用して）呼び出され、アウトバウンドゲートウェイを使用して構成ファイルを内部的にフェッチします。しかし、同じスレッドでメッセージペイロードを送受信できる単一のチャネルが見つかりませんでした。SpringのドキュメントでMessagingTemplateを見つけましたが、これをアウトバウンドゲートウェイ統合フローに接続する方法が見つかりませんでした。
sftpGetMessageTemplate.sendAndReceive(sftpGetInputChannel, new GenericMessage<>("/dir/file.csv", headers))DirectChannel（）で「ディスパッチャにチャネルのサブスクライバがありません」という例外が発生します。
次のいずれかの方法でサーバーから必要なファイルをフェッチできるソリューションを探しています。
- 適切なチャネルを使用して、MessagingTemplateをIntegrationFlowと統合します（可能な場合）。
- インバウンドアダプタフロー内のメッセージハンドラのチェーン。元のファイルをポーリングした後、sftpアウトバウンドゲートウェイを使用して別のファイルをフェッチし、両方のオブジェクト（元のファイルと構成ファイル）で最終ハンドラを呼び出します。上記のカスタムコードを使用して、同様のことを実現しようとしています。
- マルチスレッド環境でGETコマンドに送信チャネルとポーラーチャネルを使用するその他の方法。
アプリケーションは、GETコマンドを使用しているときに、実行時にディレクトリパスを決定する必要があります。
回答
おそらく@MessagingGateway、アウトバウンドゲートウェイのチャネルと相互作用するための機能とその方法を学ぶ必要があります。
詳細については、ドキュメントを参照してください。 https://docs.spring.io/spring-integration/docs/5.3.2.RELEASE/reference/html/messaging-endpoints.html#gateway
結果として本当に設定ファイルを取得したいのであれば、そうすべきではありません.channel("nullChannel")。ゲートウェイが手元にあると、ゲートウェイによって入力されreplyChannelたTemporaryReplyChannelインスタンスを含むヘッダーが作成されます。次に、コードで、その関数型インターフェースを呼び出すAPIとして使用します。
実際、メッセージングゲートウェイは前述のを使用しMessagingTemplate.sendAndReceive()ます。