SI + шаблон наблюдателя

Question

У меня есть кодовая база, которая использует интеграцию с пружиной.

<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 работает нормально, и я получаю фильтруется файлы в handleMessage() метод myDataGetter боба.

Пока все хорошо. Теперь, в myDataGetter bean, я также фильтрую файлы, основанные на дате, например. poller получит мне 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 файлов, то есть 20130816 версии для data_file1,2 и 3. Таким образом я получил логику построения в методе handleMessage построить HashMap, который будет содержать последние версии этих 3 файлов , Это было сделано с наивной логикой повторения списка данных и сравнения файла с объектом с ним. После нескольких итераций я получаю сборку HashMap, которая содержит последние 3 файла.

Теперь мое следующее требование - передать эти 3 файла в следующий бит по каналу.

Но боб, который должен читать данные с канала, должен считываться только тогда, когда HashMap полностью построен с последними 3 файлами данных. Есть ли что-нибудь, что можно сделать в SI, чтобы данные передавались на следующий канал только после обработки всех входящих данных и фильтрации из него?

Я могу думать о том, что myDataGetter был таким же безошибочным, а bean для следующего канала в качестве наблюдателя. Но это не вписывается в то, как все делается в СИ.

Любые комментарии?

Answer 1

Вы можете использовать <aggregator/> с пользовательской стратегией выпуска.