@Override
  public void handleRequest(RestRequest request, RequestContext requestContext, Callback<RestResponse> callback)
  {
    //This code path cannot accept content types or accept types that contain
    //multipart/related. This is because these types of requests will usually have very large payloads and therefore
    //would degrade server performance since RestRequest reads everything into memory.
    if (!isMultipart(request, requestContext, callback))
    {
      _restRestLiServer.handleRequest(request, requestContext, callback);
    }
  }
@Test
  public void testRequestAttachmentsResponseAttachmentsException() throws Exception
  {
    //This test verifies the server's behavior in the face of an exception. In this case the resource method
    //threw an exception AFTER setting response attachments. Additionally the resource method failed to absorb any
    //incoming request attachments. We verify in this test that StreamResponseCallbackAdaptor in RestLiServer
    //drains and absorbs all bytes from the incoming request and that the response attachments set by the resource method
    //are told to abort.

    //Define the server side resource attachments to be sent back.
    final RestLiResponseAttachments.Builder responseAttachmentsBuilder = new RestLiResponseAttachments.Builder();
    final RestLiTestAttachmentDataSource toBeAbortedDataSource = RestLiTestAttachmentDataSource.createWithRandomPayload("1");

    responseAttachmentsBuilder.appendSingleAttachment(toBeAbortedDataSource);

    Capture<ResourceContext> resourceContextCapture = EasyMock.newCapture();
    final AsyncStatusCollectionResource statusResource = getMockResource(AsyncStatusCollectionResource.class,
                                                                         capture(resourceContextCapture));

    statusResource.streamingAction(EasyMock.<String>anyObject(), EasyMock.<RestLiAttachmentReader>anyObject(),
                                   EasyMock.<Callback<Long>> anyObject());
    EasyMock.expectLastCall().andAnswer(new IAnswer<Object>()
    {
      @Override
      public Object answer() throws Throwable {
        //Verify there are still attachments to be read.
        final RestLiAttachmentReader attachmentReader = (RestLiAttachmentReader)EasyMock.getCurrentArguments()[1];
        Assert.assertFalse(attachmentReader.haveAllAttachmentsFinished());

        //Verify the action param.
        Assert.assertEquals((String)EasyMock.getCurrentArguments()[0], "someMetadata");

        //Set the response attachments
        resourceContextCapture.getValue().setResponseAttachments(responseAttachmentsBuilder.build());

        //Now throw an exception.
        @SuppressWarnings("unchecked")
        Callback<Long> callback = (Callback<Long>) EasyMock.getCurrentArguments()[2];
        callback.onError(new RestLiServiceException(HttpStatus.S_409_CONFLICT, "Some conflict"));
        return null;
      }
    });
    replay(statusResource);

    //Now we create a multipart/related payload.
    final String payload = "{\"metadata\": \"someMetadata\"}";
    final ByteStringWriter byteStringWriter = new ByteStringWriter(ByteString.copyString(payload, Charset.defaultCharset()));
    final MultiPartMIMEWriter.Builder builder = new MultiPartMIMEWriter.Builder();
    final RestLiTestAttachmentDataSource toBeDrainedDataSource = RestLiTestAttachmentDataSource.createWithRandomPayload("2");

    AttachmentUtils.appendSingleAttachmentToBuilder(builder, toBeDrainedDataSource);
    final MultiPartMIMEWriter writer = AttachmentUtils.createMultiPartMIMEWriter(byteStringWriter, "application/json", builder);

    final StreamRequest streamRequest =
        MultiPartMIMEStreamRequestFactory.generateMultiPartMIMEStreamRequest(new URI("/asyncstatuses/?action=streamingAction"),
                                                                             "related",
                                                                             writer, Collections.<String, String>emptyMap(),
                                                                             "POST",
                                                                             ImmutableMap.of(RestConstants.HEADER_ACCEPT, RestConstants.HEADER_VALUE_MULTIPART_RELATED),
                                                                             Collections.emptyList());

    final Callback<StreamResponse> callback = new Callback<StreamResponse>()
    {
      @Override
      public void onSuccess(StreamResponse streamResponse)
      {
        fail();
      }

      @Override
      public void onError(Throwable e)
      {
        //Verify the exception.
        assertTrue(e instanceof StreamException);
        StreamException streamException = (StreamException)e;
        StreamResponse streamResponse = streamException.getResponse();

        assertEquals(streamResponse.getStatus(), 409);
        final FullEntityReader fullEntityReader = new FullEntityReader(new Callback<ByteString>()
        {
          @Override
          public void onError(Throwable e)
          {
            Assert.fail();
          }

          @Override
          public void onSuccess(ByteString result)
          {
            //We have the body so assert exception made it.
            assertTrue(result.length() > 0);
            assertTrue(result.asString(Charset.defaultCharset()).contains("Some conflict"));
          }
        });
        streamResponse.getEntityStream().setReader(fullEntityReader);

        EasyMock.verify(statusResource);
        EasyMock.reset(statusResource);
      }
    };

    _server.handleRequest(streamRequest, new RequestContext(), callback);

    //Verify that the request level attachments were drained.
    Assert.assertTrue(toBeDrainedDataSource.finished());

    //Verify that response attachments were told to abort.
    Assert.assertTrue(toBeAbortedDataSource.dataSourceAborted());
  }