Клиент зависает при отправке сообщений RabbitMQ с помощью Java-клиента

Question

Я реализовал издателя и потребителя RabbitMQ в реактивной манере с Java, но мои функции публикации зависают от канала. Сама очередь, объявляя очередь и потребляя, однако, работает нормально, я тестировал ее с помощью пользовательского интерфейса управления администратора. Когда я пытаюсь отправить больше сообщений, я больше не вижу журналов типа "очередь объявления успешно" или "доставка сообщения для обмена...". Кстати, я знаю, что мне не нужно declareQueue в доставке (), но я добавил его, чтобы проверить, работает ли связь в этом конкретном канале.

Мой код:

@Slf4j
@Component
public class RabbitConfigurator {
    private TasksQueueConfig cfg;
    private ReceiverOptions recOpts;
    private List<Address> addresses;
    private Utils.ExceptionFunction<ConnectionFactory, Connection> connSupplier;

    public RabbitConfigurator(TasksQueueConfig cfg) {
        this.cfg = cfg;
        addresses = cfg
                .getHosts()
                .stream()
                .map(Address::new)
                .collect(Collectors.toList());
        connSupplier = cf -> {
            LOG.info("initializing new RabbitMQ connection");
            return cf.newConnection(addresses, "dmTasksProc");
        };
    }

    @Bean
    public ConnectionFactory rabbitMQConnectionFactory() {
        ConnectionFactory cf = new ConnectionFactory();
        cf.setHost(cfg.getHosts().get(0));
        cf.setPort(5672);
        cf.setUsername(cfg.getUsername());
        cf.setPassword(cfg.getPassword());
        return cf;
    }

    @Bean
    public Sender sender(ConnectionFactory connFactory) {
        SenderOptions sendOpts = new SenderOptions()
                .connectionClosingTimeout(Duration.parse(cfg.getConnectionTimeout()))
                .connectionFactory(connFactory)
                .connectionSupplier(connSupplier)
                .connectionSubscriptionScheduler(Schedulers.elastic());
        return RabbitFlux.createSender(sendOpts);
    }

    @Bean
    public Receiver receiver(ConnectionFactory connFactory) {
        ReceiverOptions recOpts = new ReceiverOptions()
                .connectionClosingTimeout(Duration.parse(cfg.getConnectionTimeout()))
                .connectionFactory(connFactory)
                .connectionSupplier(connSupplier)
                .connectionSubscriptionScheduler(Schedulers.elastic());
        return RabbitFlux.createReceiver(recOpts);
    }

    @Bean
    public Flux<Delivery> deliveryFlux(Receiver receiver) {
        return receiver.consumeAutoAck(cfg.getName(), new ConsumeOptions().qos(cfg.getPrefetchCount()));
    }

    @Bean
    public AmqpAdmin rabbitAmqpAdmin(ConnectionFactory connFactory) {
        return new RabbitAdmin(new CachingConnectionFactory(connFactory));
    }
}

и потребитель / издатель:

@Slf4j
@Service
public class TasksQueue implements DisposableBean {
    private TasksQueueConfig cfg;
    private ObjectMapper mapper;
    private Flux<Delivery> deliveryFlux;
    private Receiver receiver;
    private Sender sender;

    private Disposable consumer;

    public TasksQueue(TasksQueueConfig cfg, AmqpAdmin amqpAdmin, ObjectMapper mapper, Flux<Delivery> deliveryFlux,
                         Receiver receiver, Sender sender) {
        this.cfg = cfg;
        this.mapper = mapper;
        this.deliveryFlux = deliveryFlux;
        this.receiver = receiver;
        this.sender = sender;

        amqpAdmin.declareQueue(new Queue(cfg.getName(), false, false, false));
        consumer = consume();
    }

    public Mono<Void> deliver(Flux<Task> tasks) {
        var pub = sender.sendWithPublishConfirms(
                        tasks.map(task -> {
                            try {
                                String exchange = "";
                                LOG.debug("delivering message to exchange='{}', routingKey='{}'", exchange, cfg.getName());
                                return new OutboundMessage(exchange, cfg.getName(), mapper.writeValueAsBytes(task));
                            } catch(JsonProcessingException ex) {
                                throw Exceptions.propagate(ex);
                            }
                        }));

        return sender.declareQueue(QueueSpecification.queue(cfg.getName()))
                .flatMap(declareOk -> {
                    LOG.info("queue declare success");
                    return Mono.just(declareOk);
                })
                .thenMany(pub)
                .doOnError(JsonProcessingException.class, ex -> LOG.error("Cannot prepare queue message:", ex))
                .doOnError(ex -> LOG.error("Failed to send task to the queue:", ex))
                .map(res -> {
                    if(res.isAck()) {
                        LOG.info("Message {} sent successfully", new String(res.getOutboundMessage().getBody()));
                        return res;
                    } else {
                        LOG.info("todo");
                        return res;
                    }
                })
                .then();
    }

    private Disposable consume() {
        return deliveryFlux
                .retryWhen(Retry.fixedDelay(10, Duration.ofSeconds(1)))
                .doOnError(err -> {
                    LOG.error("tasks consumer error", err);
                })
                .subscribe(m -> {
                    LOG.info("Received message {}", new String(m.getBody()));
                });
    }

    @Override
    public void destroy() throws Exception {
        LOG.info("Cleaning up tasks queue resources");
        consumer.dispose();
        receiver.close();
        sender.close();
    }
}

Через пять минут после попытки отправить сообщение я получаю журнал:

r.r.ChannelCloseHandlers$SenderChannelCloseHandler:47: closing channel 1 by signal cancel
r.r.ChannelCloseHandlers$SenderChannelCloseHandler:53: Channel 1 didn't close normally: null

Большое спасибо за вклад!