Add support for reading warc-zstd files

ato · ato · commit d62906447514 · 2025-09-08T13:49:51.000+09:00
diff --git a/README.md b/README.md
@@ -305,26 +305,27 @@ Note: revisit records never have a payload so
 
 ## Comparison
 
-| Criteria            | jwarc       | [JWAT]          | [webarchive-commons]  |
-|---------------------|-------------|-----------------|---------------|
-| License             | Apache 2    | Apache 2        | Apache 2      |
-| Parser based on     | Ragel FSM   | Hand-rolled FSM | Apache HTTP   |
-| Push parsing        | Low level   | ✘               | ✘             |
-| Folded headers †    | ✔           | ✔               | ✔             | 
-| [Encoded words] †   | ✘           | ✘ (disabled)    | ✘             |
-| Validation          | The basics  | ✔               | ✘             |
-| Strict parsing  ‡   | ✔           | ✘               | ✘             |
-| Lenient parsing     | HTTP only   | ✔               | ✔             |
-| Multi-value headers | ✔           | ✔               | ✘             |
-| I/O Framework       | NIO         | IO              | IO            |
-| Record type classes | ✔           | ✘               | ✘             |
-| Typed accessors     | ✔           | ✔               | Some          |
-| GZIP detection      | ✔           | ✔               | Filename only |
-| WARC writer         | Barebones   | ✔               | ✔             |
-| ARC reader          | Auto        | Separate API    | Factory       |
-| ARC writer          | ✘           | ✔               | ✔             |
-| Speed * (.warc)     | 1x          | ~5x slower      | ~13x slower   |
-| Speed * (.warc.gz)  | 1x          | ~1.4x slower    | ~2.8x slower  |
+| Criteria            | jwarc      | [JWAT]          | [webarchive-commons] |
+|---------------------|------------|-----------------|----------------------|
+| License             | Apache 2   | Apache 2        | Apache 2             |
+| Parser based on     | Ragel FSM  | Hand-rolled FSM | Apache HTTP          |
+| Push parsing        | Low level  | ✘               | ✘                    |
+| Folded headers †    | ✔          | ✔               | ✔                    | 
+| [Encoded words] †   | ✘          | ✘ (disabled)    | ✘                    |
+| Validation          | The basics | ✔               | ✘                    |
+| Strict parsing  ‡   | ✔          | ✘               | ✘                    |
+| Lenient parsing     | HTTP only  | ✔               | ✔                    |
+| Multi-value headers | ✔          | ✔               | ✘                    |
+| I/O Framework       | NIO        | IO              | IO                   |
+| Record type classes | ✔          | ✘               | ✘                    |
+| Typed accessors     | ✔          | ✔               | Some                 |
+| GZIP detection      | ✔          | ✔               | Filename only        |
+| [zstd compression]  | read-only  | ✘               | ✘                    |
+| WARC writer         | Barebones  | ✔               | ✔                    |
+| ARC reader          | Auto       | Separate API    | Factory              |
+| ARC writer          | ✘          | ✔               | ✔                    |
+| Speed * (.warc)     | 1x         | ~5x slower      | ~13x slower          |
+| Speed * (.warc.gz)  | 1x         | ~1.4x slower    | ~2.8x slower         |
 
 (†) WARC features copied from HTTP that have since been deprecated in HTTP. I'm not aware of any software that writes
 WARCs using these features and usage of them should probably be avoided. JWAT behaves differently from jwarc and
