multipart parsing, request body access, and chunked input stream support
This commit is contained in:
Relism
@@ -0,0 +1,73 @@
|
||||
package dev.relism;
|
||||
|
||||
import java.io.ByteArrayInputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.SequenceInputStream;
|
||||
|
||||
/**
|
||||
* De-chunking {@link InputStream} for HTTP/1.1 {@code Transfer-Encoding: chunked} request bodies.
|
||||
* Handles pre-buffered bytes from the header read-ahead, chunk framing, and trailer consumption.
|
||||
* Returns -1 at end of the final chunk; the underlying socket is left positioned for the next request.
|
||||
*/
|
||||
final class ChunkedInputStream extends InputStream {
|
||||
private final InputStream src;
|
||||
private int chunkRemaining = 0;
|
||||
private boolean done = false;
|
||||
|
||||
ChunkedInputStream(InputStream socket, byte[] preBuf, int preBufOff, int preBufLen) {
|
||||
src = preBufLen > 0
|
||||
? new SequenceInputStream(new ByteArrayInputStream(preBuf, preBufOff, preBufLen), socket)
|
||||
: socket;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int read() throws IOException {
|
||||
if (done) return -1;
|
||||
while (chunkRemaining == 0) {
|
||||
chunkRemaining = readChunkSize();
|
||||
if (chunkRemaining == 0) { consumeTrailers(); done = true; return -1; }
|
||||
}
|
||||
int b = src.read();
|
||||
if (b >= 0 && --chunkRemaining == 0) { src.read(); src.read(); } // consume trailing \r\n
|
||||
return b;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int read(byte[] buf, int off, int len) throws IOException {
|
||||
if (done) return -1;
|
||||
while (chunkRemaining == 0) {
|
||||
chunkRemaining = readChunkSize();
|
||||
if (chunkRemaining == 0) { consumeTrailers(); done = true; return -1; }
|
||||
}
|
||||
int n = src.read(buf, off, Math.min(len, chunkRemaining));
|
||||
if (n > 0) {
|
||||
chunkRemaining -= n;
|
||||
if (chunkRemaining == 0) { src.read(); src.read(); } // consume trailing \r\n
|
||||
}
|
||||
return n;
|
||||
}
|
||||
|
||||
private int readChunkSize() throws IOException {
|
||||
int size = 0, b;
|
||||
while ((b = src.read()) != -1) {
|
||||
if (b >= '0' && b <= '9') size = size * 16 + (b - '0');
|
||||
else if (b >= 'a' && b <= 'f') size = size * 16 + (b - 'a' + 10);
|
||||
else if (b >= 'A' && b <= 'F') size = size * 16 + (b - 'A' + 10);
|
||||
else { // ';' (extensions) or '\r' — skip to end of line
|
||||
while ((b = src.read()) != -1 && b != '\n');
|
||||
break;
|
||||
}
|
||||
}
|
||||
return size;
|
||||
}
|
||||
|
||||
// Reads and discards trailer headers until the empty line that terminates the chunked body.
|
||||
private void consumeTrailers() throws IOException {
|
||||
while (true) {
|
||||
int b = src.read();
|
||||
if (b == -1 || b == '\r') { src.read(); return; } // empty line — done
|
||||
while ((b = src.read()) != -1 && b != '\n'); // skip non-empty trailer line
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,336 @@
|
||||
package dev.relism.api.multipart;
|
||||
|
||||
import dev.relism.models.Request;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.*;
|
||||
|
||||
/**
|
||||
* Lazy streaming {@code multipart/form-data} parser.
|
||||
*
|
||||
* <p>Reads from {@code body.stream()} — the request body is <em>never</em> fully materialised.
|
||||
* Text fields are buffered eagerly on first encounter (they are small by definition). File part
|
||||
* bodies are exposed as zero-copy {@link InputStream}s backed directly by the socket stream and
|
||||
* must be consumed before the next call to any scan method.
|
||||
*
|
||||
* <pre>{@code
|
||||
* Multipart mp = Multipart.of(req);
|
||||
*
|
||||
* // Text fields — buffered eagerly, accessible in any call order
|
||||
* String userId = mp.field("userId");
|
||||
* String width = mp.field("width");
|
||||
*
|
||||
* // File parts — zero-copy socket stream; consume before requesting the next file
|
||||
* Part avatar = mp.file("avatar");
|
||||
* Files.copy(avatar.stream(), destination);
|
||||
*
|
||||
* // Explicit materialization of a file body (opt-in heap allocation)
|
||||
* byte[] data = mp.file("doc").materialize();
|
||||
*
|
||||
* // Collect everything at once — eagerly buffers all bodies (accept the memory cost)
|
||||
* List<Part> files = mp.parts("files");
|
||||
* }</pre>
|
||||
*
|
||||
* <p><b>Scan ordering:</b> {@link #field} and {@link #file} scan forward through the stream.
|
||||
* Parts already passed cannot be re-read. Text fields encountered while scanning toward a file
|
||||
* are buffered silently; file bodies encountered while scanning toward a text field are drained
|
||||
* silently. Use {@link #parts()} only if you need everything and accept full materialisation.
|
||||
*
|
||||
* <p><b>Thread safety:</b> not thread-safe; one instance per request.
|
||||
*/
|
||||
public final class Multipart {
|
||||
|
||||
private static final int BUF_CAP = 8192;
|
||||
|
||||
private final InputStream src;
|
||||
private final byte[] crlfBound; // "\r\n--<boundary>"
|
||||
|
||||
private final byte[] win;
|
||||
private int wPos = 0;
|
||||
private int wLen = 0;
|
||||
private boolean srcEof = false;
|
||||
private boolean done = false;
|
||||
|
||||
private final List<Part> scanned = new ArrayList<>();
|
||||
private PartBodyStream active = null; // open file stream; must be drained before next scan
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Factory
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private Multipart(InputStream src, String boundary) throws IOException {
|
||||
this.src = src;
|
||||
this.crlfBound = ("\r\n--" + boundary).getBytes(StandardCharsets.US_ASCII);
|
||||
this.win = new byte[BUF_CAP + crlfBound.length];
|
||||
skipFromWindow(2 + boundary.length() + 2); // "--boundary\r\n"
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a parser for {@code request}. Uses {@code body.stream()} — zero heap allocation.
|
||||
*
|
||||
* @throws IllegalArgumentException if the request is not multipart or boundary is missing
|
||||
* @throws IOException if the initial stream read fails
|
||||
*/
|
||||
public static Multipart of(Request req) throws IOException {
|
||||
String ct = req.header("Content-Type");
|
||||
if (ct == null || !ct.startsWith("multipart/"))
|
||||
throw new IllegalArgumentException("Not a multipart request (Content-Type: " + ct + ")");
|
||||
String boundary = extractParam(ct, "boundary");
|
||||
if (boundary == null)
|
||||
throw new IllegalArgumentException("Missing boundary in Content-Type: " + ct);
|
||||
return new Multipart(req.body().stream(), boundary);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Public API
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Returns the text value of the first field named {@code name}, or {@code null}.
|
||||
* Scans forward; text parts encountered along the way are buffered, file bodies are drained.
|
||||
*/
|
||||
public String field(String name) throws IOException {
|
||||
for (Part p : scanned)
|
||||
if (name.equals(p.name()) && !p.isFile()) return p.text();
|
||||
while (!done) {
|
||||
Part p = scanNext(false);
|
||||
if (p != null && name.equals(p.name()) && !p.isFile()) return p.text();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the first file part named {@code name}, or {@code null}.
|
||||
* Scans forward; text parts encountered along the way are buffered, earlier file bodies
|
||||
* are drained. The returned part's stream must be consumed before the next scan call.
|
||||
*/
|
||||
public Part file(String name) throws IOException {
|
||||
for (Part p : scanned)
|
||||
if (name.equals(p.name()) && p.isFile()) return p; // already materialized via parts()
|
||||
while (!done) {
|
||||
Part p = scanNext(false);
|
||||
if (p != null && name.equals(p.name()) && p.isFile()) return p;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns all parts named {@code name} in declaration order.
|
||||
* Forces a full scan; all file bodies are materialised into heap.
|
||||
*/
|
||||
public List<Part> parts(String name) throws IOException {
|
||||
scanAll();
|
||||
List<Part> result = new ArrayList<>();
|
||||
for (Part p : scanned) if (name.equals(p.name())) result.add(p);
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns all parts in declaration order.
|
||||
* Forces a full scan; all file bodies are materialised into heap.
|
||||
*/
|
||||
public List<Part> parts() throws IOException {
|
||||
scanAll();
|
||||
return List.copyOf(scanned);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Scan
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private void scanAll() throws IOException { while (!done) scanNext(true); }
|
||||
|
||||
/**
|
||||
* Scans the next part.
|
||||
*
|
||||
* @param materialize if {@code true}, file part bodies are buffered immediately
|
||||
* (used by {@link #parts()} and {@link #parts(String)})
|
||||
*/
|
||||
private Part scanNext(boolean materialize) throws IOException {
|
||||
if (done) return null;
|
||||
drainActive();
|
||||
if (done) return null;
|
||||
|
||||
Map<String, String> headers = readPartHeaders();
|
||||
if (headers == null) { done = true; return null; }
|
||||
|
||||
String disp = headers.get("content-disposition");
|
||||
String name = extractParam(disp, "name");
|
||||
String filename = extractParam(disp, "filename");
|
||||
String ct = headers.get("content-type");
|
||||
|
||||
active = new PartBodyStream();
|
||||
|
||||
Part p;
|
||||
if (filename != null && !materialize) {
|
||||
// File part — expose streaming body; not cached (stream is consumed once)
|
||||
p = Part.streaming(name, filename, ct, active);
|
||||
} else {
|
||||
// Text part, or full-scan path: buffer body now
|
||||
byte[] body = active.readAllBytes();
|
||||
active = null;
|
||||
p = Part.buffered(name, filename, ct, body);
|
||||
scanned.add(p);
|
||||
}
|
||||
return p;
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// PartBodyStream — inner class sharing the window buffer
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
final class PartBodyStream extends InputStream {
|
||||
boolean bodyEof = false;
|
||||
|
||||
@Override
|
||||
public int read(byte[] buf, int off, int len) throws IOException {
|
||||
if (bodyEof) return -1;
|
||||
refill();
|
||||
if (wLen == 0) { seal(); return -1; }
|
||||
|
||||
int delimAt = findDelim(wPos, wLen);
|
||||
int available;
|
||||
if (delimAt >= 0) {
|
||||
available = delimAt - wPos;
|
||||
if (available == 0) { advancePastBoundary(); seal(); return -1; }
|
||||
} else {
|
||||
available = srcEof ? wLen : Math.max(0, wLen - (crlfBound.length - 1));
|
||||
if (available == 0) { refill(); return read(buf, off, len); }
|
||||
}
|
||||
|
||||
int n = Math.min(len, available);
|
||||
System.arraycopy(win, wPos, buf, off, n);
|
||||
wPos += n;
|
||||
wLen -= n;
|
||||
if (delimAt >= 0 && wPos == delimAt) { advancePastBoundary(); seal(); }
|
||||
return n;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int read() throws IOException {
|
||||
byte[] b = {0};
|
||||
return read(b, 0, 1) < 0 ? -1 : b[0] & 0xFF;
|
||||
}
|
||||
|
||||
/** Marks this stream as exhausted and releases the active slot. */
|
||||
private void seal() { bodyEof = true; if (active == this) active = null; }
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Window management
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private void refill() throws IOException {
|
||||
if (srcEof) return;
|
||||
if (wPos > 0) { System.arraycopy(win, wPos, win, 0, wLen); wPos = 0; }
|
||||
int space = win.length - wLen;
|
||||
if (space > 0) {
|
||||
int n = src.read(win, wLen, space);
|
||||
if (n < 0) srcEof = true; else wLen += n;
|
||||
}
|
||||
}
|
||||
|
||||
private void skipFromWindow(int bytes) throws IOException {
|
||||
int rem = bytes;
|
||||
while (rem > 0) { refill(); int s = Math.min(rem, wLen); wPos += s; wLen -= s; rem -= s; }
|
||||
}
|
||||
|
||||
/** Drains the active file stream so we can advance to the next part. */
|
||||
private void drainActive() throws IOException {
|
||||
if (active == null || active.bodyEof) return;
|
||||
PartBodyStream ps = active;
|
||||
byte[] sink = new byte[BUF_CAP];
|
||||
while (ps.read(sink) >= 0) {}
|
||||
}
|
||||
|
||||
private void advancePastBoundary() throws IOException {
|
||||
wPos += crlfBound.length;
|
||||
wLen -= crlfBound.length;
|
||||
refill();
|
||||
if (wLen >= 2) {
|
||||
if (win[wPos] == '-' && win[wPos + 1] == '-') {
|
||||
done = true;
|
||||
wPos += 2; wLen -= 2;
|
||||
if (wLen >= 2) { wPos += 2; wLen -= 2; } // optional trailing \r\n
|
||||
} else {
|
||||
wPos += 2; wLen -= 2; // \r\n before next part's headers
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Header parsing
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private Map<String, String> readPartHeaders() throws IOException {
|
||||
Map<String, String> map = new HashMap<>();
|
||||
while (true) {
|
||||
String line = readLine();
|
||||
if (line == null || line.isEmpty()) break;
|
||||
int colon = line.indexOf(':');
|
||||
if (colon > 0)
|
||||
map.put(line.substring(0, colon).trim().toLowerCase(Locale.ROOT),
|
||||
line.substring(colon + 1).trim());
|
||||
}
|
||||
return map.isEmpty() ? null : map;
|
||||
}
|
||||
|
||||
private String readLine() throws IOException {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
while (true) {
|
||||
refill();
|
||||
if (wLen == 0) return sb.length() > 0 ? sb.toString() : null;
|
||||
|
||||
int end = wPos + wLen;
|
||||
for (int i = wPos; i < end - 1; i++) {
|
||||
if (win[i] == '\r' && win[i + 1] == '\n') {
|
||||
sb.append(new String(win, wPos, i - wPos, StandardCharsets.UTF_8));
|
||||
int consumed = i - wPos + 2;
|
||||
wPos += consumed; wLen -= consumed;
|
||||
return sb.toString();
|
||||
}
|
||||
}
|
||||
// No \r\n found yet — keep the last byte (might be a split \r\n) and refill
|
||||
int append = srcEof ? wLen : (wLen > 0 ? wLen - 1 : 0);
|
||||
if (append > 0) {
|
||||
sb.append(new String(win, wPos, append, StandardCharsets.UTF_8));
|
||||
wPos += append; wLen -= append;
|
||||
}
|
||||
if (srcEof && wLen > 0) {
|
||||
sb.append(new String(win, wPos, wLen, StandardCharsets.UTF_8));
|
||||
wPos += wLen; wLen = 0;
|
||||
return sb.toString();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Utilities
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private int findDelim(int from, int searchLen) {
|
||||
int last = from + searchLen - crlfBound.length;
|
||||
outer:
|
||||
for (int i = from; i <= last; i++) {
|
||||
for (int j = 0; j < crlfBound.length; j++)
|
||||
if (win[i + j] != crlfBound[j]) continue outer;
|
||||
return i;
|
||||
}
|
||||
return -1;
|
||||
}
|
||||
|
||||
private static String extractParam(String header, String param) {
|
||||
if (header == null) return null;
|
||||
int idx = header.indexOf(param + "=");
|
||||
if (idx < 0) return null;
|
||||
idx += param.length() + 1;
|
||||
if (idx < header.length() && header.charAt(idx) == '"') {
|
||||
int end = header.indexOf('"', idx + 1);
|
||||
return end >= 0 ? header.substring(idx + 1, end) : null;
|
||||
}
|
||||
int end = header.indexOf(';', idx);
|
||||
return end >= 0 ? header.substring(idx, end).trim() : header.substring(idx).trim();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
package dev.relism.api.multipart;
|
||||
|
||||
import java.io.ByteArrayInputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
/**
|
||||
* One part of a {@code multipart/form-data} body produced by {@link Multipart}.
|
||||
*
|
||||
* <p>Two internal states:
|
||||
* <ul>
|
||||
* <li><b>Buffered</b> (text fields and {@link Multipart#parts()} results) — body is in a
|
||||
* {@code byte[]}; {@link #stream()}, {@link #materialize()}, and {@link #text()} are all
|
||||
* repeatable and free of I/O.</li>
|
||||
* <li><b>Streaming</b> (file parts returned by {@link Multipart#file(String)}) — body is a
|
||||
* bounded socket {@link InputStream}; call {@link #stream()} or {@link #materialize()} exactly
|
||||
* once, before requesting the next part from {@link Multipart}.</li>
|
||||
* </ul>
|
||||
*/
|
||||
public final class Part {
|
||||
|
||||
private final String name;
|
||||
private final String filename; // null → text field
|
||||
private final String contentType;
|
||||
private byte[] buf; // non-null once materialised
|
||||
private final InputStream socketStream; // non-null only for streaming file parts
|
||||
|
||||
static Part buffered(String name, String filename, String contentType, byte[] buf) {
|
||||
return new Part(name, filename, contentType, buf, null);
|
||||
}
|
||||
|
||||
static Part streaming(String name, String filename, String contentType, InputStream stream) {
|
||||
return new Part(name, filename, contentType, null, stream);
|
||||
}
|
||||
|
||||
private Part(String name, String filename, String contentType,
|
||||
byte[] buf, InputStream socketStream) {
|
||||
this.name = name;
|
||||
this.filename = filename;
|
||||
this.contentType = contentType;
|
||||
this.buf = buf;
|
||||
this.socketStream = socketStream;
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Metadata
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/** Field or file name from {@code Content-Disposition: form-data; name="..."}. */
|
||||
public String name() { return name; }
|
||||
|
||||
/** Original filename from {@code filename="..."}, or {@code null} for text fields. */
|
||||
public String filename() { return filename; }
|
||||
|
||||
/** {@code Content-Type} declared in the part headers, or {@code null} if absent. */
|
||||
public String contentType() { return contentType; }
|
||||
|
||||
/** {@code true} if this part has a {@code filename} attribute (i.e. a file upload). */
|
||||
public boolean isFile() { return filename != null; }
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Body access
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Returns an {@link InputStream} over the part body.
|
||||
* For buffered parts: returns a fresh reader each call.
|
||||
* For streaming file parts: returns the raw socket stream — read once only.
|
||||
*/
|
||||
public InputStream stream() {
|
||||
return buf != null ? new ByteArrayInputStream(buf) : socketStream;
|
||||
}
|
||||
|
||||
/**
|
||||
* Materialises the body into a {@code byte[]}. Result is cached; safe to call repeatedly.
|
||||
* For streaming file parts: triggers a full read from the socket on first call.
|
||||
*/
|
||||
public byte[] materialize() throws IOException {
|
||||
if (buf != null) return buf;
|
||||
return buf = socketStream.readAllBytes();
|
||||
}
|
||||
|
||||
/**
|
||||
* Decodes the body as UTF-8. Materialises if needed; result is cached.
|
||||
* For streaming file parts: triggers a full read from the socket on first call.
|
||||
*/
|
||||
public String text() throws IOException {
|
||||
return new String(materialize(), StandardCharsets.UTF_8);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,152 @@
|
||||
package dev.relism.models;
|
||||
|
||||
import java.io.*;
|
||||
|
||||
/**
|
||||
* Accessor for the HTTP request body. Supports two mutually exclusive read modes per request:
|
||||
*
|
||||
* <ul>
|
||||
* <li>{@link #bytes()} — materialises the full body into a {@code byte[]} and caches it.
|
||||
* Safe to call multiple times; the second call returns the cached array. Throws for
|
||||
* bodies larger than 2 GB.</li>
|
||||
* <li>{@link #stream()} — returns a bounded {@link InputStream} without upfront allocation.
|
||||
* For fixed-length bodies this is a view into the already-buffered header bytes stitched
|
||||
* to the socket; for chunked bodies it is the raw {@link dev.relism.ChunkedInputStream}
|
||||
* that de-chunks on the fly.</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p><b>Mutual exclusivity:</b> calling both {@code bytes()} and {@code stream()} on the same
|
||||
* request produces undefined results. Choose one mode per handler.
|
||||
*
|
||||
* <p><b>Keep-alive:</b> unread body bytes are discarded by {@link Request#drain()} after the
|
||||
* handler returns so the socket is correctly positioned for the next pipelined request.
|
||||
*/
|
||||
public final class RequestBody {
|
||||
private static final byte[] EMPTY_BYTES = new byte[0];
|
||||
|
||||
private final InputStream socket;
|
||||
private final long contentLength;
|
||||
private final byte[] preBuf;
|
||||
private final int preBufOff;
|
||||
private final int preBufLen;
|
||||
|
||||
private byte[] resolved;
|
||||
private long socketConsumed;
|
||||
|
||||
RequestBody(InputStream socket, long contentLength, byte[] preBuf, int preBufOff, int preBufLen) {
|
||||
this.socket = socket;
|
||||
this.contentLength = contentLength;
|
||||
this.preBuf = preBuf;
|
||||
this.preBufOff = preBufOff;
|
||||
this.preBufLen = preBufLen;
|
||||
}
|
||||
|
||||
static RequestBody of(byte[] bytes) {
|
||||
RequestBody b = new RequestBody(null, bytes.length, null, 0, 0);
|
||||
b.resolved = bytes;
|
||||
return b;
|
||||
}
|
||||
|
||||
private static final RequestBody EMPTY_INSTANCE;
|
||||
static {
|
||||
EMPTY_INSTANCE = new RequestBody(null, 0, null, 0, 0);
|
||||
EMPTY_INSTANCE.resolved = EMPTY_BYTES;
|
||||
}
|
||||
|
||||
static RequestBody empty() { return EMPTY_INSTANCE; }
|
||||
|
||||
/** {@code true} if the body has zero bytes ({@code Content-Length: 0} or no body). */
|
||||
public boolean isEmpty() { return contentLength == 0; }
|
||||
|
||||
/**
|
||||
* Declared body size in bytes. Returns {@code -1} for {@code Transfer-Encoding: chunked}
|
||||
* bodies where the size is not known upfront.
|
||||
*/
|
||||
public long contentLength() { return contentLength; }
|
||||
|
||||
/**
|
||||
* Materialises and caches the full body. Suitable for JSON, small form data, and any payload
|
||||
* that must be inspected in full. The result is cached — repeated calls return the same array.
|
||||
*
|
||||
* <p>For chunked bodies ({@link #contentLength()} {@code == -1}), reads until the chunked
|
||||
* stream signals EOF.
|
||||
*
|
||||
* @throws IllegalStateException if {@link #contentLength()} exceeds {@code Integer.MAX_VALUE}
|
||||
* (~2 GB); use {@link #stream()} for large bodies instead
|
||||
*/
|
||||
public byte[] bytes() {
|
||||
if (resolved != null) return resolved;
|
||||
if (contentLength < 0) { // chunked — read until ChunkedInputStream signals EOF
|
||||
try { return resolved = socket.readAllBytes(); } catch (IOException e) { throw new UncheckedIOException(e); }
|
||||
}
|
||||
if (contentLength > Integer.MAX_VALUE)
|
||||
throw new IllegalStateException("Body too large to materialize (" + contentLength + " bytes), use stream()");
|
||||
int size = (int) contentLength;
|
||||
int copied = (int) Math.min(preBufLen, contentLength);
|
||||
byte[] buf = new byte[size];
|
||||
if (copied > 0) System.arraycopy(preBuf, preBufOff, buf, 0, copied);
|
||||
if (copied < size) {
|
||||
try {
|
||||
socket.readNBytes(buf, copied, size - copied);
|
||||
} catch (IOException e) {
|
||||
throw new UncheckedIOException(e);
|
||||
}
|
||||
socketConsumed = contentLength - preBufLen;
|
||||
}
|
||||
return resolved = buf;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a bounded {@link InputStream} over the body without upfront allocation.
|
||||
*
|
||||
* <p>For fixed-length bodies: a {@link SequenceInputStream} of any already-buffered header
|
||||
* bytes followed by a bounded view of the socket stream — zero heap beyond those small
|
||||
* pre-buffered bytes.
|
||||
*
|
||||
* <p>For chunked bodies: the raw {@link dev.relism.ChunkedInputStream} that de-chunks on
|
||||
* the fly; EOF signals the end of the logical body and leaves the socket positioned for
|
||||
* the next keep-alive request.
|
||||
*
|
||||
* <p>If {@link #bytes()} was called first, returns a fresh {@link java.io.ByteArrayInputStream}
|
||||
* over the cached array.
|
||||
*/
|
||||
public InputStream stream() {
|
||||
if (resolved != null) return new ByteArrayInputStream(resolved);
|
||||
if (contentLength < 0) return socket; // ChunkedInputStream — EOF signals end of body
|
||||
int fromBuf = (int) Math.min(preBufLen, contentLength);
|
||||
long fromSocket = contentLength - fromBuf;
|
||||
InputStream bufPart = new ByteArrayInputStream(preBuf, preBufOff, fromBuf);
|
||||
return fromSocket == 0 ? bufPart : new SequenceInputStream(bufPart, bounded(socket, fromSocket));
|
||||
}
|
||||
|
||||
/** Discards unread body bytes to reposition the socket for the next keep-alive request. */
|
||||
void drain() {
|
||||
if (isEmpty() || resolved != null) return;
|
||||
if (contentLength < 0) {
|
||||
try { socket.transferTo(OutputStream.nullOutputStream()); } catch (IOException ignored) {}
|
||||
return;
|
||||
}
|
||||
long remaining = (contentLength - preBufLen) - socketConsumed;
|
||||
if (remaining > 0) try { socket.skipNBytes(remaining); } catch (IOException ignored) {}
|
||||
}
|
||||
|
||||
private InputStream bounded(InputStream src, long limit) {
|
||||
return new InputStream() {
|
||||
private long left = limit;
|
||||
|
||||
@Override public int read() throws IOException {
|
||||
if (left == 0) return -1;
|
||||
int b = src.read();
|
||||
if (b >= 0) { left--; socketConsumed++; }
|
||||
return b;
|
||||
}
|
||||
|
||||
@Override public int read(byte[] buf, int off, int len) throws IOException {
|
||||
if (left == 0) return -1;
|
||||
int n = src.read(buf, off, (int) Math.min(len, left));
|
||||
if (n > 0) { left -= n; socketConsumed += n; }
|
||||
return n;
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user