@Override
  public void handleRequest(RestRequest request, RequestContext requestContext, final Callback<RestResponse> callback)
  {
    if (HttpMethod.POST != HttpMethod.valueOf(request.getMethod()))
    {
      _log.error("POST is expected, but " + request.getMethod() + " received");
      callback.onError(RestException.forError(HttpStatus.S_405_METHOD_NOT_ALLOWED.getCode(), "Invalid method"));
      return;
    }

    // Disable server-side latency instrumentation for multiplexed requests
    requestContext.putLocalAttr(TimingContextUtil.TIMINGS_DISABLED_KEY_NAME, true);

    IndividualRequestMap individualRequests;
    try
    {
      individualRequests = extractIndividualRequests(request);
      if (_multiplexerSingletonFilter != null) {
        individualRequests = _multiplexerSingletonFilter.filterRequests(individualRequests);
      }
    }
    catch (RestException e)
    {
      _log.error("Invalid multiplexed request", e);
      callback.onError(e);
      return;
    }
    catch (Exception e)
    {
      _log.error("Invalid multiplexed request", e);
      callback.onError(RestException.forError(HttpStatus.S_400_BAD_REQUEST.getCode(), e));
      return;
    }
    // prepare the map of individual responses to be collected
    final IndividualResponseMap individualResponses = new IndividualResponseMap(individualRequests.size());
    final Map<String, HttpCookie> responseCookies = new HashMap<>();
    // all tasks are Void and side effect based, that will be useful when we add streaming
    Task<?> requestProcessingTask = createParallelRequestsTask(request, requestContext, individualRequests, individualResponses, responseCookies);
    Task<Void> responseAggregationTask = Task.action("send aggregated response", () ->
      {
        RestResponse aggregatedResponse = aggregateResponses(individualResponses, responseCookies);
        callback.onSuccess(aggregatedResponse);
      }
    );
    _engine.run(requestProcessingTask.andThen(responseAggregationTask), MUX_PLAN_CLASS);
  }
@Test(dataProvider = "multiplexerConfigurations")
  public void testResponseCookiesAggregated(MultiplexerRunMode multiplexerRunMode) throws Exception
  {
    // Per security review: We should not make cookies for each individual responses visible to the client (especially if the cookie is HttpOnly).
    // Therefore all cookies returned by individual responses will be aggregated at the envelope response level.

    // Create a mockHandler. Make it return different cookies based on the request
    SynchronousRequestHandler mockHandler = new SynchronousRequestHandler() {
      @Override
      public RestResponse handleRequestSync(RestRequest request, RequestContext requestContext)
      {
        try
        {
          URI uri = request.getURI();
          RestResponseBuilder restResponseBuilder = new RestResponseBuilder();
          restResponseBuilder.setStatus(HttpStatus.S_200_OK.getCode());
          restResponseBuilder.setEntity(jsonBodyToByteString(fakeIndividualBody("don't care")));
          List<HttpCookie> cookies = new ArrayList<>();
          if (uri.getPath().contains("req1"))
          {
            HttpCookie cookie = new HttpCookie("cookie1", "cookie1Value");
            cookie.setDomain(".www.linkedin.com");
            cookie.setSecure(false);
            cookies.add(cookie);

            HttpCookie commonCookie = new HttpCookie("commonCookie", "commonCookieValue");
            commonCookie.setDomain(".WWW.linkedin.com");
            commonCookie.setPath("/foo");
            commonCookie.setSecure(false);
            cookies.add(commonCookie);
          }
          else if (uri.getPath().contains("req2"))
          {
            HttpCookie cookie = new HttpCookie("cookie2", "cookie2Value");
            cookie.setDomain("www.linkedin.com");
            cookie.setSecure(false);
            cookies.add(cookie);

            cookie = new HttpCookie("cookie3", "cookie3Value");
            cookies.add(cookie);

            HttpCookie commonCookie = new HttpCookie("commonCookie", "commonCookieValue");
            commonCookie.setDomain(".www.linkedin.com");
            commonCookie.setPath("/foo");
            commonCookie.setSecure(true);
            cookies.add(commonCookie);
          }
          else
          {
            HttpCookie cookie = new HttpCookie("cookie2", "newCookie2Value");
            cookie.setDomain("www.linkedin.com");
            cookie.setSecure(true);
            cookies.add(cookie);
          }
          return restResponseBuilder.setCookies(CookieUtil.encodeSetCookies(cookies)).build();
        }
        catch (Exception e)
        {
          throw new RuntimeException(e);
        }
      }
    };

    // Prepare request to mux handler
    FutureCallback<RestResponse> callback = new FutureCallback<>();
    RequestContext requestContext = new RequestContext();
    Map<String, IndividualRequest> individualRequests = ImmutableMap.of(
      "0", fakeIndRequest("/req1"),
      "1", fakeIndRequest("/req2", ImmutableMap.of("2", fakeIndRequest("/req3"))));

    // Create mux handler instance
    MultiplexedRequestHandlerImpl multiplexer = createMultiplexer(mockHandler, null, Collections.<String>emptySet(), 3, multiplexerRunMode);

    try
    {
      multiplexer.handleRequest(fakeMuxRestRequest(individualRequests), requestContext, callback);
    }
    catch (Exception e)
    {
      fail("Multiplexer should not throw exception", e);
    }

    RestResponse muxRestResponse = callback.get();
    // assert multiplexed request should return a 200 status code
    assertEquals(muxRestResponse.getStatus(), 200, "Multiplexer should return 200");
    MultiplexedResponseContent muxResponseContent = new MultiplexedResponseContent(DataMapConverter.bytesToDataMap(muxRestResponse.getHeaders(), muxRestResponse.getEntity()));

    // individual response should not have set-cookie headers
    IndividualResponseMap responses = muxResponseContent.getResponses();
    for(IndividualResponse res : responses.values())
    {
      for(String headerName : res.getHeaders().keySet())
      {
        assertTrue(headerName.equalsIgnoreCase("set-cookie"), "Individual response header should not container set-cookie header: " + responses.toString());
      }
    }

    // Ensure cookies are aggregated at envelope level
    List<HttpCookie> cookies = CookieUtil.decodeSetCookies(muxRestResponse.getCookies());
    assertEquals(cookies.size(), 4);
    for(HttpCookie cookie: cookies)
    {
      if ("cookie1".equals(cookie.getName()))
      {
        assertEquals(cookie.getValue(), "cookie1Value");
        assertEquals(cookie.getDomain(), ".www.linkedin.com");
        assertEquals(cookie.getSecure(), false);
      }
      else if ("cookie2".equals(cookie.getName()))
      {
        assertEquals(cookie.getValue(), "newCookie2Value");
        assertEquals(cookie.getDomain(), "www.linkedin.com");
        assertEquals(cookie.getSecure(), true);
      }
      else if ("cookie3".equals(cookie.getName()))
      {
        assertEquals(cookie.getValue(), "cookie3Value");
      }
      else if ("commonCookie".equals(cookie.getName()))
      {
        assertEquals(cookie.getValue(), "commonCookieValue");
        assertEquals(cookie.getDomain().toLowerCase(), ".www.linkedin.com");
        assertEquals(cookie.getPath(), "/foo");
        // Since request0 and request1 are executed in parallel, depending on which request is completed first,
        // we don't know what will be its final 'secure' attribute value.
      }
      else
      {
        fail("Unknown cookie name: " + cookie.getName());
      }
    }
  }