stream improvements, fix brotli input wrapping

src/main/java/org/cobbzilla/util/http/HttpContentEncodingType.java

@@ -6,10 +6,13 @@ import com.nixxcode.jvmbrotli.dec.BrotliInputStream;
 import com.nixxcode.jvmbrotli.enc.BrotliOutputStream;
 import lombok.AllArgsConstructor;
 import org.cobbzilla.util.io.FilterInputStreamViaOutputStream;
+import org.cobbzilla.util.system.Bytes;
 
 import java.io.*;
 import java.util.zip.*;
 
+import static org.cobbzilla.util.daemon.ZillaRuntime.die;
+
 @AllArgsConstructor
 public enum HttpContentEncodingType {
 
@@ -17,14 +20,30 @@ public enum HttpContentEncodingType {
 
     gzip (GZIPInputStream::new, GZIPOutputStream::new, GZIPOutputStream.class),
 
-    deflate (in -> new InflaterInputStream(in, new Inflater(true)),
+    deflate ((in, bufsiz) -> new InflaterInputStream(in, new Inflater(true), bufsiz),
              out -> new DeflaterOutputStream(out, new Deflater(7, true)),
              in -> new FilterInputStreamViaOutputStream(in, out -> new InflaterOutputStream(out, new Inflater(true)))),
 
-    br (BrotliInputStream::new, BrotliOutputStream::new, BrotliOutputStream.class),
-    bro (BrotliInputStream::new, BrotliOutputStream::new, BrotliOutputStream.class);
+    br (HttpContentEncodingType::wrapBrotliInput, BrotliOutputStream::new, BrotliOutputStream.class),
+    bro (HttpContentEncodingType::wrapBrotliInput, BrotliOutputStream::new, BrotliOutputStream.class);
+
+    public static final int DEFAULT_IN_BUFSIZ = (int) (8 * Bytes.KB);
+
+    static {
+        if (!BrotliLoader.isBrotliAvailable()) die("BrotliLoader.isBrotliAvailable() returned false");
+    }
+
+    public interface HttpContentEncodingInputWrapper {
+        InputStream wrap(InputStream in, int bufsiz) throws IOException;
+    }
+
+    public interface HttpContentEncodingOutputWrapper {
+        OutputStream wrap(OutputStream out) throws IOException;
+    }
 
-    static { BrotliLoader.isBrotliAvailable(); }
+    public interface HttpContentEncodingInputAsOutputWrapper {
+        FilterInputStreamViaOutputStream wrap(InputStream in) throws IOException;
+    }
 
     private final HttpContentEncodingInputWrapper inputWrapper;
     private final HttpContentEncodingOutputWrapper outputWrapper;
@@ -40,22 +59,17 @@ public enum HttpContentEncodingType {
 
     @JsonCreator public static HttpContentEncodingType fromString (String v) { return valueOf(v.toLowerCase()); }
 
-    public InputStream wrapInput(InputStream in) throws IOException { return inputWrapper.wrap(in); }
+    private static InputStream wrapBrotliInput(InputStream in, int bufsiz) throws IOException {
+        final BrotliInputStream brIn = new BrotliInputStream(in, bufsiz);
+        brIn.enableEagerOutput();
+        return brIn;
+    }
+
+    public InputStream wrapInput(InputStream in) throws IOException { return inputWrapper.wrap(in, DEFAULT_IN_BUFSIZ); }
+    public InputStream wrapInput(InputStream in, int bufsiz) throws IOException { return inputWrapper.wrap(in, bufsiz); }
 
     public OutputStream wrapOutput(OutputStream out) throws IOException { return outputWrapper.wrap(out); }
 
     public FilterInputStreamViaOutputStream wrapInputAsOutput(InputStream in) throws IOException { return inputAsOutputWrapper.wrap(in); }
 
-    public interface HttpContentEncodingInputWrapper {
-        InputStream wrap(InputStream in) throws IOException;
-    }
-
-    public interface HttpContentEncodingOutputWrapper {
-        OutputStream wrap(OutputStream out) throws IOException;
-    }
-
-    public interface HttpContentEncodingInputAsOutputWrapper {
-        FilterInputStreamViaOutputStream wrap(InputStream in) throws IOException;
-    }
-
 }

src/main/java/org/cobbzilla/util/io/FilterInputStreamViaOutputStream.java

@@ -4,9 +4,9 @@ import lombok.extern.slf4j.Slf4j;
 import org.cobbzilla.util.system.Bytes;
 
 import java.io.*;
-import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.cobbzilla.util.daemon.ZillaRuntime.*;
 import static org.cobbzilla.util.reflect.ReflectionUtil.closeQuietly;
 import static org.cobbzilla.util.reflect.ReflectionUtil.instantiate;
@@ -16,9 +16,9 @@ public class FilterInputStreamViaOutputStream extends PipedInputStream implement
 
     private static final int DEFAULT_PIPE_BUFFER_SIZE = (int) (64 * Bytes.KB);
     private static final int DEFAULT_COPY_BUFFER_SIZE = (int) (8 * Bytes.KB);
-    private static final long THREAD_TERMINATE_TIMEOUT = TimeUnit.SECONDS.toMillis(10);
+    private static final long THREAD_TERMINATE_TIMEOUT = SECONDS.toMillis(10);
 
-    private InputStream in;
+    private final InputStream in;
     private PipedOutputStream pipeOut;
     private OutputStream out;
     private Thread thread;
@@ -65,9 +65,11 @@ public class FilterInputStreamViaOutputStream extends PipedInputStream implement
             final byte[] buf = new byte[DEFAULT_COPY_BUFFER_SIZE];
             int bytesRead;
             while ((bytesRead = in.read(buf)) >= 0) {
-                out.write(buf, 0, bytesRead);
+                if (bytesRead > 0) {
+                    out.write(buf, 0, bytesRead);
+                    out.flush();
+                }
             }
-            out.flush();
 
         } catch (IOException e) {
             final String msg = "run: error copying bytes: " + shortError(e);
@@ -92,4 +94,5 @@ public class FilterInputStreamViaOutputStream extends PipedInputStream implement
             background(() -> terminate(this.thread, THREAD_TERMINATE_TIMEOUT));
         }
     }
+
 }

