title: How to build a nonblocking Consumer when using AsyncRabbitTemplate with Request/Reply Pattern
body: <p>I'm new to rabbitmq and currently trying to implement a nonblocking producer with a nonblocking consumer. I've build some test producer where I played around with typereference:
  </p>
  
  <pre><code>@Service
  public class Producer {
      @Autowired
      private AsyncRabbitTemplate asyncRabbitTemplate;
  
      public &lt;T extends RequestEvent&lt;S&gt;, S&gt; RabbitConverterFuture&lt;S&gt; asyncSendEventAndReceive(final T event) {
          return asyncRabbitTemplate.convertSendAndReceiveAsType(QueueConfig.EXCHANGE_NAME, event.getRoutingKey(), event, event.getResponseTypeReference());
      }
  }
  </code></pre>
  
  <p>And in some other place the test function that gets called in a RestController</p>
  
  <pre><code>@Autowired
  Producer producer;
  
  public void test() throws InterruptedException, ExecutionException {
      TestEvent requestEvent = new TestEvent("SOMEDATA");
      RabbitConverterFuture&lt;TestResponse&gt; reply = producer.asyncSendEventAndReceive(requestEvent);
      log.info("Hello! The Reply is: {}", reply.get());
  }
  </code></pre>
  
  <p>This so far was pretty straightforward, where I'm stuck now is how to create a consumer which is non-blocking too. My current listener:</p>
  
  <pre><code>@RabbitListener(queues = QueueConfig.QUEUENAME)
  public TestResponse onReceive(TestEvent event) {
      Future&lt;TestResponse&gt; replyLater = proccessDataLater(event.getSomeData())
      return replyLater.get();
  }
  </code></pre>
  
  <p>As far as I'm aware, when using @RabbitListener this listener runs in its own thread. And I could configure the MessageListener to use more then one thread for the active listeners. Because of that, blocking the listener thread with future.get() is not blocking the application itself. Still there might be the case where all threads are blocking now and new events are stuck in the queue, when they maybe dont need to. What I would like to do is to just receive the event without the need to instantly return the result. Which is probably not possible with @RabbitListener. Something like:</p>
  
  <pre><code>@RabbitListener(queues = QueueConfig.QUEUENAME)
  public void onReceive(TestEvent event) {
     /* 
      * Some fictional RabbitMQ API call where i get a ReplyContainer which contains
      * the CorrelationID for the event. I can call replyContainer.reply(testResponse) later 
      * in the code without blocking the listener thread
      */        
      ReplyContainer replyContainer = AsyncRabbitTemplate.getReplyContainer()
  
      // ProcessDataLater calls reply on the container when done with its action
      proccessDataLater(event.getSomeData(), replyContainer);
  }
  </code></pre>
  
  <p>What is the best way to implement such behaviour with rabbitmq in spring?</p>
  
  <p>EDIT Config Class:</p>
  
  <pre><code>@Configuration
  @EnableRabbit
  public class RabbitMQConfig implements RabbitListenerConfigurer {
  
      public static final String topicExchangeName = "exchange";
  
      @Bean
      TopicExchange exchange() {
          return new TopicExchange(topicExchangeName);
      }
  
      @Bean
      public ConnectionFactory rabbitConnectionFactory() {
          CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
          connectionFactory.setHost("localhost");
          return connectionFactory;
      }
  
      @Bean
      public MappingJackson2MessageConverter consumerJackson2MessageConverter() {
          return new MappingJackson2MessageConverter();
      }
  
      @Bean
      public RabbitTemplate rabbitTemplate() {
          final RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory());
          rabbitTemplate.setMessageConverter(producerJackson2MessageConverter());
          return rabbitTemplate;
      }
  
      @Bean
      public AsyncRabbitTemplate asyncRabbitTemplate() {
          return new AsyncRabbitTemplate(rabbitTemplate());
      }
  
      @Bean
      public Jackson2JsonMessageConverter producerJackson2MessageConverter() {
          return new Jackson2JsonMessageConverter();
      }
  
      @Bean
      Queue queue() {
          return new Queue("test", false);
      }
  
      @Bean
      Binding binding() {
          return BindingBuilder.bind(queue()).to(exchange()).with("foo.#");
      }
  
      @Bean
      public SimpleRabbitListenerContainerFactory myRabbitListenerContainerFactory() {
          SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
          factory.setConnectionFactory(rabbitConnectionFactory());
          factory.setMaxConcurrentConsumers(5);
          factory.setMessageConverter(producerJackson2MessageConverter());
          factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
          return factory;
      }
  
      @Override
      public void configureRabbitListeners(final RabbitListenerEndpointRegistrar registrar) {
          registrar.setContainerFactory(myRabbitListenerContainerFactory());
      }
  
  }
  </code></pre>