java - SI+观察者模式

Question

我有一个使用 spring-integration 的代码库。

<integration-ftp:inbound-channel-adapter id="ftpInbound"
                               channel="ftpChannel"
                               session-factory="ftpClientFactory"
                               filter="myCustomFilter"                                                                                             
                               auto-create-local-directory="true"
                               delete-remote-files="false"                                                                    
                               remote-directory="/foo/bar"                                                                     
                               local-directory="file:output">
    <integration:poller fixed-rate="5000" max-messages-per-poll="-1"/>
</integration-ftp:inbound-channel-adapter>



<integration:channel id="ftpChannel">
        <integration:queue />
</integration:channel>

<integration:service-activator id="mySA"  method="handleMessage" input-channel="ftpChannel" output-channel="outputChannel" ref="myDataGetter">
    <integration:poller fixed-rate="1000" max-messages-per-poll="-1"/>  
</integration:service-activator>

myCustomFilter 的 bean 工作正常，我在 myDataGetter bean 的 handleMessage() 方法中获取已过滤的文件。

到目前为止一切顺利。现在在 myDataGetter bean 中，我根据日期进一步过滤文件，例如轮询器将为我提供 9 个文件，但实际上它们是仅适用于 3 个日期的 3 个版本。

data_file1.20130816
data_file1.20130815
data_file1.20130814
data_file2.20130816
data_file2.20130815
data_file2.20130814
data_file3.20130816
data_file3.20130815
data_file3.20130814

现在我的目标是获取最新的 3 个文件，即 data_file1,2 和 3 的 20130816 版本。因此，我在 handleMessage 方法中构建了一个逻辑来构造 HashMap，其中包含这 3 个文件的最新版本。它通过迭代数据列表并将主题中的文件与其进行比较的简单逻辑来完成。经过几次迭代后，我得到了包含最新 3 个文件的 HashMap 构建。

现在我的下一个要求是通过 channel 将这 3 个文件传递给下一个 bean。

但是应该从 channel 读取数据的 bean 只能在使用最新 3 个数据文件完全构建 HashMap 时读取。SI 中是否可以做任何事情，以便仅在处理所有传入数据并从中过滤出一组数据后才将数据放入下一个 channel ？

我可以考虑让 myDataGetter 成为可观察的，并将下一个 channel 的 bean 作为观察者。但它不符合 SI 中的工作方式。

有什么意见吗？

Answer 1

您可以使用 <aggregator/>使用自定义发布策略。