title: Handling REST request with Message Queue
body: <p>I have two applications as stated below:</p>
  
  <ol>
  <li>Spring boot application - Acts as rest end point, publishes the request to message queue. ( Apache Pulsar )</li>
  <li>Heron (Storm) topology - which processes the message received from  Message queue ( PULSAR ) and has all logic for processing.</li>
  </ol>
  
  <p>My requirement, i need to serve different user queries through Spring boot application, which emits that query to message queue, and is consumed at spout. Once spout and bolts process the requests, a message is published again from bolt. That response from Bolt is handled at Spring boot (consumer) and replies to the user request. Typlically as shown below:</p>
  
  <p><a href="https://i.stack.imgur.com/PdGQd.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/PdGQd.png" alt="enter image description here"></a></p>
  
  <p>To serve to the same request, Im right now caching the deferred result object ( I set a reqID to each message which is sent to topology and  I also maintain a key, value pair for  )  in memory and when the message arrives I parse the request id and set the result to the defferedResult (I know this is a bad design, <strong>HOW SHOULD ONE SOLVE THIS ISSUE ?</strong>).</p>
  
  <p>How can I proceed to serve the response back to the same request in this scenario where the order of messages received from topology is not sequential ( as each request which is processes takes its own time and producer bolt will fire the response as on when it is receives one ). </p>
  
  <p>Im kind of stuck with this design and not able to proceed further.</p>
  
  <pre><code>//Controller
  public DeferredResult&lt;ResponseEntity&lt;?&gt;&gt; process(//someinput) {
      DeferredResult&lt;ResponseEntity&lt;?&gt;&gt; result = new DeferredResult&lt;&gt;(config.getTimeout());
      CompletableFuture&lt;String&gt; serviceResponse = service.processAsync(inputSource);
      serviceResponse.whenComplete((response, exception) -&gt; {
          if (!ObjectUtils.isEmpty(exception))
              result.setErrorResult(//error);
          else
              result.setResult(//complete);
      });
      return result;
  }
  
  //In Service
  public CompletableFuture processAsync(//input){
      producer.send(input);
      CompletableFuture result = new CompletableFuture();
      //consumer has a listener as shown below
      // **I want to avoid below line, how can I redesign this**
      map.put(id, result);
    return result;
  }
  
  //in same service, a listener is present for consumer for reading the messages
  consumerListener(Message msg){
       int reqID = msg.getRequestID();
       map.get(reqID).complete(msg.getData);
  }
  </code></pre>
  
  <blockquote>
    <p>As shown above as soon as I get a message I get the completableFuture
    object and set the result, which interally calls the defferred result
    object and returns the response to the user.</p>
  </blockquote>