message selectors optimization and message grouping

[jmesnil]

Working on https://jira.jboss.org/jira/browse/JBMESSAGING-1505 about delivery optimization for consumers with selectors.

The idea of the optimization is to provide to consumers with selectors their own iterators on the queue so that they don't have to rescan the whole queue to find messages which they can handle (when the filter match the message)
This means that:
- consumers with filters have an associated iterator that is used when the queue deliver messages
- consumers without filters will peek the messages from the queue

In order to distinguish between the 2 types of consumers, we need to peek the consumer which will handle the message from the distributor. I added a peekConsumer() to the Distributor interface to be able to do that.
This works fine for RoundRobinDistributor but it is broken for GroupingRoundRobinDistributor (in that case, it must know the message to determine which consumer will handle it).

As message groups and queues with selectors are not compatible, there are several options:
1/ forbids to use consumers with filters on a queue with message groups
2/ degrades the delivery algorithm in case of message groups with consumers with filters. In that case, the current algorithm is used (with a full scan of the queue for each call to deliver) (+ warning log)

I think (2) is a better option: this will give us the optimization in the most common case and handles the anti pattern case with a degradation of performance.
The only real con is that this means we maintain 2 delivery algorithm:
- an optimized one (round robin distribution)
- an degraded one (messages groups distribution + consumers with filters)

depending on the distribution policy type, when QueueImpl.deliver() is called we either use the deliverWithFullScan() (aka the current deliver algorithm) in the grouping distribution case or
deliverWithIterators() in the round robin case

Some remarks on the new optimized algorithm:
* when a consumer with filter has finished iterating on the queue, I reassigned to it a new messageReferences.iterator() so that it can handle new messages.
However this means it will iterate again on all the messages it has already discarded to reach the new messages

* in the case of consumers without filters, I peek directly from the messageReferences list to deliver. There is no need to have an iterator to share between all the consumers without filters

[timfox]

Actually I think there is a solution to this problem which allows grouping and filters to work together, but it requires a rethink of the way message grouping is currently implemented.

The problem is, as already stated, that, with the way the grouping round robin distributor currently works, the message would need to be provided at peek() time in order to determine which consumer to choose, but the message is not known at that time.

However, in principle it should be possible to implement the behaviour that if a consumer has both a filter and a message group it should only receive messages which match both the filter and the message group, there's nothing impossible about implementing that.

To do this, we would have a Distributor that *always round robins* (i.e we don't have a grouping round robin distributor at all). This makes it easy to have a peek() method on it that does not need a message to be passed in.

Once the queue has the consumer it looks up the iterator for that consumer (if any) and gets the next message using that iterator.

It passes that message to the consumer which either accepts or rejects it depending on its filter and what message group (if any) it is in.

The queue then asks the distributor for the next consumer gets the iterator for that consumer and the process repeats.

In this implementation its the *consumer* that would know what message groups it is handling, i.e. the grouping logic moves from the grouping round robin distributor to the server consumer.

Each server consumer will only accept a message if both it's filter matches and the message's group id is in the shared set for itself.

Every grouping server consumer would need access to a shared map of group id-->consumer. If a consumer receives a message with a group id and no other consumer is already handling it, then it can associate itself to that group id in the map.

[timfox]

A result of this also is that you only need a peek() method on the distributor, you don't need a distribute() method at all.

Once the queue has the consumer from the peek call it can just call handle() on it directly.