Add batch message-getting to the Kombu SQS transport

[diranged]

Should solve issue #280. Patch includes unit tests as well and has been tested manually by me against the normal master/worker and solo process model. Speeds up Celery SQS message fetching by 10x by default.

[ask]

Thanks! Great work (and even with unit tests).

One problem right now is that it doesn't adhere to the QoS limits, I think it should rather
override Channel.drain_events directly, that way you can also avoid the extra "buffer by queue" mapping.
The mapping may be a problem since it may leave "orphan messages" if you stop consuming from a queue,
as there is no clean up step, so removing the mapping and using a single buffer would make the implementation
simpler.

So instead of a _get method you should define drain_events directly:

from kombu.five import Empty

def drain_events(self, timeout=None):
    queues = self._active_queues  # <-- list of queues we currently consume from
    buffer = self._buffer   # <-- the buffer deque

    # ... read messages into buffer .... #
    buffer.append((message, name_of_origin_queue))

    try:
        return buffer.popleft()
    except IndexError:
        raise Empty()

Here name_of_origin_queue is the name of the queue the message was delivered to.

As for QoS I suspect that you cannot reliably get the number of messages you are allowed
to receive since the object does not use locks (way too slow), but it will be better than nothing
even if it may consume one too many (or too few) by chance.

The following method would need to be added to kombu.transport.virtual.QoS:

    def can_consume_max_estimate(self):
        pcount = self.prefetch_count
        if pcount:
            return pcount - (len(self._delivered) - len(self._dirty))

Then you can use this to estimate the maximum number of messages drain_events may retrieve

[ask]

Note: if the above method returns None it means 'no limit':

fetch_max = self.messages_to_fetch
limit = self._qos.can_consume_max_estimate()
limit = fetch_max if limit is None else min(fetch_max, limit)

[diranged]

@ask, I've been digging into this change for a day now, and its a lot more complex than I expected. I'm particularly unclear about the use of basic_get() (which is called by Queue.get()), and the drain_events() methods.

From what I can tell, if I don't do the "buffering" logic inside of _get(), then I need to override basic_get() and drain_events(). Overriding both of those means I need to create common "get_from_cache" and "put_to_cache" methods for handling the local cache buffer. Even then, none of that solves the stranded message issue in the event that 5-10 messages are pulled down from SQS, but the worker only gets a chance to work on 1 of them.

I can see changing the "messages_to_fetch" code so that it uses QoS.can_consume_max_estimate() pretty easily, which allows the prefetch configuration to define how many messages to ask Amazon for at once... but I still think the simplest system is to handle all of the "get and buffer message" logic inside of the _get() method.

Perhaps there are two possible solutions here..

1. Never get more than 1 message at a time for a noack queue?
2. When shutting Kombu down gracefully, handle re-injecting the messages that were left in the buffer back into SQS?

[diranged]

Here you go @ask. Per our discussion earlier today, this patch now only modifies the drain_events() method and leaves basic_get() and _get() alone. The drain_events() method itself is what handles queueing message objects up and storing them now.

The unit tests are basic, but execute. I've also executed the code inside our own Celery environment and it seems to work as expected.

[diranged]

Here are my final touches for this pull request @ask. The unit tests now all execute in TravisCI where they can (Python 2.x), and are skipped on Python 3.x.

I will, in a separate pull request, clean up the SQS Mock code so that it uses dicts instead of files for storing messages.

[ask]

Need to rewrite this so that it doesn't use the filesystem. Should be easy by just using self.messages = []

[ask]

Merged but with changes so please review and test: 49bf684