@@ -345,6 +346,7 @@ See also: [Unaffiliated benchmark against other languages](https://code402.com/h
 [JWAT]: https://sbforge.org/display/JWAT/JWAT
 [webarchive-commons]: https://github.com/iipc/webarchive-commons
 [Encoded words]: https://www.ietf.org/rfc/rfc2047.txt
+[zstd compression]: https://iipc.github.io/warc-specifications/specifications/warc-zstd/
 
 ### Other WARC libraries
 
diff --git a/pom.xml b/pom.xml
@@ -77,6 +77,11 @@
           <scope>compile</scope>
           <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>com.github.luben</groupId>
+            <artifactId>zstd-jni</artifactId>
+            <version>1.5.7-4</version>
+        </dependency>
     </dependencies>
 
     <properties>
diff --git a/src/org/netpreserve/jwarc/DecompressingChannel.java b/src/org/netpreserve/jwarc/DecompressingChannel.java
@@ -0,0 +1,21 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright (C) 2025 National Library of Australia and the jwarc contributors
+ */
+
+package org.netpreserve.jwarc;
+
+import java.io.IOException;
+
+interface DecompressingChannel {
+
+    /**
+     * Number of bytes read from the underlying channel.
+     */
+    long inputPosition();
+
+    /**
+     * Reset the decompressor, discarding any buffered data.
+     */
+    void reset() throws IOException;
+}
diff --git a/src/org/netpreserve/jwarc/GunzipChannel.java b/src/org/netpreserve/jwarc/GunzipChannel.java
@@ -1,6 +1,6 @@
 /*
  * SPDX-License-Identifier: Apache-2.0
- * Copyright (C) 2018-2022 National Library of Australia and the jwarc contributors
+ * Copyright (C) 2018-2025 National Library of Australia and the jwarc contributors
  */
 
 package org.netpreserve.jwarc;
@@ -15,7 +15,7 @@
 import java.util.zip.Inflater;
 import java.util.zip.ZipException;
 
-class GunzipChannel implements ReadableByteChannel {
+class GunzipChannel implements ReadableByteChannel, DecompressingChannel {
     private static final int FHCRC = 2;
     private static final int FEXTRA = 4;
     private static final int FNAME = 8;
diff --git a/src/org/netpreserve/jwarc/IOUtils.java b/src/org/netpreserve/jwarc/IOUtils.java
@@ -1,11 +1,12 @@
 /*
  * SPDX-License-Identifier: Apache-2.0
- * Copyright (C) 2018 National Library of Australia and the jwarc contributors
+ * Copyright (C) 2018-2025 National Library of Australia and the jwarc contributors
  */
 
 package org.netpreserve.jwarc;
 
 import javax.net.ssl.SSLSocketFactory;
+import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -22,6 +23,7 @@ public final class IOUtils {
 
     /**
      * Transfers as many bytes as possible from src to dst.
+     *
      * @return the number of bytes transferred.
      */
     static int transfer(ByteBuffer src, ByteBuffer dst) {
@@ -30,10 +32,11 @@ static int transfer(ByteBuffer src, ByteBuffer dst) {
 
     /**
      * Transfers up to limits from src to dst.
+     *
      * @return the number of bytes transferred.
      */
     static int transfer(ByteBuffer src, ByteBuffer dst, long limit) {
-        return transferExactly(src, dst, (int)Math.min(Math.min(src.remaining(), dst.remaining()), limit));
+        return transferExactly(src, dst, (int) Math.min(Math.min(src.remaining(), dst.remaining()), limit));
     }
 
     private static int transferExactly(ByteBuffer src, ByteBuffer dst, int n) {
@@ -125,7 +128,7 @@ static Socket connect(String scheme, String host, int port) throws IOException {
 
     public static byte[] readNBytes(InputStream stream, int n) throws IOException {
         byte[] buffer = new byte[n];
-        for (int remaining = n; remaining > 0;) {
+        for (int remaining = n; remaining > 0; ) {
             int read = stream.read(buffer, buffer.length - remaining, remaining);
             if (read < 0) {
                 return Arrays.copyOf(buffer, buffer.length - remaining);
@@ -134,4 +137,26 @@ public static byte[] readNBytes(InputStream stream, int n) throws IOException {
         }
         return buffer;
     }
+
+    /**
+     * Ensures that at least the specified number of bytes are available in the buffer by reading from the provided channel
+     * if necessary. If the buffer already contains enough bytes, no data is read. If the channel reaches EOF before the
+     * required number of bytes is available, an EOFException is thrown.
+     *
+     * @return the total number of bytes read from the channel to ensure the required availability in the buffer.
+     * @throws IOException if an I/O error occurs while reading from the channel.
+     * @throws EOFException if the channel reaches EOF before the required number of bytes are available.
+     */
+    static int ensureAvailable(ReadableByteChannel channel, ByteBuffer buffer, int needed) throws IOException {
+        int totalRead = 0;
+        if (buffer.remaining() >= needed) return 0;
+        buffer.compact();
+        while (buffer.position() < needed) {
+            int n = channel.read(buffer);
+            if (n == -1) throw new EOFException("expected " + (needed - totalRead) + " more bytes in channel");
+            totalRead += n;
+        }
+        buffer.flip();
+        return totalRead;
+    }
 }
diff --git a/src/org/netpreserve/jwarc/WarcCompression.java b/src/org/netpreserve/jwarc/WarcCompression.java
@@ -1,18 +1,21 @@
 /*
  * SPDX-License-Identifier: Apache-2.0
- * Copyright (C) 2018 National Library of Australia and the jwarc contributors
+ * Copyright (C) 2018-2025 National Library of Australia and the jwarc contributors
  */
 
 package org.netpreserve.jwarc;
 
 import java.nio.file.Path;
 
 public enum WarcCompression {
-    NONE, GZIP;
+    NONE, GZIP, ZSTD;
 
     static WarcCompression forPath(Path path) {
-        if (path.getFileName().toString().endsWith(".gz")) {
+        String filename = path.getFileName().toString();
+        if (filename.endsWith(".gz")) {
             return GZIP;
+        } else if (filename.endsWith(".zst")) {
+            return ZSTD;
         } else {
             return NONE;
         }
diff --git a/src/org/netpreserve/jwarc/WarcReader.java b/src/org/netpreserve/jwarc/WarcReader.java
@@ -1,11 +1,12 @@
 /*
  * SPDX-License-Identifier: Apache-2.0
- * Copyright (C) 2018-2022 National Library of Australia and the jwarc contributors
+ * Copyright (C) 2018-2025 National Library of Australia and the jwarc contributors
  */
 
 package org.netpreserve.jwarc;
 
 import java.io.*;
+import java.lang.reflect.InvocationTargetException;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.nio.channels.Channels;
@@ -61,7 +62,7 @@ public WarcReader(ReadableByteChannel channel, ByteBuffer buffer) throws IOExcep
         startPosition = tryPosition(channel);
         position = startPosition;
 
-        while (buffer.remaining() < 2) {
+        while (buffer.remaining() < 4) {
             buffer.compact();
             int n = channel.read(buffer);
             buffer.flip();
@@ -78,17 +79,40 @@ public WarcReader(ReadableByteChannel channel, ByteBuffer buffer) throws IOExcep
             }
         }
 
-        if ((buffer.order() == ByteOrder.LITTLE_ENDIAN && buffer.getShort(buffer.position()) == (short) 0x8b1f)
-                || (buffer.order() == ByteOrder.BIG_ENDIAN && buffer.getShort(buffer.position()) == 0x1f8b)) {
-            this.channel = new GunzipChannel(channel, buffer);
-            this.buffer = ByteBuffer.allocate(8192);
-            this.buffer.flip();
-            compression = WarcCompression.GZIP;
-        } else {
-            this.channel = channel;
-            this.buffer = buffer;
-            compression = WarcCompression.NONE;
+        ByteOrder originalOrder = buffer.order();
+        try {
+            buffer.order(ByteOrder.LITTLE_ENDIAN);
+
+            if (buffer.getShort(buffer.position()) == GzipChannel.GZIP_MAGIC) {
+                this.channel = new GunzipChannel(channel, buffer);
+                this.buffer = ByteBuffer.allocate(8192);
+                this.buffer.flip();
+                compression = WarcCompression.GZIP;
+            } else if (buffer.getInt(buffer.position()) == 0xfd2fb528 || buffer.getInt(buffer.position()) == 0x184D2A5D) {
+                try {
+                    this.channel = (ReadableByteChannel) Class.forName("org.netpreserve.jwarc.ZstdDecompressingChannel")
+                            .getConstructor(ReadableByteChannel.class, ByteBuffer.class)
+                            .newInstance(channel, buffer);
+                } catch (InstantiationException | IllegalAccessException | InvocationTargetException |
+                         NoSuchMethodException | ClassNotFoundException e) {
+                    throw new IOException(e);
+                }
+                this.buffer = ByteBuffer.allocate(8192);
+                this.buffer.flip();
+                compression = WarcCompression.ZSTD;
+
+                // update position in case we read a dictionary frame
+                position = ((DecompressingChannel) this.channel).inputPosition();
+            } else {
+                this.channel = channel;
+                this.buffer = buffer;
+                compression = WarcCompression.NONE;
+            }
+
+        } finally {
+            buffer.order(originalOrder);
         }
+
         underlyingChannel = channel;
     }
 
@@ -150,8 +174,8 @@ public Optional<WarcRecord> next() throws IOException {
             record.body().close();
             long trailerLength = consumeTrailer();
 
-            if (channel instanceof GunzipChannel) {
-                position = startPosition + ((GunzipChannel) channel).inputPosition();
+            if (channel instanceof DecompressingChannel) {
+                position = startPosition + ((DecompressingChannel) channel).inputPosition();
             } else {
                 position += headerLength + record.body().size() + trailerLength;
             }
@@ -309,8 +333,8 @@ public void position(long newPosition) throws IOException {
             record.body().close();
             record = null;
         }
-        if (compression == WarcCompression.GZIP) {
-            ((GunzipChannel)channel).reset();
+        if (channel instanceof DecompressingChannel) {
+            ((DecompressingChannel)channel).reset();
         }
     }
 
diff --git a/src/org/netpreserve/jwarc/WarcWriter.java b/src/org/netpreserve/jwarc/WarcWriter.java
@@ -1,6 +1,6 @@
 /*
  * SPDX-License-Identifier: Apache-2.0
- * Copyright (C) 2018-2023 National Library of Australia and the jwarc contributors
+ * Copyright (C) 2018-2025 National Library of Australia and the jwarc contributors
  */
 
 package org.netpreserve.jwarc;
@@ -47,8 +47,10 @@ public WarcWriter(WritableByteChannel channel, WarcCompression compression) thro
         this.compression = compression;
         if (compression == WarcCompression.GZIP) {
             this.channel = new GzipChannel(channel);
-        } else {
+        } else if (compression == WarcCompression.NONE || compression == null) {
             this.channel = channel;
+        } else {
+            throw new IllegalArgumentException("Unsupported compression: " + compression);
         }
 
         if (channel instanceof SeekableByteChannel) {
diff --git a/src/org/netpreserve/jwarc/ZstdDecompressingChannel.java b/src/org/netpreserve/jwarc/ZstdDecompressingChannel.java
diff --git a/test/org/netpreserve/jwarc/ZstdDecompressingChannelTest.java b/test/org/netpreserve/jwarc/ZstdDecompressingChannelTest.java
