support range requests (#252)

SentryMan · rbygrave · web-flow · commit 7f1c5de18216 · 2025-04-14T21:06:19.000+12:00
Co-authored-by: Rob Bygrave &lt;robin.bygrave@gmail.com&gt;
diff --git a/avaje-jex-static-content/src/main/java/io/avaje/jex/staticcontent/StaticClassResourceHandler.java b/avaje-jex-static-content/src/main/java/io/avaje/jex/staticcontent/StaticClassResourceHandler.java
@@ -93,7 +93,7 @@ private void sendURL(Context ctx, String urlPath, URL path) {
         addCachedEntry(ctx, urlPath, fis);
         return;
       }
-      ctx.write(fis);
+      ctx.rangedWrite(fis);
     } catch (final Exception e) {
       throw404(ctx.exchange());
     }
diff --git a/avaje-jex-static-content/src/main/java/io/avaje/jex/staticcontent/StaticFileHandler.java b/avaje-jex-static-content/src/main/java/io/avaje/jex/staticcontent/StaticFileHandler.java
@@ -108,7 +108,7 @@ private void sendFile(Context ctx, HttpExchange jdkExchange, String urlPath, Fil
         addCachedEntry(ctx, urlPath, fis);
         return;
       }
-      ctx.write(fis);
+      ctx.rangedWrite(fis);
     } catch (FileNotFoundException e) {
       throw404(jdkExchange);
     }
diff --git a/avaje-jex/src/main/java/io/avaje/jex/DJexConfig.java b/avaje-jex/src/main/java/io/avaje/jex/DJexConfig.java
@@ -28,6 +28,7 @@ final class DJexConfig implements JexConfig {
   private final CompressionConfig compression = new CompressionConfig();
   private int bufferInitial = 256;
   private long bufferMax = 4096L;
+  private int rangeChunkSize = 131_072;
   private HttpServerProvider serverProvider;
 
   @Override
@@ -199,4 +200,15 @@ public JexConfig serverProvider(HttpServerProvider serverProvider) {
     this.serverProvider = serverProvider;
     return this;
   }
+
+  @Override
+  public int rangeChunkSize() {
+    return rangeChunkSize;
+  }
+
+  @Override
+  public JexConfig rangeChunkSize(int rangeChunkSize) {
+    this.rangeChunkSize = rangeChunkSize;
+    return this;
+  }
 }
diff --git a/avaje-jex/src/main/java/io/avaje/jex/JexConfig.java b/avaje-jex/src/main/java/io/avaje/jex/JexConfig.java
@@ -146,6 +146,17 @@ public interface JexConfig {
    */
   JexConfig port(int port);
 
+  /** The configured rangeChunk size */
+  int rangeChunkSize();
+
+  /**
+   * Set the chunk size on range requests, set to a high number to reduce the amount of range
+   * requests (especially for video streaming)
+   *
+   * @param rangeChunkSize chunk size on range requests
+   */
+  JexConfig rangeChunkSize(int rangeChunkSize);
+
   /**
    * Registers a template renderer for a specific file extension.
    *
@@ -185,5 +196,4 @@ public interface JexConfig {
    *     default value is used
    */
   JexConfig socketBacklog(int backlog);
-
 }
