Support concurrent links with stream filtering

[ansd]

What?

If a receiver performs stream filtering with AMQP property filters or AMQP SQL filter expressions, the following downsides can occur:

1. While the stream is being filtered, other links on the same session are blocked
2. RabbitMQ sends messages late to the receiver

As an example, let's assume a receiver attaches to the start of a multi GB stream providing a link credit of 2. Let's assume only the very first message matches the filter. In this case, RabbitMQ scans the entire stream without processing other links on the same session, and sends the matched message only once the scan completed (after many seconds or even minutes).

Instead, we want other links to be processed concurrently and the receiver might want to start processing the first matched message while RabbitMQ continues filtering the stream.

This commit fixes these two downsides.

How?

After a threshold of consecutively unmatched messages, the session "pauses" filtering on that link temporarily by:

1. sending an Erlang message resume_filtering to itself, and
2. sending any matched messages to the receiver

Any other Erlang messages then have a chance to be processed by the session before the filtering on that link is resumed by the resume_filtering Erlang message.

Once the end of the stream is reached or link credit is exhausted, the credit_reply will be returned from rabbit_stream_queue to rabbit_amqp_session.

An alternative solution would be to use separate Erlang processes for filtering links as they can be CPU bound and also block for disk I/O.