Multi channel endpoint and extending AbstractPollingEndpoint

[keidi19]

Hi,

I need some kind of mechanism for polling multiple channels (according to a certain strategy determined by the user). I was thinking of extending the AbstractPollingEndpoint in a way quite similar to the PollingConsumer class but instead of polling from a single channel, have it poll from multiple channels.

I have 2 questions in this regard:

1. Does it make sense to extend AbstractPollingEndpoint? I'm guessing this is not the kind of API that you guarantee to keep stable.
2. If it doesn't make sense, is there an alternative?

Thanks,
Shahar

[Mark Fisher]

Depending on the details of your use-case, I think there are a couple options that would preclude the need to extend AbstractPollingEndpoint.

Can you describe the nature of why you need multiple pollers?

If it's a matter of combining different poll-time settings, then perhaps it can be accomplished with a custom Trigger implementation. If it's a matter of actually polling different sources, then I would probably recommend using multiple 'inbound-channel-adapters' that share a common reference to a 'channel' for passing the resulting Messages.

Does your use-case fall into either of those categories? If not, can you describe the difference?

Thanks,
Mark

[Mark Fisher]

Actually, reading it again, I see that you are referring to an internal component - presumably polling other channels that are producing output from some upstream component.

So, I will rephrase my question: why can you not simply use the same output-channel on those upstream components? Once I understand your answer to that question, hopefully I'll be able to provide some alternatives.

[keidi19]

Hi Mark,

The idea is to be able to prioritize between channels without causing starvation on one\some of the channels. So I want to be able to poll from channels according to some predefined ratio, thus giving advantage to some channels over others.

Actually you gave me and idea. I could use the same output-channel on all upstream components, defining different polling rate to each component. So components with higher priority will poll more frequently than components with lower priority. Will that do the trick?

Thanks,
Shahar

[keidi19]

And I could probably also use the max-messages-per-poll attribute in order to poll more messages from high-priority channels, right?

I'm thinking of using the following configuration:

<channel id="InternalDistributionChannel">
		<queue capacity="500"/>
	</channel>
	
	<bridge input-channel="InternalDistributionChannel" output-channel="DistributionChannel">
		<poller send-timeout="-1" max-messages-per-poll="{distribution.rate}>
			<interval-trigger interval="${distribution.interval}"/>
		</poller>
	</bridge>

	<channel id="DistributionChannel" >
		<queue ref="DistributionQueue"/>
	</channel>

The DistributionQueue is a HazelCast queue so it is distributed across multiple JVMs. The ${distribution.interval} and ${distribution.rate} will be given different values depending on the priority of the component upstreaming to the DistributionChannel. This way the DistributionChannel will hold a mix of messages which represents a prioritized distribution.

Will appreciate any comments. Thanks,
Shahahr

[Mark Fisher]

Have you considered using a "priority-queue" within that channel rather than a simple "queue" sub-element?

[keidi19]

Yes. A priority-queue is not good for 2 reasons:
1. The queue has to be a distributed queue. I'm not using a simple queue but rather a HazelCast queue. HazelCast does not have a priority queue implementation quite yet.
2. The throughput of our system is quite high making it quite possible that low priority items will be starved in a priority queue.

[Mark Fisher]

Okay. Have you considered using a JMS-backed channel with JMSPriority properties?

[keidi19]

We preferred grid-based distribution and redundancy over JMS. JMS is not easily scalable in opposed to grid-based technologies.
Are you proposing other options because the solution above is not good enough?

[oleg.zhurakousky]

If you are using HazelCast, you can possibly implement HazelCast based MessageStore and identify it via 'message-store' attribute in the queue configuration, thus making your queue distributed.

<channel id="myDistributedQueue">
     <queue capacity=100000 message-store="hazelcastStore"/>
</channel>

<bean id="hazelcastStore" class="foo.bar.MyHazelCastStore"/>