Java Spring framework, start consuming HTTP request body before request has ended

Question

In the Java Spring framework, how could I create an end point that starts consuming the HTTP request body in chunks before the request has finished?

It seem like the default behavior i that the end point method is not executed until the request has ended.

The following Node.js server starts consuming the request body, how to do the same with the Java Spring framework?

const http = require('http');

const server = http.createServer((request, response) => {
  request.on('data', (chunk) => {
    console.log('NEW CHUNK: ', chunk.toString().length);
  });

  request.on('end', () => {
    response.end('done');
  });
});

server.listen(3000);

Outputs:

NEW CHUNK:  65189
NEW CHUNK:  65536
NEW CHUNK:  65536
NEW CHUNK:  65536
NEW CHUNK:  54212

Answer 1

I'm not sure there is a solution for mapping chunked request with spring what i'll do is something like this:

import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;

import javax.servlet.http.HttpServletRequest;

import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;

@Controller
public class ChunkController {

private static final int EOS = -1;

    @RequestMapping(method = RequestMethod.POST)
    public ResponseEntity<Void> upload(final HttpServletRequest request, @RequestParam final int chunkSize) {
        try (InputStream in = request.getInputStream()) {
            byte[] readBuffer = new byte[chunkSize];
            int nbByteRead = 0;
            int remainingByteToChunk = chunkSize;
            while ((nbByteRead = in.read(readBuffer, chunkSize - remainingByteToChunk, remainingByteToChunk)) != EOS) {
                remainingByteToChunk -= nbByteRead;
                if (remainingByteToChunk == 0) {
                    byte[] chunk = Arrays.copyOf(readBuffer, readBuffer.length);
                    remainingByteToChunk = readBuffer.length;
                    // do something with the chunk.
                }
            }
            if (remainingByteToChunk != chunkSize) {
                byte[] lastChunk = Arrays.copyOf(readBuffer, readBuffer.length - remainingByteToChunk);
                //  do something with the last chunk
            }
            return new ResponseEntity<>(HttpStatus.OK);
        } catch (IOException e) {
            return new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR);
        }
    }
}

or you can define a constant for the size of the chunk.

You can also ignore the size the chunk and just handle the result of in.read until the end of stream.

Reading like this you will need to parse data to find actual chunk sent by the client. A typical body will look like:

7\r\n
Mozilla\r\n
9\r\n
Developer\r\n
7\r\n
Network\r\n
0\r\n
\r\n

What you can do is create custom InputStream like this (adapted from Apache HttpClient ChunkedInputStream)

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Objects;


public class ChunkedInputStream extends InputStream {
    public static final byte[] EMPTY = new byte[0];
    private final Charset charset;
    private InputStream in;
    private int chunkSize;
    private int pos;
    private boolean bof = true;
    private boolean eof = false;
    private boolean closed = false;

    public ChunkedInputStream(final InputStream in, Charset charset) throws IOException {

        if (in == null) {
            throw new IllegalArgumentException("InputStream parameter may not be null");
        }
        this.in = in;
        this.pos = 0;

        this.charset = Objects.requireNonNullElse(charset, StandardCharsets.US_ASCII);
    }


    public int read() throws IOException {

        if (closed) {
            throw new IOException("Attempted read from closed stream.");
        }
        if (eof) {
            return -1;
        }
        if (pos >= chunkSize) {
            nextChunk();
            if (eof) {
                return -1;
            }
        }
        pos++;
        return in.read();
    }

    public int read(byte[] b, int off, int len) throws IOException {

        if (closed) {
            throw new IOException("Attempted read from closed stream.");
        }

        if (eof) {
            return -1;
        }
        if (pos >= chunkSize) {
            nextChunk();
            if (eof) {
                return -1;
            }
        }
        len = Math.min(len, chunkSize - pos);
        int count = in.read(b, off, len);
        pos += count;
        return count;
    }

    public int read(byte[] b) throws IOException {
        return read(b, 0, b.length);
    }

