Optimize protobuf serializer for in-memory transport

In-memory transport provides back the same input stream that was
provided, so if we notice our own object simply avoid serializing.

DeferredProtoInputStream is made package-private because 1) it doesn't
seem we need it to be public and 2) the change depends on it being
package-private so the constructor can be changed.

protobuf/src/main/java/io/grpc/protobuf/ProtoInputStream.java

@@ -34,6 +34,7 @@ package io.grpc.protobuf;
 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;
@@ -48,16 +49,18 @@ import javax.annotation.Nullable;
 /**
  * An {@link InputStream} backed by a protobuf.
  */
-public class ProtoInputStream extends InputStream implements Drainable, KnownLength {
+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) {
+  public ProtoInputStream(MessageLite message, Parser<?> parser) {
     this.message = message;
+    this.parser = parser;
   }
 
   @Override
@@ -125,4 +128,15 @@ public class ProtoInputStream extends InputStream implements Drainable, KnownLen
     }
     return 0;
   }
+
+  MessageLite message() {
+    if (message == null) {
+      throw new IllegalStateException("message not available");
+    }
+    return message;
+  }
+
+  Parser<?> parser() {
+    return parser;
+  }
 }

protobuf/src/main/java/io/grpc/protobuf/ProtoUtils.java

@@ -52,11 +52,26 @@ public class ProtoUtils {
     return new Marshaller<T>() {
       @Override
       public InputStream stream(T value) {
-        return new ProtoInputStream(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) {
+            @SuppressWarnings("unchecked")
+            T message = (T) ((ProtoInputStream) stream).message();
+            return message;
+          }
+        }
         try {
           return parser.parseFrom(stream);
         } catch (InvalidProtocolBufferException ipbe) {

protobuf/src/test/java/io/grpc/protobuf/ProtoUtilsTest.java

@@ -0,0 +1,75 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+
+import com.google.common.io.ByteStreams;
+import com.google.protobuf.Enum;
+import com.google.protobuf.Type;
+
+import io.grpc.Marshaller;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+
+/** Unit tests for {@link ProtoUtils}. */
+@RunWith(JUnit4.class)
+public class ProtoUtilsTest {
+  private Marshaller<Type> marshaller = ProtoUtils.marshaller(Type.parser());
+  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 testMismatch() throws Exception {
+    Marshaller<Enum> enumMarshaller = ProtoUtils.marshaller(Enum.parser());
+    // 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)));
+  }
+}