CXF buffering data when using chunked encoding

user_3115734 0
Viewed 2654

I've written a java REST (streaming) servlet using Apache CXF 2.5.1 and deployed it to a Tomcat 7.0.42 container. The REST endpoint is essentially an implementation of StreamingOutput, wrapped it a Response object that is handed off to the container when a client requests.

The nature of the service is to return a stream of sensor data to a client. This stream could theoretically be infinitely long because it's only terminated when the client disconnects. The issue arises when the data generated by the sensor comes in small quantities.

The service "works" but I'm running into an issue when it comes to the size of the data responses that the client receives. The client only receives data after an 8192 byte threshold has been broken by the service. Then the client receives 800 bytes, then 8192 bytes, then 800 bytes...

I would like the data to be sent to the client as soon as I invoke flush on the OutputStream which the container hands off to my implementation of StreamingOutput. However, the implementation of OutputStream that i'm given (WrappedOutputStream defined in org.apache.cxf.transport.http.AbstractHTTPDestination) has a flush method which does nothing.

Is there any way to have more control over the OutputStream that CXF uses so I can "flush" to the client on demand?

chunked-encoding cxf java
4 Answers

This is a follow-up answer to supplement @harumph's excellent idea. All credit belongs to this person. I only wanted to provide a working example. I am stuck on an ancient version of Apache CXF with no hope of upgrade to latest Glassfish/Jersey. This is my "fake it until you make it". :)

public static final class FlushableHttpServletResponseOutputStream
extends OutputStream
{
    private final OutputStream delegate;
    private final HttpServletResponse response;

    public FlushableHttpServletResponseOutputStream(OutputStream delegate, HttpServletResponse response)
    {
        this.delegate = ObjectArgs.checkNotNull(delegate, "delegate");
        this.response = ObjectArgs.checkNotNull(response, "response");
    }

    @Override
    public void write(int b)
    throws IOException
    {
        delegate.write(b);
    }

    @Override
    public void flush()
    throws IOException
    {
        delegate.flush();
        // Ref: https://stackoverflow.com/a/20708446/257299
        response.flushBuffer();
    }
}

@GET  // or whatever you like
@Path("/your/url/path/here")
@Produces(MediaType.TEXT_PLAIN)  // or whatever you like
public Response
httpGetStuff(@Context HttpServletResponse response)  // auto-magically injected by CXF framework
throws Exception
{
    // Ref: https://stackoverflow.com/a/63605927/257299
    final StreamingOutput so = new StreamingOutput()
    {
        @Override
        public void write(OutputStream os)
        throws IOException, WebApplicationException
        {
            final FlushableHttpServletResponseOutputStream fos =
                new FlushableHttpServletResponseOutputStream(os, response);

            // Do something complex with 'fos' here.
            // Call 'fos.flush()' to immediately send all buffered data via HTTP response chunk.
            // Alternatively: You may call response.flush() directly.
        }
    };
    final Response x = Response.ok(so).build();
    return x;
}

For careful readers, you can find exactly where Apache CXF ignores a flush request here: org.apache.cxf.transport.http.AbstractHTTPDestination.WrappedOutputStream.flush() (version 2.2.12):

    public void flush() throws IOException {
        //ignore until we close 
        // or we'll force chunking and cause all kinds of network packets
    }
user_2572990
Related