diff --git a/avaje-jex/src/main/java/io/avaje/jex/compression/CompressedOutputStream.java b/avaje-jex/src/main/java/io/avaje/jex/compression/CompressedOutputStream.java
@@ -36,6 +36,7 @@ private void decideCompression(int length) throws IOException {
     if (!compressionDecided) {
       boolean compressionAllowed =
           compressedStream == null
+              && ctx.responseHeader(Constants.CONTENT_RANGE) == null
               && compression.allowsForCompression(ctx.responseHeader(Constants.CONTENT_TYPE));
 
       if (compressionAllowed && length >= minSizeForCompression) {
diff --git a/avaje-jex/src/main/java/io/avaje/jex/core/BufferedOutStream.java b/avaje-jex/src/main/java/io/avaje/jex/core/BufferedOutStream.java
@@ -18,6 +18,11 @@ final class BufferedOutStream extends OutputStream {
     this.context = context;
     this.max = max;
     this.buffer = new ByteArrayOutputStream(initial);
+
+    // if content length is set, skip buffer
+    if (context.responseHeader(Constants.CONTENT_LENGTH) != null) {
+      count = max + 1;
+    }
   }
 
   @Override
@@ -52,7 +57,9 @@ public void write(byte[] b, int off, int len) throws IOException {
   /** Use responseLength 0 and chunked response. */
   private void initialiseChunked() throws IOException {
     final HttpExchange exchange = context.exchange();
-    exchange.sendResponseHeaders(context.statusCode(), 0);
+    // if a manual content-length is set, honor that instead of chunking
+    String length = context.responseHeader(Constants.CONTENT_LENGTH);
+    exchange.sendResponseHeaders(context.statusCode(), length == null ? 0 : Long.parseLong(length));
     stream = exchange.getResponseBody();
     // empty the existing buffer
     buffer.writeTo(stream);
diff --git a/avaje-jex/src/main/java/io/avaje/jex/core/Constants.java b/avaje-jex/src/main/java/io/avaje/jex/core/Constants.java
@@ -19,4 +19,9 @@ private Constants() {}
   public static final String TEXT_PLAIN_UTF8 = "text/plain;charset=utf-8";
   public static final String APPLICATION_JSON = "application/json";
   public static final String APPLICATION_X_JSON_STREAM = "application/x-json-stream";
+
+  // range
+  public static final String ACCEPT_RANGES = "Accept-ranges";
+  public static final String RANGE = "Range";
+  public static final String CONTENT_RANGE = "Content-range";
 }
diff --git a/avaje-jex/src/main/java/io/avaje/jex/core/JdkContext.java b/avaje-jex/src/main/java/io/avaje/jex/core/JdkContext.java
@@ -536,4 +536,9 @@ public void write(String content) {
   public JsonService jsonService() {
     return mgr.jsonService();
   }
+
+  @Override
+  public void rangedWrite(InputStream inputStream, long totalBytes) {
+    mgr.writeRange(this, inputStream, totalBytes);
+  }
 }
diff --git a/avaje-jex/src/main/java/io/avaje/jex/core/RangeWriter.java b/avaje-jex/src/main/java/io/avaje/jex/core/RangeWriter.java
@@ -0,0 +1,86 @@
+package io.avaje.jex.core;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.List;
+
+import io.avaje.jex.http.Context;
+import io.avaje.jex.http.HttpStatus;
+
+class RangeWriter {
+
+  private static final int DEFAULT_BUFFER_SIZE = 16384;
+
+  static void write(Context ctx, InputStream inputStream, long totalBytes, long chunkSize) {
+
+    ctx.header(Constants.ACCEPT_RANGES, "bytes");
+    final String rangeHeader = ctx.header(Constants.RANGE);
+    if (rangeHeader == null) {
+      ctx.contentLength(totalBytes).write(inputStream);
+      return;
+    }
+    final List<String> requestedRange =
+        Arrays.stream(rangeHeader.split("=")[1].split("-")).filter(s -> !s.isEmpty()).toList();
+    final long from = Long.parseLong(requestedRange.get(0));
+    final long to;
+
+    final boolean audioOrVideo = isAudioOrVideo(ctx.responseHeader(Constants.CONTENT_TYPE));
+
+    if (!audioOrVideo || from + chunkSize > totalBytes) {
+      to = totalBytes - 1; // chunk bigger than file, write all
+    } else if (requestedRange.size() == 2) {
+      to = Long.parseLong(requestedRange.get(1)); // chunk smaller than file, to/from specified
+    } else {
+      to = from + chunkSize - 1;
+    }
+
+    long contentLength;
+    if (audioOrVideo) {
+      contentLength = Math.min(to - from + 1, totalBytes);
+    } else {
+      contentLength = totalBytes - from;
+    }
+
+    final HttpStatus status;
+    if (audioOrVideo) {
+      status = HttpStatus.PARTIAL_CONTENT_206;
+    } else {
+      status = HttpStatus.OK_200;
+    }
+
+    ctx.status(status);
+    ctx.header(Constants.ACCEPT_RANGES, "bytes");
+    ctx.header(Constants.CONTENT_RANGE, "bytes " + from + "-" + to + "/" + totalBytes);
+    ctx.contentLength(contentLength);
+    try (var os = ctx.outputStream()) {
+      write(os, inputStream, from, to);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  private static void write(OutputStream outputStream, InputStream inputStream, long from, long to)
+      throws IOException {
+    byte[] buffer = new byte[DEFAULT_BUFFER_SIZE];
+    long toSkip = from;
+    while (toSkip > 0) {
+      toSkip -= inputStream.skip(toSkip);
+    }
+    long bytesLeft = to - from + 1;
+    while (bytesLeft > 0) {
+      int read = inputStream.read(buffer, 0, (int) Math.min(DEFAULT_BUFFER_SIZE, bytesLeft));
+      if (read == -1) {
+        break; // End of stream reached unexpectedly
+      }
+      outputStream.write(buffer, 0, read);
+      bytesLeft -= read;
+    }
+  }
+
+  private static boolean isAudioOrVideo(String contentType) {
+    return contentType.startsWith("audio/") || contentType.startsWith("video/");
+  }
+}
diff --git a/avaje-jex/src/main/java/io/avaje/jex/core/ServiceManager.java b/avaje-jex/src/main/java/io/avaje/jex/core/ServiceManager.java
@@ -37,6 +37,7 @@ final class ServiceManager {
   private final String scheme;
   private final int bufferInitial;
   private final long bufferMax;
+  private final int rangeChunks;
 
   static ServiceManager create(Jex jex) {
     return new Builder(jex).build();
@@ -49,14 +50,16 @@ static ServiceManager create(Jex jex) {
       TemplateManager templateManager,
       String scheme,
       long bufferMax,
-      int bufferInitial) {
+      int bufferInitial,
+      int rangeChunks) {
     this.compressionConfig = compressionConfig;
     this.jsonService = jsonService;
     this.exceptionHandler = manager;
     this.templateManager = templateManager;
     this.scheme = scheme;
     this.bufferInitial = bufferInitial;
     this.bufferMax = bufferMax;
+    this.rangeChunks = rangeChunks;
   }
 
   OutputStream createOutputStream(JdkContext jdkContext) {
@@ -97,6 +100,10 @@ <E> void toJsonStream(Iterator<E> iterator, OutputStream os) {
     }
   }
 
+  void writeRange(Context ctx, InputStream is, long totalBytes) {
+    RangeWriter.write(ctx, is, totalBytes, rangeChunks);
+  }
+
   void maybeClose(Object iterator) {
     if (iterator instanceof AutoCloseable closeable) {
       try {
@@ -177,7 +184,8 @@ ServiceManager build() {
           initTemplateMgr(),
           jex.config().scheme(),
           jex.config().maxStreamBufferSize(),
-          jex.config().initialStreamBufferSize());
+          jex.config().initialStreamBufferSize(),
+          jex.config().rangeChunkSize());
     }
 
     JsonService initJsonService() {
diff --git a/avaje-jex/src/main/java/io/avaje/jex/http/Context.java b/avaje-jex/src/main/java/io/avaje/jex/http/Context.java
@@ -3,8 +3,10 @@
 import static java.util.Collections.emptyList;
 import static java.util.Collections.emptyMap;
 
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.io.UncheckedIOException;
 import java.lang.reflect.Type;
 import java.net.URI;
 import java.time.Duration;
@@ -110,6 +112,11 @@ default <T> T bodyStreamAsClass(Class<T> beanType) {
   /** Return the request content length. */
   long contentLength();
 
+  /** Manually set the response content length. */
+  default Context contentLength(long length) {
+    return header(Constants.CONTENT_LENGTH, String.valueOf(length));
+  }
+
   /** Return the request content type. */
   String contentType();
 
@@ -192,6 +199,26 @@ default String fullUrl() {
     return uri.charAt(0) != '/' ? uri : scheme() + "://" + host() + uri;
   }
 
+  /**
+   * Reads HTTP Range headers and determines which part of the provided InputStream to write back.
+   *
+   * @param inputStream data to write
+   * @param totalBytes total size of the data
+   */
+  void rangedWrite(InputStream inputStream, long totalBytes);
+
+  /**
+   * Writes input stream to {@link #rangedWrite(InputStream, long)} with currently available data
+   * via {@link InputStream#available}
+   */
+  default void rangedWrite(InputStream inputStream) {
+    try {
+      rangedWrite(inputStream, inputStream.available());
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
   /**
    * Return the request header value by name.
    *
diff --git a/avaje-jex/src/test/java/io/avaje/jex/compression/CompressionTest.java b/avaje-jex/src/test/java/io/avaje/jex/compression/CompressionTest.java
@@ -69,4 +69,23 @@ void testNoCompression() {
     assertThat(res.statusCode()).isEqualTo(200);
     assertThat(res.headers().firstValue(Constants.CONTENT_ENCODING)).isEmpty();
   }
+
+  @Test
+  void testCompressionRange() throws IOException {
+    var res =
+        pair.request()
+            .header(Constants.ACCEPT_ENCODING, "deflate, gzip;q=1.0, *;q=0.5")
+            .path("compress")
+            .GET()
+            .asInputStream();
+    assertThat(res.statusCode()).isEqualTo(200);
+    assertThat(res.headers().firstValue(Constants.CONTENT_ENCODING)).contains("gzip");
+
+    var expected = CompressionTest.class.getResourceAsStream("/64KB.json").readAllBytes();
+
+    final var gzipInputStream = new GZIPInputStream(res.body());
+    var decompressed = gzipInputStream.readAllBytes();
+    gzipInputStream.close();
+    assertThat(decompressed).isEqualTo(expected);
+  }
 }
diff --git a/avaje-jex/src/test/java/io/avaje/jex/http/LargeSeekableInput.java b/avaje-jex/src/test/java/io/avaje/jex/http/LargeSeekableInput.java
@@ -0,0 +1,57 @@
+package io.avaje.jex.http;
+
+import java.io.InputStream;
+
+class LargeSeekableInput extends InputStream {
+    private final long prefixSize;
+    private final long contentSize;
+    private long alreadyRead = 0L;
+
+    public LargeSeekableInput(long prefixSize, long contentSize) {
+        this.prefixSize = prefixSize;
+        this.contentSize = contentSize;
+    }
+
+    private long remaining() {
+        return prefixSize + contentSize - alreadyRead;
+    }
+
+    @Override
+    public int available() {
+        long rem = remaining();
+        if (rem >= 0 && rem <= Integer.MAX_VALUE) {
+            return (int) rem;
+        } else {
+            return Integer.MAX_VALUE;
+        }
+    }
+
+    @Override
+    public long skip(long toSkip) {
+        if (toSkip <= 0) {
+            return 0;
+        } else {
+            long rem = remaining();
+            if (rem >= 0 && rem <= toSkip) {
+                alreadyRead += rem;
+                return rem;
+            } else {
+                alreadyRead += toSkip;
+                return toSkip;
+            }
+        }
+    }
+
+    @Override
+    public int read() {
+        if (remaining() == 0L) {
+            return -1;
+        } else if (alreadyRead < prefixSize) {
+            alreadyRead++;
+            return ' ';
+        } else {
+            alreadyRead++;
+            return 'J';
+        }
+    }
+}
diff --git a/avaje-jex/src/test/java/io/avaje/jex/http/RangeTest.java b/avaje-jex/src/test/java/io/avaje/jex/http/RangeTest.java

Original file line number	Diff line number	Diff line change
`@@ -93,7 +93,7 @@ private void sendURL(Context ctx, String urlPath, URL path) {`
`93`	`93`	`addCachedEntry(ctx, urlPath, fis);`
`94`	`94`	`return;`
`95`	`95`	`}`
`96`		`- ctx.write(fis);`
	`96`	`+ ctx.rangedWrite(fis);`
`97`	`97`	`} catch (final Exception e) {`
`98`	`98`	`throw404(ctx.exchange());`
`99`	`99`	`}`
Original file line number	Diff line number	Diff line change
`@@ -108,7 +108,7 @@ private void sendFile(Context ctx, HttpExchange jdkExchange, String urlPath, Fil`
`108`	`108`	`addCachedEntry(ctx, urlPath, fis);`
`109`	`109`	`return;`
`110`	`110`	`}`
`111`		`- ctx.write(fis);`
	`111`	`+ ctx.rangedWrite(fis);`
`112`	`112`	`} catch (FileNotFoundException e) {`
`113`	`113`	`throw404(jdkExchange);`
`114`	`114`	`}`
Original file line number	Diff line number	Diff line change
`@@ -536,4 +536,9 @@ public void write(String content) {`
`536`	`536`	`public JsonService jsonService() {`
`537`	`537`	`return mgr.jsonService();`
`538`	`538`	`}`
	`539`	`+`
	`540`	`+ @Override`
	`541`	`+ public void rangedWrite(InputStream inputStream, long totalBytes) {`
	`542`	`+ mgr.writeRange(this, inputStream, totalBytes);`
	`543`	`+ }`
`539`	`544`	`}`