Inserting <collection-splitter /><collection-aggregator /> into a flow produces unexpected results

Having similar issues with a different configuration. Configuring the transformer at outbound endpoint level with transformer-refs is not working either.
Everything works if the outbound-endpoint is request-response.

I took another look at this and there seems to be multiple issues in both Mule 2.x (MULE-4213) and 3.x manifesting the same underlying problems.

Afaics the implementation of DefaultMessageCollection seems to be confused whether it is a message or a collection of messages - it contains both

• private transient Object payload; inherited from DefaultMuleMessage, and
• private List messageList = new CopyOnWriteArrayList(); the messages in the list contain their own payload objects

Both inside and outside the class there is then code assuming one way or another, eg.
the inherited getPayload() returns the internal payload object, whereas
getPayload(Class) returns a list of payload values taken from the messageList object.

So what seems to happen in the case of this ticket:

• collection-aggregator produces a DefaultMessageCollection message
• transformations: these normally modify the internal payload object
• outbound-endpoint uses methods that gets payload out of the messageList objects

I think there are two design issues here:

• Should collection-aggregator emit a message with a collection payload, or a collection of messages? The latter seems intuitively like a tight coupling to the configuration of previous stages in the flow.
• What is the payload of a DefaultMessageCollection?

This issue can be worked around by reconciling the internal "payload" states in DefaultMessageCollection.
E.g. in the case of
<collection-aggregator/>
<transformations... />
<outbound-endpoint/>

and if the transformations affect the internal payload object, and the outbound-endpoint receives the payloads in the internal messageList
-then it's possible to insert a transformer before the endpoint that reconciles the internal payload states, something along the lines of this:

	<scripting:transformer name="messagePayloadToCollectionPayload">
		<scripting:script engine="groovy">
			<scripting:text>
				if (message instanceof org.mule.DefaultMessageCollection) {
					def messages = message.getMessagesAsArray()
					if (payload instanceof List) {
						for (i = 0; i &lt; payload.size(); i++) {
							p = payload.get(i)
							m = messages[i]
							m.setPayload(p)
						}
					}
				}
				return payload // by convention
			</scripting:text>
		</scripting:script>
	</scripting:transformer>

If you are in control of what transformers do it's of course also possible to write transformers that directly apply changes (by using the right DefaultMessageCollection methods) to the payload state that the outbound-endpoint will read later.

NB I haven't done any extensive testing on this so there could be side effects or corner cases I'm not aware of.

cheers
Tobias

Hi Tobias,

Can you give me an idea of what transformers you are trying to apply to the MuleMessageCollection?

Also, if I were to make the change (and update usages) so that getPayload() is overwritten and behaves similarly to getPayload(Class), do you have a test case that I could run to ensure this fixes your issue?

Hello!
Please note that you can now use CombineCollectionTransformer instead of groovy script:

  
  ...
        <custom-transformer class="org.mule.transformer.simple.CombineCollectionsTransformer" doc:name="Message[] to Message of []"/>
  ...

I have also encountered this:

<script:transformer name="multiplier">
<script:script engine="groovy">
Object.sleep(1000);
return payload * payload;
</script:script>
</script:transformer>

<flow name="work1In">
<description>
<![CDATA[
Work-1 Demo.

This demonstrates a standard synchronous processing model of using a collection splitter. Each call to
the flow work1Worker will be done serially.
]]>
</description>

<http:inbound-endpoint name="httpWork1In"
host="127.0.0.1" port="8080"
path="work-1" exchange-pattern="request-response">
</http:inbound-endpoint>

<expression-transformer evaluator="groovy" expression="1..10"/>

<collection-splitter enableCorrelation="ALWAYS"/>
<flow-ref name="work1Worker"/>
<collection-aggregator/>
<combine-collections-transformer/>

<logger message="Aggregated payload: #[groovy:payload.size()] elements" level="ERROR"/>
<logger message="Aggregated payload: #[payload]" level="ERROR"/>

<json:object-to-json-transformer/>

<logger message="" level="ERROR"/>
<logger message="#[payload]" level="ERROR"/>

<logger message="Done." level="ERROR"/>
</flow>

<flow name="work1Worker">
<expression-transformer evaluator="message" expression="correlationSequence"/>
<logger message="ID: #[payload]" level="ERROR"/>
<transformer ref="multiplier"/>
</flow>

The response to the HTTP consumer is the result of the <collection-aggregator/>, and not the json-to-object-transformer.

Increasing priority to 3.3 blocker given number of votes on this and issue it is causing.