Add native support for Protobuf Lite

Lite already worked by using the protobuf project, but would bring in
extra dependencies that are not intended to work with lite. Although
protobuf is not yet providing a lite package on Maven Central, we will
be able to swap to it once it is available.

There isn't any new original code in the Java portion, except for a new
overload in ProtoUtils that accepts Message instead of MessageLite.
Depending on Message in ProtoUtils allows us to support extra features
out-of-the-box without any changes to the generated code. For example,
JSON encoding could be supported in this way if Marshaller is enhanced.

However, now codegen must be aware of Lite in order to choose with Util
class to use. That is new code.

4 files changed, 509 insertions, 0 deletions

diff --git a/protobuf-lite/build.gradle b/protobuf-lite/build.gradle
new file mode 100644
index 000000000..1425a3bb8
--- /dev/null
+++ b/protobuf-lite/build.gradle
@@ -0,0 +1,15 @@
+plugins {
+    id "be.insaneprogramming.gradle.animalsniffer" version "1.4.0"
+}
+
+description = 'gRPC: Protobuf Lite'
+
+dependencies {
+    compile project(':grpc-core'),
+            libraries.protobuf,
+            libraries.guava
+}
+
+animalsniffer {
+    signature = "org.codehaus.mojo.signature:java16:+@signature"
+}
diff --git a/protobuf-lite/src/main/java/io/grpc/protobuf/lite/ProtoInputStream.java b/protobuf-lite/src/main/java/io/grpc/protobuf/lite/ProtoInputStream.java
new file mode 100644
index 000000000..b2e4282f5
--- /dev/null
+++ b/protobuf-lite/src/main/java/io/grpc/protobuf/lite/ProtoInputStream.java
@@ -0,0 +1,144 @@
+/*
+ * Copyright 2014, Google Inc. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *    * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *    * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *
+ *    * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package io.grpc.protobuf.lite;
+
+import com.google.common.io.ByteStreams;
+import com.google.protobuf.CodedOutputStream;
+import com.google.protobuf.MessageLite;
+import com.google.protobuf.Parser;
+
+import io.grpc.Drainable;
+import io.grpc.KnownLength;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import javax.annotation.Nullable;
+
+/**
+ * An {@link InputStream} backed by a protobuf.
+ */
+class ProtoInputStream extends InputStream implements Drainable, KnownLength {
+
+  // ProtoInputStream is first initialized with a *message*. *partial* is initially null.
+  // Once there has been a read operation on this stream, *message* is serialized to *partial* and
+  // set to null.
+  @Nullable private MessageLite message;
+  private final Parser<?> parser;
+  @Nullable private ByteArrayInputStream partial;
+
+  public ProtoInputStream(MessageLite message, Parser<?> parser) {
+    this.message = message;
+    this.parser = parser;
+  }
+
+  @Override
+  public int drainTo(OutputStream target) throws IOException {
+    int written;
+    if (message != null) {
+      written = message.getSerializedSize();
+      message.writeTo(target);
+      message = null;
+    } else if (partial != null) {
+      written = (int) ByteStreams.copy(partial, target);
+      partial = null;
+    } else {
+      written = 0;
+    }
+    return written;
+  }
+
+  @Override
+  public int read() throws IOException {
+    if (message != null) {
+      partial = new ByteArrayInputStream(message.toByteArray());
+      message = null;
+    }
+    if (partial != null) {
+      return partial.read();
+    }
+    return -1;
+  }
+
+  @Override
+  public int read(byte[] b, int off, int len) throws IOException {
+    if (message != null) {
+      int size = message.getSerializedSize();
+      if (size == 0) {
+        message = null;
+        partial = null;
+        return -1;
+      }
+      if (len >= size) {
+        // This is the only case that is zero-copy.
+        CodedOutputStream stream = CodedOutputStream.newInstance(b, off, size);
+        message.writeTo(stream);
+        stream.flush();
+        stream.checkNoSpaceLeft();
+
+        message = null;
+        partial = null;
+        return size;
+      }
+
+      partial = new ByteArrayInputStream(message.toByteArray());
+      message = null;
+    }
+    if (partial != null) {
+      return partial.read(b, off, len);
+    }
+    return -1;
+  }
+
+  @Override
+  public int available() throws IOException {
+    if (message != null) {
+      return message.getSerializedSize();
+    } else if (partial != null) {
+      return partial.available();
+    }
+    return 0;
+  }
+
+  MessageLite message() {
+    if (message == null) {
+      throw new IllegalStateException("message not available");
+    }
+    return message;
+  }
+
+  Parser<?> parser() {
+    return parser;
+  }
+}
diff --git a/protobuf-lite/src/main/java/io/grpc/protobuf/lite/ProtoLiteUtils.java b/protobuf-lite/src/main/java/io/grpc/protobuf/lite/ProtoLiteUtils.java
new file mode 100644
index 000000000..c5677ff5e
--- /dev/null
+++ b/protobuf-lite/src/main/java/io/grpc/protobuf/lite/ProtoLiteUtils.java
@@ -0,0 +1,136 @@
+/*
+ * Copyright 2014, Google Inc. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *    * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *    * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *
+ *    * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package io.grpc.protobuf.lite;
+
+import com.google.protobuf.CodedInputStream;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.MessageLite;
+import com.google.protobuf.Parser;
+
+import io.grpc.ExperimentalApi;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor.Marshaller;
+import io.grpc.Status;
+
+import java.io.InputStream;
+
+/**
+ * Utility methods for using protobuf with grpc.
+ */
+@ExperimentalApi("Experimental until Lite is stable in protobuf")
+public class ProtoLiteUtils {
+
+  /** Create a {@code Marshaller} for protos of the same type as {@code defaultInstance}. */
+  public static <T extends MessageLite> Marshaller<T> marshaller(final T defaultInstance) {
+    @SuppressWarnings("unchecked")
+    final Parser<T> parser = (Parser<T>) defaultInstance.getParserForType();
+    return new Marshaller<T>() {
+      @Override
+      public InputStream stream(T value) {
+        return new ProtoInputStream(value, parser);
+      }
+
+      @Override
+      public T parse(InputStream stream) {
+        if (stream instanceof ProtoInputStream) {
+          ProtoInputStream protoStream = (ProtoInputStream) stream;
+          // Optimization for in-memory transport. Returning provided object is safe since protobufs
+          // are immutable.
+          //
+          // However, we can't assume the types match, so we have to verify the parser matches.
+          // Today the parser is always the same for a given proto, but that isn't guaranteed. Even
+          // if not, using the same MethodDescriptor would ensure the parser matches and permit us
+          // to enable this optimization.
+          if (protoStream.parser() == parser) {
+            try {
+              @SuppressWarnings("unchecked")
+              T message = (T) ((ProtoInputStream) stream).message();
+              return message;
+            } catch (IllegalStateException ex) {
+              // Stream must have been read from, which is a strange state. Since the point of this
+              // optimization is to be transparent, instead of throwing an error we'll continue,
+              // even though it seems likely there's a bug.
+            }
+          }
+        }
+        try {
+          return parseFrom(stream);
+        } catch (InvalidProtocolBufferException ipbe) {
+          throw Status.INTERNAL.withDescription("Invalid protobuf byte sequence")
+            .withCause(ipbe).asRuntimeException();
+        }
+      }
+
+      private T parseFrom(InputStream stream) throws InvalidProtocolBufferException {
+        // Pre-create the CodedInputStream so that we can remove the size limit restriction
+        // when parsing.
+        CodedInputStream codedInput = CodedInputStream.newInstance(stream);
+        codedInput.setSizeLimit(Integer.MAX_VALUE);
+
+        T message = parser.parseFrom(codedInput);
+        try {
+          codedInput.checkLastTagWas(0);
+          return message;
+        } catch (InvalidProtocolBufferException e) {
+          e.setUnfinishedMessage(message);
+          throw e;
+        }
+      }
+    };
+  }
+
+  /**
+   * Produce a metadata marshaller for a protobuf type.
+   */
+  public static <T extends MessageLite> Metadata.BinaryMarshaller<T> metadataMarshaller(
+      final T instance) {
+    return new Metadata.BinaryMarshaller<T>() {
+      @Override
+      public byte[] toBytes(T value) {
+        return value.toByteArray();
+      }
+
+      @Override
+      @SuppressWarnings("unchecked")
+      public T parseBytes(byte[] serialized) {
+        try {
+          return (T) instance.getParserForType().parseFrom(serialized);
+        } catch (InvalidProtocolBufferException ipbe) {
+          throw new IllegalArgumentException(ipbe);
+        }
+      }
+    };
+  }
+
+  private ProtoLiteUtils() {
+  }
+}
diff --git a/protobuf-lite/src/test/java/io/grpc/protobuf/lite/ProtoLiteUtilsTest.java b/protobuf-lite/src/test/java/io/grpc/protobuf/lite/ProtoLiteUtilsTest.java
new file mode 100644
index 000000000..a5042a435
--- /dev/null
+++ b/protobuf-lite/src/test/java/io/grpc/protobuf/lite/ProtoLiteUtilsTest.java
@@ -0,0 +1,214 @@
+/*
+ * Copyright 2015, Google Inc. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *    * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *    * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *
+ *    * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package io.grpc.protobuf.lite;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.fail;
+
+import com.google.common.io.ByteStreams;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.Empty;
+import com.google.protobuf.Enum;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Type;
+
+import io.grpc.Drainable;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor.Marshaller;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
+
+/** Unit tests for {@link ProtoLiteUtils}. */
+@RunWith(JUnit4.class)
+public class ProtoLiteUtilsTest {
+  private Marshaller<Type> marshaller = ProtoLiteUtils.marshaller(Type.getDefaultInstance());
+  private Type proto = Type.newBuilder().setName("name").build();
+
+  @Test
+  public void testPassthrough() {
+    assertSame(proto, marshaller.parse(marshaller.stream(proto)));
+  }
+
+  @Test
+  public void testRoundtrip() throws Exception {
+    InputStream is = marshaller.stream(proto);
+    is = new ByteArrayInputStream(ByteStreams.toByteArray(is));
+    assertEquals(proto, marshaller.parse(is));
+  }
+
+  @Test
+  public void testInvalidatedMessage() throws Exception {
+    InputStream is = marshaller.stream(proto);
+    // Invalidates message, and drains all bytes
+    ByteStreams.toByteArray(is);
+    try {
+      ((ProtoInputStream) is).message();
+      fail("Expected exception");
+    } catch (IllegalStateException ex) {
+      // expected
+    }
+    // Zero bytes is the default message
+    assertEquals(Type.getDefaultInstance(), marshaller.parse(is));
+  }
+
+  @Test
+  public void parseInvalid() throws Exception {
+    InputStream is = new ByteArrayInputStream(new byte[] {-127});
+    try {
+      marshaller.parse(is);
+      fail("Expected exception");
+    } catch (StatusRuntimeException ex) {
+      assertEquals(Status.Code.INTERNAL, ex.getStatus().getCode());
+      assertNotNull(((InvalidProtocolBufferException) ex.getCause()).getUnfinishedMessage());
+    }
+  }
+
+  @Test
+  public void testMismatch() throws Exception {
+    Marshaller<Enum> enumMarshaller = ProtoLiteUtils.marshaller(Enum.getDefaultInstance());
+    // Enum's name and Type's name are both strings with tag 1.
+    Enum altProto = Enum.newBuilder().setName(proto.getName()).build();
+    assertEquals(proto, marshaller.parse(enumMarshaller.stream(altProto)));
+  }
+
+  @Test
+  public void marshallerShouldNotLimitProtoSize() throws Exception {
+    // The default limit is 64MB. Using a larger proto to verify that the limit is not enforced.
+    byte[] bigName = new byte[70 * 1024 * 1024];
+    Arrays.fill(bigName, (byte) 32);
+
+    proto = Type.newBuilder().setNameBytes(ByteString.copyFrom(bigName)).build();
+
+    // Just perform a round trip to verify that it works.
+    testRoundtrip();
+  }
+
+  @Test
+  public void testAvailable() throws Exception {
+    InputStream is = marshaller.stream(proto);
+    assertEquals(proto.getSerializedSize(), is.available());
+    is.read();
+    assertEquals(proto.getSerializedSize() - 1, is.available());
+    while (is.read() != -1) {}
+    assertEquals(-1, is.read());
+    assertEquals(0, is.available());
+  }
+
+  @Test
+  public void testEmpty() throws IOException {
+    Marshaller<Empty> marshaller = ProtoLiteUtils.marshaller(Empty.getDefaultInstance());
+    InputStream is = marshaller.stream(Empty.getDefaultInstance());
+    assertEquals(0, is.available());
+    byte[] b = new byte[10];
+    assertEquals(-1, is.read(b));
+    assertArrayEquals(new byte[10], b);
+    // Do the same thing again, because the internal state may be different
+    assertEquals(-1, is.read(b));
+    assertArrayEquals(new byte[10], b);
+    assertEquals(-1, is.read());
+    assertEquals(0, is.available());
+  }
+
+  @Test
+  public void testDrainTo_all() throws Exception {
+    byte[] golden = ByteStreams.toByteArray(marshaller.stream(proto));
+    InputStream is = marshaller.stream(proto);
+    Drainable d = (Drainable) is;
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    int drained = d.drainTo(baos);
+    assertEquals(baos.size(), drained);
+    assertArrayEquals(golden, baos.toByteArray());
+    assertEquals(0, is.available());
+  }
+
+  @Test
+  public void testDrainTo_partial() throws Exception {
+    final byte[] golden;
+    {
+      InputStream is = marshaller.stream(proto);
+      is.read();
+      golden = ByteStreams.toByteArray(is);
+    }
+    InputStream is = marshaller.stream(proto);
+    is.read();
+    Drainable d = (Drainable) is;
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    int drained = d.drainTo(baos);
+    assertEquals(baos.size(), drained);
+    assertArrayEquals(golden, baos.toByteArray());
+    assertEquals(0, is.available());
+  }
+
+  @Test
+  public void testDrainTo_none() throws Exception {
+    byte[] golden = ByteStreams.toByteArray(marshaller.stream(proto));
+    InputStream is = marshaller.stream(proto);
+    ByteStreams.toByteArray(is);
+    Drainable d = (Drainable) is;
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    assertEquals(0, d.drainTo(baos));
+    assertArrayEquals(new byte[0], baos.toByteArray());
+    assertEquals(0, is.available());
+  }
+
+  @Test
+  public void metadataMarshaller_roundtrip() {
+    Metadata.BinaryMarshaller<Type> metadataMarshaller =
+        ProtoLiteUtils.metadataMarshaller(Type.getDefaultInstance());
+    assertEquals(proto, metadataMarshaller.parseBytes(metadataMarshaller.toBytes(proto)));
+  }
+
+  @Test
+  public void metadataMarshaller_invalid() {
+    Metadata.BinaryMarshaller<Type> metadataMarshaller =
+        ProtoLiteUtils.metadataMarshaller(Type.getDefaultInstance());
+    try {
+      metadataMarshaller.parseBytes(new byte[] {-127});
+      fail("Expected exception");
+    } catch (IllegalArgumentException ex) {
+      assertNotNull(((InvalidProtocolBufferException) ex.getCause()).getUnfinishedMessage());
+    }
+  }
+}