    public byte[] readChunk() throws IOException {
        if (eof) {
            return EMPTY;
        }
        if (pos >= chunkSize) {
            nextChunk();
            if (eof) {
                return EMPTY;
            }
        }
        byte[] chunk = new byte[chunkSize];
        int nbByteRead = 0;
        int remainingByteToChunk = chunkSize;
        while (remainingByteToChunk > 0 && !eof) {
            nbByteRead = read(chunk, chunkSize - remainingByteToChunk, remainingByteToChunk);
            remainingByteToChunk -= nbByteRead;
        }

        if (remainingByteToChunk == 0) {
            return chunk;
        } else {
            return Arrays.copyOf(chunk, chunk.length - remainingByteToChunk);
        }
    }

    private void readCRLF() throws IOException {
        int cr = in.read();
        int lf = in.read();
        if ((cr != '\r') || (lf != '\n')) {
            throw new IOException(
                    "CRLF expected at end of chunk: " + cr + "/" + lf);
        }
    }

    private void nextChunk() throws IOException {
        if (!bof) {
            readCRLF();
        }
        chunkSize = getChunkSizeFromInputStream(in);
        bof = false;
        pos = 0;
        if (chunkSize == 0) {
            eof = true;
        }
    }

    private int getChunkSizeFromInputStream(final InputStream in)
            throws IOException {

        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        // States: 0=normal, 1=\r was scanned, 2=inside quoted string, -1=end
        int state = 0;
        while (state != -1) {
            int b = in.read();
            if (b == -1) {
                throw new IOException("chunked stream ended unexpectedly");
            }
            switch (state) {
                case 0:
                    switch (b) {
                        case '\r':
                            state = 1;
                            break;
                        case '\"':
                            state = 2;
                            /* fall through */
                        default:
                            baos.write(b);
                    }
                    break;

                case 1:
                    if (b == '\n') {
                        state = -1;
                    } else {
                        // this was not CRLF
                        throw new IOException("Protocol violation: Unexpected"
                                + " single newline character in chunk size");
                    }
                    break;

                case 2:
                    switch (b) {
                        case '\\':
                            b = in.read();
                            baos.write(b);
                            break;
                        case '\"':
                            state = 0;
                            /* fall through */
                        default:
                            baos.write(b);
                    }
                    break;
                default:
                    throw new RuntimeException("assertion failed");
            }
        }

        //parse data
        String dataString = baos.toString(charset);
        int separator = dataString.indexOf(';');
        dataString = (separator > 0)
                ? dataString.substring(0, separator).trim()
                : dataString.trim();

        int result;
        try {
            result = Integer.parseInt(dataString.trim(), 16);
        } catch (NumberFormatException e) {
            throw new IOException("Bad chunk size: " + dataString);
        }
        return result;
    }

    public void close() throws IOException {
        if (!closed) {
            try {
                if (!eof) {
                    exhaustInputStream(this);
                }
            } finally {
                eof = true;
                closed = true;
            }
        }
    }

    static void exhaustInputStream(InputStream inStream) throws IOException {
        // read and discard the remainder of the message
        byte[] buffer = new byte[1024];
        while (inStream.read(buffer) >= 0) {
            ;
        }
    }
}

In the controller you can keep the same controller code but wrap the request.getInputStream() with this but you still won't get the actual client chunk. That's why I add the readChunk() method

@PostMapping("/upload")
public ResponseEntity<Void> upload(final HttpServletRequest request, @RequestHeader HttpHeaders headers) {
    Charset charset = StandardCharsets.US_ASCII;
    if (headers.getContentType() != null) {
        charset = Objects.requireNonNullElse(headers.getContentType().getCharset(), charset);
    }
    try (ChunkedInputStream in = new ChunkedInputStream(request.getInputStream(), charset)) {
        byte[] chunk;
        while ((chunk = in.readChunk()).length > 0) {
            // do something with the chunk.
            System.out.println(new String(chunk, Objects.requireNonNullElse(charset, StandardCharsets.US_ASCII)));
        }
        return new ResponseEntity<>(HttpStatus.OK);
    } catch (IOException e) {
        return new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR);
    }
}