Spring Integration: split message again after using aggregate?

Question

In my Spring Integration powered project, I have a splitter and a useful router for sending my data to various transformers. New "converted" objects are then transferred back to the aggregator and processed.

Now I want to split my aggregated results so that they are saved correctly, since I need to direct some objects to a separate outbound channel adapter. To do this, I added a second delimiter after my aggregator; but it seems that only the first item in the aggregated collection is passed to the router.

This is my current thread:

<splitter ref="articleContentExtractor" /> <!-- This router works exactly as expected --> <payload-type-router> ... routing to various transformers ... ... results are sent to articleOutAggregateChannel ... </payload-type-router> <aggregator ref="articleAggregator" /> <splitter /> <!-- This is where it seems to go wrong, the second splitter returns only the first object in the collection --> <payload-type-router resolution-required="true"> <mapping type="xyzAbstractContent" channel="contentOutChannel" /> <mapping type="xyzStaff" channel="staffOutChannel" /> </payload-type-router> <outbound-channel-adapter id="contentSaveService" ref="contentExporter" method="persist" channel="contentOutChannel" /> <outbound-channel-adapter id="staffSaveService" ref="staffExporter" method="persist" channel="staffOutChannel" /> 

And my aggregator code:

 @Aggregator public List<? super BaseObject> compileArticle(List<? super BaseObject> parts) { // Search for the required objects for referencing Iterator<? super BaseObject> it = parts.iterator(); Article article = null; List<Staff> authors = new ArrayList<Staff>(); while (it.hasNext()) { Object part = it.next(); if (part instanceof Article) { article = (Article)part; } else if (part instanceof Staff) { authors.add((Staff)part); } } // Apply references article.setAuthors(authors); return parts; } 

What am I doing wrong? Am I using my aggregator correctly?

Note. If I just delete both the aggregator and the second splitter as a whole, the rest of the stream works fine.

Answer 1

It’s hard to say everything that happens because I don’t have all of your code, but I was able to make this thread work the way I think you wanted. BaseObjectTransformer does nothing but set the flag on objects passing through it:

 ... <context:component-scan base-package="net.grogscave.example" /> <channel id="inputChannel" /> <splitter ref="articleContentExtractor" input-channel="inputChannel" output-channel="splitArticleStaff" /> <channel id="splitArticleStaff" /> <payload-type-router input-channel="splitArticleStaff"> <mapping type="net.grogscave.example.domain.Article" channel="articleChannel" /> <mapping type="net.grogscave.example.domain.Staff" channel="staffChannel" /> </payload-type-router> <channel id="articleChannel" /> <transformer input-channel="articleChannel" output-channel="articleOutAggregateChannel" ref="baseObjectTransformer"/> <channel id="staffChannel" /> <transformer input-channel="staffChannel" output-channel="articleOutAggregateChannel" ref="baseObjectTransformer"/> <channel id="articleOutAggregateChannel" /> <aggregator ref="articleAggregator" input-channel="articleOutAggregateChannel" output-channel="splitArticleInChannel"/> <channel id="splitArticleInChannel" /> <splitter input-channel="splitArticleInChannel" output-channel="splitArticleOutChannel" /> <channel id="splitArticleOutChannel" /> <payload-type-router resolution-required="true" input-channel="splitArticleOutChannel"> <mapping type="net.grogscave.example.domain.Article" channel="contentOutChannel" /> <mapping type="net.grogscave.example.domain.Staff" channel="staffOutChannel" /> </payload-type-router> <channel id="contentOutChannel"> <queue capacity="10" /> </channel> <channel id="staffOutChannel"> <queue capacity="10" /> </channel> ... 

Then I just execute this thread with the following code in a test case:

  ... AbstractApplicationContext context = new ClassPathXmlApplicationContext( "/META-INF/spring/split-agg-split.xml", SplitAggSplitTests.class); MessageChannel inputChannel = context.getBean("inputChannel", MessageChannel.class); PollableChannel contentOutChannel = context.getBean("contentOutChannel", PollableChannel.class); PollableChannel staffOutChannel = context.getBean("staffOutChannel", PollableChannel.class); inputChannel.send(new GenericMessage<String>("Dewey Wins!,A. Fool, C. Lewlis")); logger.info("==> Article recieved: " + contentOutChannel.receive(0).getPayload()); for(int i = 0; i < 2; i++) { logger.info("==> Staff recieved: " + staffOutChannel.receive(0).getPayload()); } ... 

My log output displays all three objects.

All that was said, did you consider just removing the extra splitter / aggregator and just moving the setAuthors logic to your first splitter? I don’t know all the details of your flow, but it seems like it would make it easier.