public void onStreamRequest(StreamRequest req, final RequestContext requestContext, final Map<String, String> wireAttrs,
                              final NextFilter<StreamRequest, StreamResponse> nextFilter)
  {
    //Set accepted encoding for compressed response
    String operation = (String) requestContext.getLocalAttr(R2Constants.OPERATION);
    if (!_acceptEncodingHeader.isEmpty() && _helper.shouldCompressResponseForOperation(operation))
    {
      CompressionOption responseCompressionOverride =
          (CompressionOption) requestContext.getLocalAttr(R2Constants.RESPONSE_COMPRESSION_OVERRIDE);
      req = addResponseCompressionHeaders(responseCompressionOverride, req);
    }

    if (_requestContentEncoding != StreamEncodingType.IDENTITY)
    {
      final StreamRequest request = req;
      final StreamingCompressor compressor = _requestContentEncoding.getCompressor(_executor);
      CompressionOption option = (CompressionOption) requestContext.getLocalAttr(R2Constants.REQUEST_COMPRESSION_OVERRIDE);
      if (option == null || option != CompressionOption.FORCE_OFF)
      {
        final int threshold = option == CompressionOption.FORCE_ON ? 0 : _requestCompressionConfig.getCompressionThreshold();
        PartialReader reader = new PartialReader(threshold, new Callback<EntityStream[]>()
        {
          @Override
          public void onError(Throwable ex)
          {
            nextFilter.onError(ex, requestContext, wireAttrs);
          }

          @Override
          public void onSuccess(EntityStream[] result)
          {
            if (result.length == 1)
            {
              StreamRequest uncompressedRequest = request.builder().build(result[0]);
              nextFilter.onRequest(uncompressedRequest, requestContext, wireAttrs);
            }
            else
            {
              StreamRequestBuilder builder = request.builder();
              EntityStream compressedStream = compressor.deflate(EntityStreams.newEntityStream(new CompositeWriter(result)));
              Map<String, String> headers = stripHeaders(builder.getHeaders(), HttpConstants.CONTENT_LENGTH);
              StreamRequest compressedRequest = builder.setHeaders(headers)
                  .setHeader(HttpConstants.CONTENT_ENCODING, compressor.getContentEncodingName())
                  .build(compressedStream);
              nextFilter.onRequest(compressedRequest, requestContext, wireAttrs);
            }
          }
        });
        req.getEntityStream().setReader(reader);
        return;
      }
    }

    nextFilter.onRequest(req, requestContext, wireAttrs);
  }
@Test(dataProvider = "requestData")
  public void testAcceptEncodingHeader(CompressionConfig requestCompressionConfig,
                                       CompressionOption requestCompressionOverride,
                                       boolean headerShouldBePresent,
                                       String operation)
      throws CompressionException, URISyntaxException, InterruptedException, ExecutionException, TimeoutException {
    Executor executor = Executors.newCachedThreadPool();
    ClientStreamCompressionFilter clientCompressionFilter = new ClientStreamCompressionFilter(
        StreamEncodingType.GZIP.getHttpName(),
        requestCompressionConfig,
        ACCEPT_COMPRESSIONS,
        new CompressionConfig(Integer.MAX_VALUE),
        Arrays.asList(ClientCompressionHelper.COMPRESS_ALL_RESPONSES_INDICATOR),
        executor);
    // The entity should be compressible for this test.
    int original = 100;
    byte[] entity = new byte[original];
    Arrays.fill(entity, (byte)'A');
    StreamRequest streamRequest =
        new StreamRequestBuilder(new URI(URI))
            .setMethod(RestMethod.POST)
            .build(EntityStreams.newEntityStream(new ByteStringWriter(ByteString.copy(entity))));

    int compressed = EncodingType.GZIP.getCompressor().deflate(new ByteArrayInputStream(entity)).length;
    RequestContext context = new RequestContext();
    if (operation != null)
    {
      context.putLocalAttr(R2Constants.OPERATION, operation);
    }
    context.putLocalAttr(R2Constants.REQUEST_COMPRESSION_OVERRIDE, requestCompressionOverride);
    int entityLength = headerShouldBePresent ? compressed : original;

    clientCompressionFilter.onStreamRequest(streamRequest, context, Collections.<String, String>emptyMap(),
                                            new HeaderCaptureFilter(HttpConstants.ACCEPT_ENCODING, true, null));
  }