Add JsonWriter.valueSink for streaming data

Unlike `value(BufferedSource)`, this retains the push-based nature of JsonWriter and allows layering other sinks (such as a base64-encoding Sink for binary data). Since `value(BufferedSource)` is trivially replaced by reading the source into the value sink, it is deprecated.

moshi/src/main/java/com/squareup/moshi/JsonScope.java

@@ -44,6 +44,9 @@ final class JsonScope {
   /** A document that's been closed and cannot be accessed. */
   static final int CLOSED = 8;
 
+  /** Sits above the actual state to indicate that a value is currently being streamed in. */
+  static final int STREAMING_VALUE = 9;
+
   /**
    * Renders the path in a JSON document to a string. The {@code pathNames} and {@code pathIndices}
    * parameters corresponds directly to stack: At indices where the stack contains an object

moshi/src/main/java/com/squareup/moshi/JsonUtf8Writer.java

@@ -17,9 +17,11 @@ package com.squareup.moshi;
 
 import java.io.IOException;
 import javax.annotation.Nullable;
+import okio.Buffer;
 import okio.BufferedSink;
-import okio.BufferedSource;
+import okio.Okio;
 import okio.Sink;
+import okio.Timeout;
 
 import static com.squareup.moshi.JsonScope.DANGLING_NAME;
 import static com.squareup.moshi.JsonScope.EMPTY_ARRAY;
@@ -28,6 +30,7 @@ import static com.squareup.moshi.JsonScope.EMPTY_OBJECT;
 import static com.squareup.moshi.JsonScope.NONEMPTY_ARRAY;
 import static com.squareup.moshi.JsonScope.NONEMPTY_DOCUMENT;
 import static com.squareup.moshi.JsonScope.NONEMPTY_OBJECT;
+import static com.squareup.moshi.JsonScope.STREAMING_VALUE;
 
 final class JsonUtf8Writer extends JsonWriter {
 
@@ -273,16 +276,35 @@ final class JsonUtf8Writer extends JsonWriter {
     return this;
   }
 
-  @Override public JsonWriter value(BufferedSource source) throws IOException {
+  @Override public BufferedSink valueSink() throws IOException {
     if (promoteValueToName) {
       throw new IllegalStateException(
-          "BufferedSource cannot be used as a map key in JSON at path " + getPath());
+          "BufferedSink cannot be used as a map key in JSON at path " + getPath());
     }
     writeDeferredName();
     beforeValue();
-    sink.writeAll(source);
-    pathIndices[stackSize - 1]++;
-    return this;
+    pushScope(STREAMING_VALUE);
+    return Okio.buffer(new Sink() {
+      @Override public void write(Buffer source, long byteCount) throws IOException {
+        sink.write(source, byteCount);
+      }
+
+      @Override public void close() {
+        if (peekScope() != STREAMING_VALUE) {
+          throw new AssertionError();
+        }
+        stackSize--; // Remove STREAMING_VALUE from the stack.
+        pathIndices[stackSize - 1]++;
+      }
+
+      @Override public void flush() throws IOException {
+        sink.flush();
+      }
+
+      @Override public Timeout timeout() {
+        return Timeout.NONE;
+      }
+    });
   }
 
   /**
@@ -380,6 +402,7 @@ final class JsonUtf8Writer extends JsonWriter {
    */
   @SuppressWarnings("fallthrough")
   private void beforeValue() throws IOException {
+    int nextTop;
     switch (peekScope()) {
       case NONEMPTY_DOCUMENT:
         if (!lenient) {
@@ -388,26 +411,28 @@ final class JsonUtf8Writer extends JsonWriter {
         }
         // fall-through
       case EMPTY_DOCUMENT: // first in document
-        replaceTop(NONEMPTY_DOCUMENT);
-        break;
-
-      case EMPTY_ARRAY: // first in array
-        replaceTop(NONEMPTY_ARRAY);
-        newline();
+        nextTop = NONEMPTY_DOCUMENT;
         break;
 
       case NONEMPTY_ARRAY: // another in array
         sink.writeByte(',');
+        // fall-through
+      case EMPTY_ARRAY: // first in array
         newline();
+        nextTop = NONEMPTY_ARRAY;
         break;
 
       case DANGLING_NAME: // value for name
+        nextTop = NONEMPTY_OBJECT;
         sink.writeUtf8(separator);
-        replaceTop(NONEMPTY_OBJECT);
         break;
 
+      case STREAMING_VALUE:
+        throw new IllegalStateException("Sink from valueSink() was not closed");
+
       default:
         throw new IllegalStateException("Nesting problem.");
     }
+    replaceTop(nextTop);
   }
 }

moshi/src/main/java/com/squareup/moshi/JsonValueWriter.java

@@ -21,12 +21,16 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import javax.annotation.Nullable;
-import okio.BufferedSource;
+import okio.Buffer;
+import okio.BufferedSink;
+import okio.ForwardingSink;
+import okio.Okio;
 
 import static com.squareup.moshi.JsonScope.EMPTY_ARRAY;
 import static com.squareup.moshi.JsonScope.EMPTY_DOCUMENT;
 import static com.squareup.moshi.JsonScope.EMPTY_OBJECT;
 import static com.squareup.moshi.JsonScope.NONEMPTY_DOCUMENT;
+import static com.squareup.moshi.JsonScope.STREAMING_VALUE;
 import static java.lang.Double.NEGATIVE_INFINITY;
 import static java.lang.Double.POSITIVE_INFINITY;
 
@@ -226,21 +230,35 @@ final class JsonValueWriter extends JsonWriter {
     return this;
   }
 
-  @Override public JsonWriter value(BufferedSource source) throws IOException {
+  @Override public BufferedSink valueSink() {
     if (promoteValueToName) {
       throw new IllegalStateException(
-          "BufferedSource cannot be used as a map key in JSON at path " + getPath());
+          "BufferedSink cannot be used as a map key in JSON at path " + getPath());
     }
-    Object value = JsonReader.of(source).readJsonValue();
-    boolean serializeNulls = this.serializeNulls;
-    this.serializeNulls = true;
-    try {
-      add(value);
-    } finally {
-      this.serializeNulls = serializeNulls;
+    if (peekScope() == STREAMING_VALUE) {
+      throw new IllegalStateException("Sink from valueSink() was not closed");
     }
-    pathIndices[stackSize - 1]++;
-    return this;
+    pushScope(STREAMING_VALUE);
+
+    final Buffer buffer = new Buffer();
+    return Okio.buffer(new ForwardingSink(buffer) {
+      @Override public void close() throws IOException {
+        if (peekScope() != STREAMING_VALUE || stack[stackSize] != null) {
+          throw new AssertionError();
+        }
+        stackSize--; // Remove STREAMING_VALUE from the stack.
+
+        Object value = JsonReader.of(buffer).readJsonValue();
+        boolean serializeNulls = JsonValueWriter.this.serializeNulls;
+        JsonValueWriter.this.serializeNulls = true;
+        try {
+          add(value);
+        } finally {
+          JsonValueWriter.this.serializeNulls = serializeNulls;
+        }
+        pathIndices[stackSize - 1]++;
+      }
+    });
   }
 
   @Override public void close() throws IOException {
@@ -284,6 +302,9 @@ final class JsonValueWriter extends JsonWriter {
       List<Object> list = (List<Object>) stack[stackSize - 1];
       list.add(newTop);
 
+    } else if (scope == STREAMING_VALUE) {
+      throw new IllegalStateException("Sink from valueSink() was not closed");
+
     } else {
       throw new IllegalStateException("Nesting problem.");
     }

moshi/src/main/java/com/squareup/moshi/JsonWriter.java

@@ -359,13 +359,32 @@ public abstract class JsonWriter implements Closeable, Flushable {
   public abstract JsonWriter value(@Nullable Number value) throws IOException;
 
   /**
-   * Writes {@code source} directly without encoding its contents.
-   * Since no validation is performed, {@link #setSerializeNulls} and other writer configurations
-   * are not respected.
+   * Writes {@code source} directly without encoding its contents. Equivalent to
+   * {@code try (BufferedSink sink = writer.valueSink()) { source.readAll(sink): }}
    *
-   * @return this writer.
+   * @see #valueSink()
    */
-  public abstract JsonWriter value(BufferedSource source) throws IOException;
+  public final JsonWriter value(BufferedSource source) throws IOException {
+    if (promoteValueToName) {
+      throw new IllegalStateException(
+          "BufferedSource cannot be used as a map key in JSON at path " + getPath());
+    }
+    try (BufferedSink sink = valueSink()) {
+      source.readAll(sink);
+    }
+    return this;
+  }
+
+  /**
+   * Returns a {@link BufferedSink} into which arbitrary data can be written without any additional
+   * encoding. You <b>must</b> call {@link BufferedSink#close()} before interacting with this
+   * {@code JsonWriter} instance again.
+   * <p>
+   * Since no validation is performed, options like {@link #setSerializeNulls} and other writer
+   * configurations are not respected.
+   */
+  @CheckReturnValue
+  public abstract BufferedSink valueSink() throws IOException;
 
   /**
    * Changes the writer to treat the next value as a string name. This is useful for map adapters so

moshi/src/test/java/com/squareup/moshi/JsonWriterTest.java

@@ -19,6 +19,7 @@ import java.io.IOException;
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.util.List;
+import okio.BufferedSink;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -616,4 +617,129 @@ public final class JsonWriterTest {
       assertThat(expected).hasMessage("Dangling name: a");
     }
   }
+
+  @Test public void streamingValueInObject() throws IOException {
+    JsonWriter writer = factory.newWriter();
+    writer.beginObject();
+    writer.name("a");
+    BufferedSink value = writer.valueSink();
+    value.writeByte('"');
+    value.writeHexadecimalUnsignedLong(-1L);
+    value.writeUtf8("sup");
+    value.writeDecimalLong(-1L);
+    value.writeByte('"');
+    value.close();
+    writer.endObject();
+    assertThat(factory.json()).isEqualTo("{\"a\":\"ffffffffffffffffsup-1\"}");
+  }
+
+  @Test public void streamingValueInArray() throws IOException {
+    JsonWriter writer = factory.newWriter();
+    writer.beginArray();
+    writer.valueSink()
+        .writeByte('"')
+        .writeHexadecimalUnsignedLong(-1L)
+        .writeByte('"')
+        .close();
+    writer.valueSink()
+        .writeByte('"')
+        .writeUtf8("sup")
+        .writeByte('"')
+        .close();
+    writer.valueSink()
+        .writeUtf8("-1.0")
+        .close();
+    writer.endArray();
+    assertThat(factory.json()).isEqualTo("[\"ffffffffffffffff\",\"sup\",-1.0]");
+  }
+
+  @Test public void streamingValueTopLevel() throws IOException {
+    JsonWriter writer = factory.newWriter();
+    writer.valueSink()
+        .writeUtf8("-1.0")
+        .close();
+    assertThat(factory.json()).isEqualTo("-1.0");
+  }
+
+  @Test public void streamingValueTwiceBeforeCloseFails() throws IOException {
+    JsonWriter writer = factory.newWriter();
+    writer.beginObject();
+    writer.name("a");
+    BufferedSink sink = writer.valueSink();
+    try {
+      writer.valueSink();
+      fail();
+    } catch (IllegalStateException e) {
+      assertThat(e).hasMessage("Sink from valueSink() was not closed");
+    }
+  }
+
+  @Test public void streamingValueTwiceAfterCloseFails() throws IOException {
+    JsonWriter writer = factory.newWriter();
+    writer.beginObject();
+    writer.name("a");
+    writer.valueSink().writeByte('0').close();
+    try {
+      // TODO currently UTF-8 fails eagerly on valueSink() but value does not fail until close().
+      writer.valueSink().writeByte('0').close();
+      fail();
+    } catch (IllegalStateException e) {
+      assertThat(e).hasMessage("Nesting problem.");
+    }
+  }
+
+  @Test public void streamingValueAndScalarValueFails() throws IOException {
+    JsonWriter writer = factory.newWriter();
+    writer.beginObject();
+    writer.name("a");
+    BufferedSink sink = writer.valueSink();
+    try {
+      writer.value("b");
+      fail();
+    } catch (IllegalStateException e) {
+      assertThat(e).hasMessage("Sink from valueSink() was not closed");
+    }
+  }
+
+  @Test public void streamingValueAndNameFails() throws IOException {
+    JsonWriter writer = factory.newWriter();
+    writer.beginObject();
+    writer.name("a");
+    BufferedSink sink = writer.valueSink();
+    try {
+      writer.name("b");
+      fail();
+    } catch (IllegalStateException e) {
+      assertThat(e).hasMessage("Nesting problem.");
+    }
+  }
+
+  @Test public void streamingValueInteractionAfterCloseFails() throws IOException {
+    JsonWriter writer = factory.newWriter();
+    writer.beginObject();
+    writer.name("a");
+    BufferedSink sink = writer.valueSink();
+    sink.writeUtf8("1.0");
+    sink.close();
+    try {
+      sink.writeByte('1');
+      fail();
+    } catch (IllegalStateException e) {
+      assertThat(e).hasMessage("closed");
+    }
+  }
+
+  @Test public void streamingValueCloseIsIdempotent() throws IOException {
+    JsonWriter writer = factory.newWriter();
+    writer.beginObject();
+    writer.name("a");
+    BufferedSink sink = writer.valueSink();
+    sink.writeUtf8("1.0");
+    sink.close();
+    sink.close();
+    writer.endObject();
+    sink.close();
+    assertThat(factory.json()).isEqualTo("{\"a\":1.0}");
+    sink.close();
+  }
 }

moshi/src/test/java/com/squareup/moshi/PromoteNameToValueTest.java

@@ -351,4 +351,21 @@ public final class PromoteNameToValueTest {
     writer.endObject();
     assertThat(factory.json()).isEqualTo("{\"a\":\"a value\"}");
   }
+
+  @Test public void writerValueSinkFails() throws Exception {
+    JsonWriter writer = factory.newWriter();
+    writer.beginObject();
+    writer.promoteValueToName();
+    try {
+      writer.valueSink();
+      fail();
+    } catch (IllegalStateException expected) {
+      assertThat(expected).hasMessage(
+          "BufferedSink cannot be used as a map key in JSON at path $.");
+    }
+    writer.value("a");
+    writer.value("a value");
+    writer.endObject();
+    assertThat(factory.json()).isEqualTo("{\"a\":\"a value\"}");
+  }
 }