src/main/java/org/cobbzilla/util/io/regex/RegexFilterReader.java

@@ -21,13 +21,11 @@ public class RegexFilterReader extends BufferedReader {
     public static final int DEFAULT_BUFFER_SIZE = (int) (8 * Bytes.KB);
 
     private final int bufsiz;
-    private RegexStreamFilter filter;
+    private final RegexStreamFilter filter;
     @Getter @Setter private Integer maxMatches;
     @Getter @Setter private String name; // for debugging/identifying which reader
 
-    public RegexFilterReader(Reader in, RegexStreamFilter filter) {
-        this(in, DEFAULT_BUFFER_SIZE, filter);
-    }
+    public RegexFilterReader(Reader in, RegexStreamFilter filter) { this(in, DEFAULT_BUFFER_SIZE, filter); }
 
     public RegexFilterReader(Reader in, int bufsiz, RegexStreamFilter filter) {
         super(in, DEFAULT_BUFFER_SIZE);
@@ -35,9 +33,7 @@ public class RegexFilterReader extends BufferedReader {
         this.filter = filter;
     }
 
-    public RegexFilterReader(InputStream in, RegexStreamFilter filter) {
-        this(in, DEFAULT_BUFFER_SIZE, filter);
-    }
+    public RegexFilterReader(InputStream in, RegexStreamFilter filter) { this(in, DEFAULT_BUFFER_SIZE, filter); }
 
     public RegexFilterReader(InputStream in, int bufsiz, RegexStreamFilter filter) {
         super(new InputStreamReader(in, UTF8cs), bufsiz);
@@ -107,7 +103,7 @@ public class RegexFilterReader extends BufferedReader {
 
         private int readPos = 0;
         private boolean eof = false;
-        private RegexStreamFilter filter;
+        private final RegexStreamFilter filter;
         private int matchCount = 0;
         private Integer maxMatches = null;