Allow to create LZ4BlockInputStream with LZ4SafeDecompressor

- added builder for LZ4BlockInputStream which accepts both fast and safe decompressor (prefers fast one)

Author: mihhailmeshkov

src/java/net/jpountz/lz4/LZ4BlockInputStream.java

@@ -43,7 +43,8 @@
  */
 public class LZ4BlockInputStream extends FilterInputStream {
 
-  private final LZ4FastDecompressor decompressor;
+  private final LZ4FastDecompressor fastDecompressor;
+  private final LZ4SafeDecompressor safeDecompressor;
   private final Checksum checksum;
   private final boolean stopOnEmptyBlock;
   private byte[] buffer;
@@ -56,52 +57,45 @@ public class LZ4BlockInputStream extends FilterInputStream {
    * Creates a new LZ4 input stream to read from the specified underlying InputStream.
    *
    * @param in                the {@link InputStream} to poll
-   * @param decompressor      the {@link LZ4FastDecompressor decompressor} instance to
+   * @param fastDecompressor  the {@link LZ4FastDecompressor} instance to
    *                          use
    * @param checksum          the {@link Checksum} instance to use, must be
    *                          equivalent to the instance which has been used to
    *                          write the stream
    * @param stopOnEmptyBlock  whether read is stopped on an empty block
    */
-  public LZ4BlockInputStream(InputStream in, LZ4FastDecompressor decompressor, Checksum checksum, boolean stopOnEmptyBlock) {
-    super(in);
-    this.decompressor = decompressor;
-    this.checksum = checksum;
-    this.stopOnEmptyBlock = stopOnEmptyBlock;
-    this.buffer = new byte[0];
-    this.compressedBuffer = new byte[HEADER_LENGTH];
-    o = originalLen = 0;
-    finished = false;
+  public LZ4BlockInputStream(InputStream in, LZ4FastDecompressor fastDecompressor, Checksum checksum, boolean stopOnEmptyBlock) {
+    this(in, fastDecompressor, null, checksum, stopOnEmptyBlock);
   }
 
   /**
    * Creates a new LZ4 input stream to read from the specified underlying InputStream.
    *
    * @param in                the {@link InputStream} to poll
-   * @param decompressor      the {@link LZ4FastDecompressor decompressor} instance to
+   * @param fastDecompressor  the {@link LZ4FastDecompressor} instance to
    *                          use
    * @param checksum          the {@link Checksum} instance to use, must be
    *                          equivalent to the instance which has been used to
    *                          write the stream
    *
    * @see #LZ4BlockInputStream(InputStream, LZ4FastDecompressor, Checksum, boolean)
    */
-  public LZ4BlockInputStream(InputStream in, LZ4FastDecompressor decompressor, Checksum checksum) {
-    this(in, decompressor, checksum, true);
+  public LZ4BlockInputStream(InputStream in, LZ4FastDecompressor fastDecompressor, Checksum checksum) {
+    this(in, fastDecompressor, checksum, true);
   }
 
   /**
    * Creates a new LZ4 input stream to read from the specified underlying InputStream, using {@link XXHash32} for checksuming.
    *
    * @param in                the {@link InputStream} to poll
-   * @param decompressor      the {@link LZ4FastDecompressor decompressor} instance to
+   * @param fastDecompressor  the {@link LZ4FastDecompressor} instance to
    *                          use
    *
    * @see #LZ4BlockInputStream(InputStream, LZ4FastDecompressor, Checksum, boolean)
    * @see StreamingXXHash32#asChecksum()
    */
-  public LZ4BlockInputStream(InputStream in, LZ4FastDecompressor decompressor) {
-    this(in, decompressor, XXHashFactory.fastestInstance().newStreamingHash32(DEFAULT_SEED).asChecksum(), true);
+  public LZ4BlockInputStream(InputStream in, LZ4FastDecompressor fastDecompressor) {
+    this(in, fastDecompressor, XXHashFactory.fastestInstance().newStreamingHash32(DEFAULT_SEED).asChecksum(), true);
   }
 
   /**
@@ -130,6 +124,48 @@ public LZ4BlockInputStream(InputStream in) {
     this(in, LZ4Factory.fastestInstance().fastDecompressor());
   }
 
+  /**
+   * Creates a new LZ4 input stream to read from the specified underlying InputStream.
+   *
+   * @param in                the {@link InputStream} to poll
+   * @param fastDecompressor  the {@link LZ4FastDecompressor} instance to
+   *                          use
+   * @param safeDecompressor  the {@link LZ4SafeDecompressor} instance to
+   *                          use (if both fastDecompressor and safeDecompressor are
+   *                          specified then the fastDecompressor gets used)
+   * @param checksum          the {@link Checksum} instance to use, must be
+   *                          equivalent to the instance which has been used to
+   *                          write the stream
+   * @param stopOnEmptyBlock  whether read is stopped on an empty block
+   */
+  public LZ4BlockInputStream(InputStream in, LZ4FastDecompressor fastDecompressor, LZ4SafeDecompressor safeDecompressor,
+                              Checksum checksum, boolean stopOnEmptyBlock) {
+    super(in);
+
+    this.fastDecompressor = fastDecompressor;
+    this.safeDecompressor = safeDecompressor;
+    this.checksum = checksum;
+    this.stopOnEmptyBlock = stopOnEmptyBlock;
+    this.buffer = new byte[0];
+    this.compressedBuffer = new byte[HEADER_LENGTH];
+    o = originalLen = 0;
+    finished = false;
+  }
+
+  /**
+   * Creates a new LZ4 input stream builder. The following are defaults:
+   * <ul>
+   * <ol> decompressor - {@code LZ4Factory.fastestInstance().fastDecompressor()} </ol>
+   * <ol> checksum - {@link XXHash32} </ol>
+   * <ol> stopOnEmptyBlock - {@code true} </ol>
+   * </ul>
+   * @param in  the {@link InputStream} to poll
+   * @return new instance of {@link LZ4BlockInputStreamBuilder} to be used to configure and build new LZ4 input stream
+   */
+  public static LZ4BlockInputStreamBuilder newBuilder(InputStream in) {
+    return new LZ4BlockInputStreamBuilder(in);
+  }
+
   @Override
   public int available() throws IOException {
     return originalLen - o;
@@ -213,11 +249,11 @@ private void refill() throws IOException {
     final int check = SafeUtils.readIntLE(compressedBuffer, MAGIC_LENGTH + 9);
     assert HEADER_LENGTH == MAGIC_LENGTH + 13;
     if (originalLen > 1 << compressionLevel
-        || originalLen < 0
-        || compressedLen < 0
-        || (originalLen == 0 && compressedLen != 0)
-        || (originalLen != 0 && compressedLen == 0)
-        || (compressionMethod == COMPRESSION_METHOD_RAW && originalLen != compressedLen)) {
+      || originalLen < 0
+      || compressedLen < 0
+      || (originalLen == 0 && compressedLen != 0)
+      || (originalLen != 0 && compressedLen == 0)
+      || (compressionMethod == COMPRESSION_METHOD_RAW && originalLen != compressedLen)) {
       throw new IOException("Stream is corrupted");
     }
     if (originalLen == 0 && compressedLen == 0) {
@@ -235,25 +271,32 @@ private void refill() throws IOException {
       buffer = new byte[Math.max(originalLen, buffer.length * 3 / 2)];
     }
     switch (compressionMethod) {
-    case COMPRESSION_METHOD_RAW:
-      readFully(buffer, originalLen);
-      break;
-    case COMPRESSION_METHOD_LZ4:
-      if (compressedBuffer.length < compressedLen) {
-        compressedBuffer = new byte[Math.max(compressedLen, compressedBuffer.length * 3 / 2)];
-      }
-      readFully(compressedBuffer, compressedLen);
-      try {
-        final int compressedLen2 = decompressor.decompress(compressedBuffer, 0, buffer, 0, originalLen);
-        if (compressedLen != compressedLen2) {
-          throw new IOException("Stream is corrupted");
+      case COMPRESSION_METHOD_RAW:
+        readFully(buffer, originalLen);
+        break;
+      case COMPRESSION_METHOD_LZ4:
+        if (compressedBuffer.length < compressedLen) {
+          compressedBuffer = new byte[Math.max(compressedLen, compressedBuffer.length * 3 / 2)];
         }
-      } catch (LZ4Exception e) {
-        throw new IOException("Stream is corrupted", e);
-      }
-      break;
-    default:
-      throw new AssertionError();
+        readFully(compressedBuffer, compressedLen);
+        try {
+          if (fastDecompressor == null) {
+            final int decompressedLen = safeDecompressor.decompress(compressedBuffer, 0, compressedLen, buffer, 0, originalLen);
+            if (decompressedLen != originalLen) {
+              throw new IOException("Stream is corrupted");
+            }
+          } else {
+            final int compressedLen2 = fastDecompressor.decompress(compressedBuffer, 0, buffer, 0, originalLen);
+            if (compressedLen != compressedLen2) {
+              throw new IOException("Stream is corrupted");
+            }
+          }
+        } catch (LZ4Exception e) {
+          throw new IOException("Stream is corrupted", e);
+        }
+        break;
+      default:
+        throw new AssertionError();
     }
     checksum.reset();
     checksum.update(buffer, 0, originalLen);
@@ -304,7 +347,92 @@ public void reset() throws IOException {
   @Override
   public String toString() {
     return getClass().getSimpleName() + "(in=" + in
-        + ", decompressor=" + decompressor + ", checksum=" + checksum + ")";
+      + ", decompressor=" + (fastDecompressor != null ? fastDecompressor : safeDecompressor)
+      + ", checksum=" + checksum + ")";
   }
 
+  public static final class LZ4BlockInputStreamBuilder {
+    private final InputStream in;
+    private boolean stopOnEmptyBlock = true;
+    private LZ4FastDecompressor fastDecompressor;
+    private LZ4SafeDecompressor safeDecompressor;
+    private Checksum checksum;
+
+    /**
+     * Creates a new LZ4 block input stream builder to read from the specified underlying InputStream.
+     *
+     * @param in                the {@link InputStream} to poll
+     */
+    public LZ4BlockInputStreamBuilder(InputStream in) {
+      this.in = in;
+    }
+
+    /**
+     * Registers value of stopOnEmptyBlock to be used by the builder
+     *
+     * @param stopOnEmptyBlock  whether read is stopped on an empty block
+     */
+    public LZ4BlockInputStreamBuilder withStopOnEmptyBlock(boolean stopOnEmptyBlock) {
+      this.stopOnEmptyBlock = stopOnEmptyBlock;
+      return this;
+    }
+
+    /**
+     * Registers {@link LZ4FastDecompressor} to be used by the builder as a decompressor. Overrides one set by
+     * {@link #withDecompressor(LZ4SafeDecompressor)}
+     *
+     * @param fastDecompressor      the {@link LZ4FastDecompressor} instance to use
+     */
+    public LZ4BlockInputStreamBuilder withDecompressor(LZ4FastDecompressor fastDecompressor) {
+      this.fastDecompressor = fastDecompressor;
+      this.safeDecompressor = null;
+      return this;
+    }
+
+    /**
+     * Registers {@link LZ4SafeDecompressor} to be used by the builder as a decompressor. Overrides one set by
+     * {@link #withDecompressor(LZ4FastDecompressor)}
+     *
+     * @param safeDecompressor      the {@link LZ4SafeDecompressor} instance to use.
+     */
+    public LZ4BlockInputStreamBuilder withDecompressor(LZ4SafeDecompressor safeDecompressor) {
+      this.safeDecompressor = safeDecompressor;
+      this.fastDecompressor = null;
+      return this;
+    }
+
+    /**
+     * Registers {@link Checksum} to be used by the builder
+     *
+     * @param checksum          the {@link Checksum} instance to use, must be
+     *                          equivalent to the instance which has been used to
+     *                          write the stream
+     */
+    public LZ4BlockInputStreamBuilder withChecksum(Checksum checksum) {
+      this.checksum = checksum;
+      return this;
+    }
+
+    /**
+     * Creates a new LZ4 input stream to read from the specified underlying InputStream with specified parameters
+     *
+     * @see #withChecksum(Checksum)
+     * @see #withDecompressor(LZ4FastDecompressor)
+     * @see #withDecompressor(LZ4SafeDecompressor)
+     * @see #withStopOnEmptyBlock(boolean)
+     */
+    public LZ4BlockInputStream build() {
+      Checksum checksum = this.checksum;
+      LZ4FastDecompressor fastDecompressor = this.fastDecompressor;
+      LZ4SafeDecompressor safeDecompressor = this.safeDecompressor;
+
+      if (checksum == null) {
+        checksum = XXHashFactory.fastestInstance().newStreamingHash32(DEFAULT_SEED).asChecksum();
+      }
+      if (fastDecompressor == null && safeDecompressor == null) {
+        fastDecompressor = LZ4Factory.fastestInstance().fastDecompressor();
+      }
+      return new LZ4BlockInputStream(in, fastDecompressor, safeDecompressor, checksum, stopOnEmptyBlock);
+    }
+  }
 }