Extending AbstractMessageAggregator

[cmdunn01886]

I would like to extend AbstractMessageAggregator (which extends AbstractMessageBarrierHandler), so I can flush the messages when my processing is complete.

What is the best way to configure this? If I configure it through the configuration file, it doesn't appear to work correctly. It appears i creates a MethodInvokingAggregator and invokes my methods, but doesn't create my class as a subclass of AbstractMessageAggregator. I say this because the barriers member of AbstractMessageBarrierHandler is always empty.

Do I need to start the Aggregator as a bean and start it manually?

Any help would be greatly appreciated.

Thanks in advance,
Mike

[iwein]

Currently the namespace doesn't support creating custom Aggregator types. I've tried extensions of the aggregator on a few occasions, and it seems a bit clunky. Obviously you can wire your extension as a bean, but you will probably come up with a few idea's as to making extensions to aggregator more convenient.

Make sure to create JIRA issues for these cases, that is the way to go to get them in the next release.

[cmdunn01886]

It's not clear to me how to set the input channel when extending the AbstractMessageAggregator. The input channel does not appear to be set via a method in the inheritance hierarchy. What is the easiest way to accomplish this?

[Mark Fisher]

The input-channel is actually configured on the "consumer" object that contains the handler (aggregator is a handler). When using the namespace support, each consumer is created with a reference to a handler and a reference to a channel.

There are actually two kinds of consumers depending on the type of input-channel: PollingConsumer (for PollableChannels) and EventDrivenConsumer (for SubscribableChannels). The easiest way to build your own consumer is to use the ConsumerEndpointFactoryBean. The FactoryBean decides which type of consumer to create based on the type of channel.

Hope that helps.
-Mark