[th2-2552] backpressure: added check for queue size limit

[andrew-drobynin]

New version of #137

[Nikita-Smirnov-Exactpro]

Redundant, whole common or each pin (routing key) have a publish limit.

[Nikita-Smirnov-Exactpro]

private Map<String, QueuesWithVirtualPublishLimit> groupQueuesByRoutingKey() {
Map<String, QueueInfo> queueNameToInfo = client.getQueues().stream()
.collect(toMap(QueueInfo::getName, Function.identity()));

    Map<String, List<BindingInfo>> bindings = knownExchangesToRoutingKeys.entrySet().stream()
            .flatMap(entry -> {
                        String exchange = entry.getKey();
                        Set<String> routingKeys = entry.getValue();
                        return client.getBindingsBySource(rabbitMQConfiguration.getVHost(), exchange).stream()
                                .filter(it -> it.getDestinationType() == QUEUE && routingKeys.contains(it.getRoutingKey()));
                    }
            ).collect(groupingBy(BindingInfo::getRoutingKey));

    Map<String, QueuesWithVirtualPublishLimit> routingKeyToQueuesWithLimit = new HashMap<>();
    bindings.forEach((routingKey, bindingsForRoutingKey) ->
                    routingKeyToQueuesWithLimit.put(
                            routingKey,
                            new QueuesWithVirtualPublishLimit(
                                    bindingsForRoutingKey.stream().map(bindingInfo -> queueNameToInfo.get(bindingInfo.getDestination())).collect(toList()),
                                    connectionManagerConfiguration.getVirtualPublishLimit()
                            )
                    )
            );
    return routingKeyToQueuesWithLimit;
}

[Nikita-Smirnov-Exactpro]

I think we can combine lockSendingIfSizeLimitExceeded and groupQueuesByRoutingKey into continuous stream or Kotlin sequence

[andrew-drobynin]

I didn't want to do it. I wanted to keep all logic with com.rabbitmq.http.client.Client in one place and do only locking in lockSendingIfSizeLimitExceeded().
But I'll try if you want.

[andrew-drobynin]

Please see commit 18469f3.

[andrew-drobynin]

I migrated ConnectionManager to Kotlin. Please see commit fbae68d.
But I don't think we can do it in one stream because we should take all queues for each routing key. I mean we can't do anything before check all queues.

[andrew-drobynin]

I think it may be helpful, we'll have something like:

Size limit lock for routing key 'parsed-key': java.util.concurrent.locks.ReentrantLock@5744e6e4[Unlocked]
Size limit lock for routing key 'parsed-key': java.util.concurrent.locks.ReentrantLock@5744e6e4[Locked by thread pool-3-thread-1]