CryptoStreamFactory creates sympathetic chunking OutputStreams

For detailed analysis of the problem, see
#586

CipherOutputStream appears to perform _very_ poorly on large
buffers, but substantially better when data is segmented into
small chunks which can be done by looping over the original buffer.

With this in place, the openssl wrappeer no longer provides any
benefit over JCE.

crypto-core/src/main/java/com/palantir/crypto2/io/CryptoStreamFactory.java

@@ -22,6 +22,7 @@
 import com.palantir.crypto2.cipher.SeekableCipherFactory;
 import com.palantir.crypto2.keys.KeyMaterial;
 import com.palantir.seekio.SeekableInput;
+import java.io.FilterOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -120,7 +121,7 @@ private static OutputStream createApacheEncryptedStream(OutputStream output, Key
     private static OutputStream createDefaultEncryptedStream(OutputStream output, KeyMaterial keyMaterial,
             String algorithm) {
         SeekableCipher cipher = SeekableCipherFactory.getCipher(algorithm, keyMaterial);
-        return new CipherOutputStream(output, cipher.initCipher(Cipher.ENCRYPT_MODE));
+        return new ChunkingOutputStream(new CipherOutputStream(output, cipher.initCipher(Cipher.ENCRYPT_MODE)));
     }
 
     private static class StreamSeekableInput implements SeekableInput {
@@ -150,4 +151,44 @@ public void close() throws IOException {
             input.close();
         }
     }
+
+    /**
+     * {@link ChunkingOutputStream} limits the size of individual writes to the wrapped {@link OutputStream}
+     * in order to prevent degraded performance on large buffers as described in
+     * <a href="https://github.com/palantir/hadoop-crypto/pull/586">hadoop-crypto#586</a>.
+     */
+    static final class ChunkingOutputStream extends FilterOutputStream {
+
+        private static final int CHUNK_SIZE = 16 * 1024;
+
+        ChunkingOutputStream(OutputStream delegate) {
+            super(delegate);
+        }
+
+        @Override
+        public void write(byte[] buffer, int off, int len) throws IOException {
+            validateArgs(buffer, off, len);
+            doWrite(buffer, off, len);
+        }
+
+        private void doWrite(byte[] buffer, int off, int len) throws IOException {
+            int currentOffset = off;
+            int remaining = len;
+            while (remaining > 0) {
+                int toWrite = Math.min(remaining, CHUNK_SIZE);
+                out.write(buffer, currentOffset, toWrite);
+                currentOffset += toWrite;
+                remaining -= toWrite;
+            }
+        }
+
+        private static void validateArgs(byte[] buffer, int off, int len) {
+            if (buffer == null) {
+                throw new NullPointerException("buffer is required");
+            }
+            if (off < 0 || len < 0 || buffer.length < off + len) {
+                throw new IndexOutOfBoundsException();
+            }
+        }
+    }
 }

crypto-core/src/test/java/com/palantir/crypto2/io/CryptoStreamFactoryTest.java

@@ -29,6 +29,7 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
+import java.util.concurrent.ThreadLocalRandom;
 import org.apache.commons.crypto.stream.CtrCryptoInputStream;
 import org.apache.commons.crypto.stream.CtrCryptoOutputStream;
 import org.junit.Before;
@@ -85,4 +86,23 @@ public void testEncryptDecryptJce() throws IOException {
         assertThat(bytesRead).isEqualTo(BYTES.length);
         assertThat(readBytes).isEqualTo(BYTES);
     }
+
+    @Test
+    public void testChunkingOutputStream() throws IOException {
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        byte[] data = new byte[100 * 1024 * 1024];
+        ThreadLocalRandom.current().nextBytes(data);
+        int chunkSize = 1;
+        int dataOffset = 0;
+        try (OutputStream out = new CryptoStreamFactory.ChunkingOutputStream(baos)) {
+            while (data.length - dataOffset > 0) {
+                int remaining = data.length - dataOffset;
+                int toWrite = Math.min(chunkSize, remaining);
+                out.write(data, dataOffset, toWrite);
+                dataOffset += toWrite;
+                chunkSize += ThreadLocalRandom.current().nextInt(1024);
+            }
+        }
+        assertThat(baos.toByteArray()).isEqualTo(data);
+    }
 }