  @Test
  public void testRequestAttachmentsAndResponseAttachments() throws Exception
  {

    //This test verifies the server's ability to accept request attachments and send back response attachments. This is the
    //main test to verify the wire protocol for streaming. We send a payload that contains the rest.li payload and some attachments
    //and we send a response back with a rest.li payload and some attachments.

    //Define the server side resource attachments to be sent back.
    final RestLiResponseAttachments.Builder responseAttachmentsBuilder = new RestLiResponseAttachments.Builder();
    responseAttachmentsBuilder.appendSingleAttachment(new RestLiTestAttachmentDataSource("1",
                                                                                         ByteString.copyString("one", Charset.defaultCharset())));

    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 respond back to the request.
        @SuppressWarnings("unchecked")
        Callback<Long> callback = (Callback<Long>) EasyMock.getCurrentArguments()[2];
        callback.onSuccess(1234l);
        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();
    AttachmentUtils.appendSingleAttachmentToBuilder(builder,
                                                    new RestLiTestAttachmentDataSource("2", ByteString.copyString("two", Charset.defaultCharset())));
    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)
      {
        //Before reading the data make sure top level type is multipart/related
        Assert.assertEquals(streamResponse.getStatus(), 200);
        try
        {
          ContentType contentType = new ContentType(streamResponse.getHeader(RestConstants.HEADER_CONTENT_TYPE));
          Assert.assertEquals(contentType.getBaseType(), RestConstants.HEADER_VALUE_MULTIPART_RELATED);
        }
        catch (ParseException parseException)
        {
          Assert.fail();
        }

        final CountDownLatch countDownLatch = new CountDownLatch(1);
        MultiPartMIMEFullReaderCallback fullReaderCallback = new MultiPartMIMEFullReaderCallback(countDownLatch);
        final MultiPartMIMEReader reader = MultiPartMIMEReader.createAndAcquireStream(streamResponse);
        reader.registerReaderCallback(fullReaderCallback);
        try
        {
          countDownLatch.await(3000, TimeUnit.MILLISECONDS);
        }
        catch (InterruptedException interruptedException)
        {
          Assert.fail();
        }

        final List<SinglePartMIMEFullReaderCallback> singlePartMIMEReaderCallbacks = fullReaderCallback.getSinglePartMIMEReaderCallbacks();
        Assert.assertEquals(singlePartMIMEReaderCallbacks.size(), 2);

        //Verify first part is Action response.
        Assert.assertEquals(singlePartMIMEReaderCallbacks.get(0).getHeaders().get(RestConstants.HEADER_CONTENT_TYPE), RestConstants.HEADER_VALUE_APPLICATION_JSON);
        Assert.assertEquals(singlePartMIMEReaderCallbacks.get(0).getFinishedData().asAvroString(), "{\"value\":1234}");
        //Verify the second part matches what the server should have sent back
        Assert.assertEquals(singlePartMIMEReaderCallbacks.get(1).getHeaders().get(RestConstants.HEADER_CONTENT_ID), "1");
        Assert.assertEquals(singlePartMIMEReaderCallbacks.get(1).getFinishedData().asString(Charset.defaultCharset()), "one");

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

      @Override
      public void onError(Throwable e)
      {
        fail();
      }
    };

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