title: Spring Boot REST API and SQS response queue
body: <p>I am fairly new to Spring (3 weeks to be precise), so bear with me...</p>
  
  <p>I have an application flow consisting of a Spring Boot REST API, an SQS request queue, a Spring Boot Command Handler application and an SQS response queue.</p>
  
  <p>The API will receive a client request and publish it to the request queue.
  The command handler is a subscriber pooling from the request queue. It receives a request message, handles it and publishes a response to the response queue.
  The API is a subscriber pooling from the response queue. It should receive the response message from the queue and only then respond to the client, according to the content of the message it received (success or failure).</p>
  
  <p>Searching online, all implementations respond to the client immediately, somehow indicating that the request is being processed. It is then up to the client to pool for the actual response to its initial request. This is not the approach I desire.</p>
  
  <p>The Command Handler is working as expected, so I'm only posting code for the API.</p>
  
  <p>I have the following Configuration class on the API (along with some entries on the application.properties file, it's all I need to be able to authenticate to AWS and become a publisher):</p>
  
  <pre><code>@Configuration
  public class AWSSQSConfig {
      @Value("${aws.sqs.endpoint}")
      private String AWSSqsEndpoint;
  
      // Producer QueueMessageTemplate
      @Bean
      public QueueMessagingTemplate queueMessagingTemplate(AmazonSQSAsync amazonSqs, ResourceIdResolver resourceIdResolver) {
          if (!AWSSqsEndpoint.isEmpty())
              amazonSqs.setEndpoint(AWSSqsEndpoint);
          return new QueueMessagingTemplate(amazonSqs, resourceIdResolver);
      }
  }
  </code></pre>
  
  <p>My rest client endpoint (not as desired, since it responds immediately...):</p>
  
  <pre><code>@RestController
  public class SchemaController {
      @Value("${aws.sqs.requestQueue}")
      private String RequestQueue;
  
      @Autowired
      private MessageProducer mp;
  
      @RequestMapping(path = "/something", method = RequestMethod.POST)
      public @ResponseBody
      ResponseEntity&lt;String&gt; CreateSomething(@RequestBody String json) {
  
          //QueueMessage contains a GUID that should be present in the response message to be published by the Command Handler
          QueueMessage message = new QueueMessage(Enums.ActionType.CREATE, json);
  
          mp.SendMessage(RequestQueue, message);
  
          return new ResponseEntity&lt;&gt;(message.getGUID(), HttpStatus.CREATED);
      }
  }
  </code></pre>
  
  <p>My message producer component:</p>
  
  <pre><code>@Component
  public class MessageProducer {
      @Autowired
      private QueueMessagingTemplate sqsMsgTemplate;
  
      public void SendMessage(String requestQueue, QueueMessage payload) {
          this.sqsMsgTemplate.convertAndSend(requestQueue, payload);
      }
  }
  </code></pre>
  
  <p>And finally, my message consumer component (independent from the rest endpoint):</p>
  
  <pre><code>@Component
  public class MessageConsumer {
  
      @SqsListener(value = "${aws.sqs.responseQueue}", deletionPolicy = SqsMessageDeletionPolicy.NEVER)
      public void onReceiveMessage(String payload, Acknowledgment acknowledgment) {
  
          //TODO: Route response to the proper client request
          ...
          ... This code probably doesn't even go here...
          ...
  
          // Should only acknowledge once message as been routed to proper client
          acknowledgment.acknowledge();
      }
  }
  </code></pre>
  
  <p>As mentioned in a code comment above, the request QueueMessage contains a GUID that should be present in the response QueueMessage to be published by the Command Handler once finished with the request, so that the API can match the response to the proper request (since the request knows what GUID was generated by the QueueMessage constructor).</p>
  
  <p>I thought of creating a local queue (list, array, whatever...) in my spring API. The MessageConsumer would write all messages to that local queue and acknowledge the SQS messages so they get deleted from the remote queue. while the endpoint will pool that queue for its response, then delete the local queue entry and then respond back to the client.</p>
  
  <p>Not sure if there is a more efficient/Spring way of accomplishing this. Maybe DeferredResult or some other construct in the Spring Framework... ListenableFuture maybe?</p>
  
  <p>Thanks in advance.</p>
  
  <p><strong>UPDATE</strong></p>
  
  <p>So, I've come up with my own solution to the problem, which works, but I'm still not sure it's too naive or inefficient... </p>
  
  <ul>
  <li>I made my rest endpoint asynchronously wait for a result from a Callable.</li>
  <li>I introduced a Local Queue of type PublishSubscribeChannel (on a configuration class) from spring integration that my SQS message consumer writes to when it receives a response.</li>
  <li>The delegate fired by the Callable sends a request message to the SQS request queue, then subscribes to the local queue and loops waiting for a response from the local queue/channel, then unsubscribes from the channel and resolves the callable.</li>
  <li>I also changed my autowiring to constructor injection.</li>
  </ul>
  
  <p>My local queue configuration class:</p>
  
  <pre><code>@Configuration
  public class LocalQueueConfig {
      @Bean(name="localQueue")
      public PublishSubscribeChannel localQueue() {
          return new PublishSubscribeChannel(executor());
      }
  
      @Bean
      public ThreadPoolTaskExecutor executor() {
  
          ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
          pool.setCorePoolSize(10);
          pool.setMaxPoolSize(10);
          pool.setWaitForTasksToCompleteOnShutdown(true);
          return pool;
      }
  }
  </code></pre>
  
  <p>My REST endpoint now is:</p>
  
  <pre><code>@RequestMapping(path = "/something", method = RequestMethod.POST)
  public @ResponseBody
  Callable&lt;ResponseEntity&lt;String&gt;&gt; CreateSomething(@RequestBody String json) {
  
      // RequestQueue is an @Value string and delegate is autowired in the constructor and is of type HandleRequestDelegate
      return () -&gt; delegate.processRequest(RequestQueue, json, Enums.ActionType.CREATE);
  }
  </code></pre>
  
  <p>My HandleRequestDelegate class:</p>
  
  <pre><code>public ResponseEntity&lt;String&gt; processRequest(String requestQueue, String json, Enums.ActionType actionType) {
      QueueMessage requesrMessage = new QueueMessage(actionType, json);
  
      // We will look for this particular requestId when we receive a response
      requesrMessage.setRequestId(UUID.randomUUID().toString());
  
      // messageProducer is autowired
      messageProducer.sendMessage(requestQueue, requesrMessage);
  
      final QueueMessage[] responseMessage = {null};
  
      // This handler will receive messages from the local channel and we create one for each request the API receives
      MessageHandler handler = message -&gt; {
          MessageHeaders headers = message.getHeaders();
  
          if (headers.containsKey("formSchemaRequestId") &amp;&amp;
                  headers.get("formSchemaRequestId").equals(requestId)) {
              responseMessage[0] = (QueueMessage) message.getPayload();
  
              System.out.println("handle: " + ((QueueMessage) message.getPayload()).getEntityId());
          }
      };
  
      // Subscribes to the local changed using the above handler
      // localChannel is autowired
      localChannel.subscribe(handler);
  
      while (true){
          // if the message handler received a response bearing the matching requestId we can stop the loop and unsubscribe then handler
          if (responseMessage[0] != null) {
              System.out.println("while: " + responseMessage[0].getEntityId());
              localChannel.unsubscribe(handler);
              break;
          }
      }
  
      return new ResponseEntity&lt;&gt;(responseMessage[0].getEntityId(), HttpStatus.CREATED);
  }
  </code></pre>
  
  <p>The new message consumer:</p>
  
  <pre><code>// We no longer need to acknowledge the SQS messages. Because we are sending them to the local channel we can safely delete them from SQS (default behaviour)
  @SqsListener(value = "${aws.sqs.formSchemaResponseQueue}")
  public void onReceiveSchemaMessage(QueueMessage message) {
      // localChannel is autowired
      localChannel.send(
              MessageBuilder.withPayload(message)
                      .setHeader("formSchemaRequestId", message.getRequestId()).build());
  }
  </code></pre>
  
  <p>Any comments or concerns about this implementation would be greatly appreciated.</p>