Add getByteBufferSupported method for HasByteBuffer so that it can be…

… used for okhttp as well.

api/src/main/java/io/grpc/HasByteBuffer.java

@@ -31,13 +31,21 @@
 // TODO(chengyuanzhang): add ExperimentalApi annotation.
 public interface HasByteBuffer {
 
+  /**
+   * Indicates whether or not {@link #getByteBuffer} operation is supported.
+   */
+  boolean getByteBufferSupported();
+
   /**
    * Gets a {@link ByteBuffer} containing some bytes of the content next to be read, or {@code
    * null} if has reached end of the content. The number of bytes contained in the returned buffer
    * is implementation specific. Calling this method does not change the position of the input
    * stream. The returned buffer's content should not be modified, but the position, limit, and
    * mark may be changed. Operations for changing the position, limit, and mark of the returned
-   * buffer does not affect the position, limit, and mark of this input stream.
+   * buffer does not affect the position, limit, and mark of this input stream. This is an optional
+   * method, so callers should first check {@link #getByteBufferSupported}.
+   *
+   * @throws UnsupportedOperationException if this operation is not supported.
    */
   @Nullable
   ByteBuffer getByteBuffer();

core/src/main/java/io/grpc/internal/ReadableBuffers.java

@@ -105,11 +105,7 @@ public static String readAsStringUtf8(ReadableBuffer buffer) {
    * @param owner if {@code true}, the returned stream will close the buffer when closed.
    */
   public static InputStream openStream(ReadableBuffer buffer, boolean owner) {
-    if (!owner) {
-      buffer = ignoreClose(buffer);
-    }
-    return buffer.canUseByteBuffer()
-        ? new ByteBufferInputStream(buffer) : new BufferInputStream(buffer);
+    return new BufferInputStream(owner ? buffer : ignoreClose(buffer));
   }
 
   /**
@@ -334,7 +330,8 @@ public ByteBuffer getByteBuffer() {
   /**
    * An {@link InputStream} that is backed by a {@link ReadableBuffer}.
    */
-  private static class BufferInputStream extends InputStream implements KnownLength {
+  private static final class BufferInputStream extends InputStream
+      implements KnownLength, HasByteBuffer {
     final ReadableBuffer buffer;
 
     public BufferInputStream(ReadableBuffer buffer) {
@@ -390,26 +387,20 @@ public boolean markSupported() {
     }
 
     @Override
-    public void close() throws IOException {
-      buffer.close();
-    }
-  }
-
-  /**
-   * A {@link ReadableBuffer}-backed {@link InputStream} that supports the operation of getting
-   * bytes via {@link ByteBuffer}s.
-   */
-  private static class ByteBufferInputStream extends BufferInputStream implements HasByteBuffer {
-
-    ByteBufferInputStream(ReadableBuffer buffer) {
-      super(buffer);
+    public boolean getByteBufferSupported() {
+      return buffer.canUseByteBuffer();
     }
 
     @Nullable
     @Override
     public ByteBuffer getByteBuffer() {
       return buffer.getByteBuffer();
     }
+
+    @Override
+    public void close() throws IOException {
+      buffer.close();
+    }
   }
 
   private ReadableBuffers() {}

protobuf-lite/src/main/java/io/grpc/protobuf/lite/ProtoLiteUtils.java

@@ -179,7 +179,8 @@ public T parse(InputStream stream) {
           int size = stream.available();
           // TODO(chengyuanzhang): we may still want to go with the byte array approach for small
           //  messages.
-          if (stream instanceof HasByteBuffer && stream.markSupported()) {
+          if (stream instanceof HasByteBuffer
+              && stream.markSupported() && ((HasByteBuffer) stream).getByteBufferSupported()) {
             List<ByteBuffer> buffers = new ArrayList<>();
             stream.mark(size);
             while (stream.available() != 0) {