@Test void concurrentRequestOnChannelMonoLeadToPoolWaitTimeoutException() { Mono<? extends Channel> channelMono = getChannelPool(99).getChannelMono(); ConcurrentLinkedQueue<Channel> listChannels = new ConcurrentLinkedQueue<>(); assertThatThrownBy(() -> ConcurrentTestRunner.builder() .operation((threadNumber, step) -> listChannels.add(channelMono.block())) .threadCount(10) .operationCount(10) .runSuccessfullyWithin(Duration.ofSeconds(30))) .isInstanceOf(ExecutionException.class) .hasMessageContaining("java.util.NoSuchElementException: Timeout waiting for idle object"); }
@Override public Mono<? extends Channel> getChannelMono() { return borrow(); }
ReactorRabbitMQChannelPool implements ChannelPool, Startable { @Override public Mono<? extends Channel> getChannelMono() { return borrow(); } }
ReactorRabbitMQChannelPool implements ChannelPool, Startable { @Override public Mono<? extends Channel> getChannelMono() { return borrow(); } ReactorRabbitMQChannelPool(Mono<Connection> connectionMono, Configuration configuration); }
ReactorRabbitMQChannelPool implements ChannelPool, Startable { @Override public Mono<? extends Channel> getChannelMono() { return borrow(); } ReactorRabbitMQChannelPool(Mono<Connection> connectionMono, Configuration configuration); void start(); Sender getSender(); Receiver createReceiver(); @Override Mono<? extends Channel> getChannelMono(); @Override BiConsumer<SignalType, Channel> getChannelCloseHandler(); Mono<Void> createWorkQueue(QueueSpecification queueSpecification, BindingSpecification bindingSpecification); @PreDestroy @Override void close(); Mono<Boolean> tryChannel(); }
ReactorRabbitMQChannelPool implements ChannelPool, Startable { @Override public Mono<? extends Channel> getChannelMono() { return borrow(); } ReactorRabbitMQChannelPool(Mono<Connection> connectionMono, Configuration configuration); void start(); Sender getSender(); Receiver createReceiver(); @Override Mono<? extends Channel> getChannelMono(); @Override BiConsumer<SignalType, Channel> getChannelCloseHandler(); Mono<Void> createWorkQueue(QueueSpecification queueSpecification, BindingSpecification bindingSpecification); @PreDestroy @Override void close(); Mono<Boolean> tryChannel();  }