Spring Integration: Split message again after using aggregate?

Question

In my Spring Integration powered project I have a splitter and payload-router for sending my data to various transformers. The new "transformed" objects are then passed back to an aggregator and processed.

Now I want to split up my aggregated results so they are persisted properly, since I need to route some of the objects to a seperate outbound-channel-adapter. To achieve this, I added a second splitter after my aggregator; but it seems only the first element in the aggregated collection is passed to the router.

This is my current flow:

<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="x.y.z.AbstractContent" channel="contentOutChannel" />
    <mapping type="x.y.z.Staff" 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 properly?

Note: If I just remove both the aggregator and second splitter altogether, the rest of the flow works perfectly.

It works (i.e. no errors), but only the first element in the collection is returned by the second splitter — seanhodges
I've edited my question to make it easier to understand my problem — seanhodges
Yes, the return object parts contains the whole collection when I debug. As an example, if I have {Article, Staff, Staff} in my parts object, my payload-type-router is fired only once with the Article object, the Staff objects are lost completely — seanhodges

Chris DeLashmutt Chris DeLashmutt · Accepted Answer · 2011-08-18T06:23:40

It is hard to tell everything that is going on since I don't have all your code, but I was able to make this flow work the way I think you wanted. The baseObjectTransformer does nothing but set a flag on the 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>
...

I then just exercise this flow 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 logging output produces all three entities.

All that being said, have you considered simply removing the extra splitter/aggregator and simply moving the setAuthors logic to your first splitter? I don't know all the details of your flow, but it would seem to simplify things.

Spring Integration: Split message again after using aggregate?

1 Answers