Merge pull request #538 from square/jwilson_0219_deflatersink

DeflaterSink.

okhttp-protocols/pom.xml

@@ -23,5 +23,11 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.codehaus.mojo</groupId>
+      <artifactId>animal-sniffer-annotations</artifactId>
+      <version>1.10</version>
+      <optional>true</optional>
+    </dependency>
   </dependencies>
 </project>

okhttp-protocols/src/main/java/com/squareup/okhttp/internal/bytes/BufferedSink.java

@@ -73,7 +73,7 @@ public final class BufferedSink implements Sink {
     emitCompleteSegments(deadline);
   }
 
-  private void emitCompleteSegments(Deadline deadline) throws IOException {
+  void emitCompleteSegments(Deadline deadline) throws IOException {
     long byteCount = buffer.byteCount;
     if (byteCount == 0) return;
 

okhttp-protocols/src/main/java/com/squareup/okhttp/internal/bytes/DeflaterSink.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright (C) 2014 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.squareup.okhttp.internal.bytes;
+
+import java.io.IOException;
+import java.util.zip.Deflater;
+import org.codehaus.mojo.animal_sniffer.IgnoreJRERequirement;
+
+import static com.squareup.okhttp.internal.Util.checkOffsetAndCount;
+
+/**
+ * A sink that uses <a href="http://tools.ietf.org/html/rfc1951">DEFLATE</a> to
+ * compress data written to another source.
+ *
+ * <h3>Sync flush</h3>
+ * Aggressive flushing of this stream may result in reduced compression. Each
+ * call to {@link #flush} immediately compresses all currently-buffered data;
+ * this early compression may be less effective than compression performed
+ * without flushing.
+ *
+ * <p>This is equivalent to using {@link Deflater} with the sync flush option.
+ * This class does not offer any partial flush mechanism. For best performance,
+ * only call {@link #flush} when application behavior requires it.
+ */
+public final class DeflaterSink implements Sink {
+  private final BufferedSink sink;
+  private final Deflater deflater;
+
+  public DeflaterSink(Sink sink, Deflater deflater) {
+    this.sink = new BufferedSink(sink);
+    this.deflater = deflater;
+  }
+
+  @Override public void write(OkBuffer source, long byteCount, Deadline deadline)
+      throws IOException {
+    checkOffsetAndCount(source.byteCount, 0, byteCount);
+    while (byteCount > 0) {
+      // Share bytes from the head segment of 'source' with the deflater.
+      Segment head = source.head;
+      int toDeflate = (int) Math.min(byteCount, head.limit - head.pos);
+      deflater.setInput(head.data, head.pos, toDeflate);
+
+      // Deflate those bytes into sink.
+      deflate(deadline, false);
+
+      // Mark those bytes as read.
+      source.byteCount -= toDeflate;
+      head.pos += toDeflate;
+      if (head.pos == head.limit) {
+        source.head = head.pop();
+        SegmentPool.INSTANCE.recycle(head);
+      }
+
+      byteCount -= toDeflate;
+    }
+  }
+
+  @IgnoreJRERequirement
+  private void deflate(Deadline deadline, boolean syncFlush) throws IOException {
+    while (true) {
+      Segment s = sink.buffer.writableSegment(1);
+
+      // The 4-parameter overload of deflate() doesn't exist in the RI until
+      // Java 1.7, and is public (although with @hide) on Android since 2.3.
+      // The @hide tag means that this code won't compile against the Android
+      // 2.3 SDK, but it will run fine there.
+      int deflated = syncFlush
+          ? deflater.deflate(s.data, s.limit, Segment.SIZE - s.limit, Deflater.SYNC_FLUSH)
+          : deflater.deflate(s.data, s.limit, Segment.SIZE - s.limit);
+
+      if (deflated == 0) return;
+      s.limit += deflated;
+      sink.buffer.byteCount += deflated;
+      sink.emitCompleteSegments(deadline);
+    }
+  }
+
+  @Override public void flush(Deadline deadline) throws IOException {
+    deflate(deadline, true);
+    sink.flush(deadline);
+  }
+
+  @Override public void close(Deadline deadline) throws IOException {
+    deflater.finish();
+    deflate(deadline, false);
+    sink.close(deadline);
+  }
+}

okhttp-protocols/src/main/java/com/squareup/okhttp/internal/bytes/InflaterSource.java

@@ -20,7 +20,10 @@ import java.io.IOException;
 import java.util.zip.DataFormatException;
 import java.util.zip.Inflater;
 
-/** A source that inflates another source. */
+/**
+ * A source that uses <a href="http://tools.ietf.org/html/rfc1951">DEFLATE</a>
+ * to decompress data read from another source.
+ */
 public final class InflaterSource implements Source {
   private final BufferedSource source;
   private final Inflater inflater;

okhttp-protocols/src/test/java/com/squareup/okhttp/internal/bytes/DeflaterSinkTest.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright (C) 2014 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.squareup.okhttp.internal.bytes;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
+import java.util.Random;
+import java.util.zip.Deflater;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public final class DeflaterSinkTest {
+  @Test public void deflateWithClose() throws Exception {
+    OkBuffer data = new OkBuffer();
+    String original = "They're moving in herds. They do move in herds.";
+    data.writeUtf8(original);
+    OkBuffer sink = new OkBuffer();
+    DeflaterSink deflaterSink = new DeflaterSink(sink, new Deflater());
+    deflaterSink.write(data, data.byteCount(), Deadline.NONE);
+    deflaterSink.close(Deadline.NONE);
+    OkBuffer inflated = inflate(sink);
+    assertEquals(original, inflated.readUtf8((int) inflated.byteCount()));
+  }
+
+  @Test public void deflateWithSyncFlush() throws Exception {
+    String original = "Yes, yes, yes. That's why we're taking extreme precautions.";
+    OkBuffer data = new OkBuffer();
+    data.writeUtf8(original);
+    OkBuffer sink = new OkBuffer();
+    DeflaterSink deflaterSink = new DeflaterSink(sink, new Deflater());
+    deflaterSink.write(data, data.byteCount(), Deadline.NONE);
+    deflaterSink.flush(Deadline.NONE);
+    OkBuffer inflated = inflate(sink);
+    assertEquals(original, inflated.readUtf8((int) inflated.byteCount()));
+  }
+
+  @Test public void deflateWellCompressed() throws IOException {
+    String original = repeat('a', 1024 * 1024);
+    OkBuffer data = new OkBuffer();
+    data.writeUtf8(original);
+    OkBuffer sink = new OkBuffer();
+    DeflaterSink deflaterSink = new DeflaterSink(sink, new Deflater());
+    deflaterSink.write(data, data.byteCount(), Deadline.NONE);
+    deflaterSink.close(Deadline.NONE);
+    OkBuffer inflated = inflate(sink);
+    assertEquals(original, inflated.readUtf8((int) inflated.byteCount()));
+  }
+
+  @Test public void deflatePoorlyCompressed() throws IOException {
+    ByteString original = randomBytes(1024 * 1024);
+    OkBuffer data = new OkBuffer();
+    data.write(original);
+    OkBuffer sink = new OkBuffer();
+    DeflaterSink deflaterSink = new DeflaterSink(sink, new Deflater());
+    deflaterSink.write(data, data.byteCount(), Deadline.NONE);
+    deflaterSink.close(Deadline.NONE);
+    OkBuffer inflated = inflate(sink);
+    assertEquals(original, inflated.readByteString((int) inflated.byteCount()));
+  }
+
+  /**
+   * Uses streaming decompression to inflate {@code deflated}. The input must
+   * either be finished or have a trailing sync flush.
+   */
+  private OkBuffer inflate(OkBuffer deflated) throws IOException {
+    InputStream deflatedIn = new BufferedSource(deflated).inputStream();
+    Inflater inflater = new Inflater();
+    InputStream inflatedIn = new InflaterInputStream(deflatedIn, inflater);
+    OkBuffer result = new OkBuffer();
+    byte[] buffer = new byte[8192];
+    while (!inflater.needsInput() || deflated.byteCount() > 0 || deflatedIn.available() > 0) {
+      int count = inflatedIn.read(buffer, 0, buffer.length);
+      result.write(buffer, 0, count);
+    }
+    return result;
+  }
+
+  private ByteString randomBytes(int length) {
+    Random random = new Random(0);
+    byte[] randomBytes = new byte[length];
+    random.nextBytes(randomBytes);
+    return ByteString.of(randomBytes);
+  }
+
+  private String repeat(char c, int count) {
+    char[] array = new char[count];
+    Arrays.fill(array, c);
+    return new String(array);
+  }
+}

okhttp-protocols/src/test/java/com/squareup/okhttp/internal/bytes/InflaterSourceTest.java

@@ -26,7 +26,7 @@ import org.junit.Test;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 
-public class InflaterSourceTest {
+public final class InflaterSourceTest {
   @Test public void inflate() throws Exception {
     OkBuffer deflated = decodeBase64("eJxzz09RyEjNKVAoLdZRKE9VL0pVyMxTKMlIVchIzEspVshPU0jNS8/MS00tK"
         + "tYDAF6CD5s=");

pom.xml

@@ -171,7 +171,7 @@
       <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>animal-sniffer-maven-plugin</artifactId>
-        <version>1.9</version>
+        <version>1.10</version>
         <executions>
           <execution>
             <phase>test</phase>